（五）用原始快照连接采集与导入

1. 分开保存原始输入与已经接受的状态

清洗规则会修改，数据库也可能写入失败。只留下最后的六本书，就失去了核对这次转换的输入。因此每批数据先保存为 archives/<run_id>.json，包含来源、时间、页数和十张卡片。

# 项目文件：catalog_project/src/cataloghub/pipeline.py
from datetime import datetime, timezone
import json
import os
from pathlib import Path
from uuid import uuid4

import httpx

from .cleaning import clean_cards
from .collecting import collect
from .storage import ingest, migrate


def write_json(path, data):
    path.parent.mkdir(parents=True, exist_ok=True)
    temporary = path.with_suffix(path.suffix + ".tmp")
    temporary.write_text(
        json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8"
    )
    os.replace(temporary, path)  # 正式路径只在整份 JSON 写完后出现。


def import_snapshot(settings, snapshot):
    batch = clean_cards(snapshot["cards"])
    migrate(settings.database)
    report = ingest(settings.database, batch, snapshot["run_id"], snapshot["source"])
    return {**report, "pages": snapshot["pages"]}

os.replace() 对应第五章讲过的完整替换：先写临时文件，再替换正式路径。调用方读取正式 JSON 时，看到的是已经写完的一份快照。import_snapshot() 只接收快照，不发网络请求，所以采集与重放可以共用这条清洗入库流程。

# 续写同一文件：catalog_project/src/cataloghub/pipeline.py
def run(settings, client=None):
    if client is None:
        with httpx.Client(
            timeout=5, follow_redirects=False, headers={"User-Agent": "ReadingLab/1.0"}
        ) as owned:
            return run(settings, owned)
    cards, pages = collect(settings.source, client)
    snapshot = {
        "format": 1,
        "run_id": uuid4().hex,
        "source": settings.source,
        "pages": pages,
        "created_at": datetime.now(timezone.utc).isoformat(),
        "cards": cards,
    }
    path = settings.archives / (snapshot["run_id"] + ".json")
    write_json(path, snapshot)  # 先留下原始字段，再进入清洗与数据库事务。
    report = import_snapshot(settings, snapshot)
    return {**report, "archive": str(path)}


def replay(settings, path):
    snapshot = json.loads(Path(path).read_text(encoding="utf-8"))
    if snapshot.get("format") != 1:
        raise ValueError("无法识别这份采集快照")
    return import_snapshot(settings, snapshot)

uuid4().hex 生成本批编号，UTC 时间记录采集时刻。JSON 先保存成功，再导入；导入失败时可修正原因并重放。页面尚未收齐就失败时，还没有产生这个完整批次快照，再运行采集即可。

沿 cards、snapshot、临时 JSON、完整 JSON 和入库事务追踪，失败后由 replay 接回同一批次

# 从第一次采集报告的 archive 字段取得实际文件名，再交给 replay。
uv run catalog replay instance/archives/本批编号.json

命令里的“本批编号”替换为报告中已经存在的文件名。重放保留原 run_id；重新请求目录则生成新的编号。后面利用这个区别实现幂等导入。