（七）把幂等、计数和提交放进同一事务

1. 先判断本批是否已经完成

# 续写同一文件：catalog_project/src/cataloghub/storage.py
def ingest(path, batch, run_id, source):
    with closing(open_db(path)) as connection, connection:
        # 同一快照重放返回原报告，新一轮快照另有自己的批次编号。
        connection.execute("BEGIN IMMEDIATE")
        existing = connection.execute(
            "SELECT report_json FROM runs WHERE id=?", (run_id,)
        ).fetchone()
        if existing:
            return json.loads(existing[0])
        report = {
            "run_id": run_id,
            "raw": batch.raw_count,
            "unique": len(batch.books),
            "duplicates": batch.duplicates,
            "rejected": len(batch.rejected),
            "inserted": 0,
            "updated": 0,
            "unchanged": 0,
        }

BEGIN IMMEDIATE 先取得本次写入所需的锁。随后查询 runs：已经保存这个 run_id，直接返回原报告。同一快照重放不会额外产生批次，也不会把原报告的“新增 6”改写成“未变 6”。

这就是这里的幂等：同一个操作标识重复执行，持久状态保持为已经完成的一次。重新采集带来新的编号，仍会正常比较这次数据与当前书目。

2. 逐本比较后计算报告

# 续写同一文件：catalog_project/src/cataloghub/storage.py
        # 比较当前字段，只把新增和确有变化的书目交给后面的写入。
        changes = []
        for book in batch.books:
            tags = json.dumps(book.tags, ensure_ascii=False)
            previous = connection.execute(
                "SELECT title,finished,tags_json FROM books WHERE source_id=?",
                (book.source_id,),
            ).fetchone()
            values = (book.title, book.finished, tags)
            if previous is None:
                report["inserted"] += 1
            elif tuple(previous) == values:
                report["unchanged"] += 1
                continue
            else:
                report["updated"] += 1
            changes.append((book.source_id, *values, run_id))

三种计数互相分开：编号不存在，新增；编号存在但字段不同，更新；字段都相同，未变。未变的书保留上次实际改变内容的 updated_run。这一批没有出现的旧书仍保留在库里；“来源是否删除了书”可以扩展为单独的下架流程。

对每个 source_id 分别比较不存在、相同和变化的记录，再形成整批报告与待写列表

3. 一起保存报告、书目和拒收原因

# 续写同一文件：catalog_project/src/cataloghub/storage.py
        # 先保存被书目引用的批次，再一起写入变更和拒收信息。
        connection.execute(
            "INSERT INTO runs(id,source,report_json) VALUES(?,?,?)",
            (run_id, source, json.dumps(report)),
        )
        for values in changes:
            connection.execute(
                "INSERT INTO books VALUES(?,?,?,?,?) ON CONFLICT(source_id) DO UPDATE SET title=excluded.title,finished=excluded.finished,tags_json=excluded.tags_json,updated_run=excluded.updated_run",
                values,
            )
        connection.executemany(
            "INSERT INTO rejected VALUES(?,?,?,?)",
            [(run_id, r["index"], r["page"], r["reason"]) for r in batch.rejected],
        )
    return report

先插入 runs，再让更新的书目指向它。ON CONFLICT(source_id) DO UPDATE 使用来源主键更新指定字段；excluded 表示本次原本准备插入的值。拒收记录最后一起写入。

with connection 正常离开时提交，异常离开时回滚；closing(...) 再关闭连接。无论失败发生在第几本书，报告和书目都不会只保存其中一部分。网络请求已在事务之前结束，数据库写锁不会跨越网络等待。