o iy!@sndZddlZddlZddlmZddlmZmZmZddl m Z ddl m Z edZ dZ d ZGd d d ZdS) u断点续爬管理器 维护 {category_id}:{page}:{doc_id} 三级游标，持久化到本地 SQLite。 支持进程崩溃/中断后从最近检查点恢复，避免从头重跑。 N)Path)OptionalSetTuple)datetime)loggerz/tmp/tax_documents/indexz checkpoint.db2c@seZdZdZ d!dededeefddZde efd d Z d eddfd d Z d"ddZ d ede fddZdeddfddZdefddZd"ddZdefddZd"ddZdejfddZd"dd ZdS)#CheckpointManageru 断点续爬管理器 使用示例： cp = CheckpointManager(task_id="task-abc", category_id=5) # 启动时恢复 completed_urls = cp.load_completed_urls() # 处理文档时 for doc in all_documents: if doc["url"] in completed_urls: continue # 跳过已完成 process(doc) cp.mark_done(doc["url"]) # 记录完成 # 任务结束时清理 cp.clear() Ntask_id category_iddb_dircCsF||_||_|p t|_|jt|_t|_g|_ t |_ | dSN) r r DEFAULT_CHECKPOINT_DIRr CHECKPOINT_DB_NAMEdb_path threadingLock_lock _pending_urlsset_completed_urls _ensure_db)selfr r r r`/lsinfo/ai/hellotax_ai/data_center/backend/app/services/tax_data_processor/checkpoint_manager.py__init__,s    zCheckpointManager.__init__returnc Cs~|}|d|j|jf}Wdn1swYdd|D|_td|jd|jdt|jd|jS) uD启动时加载已完成的 URL 集合，用于跳过已处理文档zESELECT doc_url FROM checkpoints WHERE task_id = ? AND category_id = ?NcSsh|]}|dqS)rr).0rowrrr Hsz8CheckpointManager.load_completed_urls..[CheckpointManager] task= cat=u 已恢复  条断点记录) _connectexecuter r fetchallrrinfolen)rconnrowsrrrload_completed_urlsAs  z%CheckpointManager.load_completed_urlsdoc_urlcCsj|j||j"|j|t|jtkr#|WddSWddS1s.wYdS)uH标记单条文档处理完成，每 CHECKPOINT_INTERVAL 条批量落盘N)raddrrappendr'CHECKPOINT_INTERVAL_flushrr+rrr mark_doneOs   "zCheckpointManager.mark_donecCs4|j |WddS1swYdS)uH强制将缓冲区写入磁盘（任务完成或异常退出时调用）N)rr/rrrrflushWs "zCheckpointManager.flushcCs ||jvS)u判断某 URL 是否已完成)rr0rrris_done\s zCheckpointManager.is_donepagec CsN|}|d|j|j|tfWddS1s wYdS)uH保存当前列表页游标（可选，用于从中断的页码恢复）a INSERT INTO page_cursors (task_id, category_id, last_page, updated_at) VALUES (?, ?, ?, ?) ON CONFLICT(task_id, category_id) DO UPDATE SET last_page = excluded.last_page, updated_at = excluded.updated_at N)r#r$r r rutcnow isoformat)rr5r(rrrsave_page_cursor`s "z"CheckpointManager.save_page_cursorc Cs|}|d|j|jf}Wdn1swY|r&|dnd}|dkrBtd|jd|jd|dd|d |S) u<加载已完成的最后页码，返回 0 表示从头开始zHSELECT last_page FROM page_cursors WHERE task_id = ? AND category_id = ?Nrr r!u 从第 u 页继续（已完成第 1~u 页）)r#r$r r fetchonerr&)rr(rr5rrrload_page_cursorms$  z"CheckpointManager.load_page_cursorcCsz|}|d|j|jf|d|j|jfWdn1s#wY|jtd|jd|jddS)uB任务成功完成后清理检查点记录（节省磁盘空间）z=DELETE FROM checkpoints WHERE task_id = ? AND category_id = ?z>DELETE FROM page_cursors WHERE task_id = ? AND category_id = ?Nr r!u 检查点已清理)r#r$r r rclearrr&rr(rrrr<|s    zCheckpointManager.clearcCs|j|jt|jt|jdS)u返回当前断点统计信息)r r completed_count pending_flush)r r r'rrr2rrrstatss zCheckpointManager.statscCs\|jjddd|}|d|d|dWddS1s'wYdS)u&初始化 SQLite 数据库和表结构T)parentsexist_oka CREATE TABLE IF NOT EXISTS checkpoints ( id INTEGER PRIMARY KEY AUTOINCREMENT, task_id TEXT NOT NULL, category_id INTEGER NOT NULL, doc_url TEXT NOT NULL, created_at TEXT NOT NULL DEFAULT (datetime('now')), UNIQUE(task_id, category_id, doc_url) ) z CREATE INDEX IF NOT EXISTS idx_cp_task_cat ON checkpoints(task_id, category_id) af CREATE TABLE IF NOT EXISTS page_cursors ( task_id TEXT NOT NULL, category_id INTEGER NOT NULL, last_page INTEGER NOT NULL DEFAULT 0, updated_at TEXT NOT NULL, PRIMARY KEY (task_id, category_id) ) N)r mkdirr#r$r=rrrrs  "zCheckpointManager._ensure_dbcCs,tjt|jdd}|d|d|S)uE返回 WAL 模式连接（上下文管理器，自动提交/回滚） )timeoutzPRAGMA journal_mode=WALzPRAGMA synchronous=NORMAL)sqlite3connectstrrr$r=rrrr#s  zCheckpointManager._connectc sjsdStfddjD}zSz(}|d|Wdn1s-wYtdtjdWnt yY}zt d|WYd}~n d}~wwWj dSWj dSj w)uE将 _pending_urls 批量写入数据库（调用方须持有 _lock）Ncsg|] }jj|fqSr)r r )rurlnowrrr sz,CheckpointManager._flush..z INSERT OR IGNORE INTO checkpoints (task_id, category_id, doc_url, created_at) VALUES (?, ?, ?, ?) uheckpointManager] 写入 r"u([CheckpointManager] 写入断点失败: ) rrr6r7r# executemanyrdebugr' Exceptionerrorr<)rr)r(errJrr/s0    zCheckpointManager._flushr )rN)__name__ __module__ __qualname____doc__rHintrrrrr*r1r3boolr4r8r;r<dictr@rrF Connectionr#r/rrrrr s*    "r )rUrFrpathlibrtypingrrrrlogururrrr.r rrrrs