《Python编程实战》12.3 稳定性、限速与合规边界

把采集从「能跑」做到「能长期跑」:用 urllib.robotparser 实测解析 robots.txt,真跑令牌桶限速器与指数退避重试,用 JSONL 做断点续采,最后收束成一页合规边界清单——只抓公开数据、尊重 robots 与 ToS、控制频率、不绕过反爬。

本节目标:把采集任务从「本地能跑一次」升级为「长期稳定、对目标友好、经得起合规审查」——实测 robots.txt 解析、令牌桶限速、指数退避重试与断点续采,并建立清晰的合规红线。
适用版本:Python 3.12+(实测 3.14.6);requests 2.34.2、httpx 0.28.1

12.3 稳定性、限速与合规边界

12.1 讲了「怎么抓」,12.2 讲了「抓不到时怎么办」。但决定一个采集系统能不能上线的,从来不是单次能不能抓到,而是三件事:稳不稳(网络抖动、对方限流时能不能自愈)、客不客气(会不会把别人服务器打崩)、合不合法(有没有越过红线)。这一节把这三件事落成可运行的代码与可执行的规范。

12.3.1 第一步永远是读 robots.txt

robots.txt 是网站用明文声明的抓取规则——它放在站点根目录,任何爬虫动手前都该先读。标准库 urllib.robotparser 直接解析:

from urllib.robotparser import RobotFileParser

rp = RobotFileParser()
rp.set_url(BASE + "/robots.txt")
rp.read()                                        # 拉取并解析
for ua, path in [("plumephp-scraper/1.0", "/"),
                 ("plumephp-scraper/1.0", "/private/x"),
                 ("plumephp-scraper/1.0", "/admin"),
                 ("evil-bot", "/")]:
    print(f"can_fetch({ua!r}, {path!r}) = {rp.can_fetch(ua, BASE + path)}")
print("crawl_delay:", rp.crawl_delay("plumephp-scraper/1.0"))

配一个真实的 robots.txt:

User-agent: *
Disallow: /private/
Disallow: /admin
Crawl-delay: 1

User-agent: evil-bot
Disallow: /

实测输出:

can_fetch('plumephp-scraper/1.0', '/') = True
can_fetch('plumephp-scraper/1.0', '/private/x') = False
can_fetch('plumephp-scraper/1.0', '/admin') = False
can_fetch('evil-bot', '/') = False
crawl_delay: 1

几点必须理解:can_fetch 的第一个参数是你的 User-Agent 名,不是 URL;* 是通配规则,特定 UA(evil-bot)优先于 *;crawl_delay 返回该 UA 建议的请求间隔(秒),拿到后要真的照着做。规则是「声明式」的——robots.txt 本身没有强制力,遵守它是一种行业契约,也是法律上「你是否善意抓取」的重要证据。

12.3.2 限速:令牌桶真跑

「别把对方打崩」的最直接手段是限速。最简单的做法是每次请求前 sleep 固定间隔,但那样既浪费又死板(突发请求也要等)。更工程化的是令牌桶:桶里按固定速率补充令牌,取到令牌才允许发请求,桶容量允许一定突发:

import time

class TokenBucket:
    def __init__(self, rate: float, capacity: int):
        self.rate = rate            # 每秒补充的令牌数
        self.capacity = capacity    # 桶容量(允许的突发量)
        self.tokens = float(capacity)
        self.updated = time.monotonic()

    def acquire(self, n: int = 1) -> None:
        while True:
            now = time.monotonic()
            self.tokens = min(self.capacity,
                              self.tokens + (now - self.updated) * self.rate)
            self.updated = now
            if self.tokens >= n:
                self.tokens -= n
                return
            time.sleep((n - self.tokens) / self.rate)

用「每秒 5 个、突发 5 个」的桶打 12 次本地请求,实测:

12 次请求耗时(秒): [0.01, 0.01, 0.01, 0.02, 0.02, 0.21, 0.41, 0.61, 0.81, 1.01, 1.21, 1.41]
总耗时: 1.41 s

