稳定性：断点续爬、去重与失败重试

跑十几页的脚本不需要稳定性，跑十万页的采集任务必须能被 Ctrl+C 中断、能重启接着跑、能只重跑失败的部分。本章讲三件事：状态落盘（断点续爬）、去重、失败重试。合规前提同样适用：遵守 robots.txt 与网站服务条款，控制请求频率，不采集个人隐私与受版权保护的付费内容，不绕过登录、付费墙、验证码等访问控制，数据使用遵守《个人信息保护法》《数据安全法》《著作权法》。

断点续爬：状态写在哪

状态只有两类内容：已处理 URL 集合（判重与跳过的依据）与分页游标（下次从哪里继续），外加一份失败明细。

方案	写什么	优点	缺点	适合
JSON 状态文件	已处理 URL 列表、游标、计数	无依赖、肉眼可读、便于手工修正	每次全量重写，量级上去后越来越慢	万级以内、单人脚本
SQLite 状态表	每个 URL 一行、状态、失败原因、时间	增量写入、可用 SQL 统计、事务安全	需要设计主键与索引	十万级以上、要按状态查询

重启流程固定：读状态 → 重建待办队列 → 跳过已处理 → 从游标继续。不要用「文件存在就重跑」这种粗暴判断。

去重三件套

• URL 规范化：去掉 #fragment、host 转小写、query 排序、剔除无意义追踪参数（utm_*、spm、from）。不规范化，同一个页面会因写法不同被当成两个任务。
• 内存 set 与布隆过滤器：set 精确且零成本，但千万级 URL 要占 GB 量级内存；布隆过滤器只占几 MB，代价是有极小误判（漏采个别页面），需要精确判重时不要用。
• 唯一键兜底：入库用唯一索引或 INSERT ... ON CONFLICT DO NOTHING，让「重复处理」在存储层也变成无害操作。

重试策略：哪些该重试

错误类型	典型表现	处理方式
可重试	超时、连接错误、连接被重置、429、5xx	指数退避重试，次数封顶
不可重试	400、401、403、404、页面结构解析失败	记入失败队列，修复规则后单独重跑

退避公式 delay = min(base * 2 ** attempt, cap) * random.uniform(0.5, 1.5)：抖动是为了避免多个任务在同一秒集体重试，把刚恢复的服务再打垮；cap 兜住上限，别出现等 40 分钟的重试。

完整脚本：断点续爬 + 去重 + 退避重试

import json, random, time
from pathlib import Path
from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit

import httpx

STATE = Path("state.json")     # 断点状态：已处理 URL 与分页游标
FAILED = Path("failed.jsonl")  # 失败队列：重跑时只读它
HEADERS = {"User-Agent": "MyCrawler/1.0 (+https://example.com/bot)"}
RETRYABLE = {429, 500, 502, 503, 504}
TRACKING = ("utm_source", "utm_medium", "utm_campaign", "spm", "from")

def normalize(url: str) -> str:
    """URL 规范化：去 fragment、host 小写、剔除追踪参数、query 排序。"""
    parts = urlsplit(url)
    pairs = sorted((k, v) for k, v in parse_qsl(parts.query) if k not in TRACKING)
    return urlunsplit((parts.scheme.lower(), parts.netloc.lower(), parts.path, urlencode(pairs), ""))

def load_state() -> dict:
    if STATE.exists():
        return json.loads(STATE.read_text(encoding="utf-8"))
    return {"done": [], "cursor": 1}

def fetch(client: httpx.Client, url: str, attempts: int = 4) -> str:
    """带指数退避的取页：可重试错误退避重试，其余直接抛出。"""
    last: Exception = RuntimeError(url)
    for attempt in range(attempts):
        try:
            resp = client.get(url, timeout=15.0)
            if resp.status_code in RETRYABLE:
                raise httpx.HTTPStatusError(str(resp.status_code), request=resp.request, response=resp)
            resp.raise_for_status()  # 400/401/403/404 到这里直接失败，不重试
            return resp.text
        except (httpx.TransportError, httpx.HTTPStatusError) as exc:
            last = exc
            if attempt == attempts - 1:
                break
            delay = min(1.0 * 2 ** attempt, 30.0) * random.uniform(0.5, 1.5)  # 退避 + 抖动
            print(f"第 {attempt + 1} 次失败，{delay:.1f}s 后重试：{url}")
            time.sleep(delay)
    raise last

