（二十）运行完整目录采集器并检查恢复

1. 命令入口组合已有动作

完整目录如下。前面分段讲解的解析器、进度类与运行器，在配套文件中已经组合为完整模块：

crawler_project/
├── pyproject.toml、uv.lock、requirements-dev.txt
├── src/
│   ├── crawlsite/           目录、详情、JSON、动态页面与故障开关
│   ├── bookdata/            已有记录检查、CSV 与 SQLite
│   └── catalogcrawl/
│       ├── models.py、errors.py、urls.py、parsers.py
│       ├── pacing.py、transport.py
│       ├── state.py、runner.py、exporting.py
│       └── cli.py、__init__.py、__main__.py
├── tests/
└── spiders/books.py

# 项目文件：crawler_project/src/catalogcrawl/cli.py
# crawl 恢复采集，status 查看进度，export 从已保存结果生成 CSV 和业务数据库。
import argparse
import asyncio
import json
import logging
from pathlib import Path
import sqlite3
import sys

from .exporting import export_results
from .runner import crawl
from .state import CrawlState


def main(argv=None):
    parser = argparse.ArgumentParser(description="书籍目录采集器")
    parser.add_argument(
        "action", choices=["crawl", "status", "export"], nargs="?", default="crawl"
    )
    parser.add_argument("--base-url", default="http://127.0.0.1:8766")
    parser.add_argument("--state", type=Path, default=Path("data/crawl.sqlite3"))
    parser.add_argument("--csv", type=Path, default=Path("data/books.csv"))
    parser.add_argument("--db", type=Path, default=Path("data/library.sqlite3"))
    parser.add_argument("--concurrency", type=int, default=2)
    parser.add_argument("--interval", type=float, default=0.02)
    parser.add_argument("--attempts", type=int, default=3)
    parser.add_argument("--deadline", type=float, default=15)
    parser.add_argument("--request-timeout", type=float, default=1)
    parser.add_argument("--stop-after", type=int)
    args = parser.parse_args(argv)
    logging.basicConfig(level=logging.INFO, format="%(levelname)s %(name)s %(message)s")
    start_url = (
        args.base_url.rstrip("/") + "/catalog/?page=1"
        if args.action == "crawl"
        else None
    )
    try:
        with CrawlState(args.state, start_url) as state:
            if args.action == "crawl":
                result = asyncio.run(
                    crawl(
                        state,
                        args.concurrency,
                        args.interval,
                        args.attempts,
                        args.deadline,
                        args.request_timeout,
                        args.stop_after,
                    )
                )
                print(json.dumps(result, ensure_ascii=False))
                return 2 if result["jobs_pending"] or result["pages_pending"] else 0
            if args.action == "status":
                print(json.dumps(state.stats(), ensure_ascii=False))
            else:
                print("已导出", export_results(state, args.csv, args.db), "本")
        return 0
    except (OSError, ValueError, sqlite3.Error, TimeoutError, ExceptionGroup) as error:
        logging.getLogger(__name__).exception("本次采集操作失败")
        print("错误：" + (str(error) or type(error).__name__), file=sys.stderr)
        return 1


if __name__ == "__main__":
    raise SystemExit(main())

默认动作是 crawl；status 只看进度，export 只从保存的数据生成输出。返回 0 表示动作完成，2 表示本次采集结束后还有待处理任务，1 表示入口遇到错误。日志写向标准错误，标准输出保留 JSON 统计或简短操作结果，便于第七章的命令与自动检查继续使用。

# 项目文件：crawler_project/src/catalogcrawl/__init__.py
# 导入包不发送请求，也不打开进度库。
"""目录解析、采集与进度保存。"""

# 项目文件：crawler_project/src/catalogcrawl/__main__.py
# 包启动与 crawl-books 命令共用相同入口和退出状态。
from .cli import main

if __name__ == "__main__":
    raise SystemExit(main())

2. 在两个终端暂停与恢复

# 服务终端：本次练习沿固定地址运行，客户端重新启动时仍能访问同一站点。
uv run python -m crawlsite --port 8766

# 客户端终端：先保存两本，返回状态 2，查看还剩四项详情。
uv run crawl-books crawl --stop-after 2
uv run crawl-books status
# 同一份 data/crawl.sqlite3 继续，已有两本与目录页不再请求。
uv run crawl-books crawl
# 不再请求网络，从采集库导出 CSV 和业务数据库。
uv run crawl-books export
uv run python -m bookdata --db data/library.sqlite3 list

--state 可以为另一轮任务选新的进度文件。想主动刷新整份目录时，为这一轮创建新状态，程序就会重新发现和读取；继续使用已有进度，则延续其中的完成记录。多次运行时先确认自己要的是继续工作还是发起一轮新的采集。

3. 用请求记录验证恢复过程

先用 pytest fixture 建立独立练习站，再真正关闭并重开进度库：

# 项目文件：crawler_project/tests/conftest.py
# 每个测试都用独立站点和随机端口，故障开关不会污染下一项检查。
import pytest

from crawlsite import running_site


@pytest.fixture
def site():
    with running_site() as opened:
        yield opened

# 项目文件：crawler_project/tests/test_resume.py
# 真正关闭进度库再打开，比较第二轮实际请求，检查恢复的是持久化进度。
import asyncio

from catalogcrawl.runner import crawl
from catalogcrawl.state import CrawlState


def test_resume_only_remaining_details(site, tmp_path):
    path = tmp_path / "crawl.sqlite3"
    start = site.base_url + "/catalog/?page=1"
    with CrawlState(path, start) as state:
        first = asyncio.run(crawl(state, interval=0, stop_after=2))
        assert first["jobs_done"] == 2
        assert first["jobs_pending"] == 4
    before = len(site.state.requests)
    with CrawlState(path, start) as state:
        second = asyncio.run(crawl(state, interval=0))
        assert second["items"] == second["jobs_done"] == 6
        assert second["jobs_pending"] == 0
    paths = [url.split("?", 1)[0] for url, started in site.state.requests[before:]]
    assert sorted(paths) == ["/books/b103", "/books/b104", "/books/b105", "/books/b106"]
    assert site.state.counts["/catalog/"] == 3
    assert site.state.counts["/books/b101"] == 1

测试检查第二轮具体只出现四个详情地址，而不只比较最终数量。配套检查还覆盖分页绕回、重试次数、解析失败、别名记录、同次提交回滚、取消后的待处理状态，以及独立进程中的 CSV 与数据库导出。

# 在 crawler_project 执行全部核心检查，不需要启动常驻服务。
uv run python -m pytest