darren_utils

darren_utils 是一个面向爬虫、自动化与通用工具场景的 Python 工具库，重点提供：

• 对称/非对称加密与摘要签名（AES / DES / 3DES / RC4 / RSA / SM2 / SM3 / SM4 / HASH / HMAC）
• 基于 httpx 的同步 HTTP 封装（Session 与非 Session、代理、下载、统一返回）
• 代理池管理（API 拉取、可用性验证、内存/Redis 模式）
• 字符串、时间、文件、剪贴板、设备信息等高频工具
• 通用返回对象 DarrenRet（success/message/data/error_code）

包布局（0.4+）：源码在 src/darren_utils/；公开入口为 import darren 或 import darren_utils。

安装

方式 1：本地开发安装（推荐）

pip install -e .

方式 2：按依赖安装

pip install -r requirements.txt

核心依赖

• httpx[http2]
• cryptography
• gmssl
• pyperclip
• send2trash
• loguru
• socksio

可选 Redis 代理池：

pip install "darren_utils[redis]"

快速开始

import darren
# 或（发行名别名）
import darren_utils as darren

1) 加密示例（AES）

cipher = darren.aes.encrypt(
    "cbc",
    "hello",
    "1234567890abcdef",
    "abcdef1234567890",
    "pkcs7",
)
plain = darren.aes.decrypt(
    "cbc",
    cipher,
    "1234567890abcdef",
    "abcdef1234567890",
    "pkcs7",
)
print(cipher, plain)

2) HTTP Session 与多线程（重要）

• 同一个 Session 不要跨线程共享（httpx Client 非线程安全）。
• 多线程：每线程 darren.http.Session(proxy_tool=darren.proxy)，进程内只 proxy.set_config() 一次。
• proxy_wait_timeout 默认 30 秒；需要无限等待时显式传 -1。
• 代理 API 请用环境变量 DARREN_PROXY_API_URL，勿把密钥写进源码；若曾硬编码请立刻轮换 key。

初始化顺序示例：

import os
import darren
from darren import ProxyConfig

cfg = ProxyConfig()
cfg.set_api_url(os.environ["DARREN_PROXY_API_URL"])
darren.proxy.set_config(cfg)  # 进程内只做一次

session = darren.http.Session(proxy_tool=darren.proxy, proxy_wait_timeout=30)
ret = session.get_ret("https://example.com", use_proxy=True)

详见 examples/proxy_http_basic.py 与 examples/proxy_http_multithread.py。

3) HTTP 基础（保持旧接口）

resp = darren.http.get("https://example.com", timeout=10, max_retries=1)
if resp:
    print(resp.status_code)
    print(resp.text_trunc(120))
    print(resp.get_location())
    print(resp.get_proxy_used())

4) HTTP 统一返回（推荐新接口）

ret = darren.http.get_ret("https://example.com", timeout=10, max_retries=0)
if ret.is_success():
    data = ret.get_data({})
    print(data.get("status_code"))
    print(data.get("proxy_used"))
else:
    print(ret.get_error_code(), ret.get_message(), ret.get_error_detail())

5) 代理配置与使用

cfg = darren.ProxyConfig()
cfg.set_api_url("http://your-proxy-api")
cfg.set_timeout(8)
cfg.set_verify_proxy(True)
cfg.set_threshold(5)   # >0：维持水位，取走后自动补；设 0 则按需（空池才拉，不预囤）

darren.proxy.set_config(cfg)
proxy_dict = darren.proxy.get_one_proxy(timeout=-1)  # -1：一直等到池中有 IP
print(proxy_dict)

# HTTP 自动取代理（与 set_timeout 无关）：
# darren.http.get(url, use_proxy=True)  # 默认 proxy_wait_timeout=30；无限等待请显式传 -1
# darren.http.get(url, use_proxy=True, proxy_wait_timeout=10)  # 最多等 10 秒

说明：cfg.set_timeout(8) 仅用于代理池内部（验 IP、拉 API），不是业务请求超时，也不是从池取 IP 的等待时间。set_threshold(0) 关闭后台囤积，仅在调用 get_one_proxy 且池空时拉一批。

