bw-retry-downloader 0.1.0

轻量级重试下载器，专为爬虫工程师设计

bw-retry-downloader

轻量级重试下载器，专为爬虫工程师设计。

确保每个请求都有结果 —— 成功返回数据，失败返回原因，任务不丢失。

特性

• 三层重试补偿：快速重试 → 指数退避 → 放弃并记录，覆盖各类失败场景
• 完整的失败上报：每个失败任务都有原因、尝试次数、业务元数据（meta）
• 失败任务可重跑：report.failed_tasks() 直接取出失败任务重新提交
• 线程池并发：开箱即用，适合 500-1000 个请求的小规模采集
• 钩子机制：请求前/响应后钩子，支持动态代理、验证码检测等
• 429 智能处理：自动读取 Retry-After 响应头
• 确定性 task_id：相同请求生成相同 ID，支持幂等去重

安装

pip install bw-retry-downloader

快速上手

from bw_retry_downloader import Downloader, Task

# 创建下载器
dl = Downloader(workers=10)

# 添加任务（meta 挂载业务数据，失败时原样带回）
tasks = [
    Task(
        url=f"https://api.example.com/item/{i}",
        meta={"product_id": i}
    )
    for i in range(500)
]
dl.add_tasks(tasks)

# 运行并获取报告
report = dl.run()
print(report.summary())
# 下载完成: 总计=500 成功=487 失败=13

# 失败上报给业务
for r in report.failed:
    print(f"失败: {r.task.url} | 原因: {r.error} | meta: {r.task.meta}")

# 重跑失败任务
if report.failed_count > 0:
    dl.add_tasks(report.failed_tasks())
    report2 = dl.run()

进阶用法

自定义重试策略

from bw_retry_downloader import Downloader, RetryPolicy

policy = RetryPolicy(
    fast_retries=3,           # 快速重试 3 次
    fast_delay=0.3,           # 快速间隔 0.3s
    slow_retries=2,           # 慢速重试 2 次
    slow_delay_base=5.0,      # 慢速起始 5s
    retry_on_status=(429, 500, 502, 503),
)
dl = Downloader(workers=5, retry_policy=policy)

进度回调

def on_progress(done, total, result):
    status = "OK" if result.success else "FAIL"
    print(f"[{done}/{total}] {result.task.url} {status}")

report = dl.run(on_progress=on_progress)

请求前/响应后钩子

from dataclasses import replace

def add_proxy(task):
    """请求前动态设置代理"""
    return replace(task, proxies={"https": "http://proxy:8080"})

def check_captcha(result):
    """响应后检测验证码"""
    if result.success and "验证码" in (result.text or ""):
        result.success = False
        result.error = "触发验证码"
    return result

dl = Downloader(
    workers=5,
    before_request=add_proxy,
    after_response=check_captcha,
)

上下文管理器

with Downloader(workers=10) as dl:
    dl.add_tasks(tasks)
    report = dl.run()
# 自动关闭 session

自定义重试判断

policy = RetryPolicy(
    retry_on=lambda status, exc, resp: (
        status in (429, 503) or
        exc is not None or
        (resp and "rate limit" in resp.text.lower())
    )
)

API 参考

Task

参数	类型	默认值	说明
url	str	必填	请求 URL
method	str	"GET"	HTTP 方法
headers	dict	None	请求头
params	dict	None	URL 参数
data	Any	None	请求体（与 json 互斥）
json	Any	None	JSON 请求体（与 data 互斥）
timeout	float	10.0	超时时间（秒）
proxies	dict	None	代理配置
meta	dict	None	业务自定义数据，结果中原样带回
task_id	str	自动生成	任务 ID，相同请求生成相同 ID

Downloader

参数	类型	默认值	说明
workers	int	5	线程池大小
retry_policy	RetryPolicy	默认策略	重试策略
queue	BaseQueue	MemoryQueue	任务队列
session	Session	自动创建	requests Session
before_request	Callable	None	请求前钩子
after_response	Callable	None	响应后钩子

DownloadReport

属性/方法	说明
total	总任务数
success_count	成功数
failed_count	失败数
success	成功的 Result 列表
failed	失败的 Result 列表
failed_tasks()	失败的 Task 列表，可直接重新提交
failed_urls()	失败的 URL 列表
summary()	摘要字符串

开发

# 安装开发依赖
pip install -e ".[dev]"

# 运行测试
pytest tests/ -v

# 带覆盖率
pytest tests/ -v --cov=bw_retry_downloader

License

MIT