采集脚本的三个稳定性细节：超时、重试、幂等

写一个能跑一次的采集脚本大概二十分钟；写一个能跑三个月的，问题全在这三件事上。

一、超时：每个外部调用都要有上限，包括"看起来不会卡"的那个

requests 默认不设超时——不是"默认 30 秒"，是不设。DNS 卡住、对端半开连接、代理静默丢包，任何一种都能让脚本永久挂起。正确写法是区分连接超时和读取超时：

import requests

resp = requests.get(url, timeout=(3.05, 15))

3.05 是连接超时，15 是两次读取之间的最大间隔。只写一个数字的写法（timeout=15）对连接和读取都用同一个值，通常不是想要的。

浏览器渲染的场景更麻烦：page.goto() 的超时和"页面渲染完成"是两件事。经验值是导航超时 30 秒，然后轮询目标元素，再给一个整体截止时间：

import time

deadline = time.time() + 60
while time.time() < deadline:
    if js("document.querySelector('#target')"):
        break
    time.sleep(1)
else:
    raise TimeoutError("target element not found in 60s")

二、重试：只重试"可能变好"的错误

把重试写成 for i in range(3): try: ... except: pass 是最常见的错误——它会把"权限不足"、"参数错误"这种确定性失败也重试三遍，浪费时间还掩盖问题。

分类原则：

• 重试：连接超时、5xx、429、临时 DNS 失败
• 不重试：4xx（除 429）、解析失败、字段缺失、鉴权失败

import time
import requests

RETRYABLE = (requests.ConnectionError, requests.Timeout)

def fetch(url, attempts=4):
    delay = 1.0
    for i in range(attempts):
        try:
            r = requests.get(url, timeout=(3.05, 15))
            if r.status_code == 429 or 500 <= r.status_code < 600:
                raise requests.HTTPError(f"retryable {r.status_code}")
            r.raise_for_status()
            return r
        except (RETRYABLE, requests.HTTPError):
            if i == attempts - 1:
                raise
            time.sleep(delay)
            delay *= 2

指数退避加一个上限（比如 30 秒），避免把上游按在地上摩擦；如果服务端给了 Retry-After，优先用它。

三、幂等：跑第二遍不会产生第二份数据

采集任务最怕的是"跑到一半失败，重跑后数据库里有半份脏数据"。三条实践：

1）以业务主键去重，而不是自增 ID

import sqlite3

conn = sqlite3.connect("data.db")
conn.execute("CREATE TABLE IF NOT EXISTS item (key TEXT PRIMARY KEY, payload TEXT, fetched_at TEXT)")
conn.execute("INSERT INTO item(key, payload, fetched_at) VALUES(?,?,datetime('now')) "
             "ON CONFLICT(key) DO UPDATE SET payload=excluded.payload, fetched_at=excluded.fetched_at",
             (key, payload))
conn.commit()

2）先写临时表，成功后原子替换

分页抓取时，把这一轮的结果写进 item_staging，全部页面成功后再 INSERT ... SELECT 到正式表并清空暂存。中途失败，正式表不受影响。

3）记录"已完成到哪一页"

state = {"cursor": 0}
# 每页成功后立刻持久化 cursor
json.dump(state, open("cursor.json", "w"))

重跑从 cursor 继续，而不是从头再来。这一条对限流的站点尤其重要——重复抓取会直接触发封禁。

四、一个可以直接抄的骨架

def run_once():
    state = load_state()
    while True:
        page = fetch_page(state["cursor"])     # 内部已带超时与退避
        if not page["items"]:
            break
        with tx():                              # 临时表 + 原子替换
            upsert(page["items"])
            state["cursor"] = page["next"]
        if page["next"] is None:
            break
    save_state(state)
    return summary(state)

配合外层 set -euo pipefail 的 wrapper 和 systemd timer（Type=oneshot，天然串行），整个任务就有了可预期的行为：卡住会被超时打断、失败会退避重试、重跑不会重复、进度不会丢。

小结

超时解决"卡住"，重试解决"抖动"，幂等解决"重跑"。这三件事都不难，难的是在写第一版的时候就留出位置——否则等到线上开始丢数据，补起来的成本要高得多。