v1.0.11: 新增历史记录回填脚本 backfill.py(幂等, 支持指定任务), 补齐数据库功能上线前的爬取记录

README.md

@@ -64,7 +64,7 @@
 - 详情页结果表中点「📋 元数据」即可在线查看
 
 ### 6. 数据库记录（MySQL）
-任务、运行记录、爬取结果实时写入 MySQL（`121.40.164.32:16006`，库 `crawler`），**网页完整内容不入库**（存磁盘文件），库中只存标题、网址、状态、文件路径等元数据：
+任务、运行记录、爬取结果实时写入 MySQL（`121.40.164.32:16006`，账号 `uni_crawler`，库 `uni_crawler`），**网页完整内容不入库**（存磁盘文件），库中只存标题、网址、状态、文件路径等元数据：
 - `crawl_tasks` — 任务信息（含回收站标记 deleted_at）
 - `crawl_runs` — 每次运行记录（状态/进度/成功失败数/图片数/时间）
 - `crawl_results` — 每页一条（**status: OK/FAIL** 成功失败标记、标题、网址、来源链接、深度、错误信息、文件路径、图片数）

backfill.py

@@ -0,0 +1,26 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""
+历史爬取记录回填数据库 (幂等, 可重复执行)
+
+用法:
+    python backfill.py                  # 回填所有任务的历史记录
+    python backfill.py <task_id> ...    # 只回填指定任务
+
+说明:
+    - 将本地 data/ 中的任务/运行/爬取结果(含成功与失败)全量写入 MySQL
+    - 网页内容不入库, 只写元数据; 重复执行不会产生重复记录
+"""
+import sys
+
+import db
+import store
+
+if __name__ == "__main__":
+    ids = [a for a in sys.argv[1:] if a.strip()] or None
+    db.init_db()
+    stats = db.sync_all_history(ids)
+    if ids:
+        print(f"[backfill] 已回填 {len(ids)} 个任务: {stats}")
+    else:
+        print(f"[backfill] 已回填全部任务: {stats}")

db.py

@@ -14,12 +14,12 @@ import pymysql
 DB_CONFIG = dict(
     host="121.40.164.32",
     port=16006,
-    user="root",
-    password="hz_123",
+    user="uni_crawler",
+    password="wleD6x2T",
     charset="utf8mb4",
 )
 
-DB_NAME = "crawler"
+DB_NAME = "uni_crawler"
 
 DDL = [
     """
@@ -262,3 +262,30 @@ def sync_run(run, persist):
         insert_results(run, results[synced:])
         run["_db_count"] = len(results)
         persist(run.get("task_id"), run)
+
+
+# ---------------- 历史数据回填 (幂等) ----------------
+
+def sync_all_history(task_ids=None):
+    """把本地 JSON 中的历史任务/运行/结果全量回填数据库
+    可重复执行 (INSERT IGNORE + 唯一键去重)
+    task_ids: 指定只回填的任务ID列表, 默认全部
+    返回统计 dict
+    """
+    import store as _store
+    stats = {"tasks": 0, "runs": 0, "results": 0}
+    for task in _store.load_tasks():
+        if task_ids and task["id"] not in task_ids:
+            continue
+        upsert_task(task)
+        stats["tasks"] += 1
+        for run in _store.get_runs(task["id"]):
+            upsert_run(run)
+            stats["runs"] += 1
+            results = run.get("results", [])
+            if results:
+                insert_results(run, results)
+                stats["results"] += len(results)
+                run["_db_count"] = len(results)
+                _store.save_run(task["id"], run)   # 记录已同步标记, 避免运行中重复插入
+    return stats