6) DarrenRet 使用

ok_ret = darren.DarrenRet.success_ret(data={"id": 1}, message="ok")
fail_ret = darren.DarrenRet.failure_ret(
    error_code="TIMEOUT",
    message="request timeout",
    error_detail="connect timeout 5s",
)
print(ok_ret.is_success(), ok_ret.to_dict())
print(fail_ret.is_failure(), fail_ret.to_error_string())

模块概览

HTTP 与返回模型

返回模型

• 旧接口不变：get/post/... -> DarrenResponse | None
• 新接口统一：get_ret/post_ret/... -> DarrenRet

DarrenRet 结构

{
  "success": bool,
  "message": str,
  "data": Any,
  "error_code": str,
  "error_detail": str,
  "meta": dict
}

常见错误码

• OK
• TIMEOUT
• NETWORK_ERROR
• PROXY_ERROR
• HTTP_STATUS_ERROR
• INTERNAL_ERROR

代理与 SOCKS 支持

• HTTP 代理：http://ip:port
• 带认证代理：http://user:pass@ip:port
• SOCKS 代理：socks5://ip:port
• 兼容别名：socket:// / socket5:// / socks://（内部归一化为 socks5://）

示例：

proxies = {
    "http": "socks5://user:pass@127.0.0.1:7890",
    "https": "socks5://user:pass@127.0.0.1:7890",
}
ret = darren.http.get_ret("http://utils.darren8.com/ip/getIP", proxies=proxies, use_proxy=False)
print(ret.get_meta("proxy_used"))

本地开发（推荐）

在 conda 环境 pip_darren 中，以可编辑模式安装本项目，确保 import darren 加载的是当前源码（而非 site-packages 里的旧 wheel）：

conda activate pip_darren
cd F:\PythonProject\pip_darren
pip install -e ".[dev]"

验证加载路径：

python -c "import darren; print(darren.__file__)"

测试

自动化测试（pytest）：

pytest

手工联调 / 多线程代理示例：

python examples/proxy_http_multithread.py --mode non-session
python examples/proxy_http_multithread.py --mode taobao   # 需自行准备 cookie/签名

发布

# 默认发布到正式 PyPI（https://pypi.org/project/darren-utils）
python release.py

# 发布到 TestPyPI
python release.py --target testpypi

上传使用 API token：环境变量 __token__ / PYPI_TOKEN / TWINE_PASSWORD（值为 pypi-...）。

FAQ

1) 为什么设备字段有时为空？

不同系统权限、硬件厂商暴露能力不同，空值属于正常情况。建议组合多个字段生成设备指纹。

2) 为什么请求失败但程序没有崩溃？

HTTP 封装默认采用安全返回策略（旧接口返回 None，*_ret 返回失败对象），便于上层统一处理。

3) 代理一定要 Redis 吗？

不需要。默认内存模式即可使用；仅在你需要跨进程共享代理池时启用 Redis（pip install "darren_utils[redis]"）。

模块速查

import darren

darren.hash.md5("123456")
darren.string.get_between("a[x]b", "[", "]")
darren.file.exists(".")
pool = darren.thread_pool.create(2)
pool.submit(lambda: 1)
pool.shutdown()

# HTTP
ret = darren.http.get_ret("https://example.com")

注意事项

• gmssl 为国密相关模块必需依赖。
• SOCKS 代理需 socksio。
• 代理密码若含特殊字符（如 @, :, /），建议先进行 URL 编码。
• Redis 代理池：pip install "darren_utils[redis]"，再 ProxyConfig(use_redis_mode=True)。
• HTTP 语义：
  • get/post/... 默认不因 4xx/5xx 返回 None（传输失败才为 None）；可用 raise_for_status=True。
  • get_ret/post_ret/... 将非 2xx/3xx 判为失败；可用 raise_on_error=True / ok_statuses=...。
  • download() 会对非成功状态 raise_for_status 并重试。
  • stream=True 不可与代理同时用于 request()（请用 download()）。