读法:前 5 次几乎瞬间完成(桶里初始有 5 个令牌,允许突发),之后被压到每 0.2 秒一个(速率 5/秒)。总耗时 1.41 秒 ≈ 5 个突发 + 7 个 × 0.2 秒。这就是令牌桶的取舍——既限住平均速率,又不惩罚合理突发。用 time.monotonic() 而非 time.time(),避免系统时钟回拨导致计算出负数。

12.3.3 礼貌抓取:User-Agent 与 Crawl-Delay

除了限速,还有几条「礼貌」约定,成本极低但很关键:

  • 带真实、可联系的 User-Agent:plumephp-scraper/1.0 (+https://plumephp.com/bot)——出问题时对方能联系到你,而不是只能一刀切封 IP。不要伪装成浏览器 UA(Mozilla/5.0...),那是欺骗,且与「善意抓取」相悖。
  • 遵守 Crawl-Delay:从 robots.txt 读到后,把它作为令牌桶速率的输入。
  • 只抓需要的:别把整站拖下来存着,按需抓取,用完即弃。
  • 优先官方 API:很多站点提供官方接口,走它是双赢。
  • 避开高峰:目标站忙时降速或错峰。

这几条合起来,就是把「封禁往往源于抓太狠」这句话落实到代码里。

12.3.4 重试:指数退避 + 抖动

12.1 用 HTTPAdapter 做了声明式重试,这里给出手写版,因为它能加两样东西:抖动(jitter)和按状态码决定策略。打一个「前两次 503、第三次成功」的端点:

import random, time, requests

def get_with_backoff(url, max_retries=4, base=0.2):
    for attempt in range(max_retries + 1):
        resp = requests.get(url, timeout=5)
        if resp.status_code < 500:            # 4xx/2xx 都不重试
            return resp
        if attempt == max_retries:
            return resp
        wait = base * (2 ** attempt) + random.uniform(0, 0.1)   # 退避 + 抖动
        print(f"  {resp.status_code},第 {attempt + 1} 次重试,等待 {wait:.2f}s")
        time.sleep(wait)

实测:

  503,第 1 次重试,等待 0.30s
  503,第 2 次重试,等待 0.45s
最终: 200 ok 耗时 0.79s

两个设计点:指数退避(0.2 → 0.4 → 0.8 秒)让等待时间随失败次数拉长,给对方喘息;抖动(random.uniform(0, 0.1))打散多个客户端同时重试的「惊群」——没有抖动,一批任务会在同一时刻齐刷刷重试,等于二次打击。4xx 不重试这条同样重要:403 是对方明确说「不」,重试只会让事情更糟。

12.3.5 断点续采:别让一次崩溃全丢

采集任务动辄跑几小时,中途断网、进程被杀是常态。必须把「已完成的 URL」持久化,重启后跳过。最小实现是一个追加写的 JSONL 检查点:

import json, pathlib

def load_done(path: pathlib.Path) -> set[str]:
    if not path.exists():
        return set()
    return {json.loads(line)["url"]
            for line in path.read_text().splitlines() if line}

def crawl(urls, path):
    done = load_done(path)
    with path.open("a", encoding="utf-8") as fh:
        for url in urls:
            if url in done:
                print("跳过(已完成):", url.rsplit("/", 1)[-1])
                continue
            fh.write(json.dumps({"url": url}, ensure_ascii=False) + "\n")
            fh.flush()                        # 立刻落盘,别等缓冲区
            print("采集:", url.rsplit("/", 1)[-1])

模拟「第一轮跑 3 个后中断,第二轮续采」,实测:

== 第一轮(跑 3 个后模拟中断)==
采集: 0
采集: 1
采集: 2
== 第二轮(续采剩余,前 3 个应被跳过)==
跳过(已完成): 0
跳过(已完成): 1
跳过(已完成): 2
采集: 3
采集: 4
采集: 5

关键在 fh.flush():JSONL 每写一行就落盘,进程被 kill -9 也不会丢进度(只丢当前这一条)。生产里还可给每条记录加状态(done/failed)与重试次数,失败项单独重跑;规模更大时换 SQLite(url 建唯一索引,天然幂等)。

12.3.6 合规边界:技术可行 ≠ 可以抓

这是本节最重要的一节。前面所有技术都只是工具,用不用、怎么用,取决于边界。把红线列清楚:

事项该做不该做
robots.txt先读、遵守 Disallow 与 Crawl-Delay无视 Disallow 硬抓
服务条款 ToS读一遍,确认允许抓取明知禁止仍批量采集
数据性质只抓公开数据抓登录后/非公开数据
频率限速 + 退避,像正常用户高并发狂打,压垮对方
反爬尊重,遇阻停下评估绕验证码、伪造指纹、破解加密参数
官方 API优先使用有 API 还硬爬页面
个人信息最小化、脱敏、不转卖采集身份证/手机号等敏感信息
数据用途明确、合法、尊重版权整站搬运、再分发牟利
标识带可联系的 UA伪装成浏览器欺骗

几条必须记住的原则:

  • 只抓公开数据:需要登录才能看到的内容,不属于「公开」,采集它可能触碰《个人信息保护法》等法规。
  • 尊重 robots 与 ToS:robots 是技术契约,ToS 是法律契约,两者都要看。
  • 控制频率:这是对目标服务器最基本的尊重,也是「善意」的证明。
  • 不绕过反爬:验证码、指纹、加密参数是网站明确的拒绝信号。遇到它们不是「技术挑战」,而是「该停手了」。本节和本书都不提供绕过手段。
  • 有疑问就找官方:很多数据需求可以通过官方 API、数据授权、合作渠道合法满足。抓之前先问一句「有没有正路」。

12.3.7 上线前的稳定性检查清单

把前面几节收束成一张可勾选的清单:

检查项对应手段
超时都设了吗requests (连接, 读取) / httpx 四段 Timeout
连接复用了吗Session / Client
5xx/超时重试了吗指数退避 + 抖动,4xx 不重试
限速了吗令牌桶,速率参考 Crawl-Delay
断点续采了吗JSONL/SQLite 检查点 + flush
编码兜底了吗header → apparent_encoding 逐级降级
解析判空了吗select_one 可能返回 None
合规确认了吗robots / ToS / 公开性 / 频率 / 不绕反爬
可观测吗记录成功/失败数、耗时、限流次数(第 3 章)

延伸阅读

小结

  • 动手前先读 robots.txt:urllib.robotparser 的 can_fetch(UA, url) 与 crawl_delay(UA) 都能实测,规则要真的遵守。
  • 令牌桶限速既限平均速率又允许合理突发(实测 12 次请求 1.41 秒),用 time.monotonic() 计时。
  • 重试用指数退避 + 抖动,只对 5xx/超时重试,4xx 不重试;抖动避免惊群。
  • 断点续采靠持久化检查点(JSONL 每行 flush),进程被杀也不丢进度。
  • 合规红线:只抓公开数据、尊重 robots 与 ToS、控制频率、不绕过反爬、优先官方 API、最小化个人信息。
  • 上线前用「稳定性检查清单」逐项过一遍:超时、复用、重试、限速、续采、编码、判空、合规、可观测。

第 12 章到这里收尾:12.1 解决「怎么抓和解析」,12.2 解决「动态页面怎么办」,12.3 解决「怎么长期、稳定、合规地抓」。采集只是数据进入系统的入口——下一章我们转向文件与文档自动化,把本地文件、Excel、Word、PDF 这些「存量数据」也纳入自动化流程。

阅读导航:上一节:动态页面与浏览器自动化 · 下一节:文件批处理与目录治理 。

继续阅读

探索更多技术文章

浏览归档,发现更多关于系统设计、工具链和工程实践的内容。

全部文章 返回首页

「python」更多文章

  1. 《Python高级编程》目录
  2. 《Python高级编程》11.3 PEP 流程与版本迁移策略
  3. 《Python高级编程》11.2 嵌入式与自由线程运行时