def main(max_pages: int = 200) -> None:
    state = load_state()
    seen = set(state["done"])          # 内存 set 负责精确判重
    page, ok, fail = state["cursor"], 0, 0
    with httpx.Client(headers=HEADERS, follow_redirects=True) as client, \
            FAILED.open("a", encoding="utf-8") as fout:
        while page <= max_pages:
            url = normalize(f"https://example.com/list?page={page}&utm_source=nav")
            page += 1
            if url in seen:
                continue                   # 已处理，直接跳过
            try:
                html = fetch(client, url)  # 此处解析 html 并写入自己的数据表
                ok += 1
            except Exception as exc:
                fail += 1
                fout.write(json.dumps({"url": url, "error": f"{type(exc).__name__}: {exc}"},
                                      ensure_ascii=False) + "\n")
            seen.add(url)
            state["done"].append(url)      # 只追加，避免每次重排整个集合
            state["cursor"] = page
            STATE.write_text(json.dumps(state, ensure_ascii=False), encoding="utf-8")
    print(f"完成：成功 {ok}，失败 {fail}，失败明细见 {FAILED.name}")

if __name__ == "__main__":
    main()

SQLite 状态表与进度 ETA

import sqlite3, time

DDL = """
CREATE TABLE IF NOT EXISTS page_state (
  url        TEXT PRIMARY KEY,          -- 规范化 URL，主键天然去重
  status     TEXT NOT NULL,             -- ok / failed
  error      TEXT,                      -- 失败原因，成功为空
  updated_at TEXT NOT NULL DEFAULT (datetime('now'))
);
"""

def mark(conn: sqlite3.Connection, url: str, status: str, error: str | None = None) -> None:
    """UPSERT：同一个 URL 重复处理也不会产生第二行。"""
    conn.execute("INSERT INTO page_state (url, status, error) VALUES (?, ?, ?) "
                 "ON CONFLICT(url) DO UPDATE SET status=excluded.status", (url, status, error))
    conn.commit()

def progress(conn: sqlite3.Connection, total: int, started: float) -> None:
    """进度与 ETA：长任务必须能看出还剩多久。"""
    ok, failed = conn.execute("SELECT count(*) FILTER (WHERE status='ok'), "
                              "count(*) FILTER (WHERE status='failed') FROM page_state").fetchone()
    rate = (ok + failed) / (time.monotonic() - started or 1)
    print(f"进度 {ok + failed}/{total} 速率 {rate:.2f} 条/秒 "
          f"ETA {(total - ok - failed) / rate / 60:.1f} 分钟")

失败队列用 JSONL 而不是 JSON 数组：追加写不用读回整个文件，崩溃最多丢最后一行，grep 就能看。跑完先打印统计（成功数、失败数、失败原因按类型分组），再决定是否重跑；重跑时把 load_state() 换成只读失败队列，成功过的 URL 一条都不要碰。

常见坑

• 状态只留在内存里、跑完才落盘：中断等于全丢，必须边跑边写。
• 只去重不规范化：同一个页面被抓两次，浪费流量还污染统计。
• 重试所有错误：404 重试三次只是白白多打三次请求，不可重试错误应立刻进失败队列。

小结：把已处理 URL 集合与游标边跑边落盘（万级用 JSON、十万级以上用 SQLite），用规范化 + set/布隆过滤器 + 唯一键三层去重，只对超时、连接错误、429、5xx 做带抖动的指数退避重试，失败项进 JSONL 队列并在跑完后按统计单独重跑。