Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
aa0c2f56d0 | ||
|
|
bf6be47c01 |
+26
@@ -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}")
|
||||||
@@ -67,9 +67,7 @@ DDL = [
|
|||||||
source_url VARCHAR(2000),
|
source_url VARCHAR(2000),
|
||||||
depth INT,
|
depth INT,
|
||||||
attempts INT DEFAULT 1,
|
attempts INT DEFAULT 1,
|
||||||
html_file VARCHAR(500),
|
base_file VARCHAR(500),
|
||||||
txt_file VARCHAR(500),
|
|
||||||
meta_file VARCHAR(500),
|
|
||||||
image_count INT DEFAULT 0,
|
image_count INT DEFAULT 0,
|
||||||
image_files TEXT,
|
image_files TEXT,
|
||||||
UNIQUE KEY uk_run_url (run_id, url(500))
|
UNIQUE KEY uk_run_url (run_id, url(500))
|
||||||
@@ -78,6 +76,17 @@ DDL = [
|
|||||||
]
|
]
|
||||||
|
|
||||||
|
|
||||||
|
def _migrate(conn):
|
||||||
|
"""存量表结构迁移: 合并 html_file/txt_file/meta_file 为 base_file"""
|
||||||
|
with conn.cursor() as cur:
|
||||||
|
cur.execute("SHOW COLUMNS FROM crawl_results LIKE 'html_file'")
|
||||||
|
if cur.fetchone():
|
||||||
|
cur.execute("ALTER TABLE crawl_results ADD COLUMN base_file VARCHAR(500) NULL AFTER meta_file")
|
||||||
|
cur.execute("UPDATE crawl_results SET base_file = REPLACE(html_file, '.html', '') WHERE base_file IS NULL")
|
||||||
|
cur.execute("ALTER TABLE crawl_results DROP COLUMN html_file, DROP COLUMN txt_file, DROP COLUMN meta_file")
|
||||||
|
print("[db] 表结构迁移完成: html_file/txt_file/meta_file -> base_file")
|
||||||
|
|
||||||
|
|
||||||
def _conn():
|
def _conn():
|
||||||
cfg = dict(DB_CONFIG)
|
cfg = dict(DB_CONFIG)
|
||||||
cfg["database"] = DB_NAME
|
cfg["database"] = DB_NAME
|
||||||
@@ -100,6 +109,7 @@ def init_db():
|
|||||||
cur.execute(f"CREATE DATABASE IF NOT EXISTS `{DB_NAME}` DEFAULT CHARACTER SET utf8mb4")
|
cur.execute(f"CREATE DATABASE IF NOT EXISTS `{DB_NAME}` DEFAULT CHARACTER SET utf8mb4")
|
||||||
conn.close()
|
conn.close()
|
||||||
conn = _conn()
|
conn = _conn()
|
||||||
|
_migrate(conn)
|
||||||
with conn.cursor() as cur:
|
with conn.cursor() as cur:
|
||||||
for ddl in DDL:
|
for ddl in DDL:
|
||||||
cur.execute(ddl)
|
cur.execute(ddl)
|
||||||
@@ -219,8 +229,21 @@ def upsert_run(run):
|
|||||||
|
|
||||||
# ---------------- 爬取结果 (每页一条, 成功失败均记录) ----------------
|
# ---------------- 爬取结果 (每页一条, 成功失败均记录) ----------------
|
||||||
|
|
||||||
|
def _base_of(entry):
|
||||||
|
"""从结果条目提取基础文件名 (html/txt/meta 三个后缀共用同一前缀)"""
|
||||||
|
for f in (entry.get("meta_file"), entry.get("html_file"), entry.get("txt_file")):
|
||||||
|
f = f or ""
|
||||||
|
if f.endswith(".meta.json"):
|
||||||
|
return f[:-10]
|
||||||
|
if f.endswith(".html"):
|
||||||
|
return f[:-5]
|
||||||
|
if f.endswith(".txt"):
|
||||||
|
return f[:-4]
|
||||||
|
return ""
|
||||||
|
|
||||||
|
|
||||||
def insert_results(run, results):
|
def insert_results(run, results):
|
||||||
"""批量插入爬取结果 (增量)"""
|
"""批量插入爬取结果 (增量); 文件只记基础名 base_file"""
|
||||||
if not results:
|
if not results:
|
||||||
return
|
return
|
||||||
|
|
||||||
@@ -234,7 +257,7 @@ def insert_results(run, results):
|
|||||||
r.get("status", "FAIL"), r.get("error"),
|
r.get("status", "FAIL"), r.get("error"),
|
||||||
_dt(r.get("crawl_time")), (r.get("source_url") or "")[:2000],
|
_dt(r.get("crawl_time")), (r.get("source_url") or "")[:2000],
|
||||||
r.get("depth"), r.get("attempts", 1),
|
r.get("depth"), r.get("attempts", 1),
|
||||||
r.get("html_file", ""), r.get("txt_file", ""), r.get("meta_file", ""),
|
_base_of(r),
|
||||||
len(r.get("images", []) or []),
|
len(r.get("images", []) or []),
|
||||||
_j([im.get("file") for im in (r.get("images") or [])]),
|
_j([im.get("file") for im in (r.get("images") or [])]),
|
||||||
))
|
))
|
||||||
@@ -243,8 +266,8 @@ def insert_results(run, results):
|
|||||||
"""INSERT IGNORE INTO crawl_results
|
"""INSERT IGNORE INTO crawl_results
|
||||||
(run_id, task_id, mode, url, title, status, error,
|
(run_id, task_id, mode, url, title, status, error,
|
||||||
crawl_time, source_url, depth, attempts,
|
crawl_time, source_url, depth, attempts,
|
||||||
html_file, txt_file, meta_file, image_count, image_files)
|
base_file, image_count, image_files)
|
||||||
VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)""",
|
VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)""",
|
||||||
rows,
|
rows,
|
||||||
)
|
)
|
||||||
conn.close()
|
conn.close()
|
||||||
@@ -262,3 +285,30 @@ def sync_run(run, persist):
|
|||||||
insert_results(run, results[synced:])
|
insert_results(run, results[synced:])
|
||||||
run["_db_count"] = len(results)
|
run["_db_count"] = len(results)
|
||||||
persist(run.get("task_id"), run)
|
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
|
||||||
Reference in New Issue
Block a user