（十六）把结果保存与完成标记一起提交
1. 一页的发现结果组成同一次提交
继续把下面的方法放进同一个 CrawlState 类：
# 续写同一文件：crawler_project/src/catalogcrawl/state.py
# 把本页的新任务、下一页地址和完成标记放进同一事务，恢复时便能接着读取待办任务。
def next_page(self):
row = self.connection.execute(
"SELECT url FROM pages WHERE done=0 ORDER BY seq LIMIT 1"
).fetchone()
return row["url"] if row else None
def save_page(self, url, links, next_url):
with self.connection:
self.connection.executemany(
"INSERT OR IGNORE INTO jobs(url) VALUES (?)",
((link,) for link in links),
)
if next_url is not None:
self.connection.execute(
"INSERT OR IGNORE INTO pages(url) VALUES (?)", (next_url,)
)
cursor = self.connection.execute(
"UPDATE pages SET done=1 WHERE url=?", (url,)
)
if cursor.rowcount != 1:
raise CrawlError("准备保存的目录页不在任务表中")
def pending_jobs(self, after_seq=0, limit=2):
return self.connection.execute(
"SELECT seq, url FROM jobs WHERE done=0 AND seq>? ORDER BY seq LIMIT ?",
(after_seq, limit),
).fetchall()save_page 一起保存本页的详情链接、下一页任务与本页完成标记。提交成功后，这一页才算处理过。若写入下一页时发生数据库错误，事务会撤销本次写入，本页仍然待处理，下一次启动可以重新发现。
pending_jobs 按首次插入的 seq 选择一批。after_seq 是本轮扫描位置：某一本已经用完重试机会后，本轮先继续后面的书；下一次启动重新从待处理任务里选，它仍然会被看到。
2. 一项详情的记录与完成标记组成同一次提交
# 续写同一文件：crawler_project/src/catalogcrawl/state.py
def complete_job(self, url, item: CollectedBook):
with self.connection:
self.connection.execute(
"""
INSERT INTO items VALUES (?, ?, ?, ?, ?)
ON CONFLICT(source_id) DO NOTHING
""",
(
item.source_id,
item.book.title,
int(item.book.finished),
json.dumps(item.book.tags, ensure_ascii=False),
item.source_url,
),
)
cursor = self.connection.execute(
"UPDATE jobs SET done=1 WHERE url=?", (url,)
)
if cursor.rowcount != 1:
raise CrawlError("准备完成的详情不在任务表中")
def collected(self):
rows = self.connection.execute("SELECT * FROM items ORDER BY source_id")
return [
CollectedBook(
row["source_id"],
Book(row["title"], bool(row["finished"]), json.loads(row["tags"])),
row["source_url"],
)
for row in rows
]
def stats(self):
result = {}
for table in ("pages", "jobs"):
# table 来自上面固定的两个名字；URL 等输入仍通过占位参数传入。
row = self.connection.execute(
f"SELECT COUNT(*) AS total, COALESCE(SUM(done), 0) AS done FROM {table}"
).fetchone()
result[table + "_done"] = row["done"]
result[table + "_pending"] = row["total"] - row["done"]
result["items"] = self.connection.execute(
"SELECT COUNT(*) FROM items"
).fetchone()[0]
return result保存时先插入记录，再把对应 jobs 行改为完成，两条 SQL 都在同一次事务里。如果刚插入记录就发生错误，事务撤销后仍是“没有记录、任务待处理”；成功提交后则是“记录存在、任务完成”。不会先持久化一个完成标记，再丢掉还没保存的书。
源站编号已经存在时，ON CONFLICT 保留第一次取得的记录，但当前 URL 的任务仍可完成。最后按 source_id 排序交回结果，导出次序因此不会跟着网络完成先后变化。
stats 中的表名来自代码里的两个固定名字；URL 和其他输入仍使用占位参数。完成量、待处理量与唯一记录量分别统计，例如两个别名任务都完成，items 仍可能只有一本。
