#!/usr/bin/env python3 # -*- coding: utf-8 -*- """script_name 空值清洁工 (janitor) — 每日收尾跑，幂等。 背景：812 个脚本手写 `INSERT INTO gov_raw (...)` 时漏了 script_name 字段， 写入的记录 script_name 落到 DEFAULT ''。逐个改脚本风险大， 改为「事后自动归属」：每天把空值按三级口径补上。 三级口径（从严到宽，只取唯一映射，歧义一律不填）： ① site_name → 脚本 （取自库内同 site_name 的非空记录） ② 域名 → 脚本 （取自库内同域名的非空记录） ③ 源码域名 → 脚本 （扫 crawl_*.py 的 BASE_URL/LIST_URL 等常量行，覆盖库内无参照的站） 用法： python3 backfill_script_name.py [--dry-run] 建议： 挂 crontab 在日跑收尾之后（日跑约 08:30 结束） """ import ast import collections import glob import os import re import sqlite3 import sys import time DB = os.getenv("SEARCH_DB", "/root/search.db") DIR = os.path.dirname(os.path.abspath(__file__)) or "/root/gov_crawler" DRY = "--dry-run" in sys.argv URL_RE = re.compile(r"https?://([A-Za-z0-9\.\-]+)") def log(m): print("[%s] %s" % (time.strftime("%Y-%m-%d %H:%M:%S"), m), flush=True) def norm_dom(u): m = re.match(r"https?://([^/]+)", u or "") return m.group(1).lower().lstrip("www.") if m else "" def main(): t0 = time.time() db = sqlite3.connect(DB, timeout=300) db.execute("PRAGMA busy_timeout=300000") c = db.cursor() empty = c.execute("SELECT COUNT(*) FROM gov_raw WHERE script_name IS NULL OR script_name=''").fetchone()[0] if empty == 0: log("无空值，无需处理") return log("发现 %d 条空 script_name" % empty) # ---- 口径① site_name ---- site_map = {} rows = list(c.execute("""SELECT site_name, script_name, COUNT(*) FROM gov_raw WHERE script_name IS NOT NULL AND script_name != '' AND site_name IS NOT NULL AND site_name != '' GROUP BY site_name, script_name""")) agg = collections.defaultdict(collections.Counter) for sn, sc, n in rows: agg[sn][sc] += n site_map = {sn: list(v)[0] for sn, v in agg.items() if len(v) == 1} # 口径①的补充：库内「压倒性多数」也可采信（唯一映射失败时用）。 # 例：宣威市-公示公告 → crawl_xw_shx.py 2130 条 vs crawl_xw.py 0 条。 # 阈值：占比 ≥90% 且 主流样本 ≥10 条，避免小样本噪声定调。 site_map_major = {} for sn, cc in agg.items(): if len(cc) < 2: continue top, top_n = cc.most_common(1)[0] total = sum(cc.values()) if top_n >= 10 and top_n / total >= 0.90: site_map_major[sn] = top # ---- 口径② 源码域名（覆盖库内无参照的站）---- # 注：曾试过「库内同域名非空记录」作为中间口径，实测收益 0 条（有非空记录的站 # 基本都能被①命中），却要全表 GROUP BY page_url，白跑 7 分钟 → 已删除。 src_agg = collections.defaultdict(set) for p in glob.glob(os.path.join(DIR, "crawl_*.py")): name = os.path.basename(p) try: src = open(p, encoding="utf-8", errors="ignore").read() except Exception: continue for line in src.splitlines(): s = line.strip() if s.startswith("#"): continue if not re.search(r"(BASE_URL|LIST_URL|LIST_TPL|BASE|API|URL|HOST|SITE)\s*=", s): continue for m in URL_RE.finditer(s): d = m.group(1).lower().lstrip("www.") if d.replace(".", "").isdigit(): continue src_agg[d].add(name) src_map = {d: list(v)[0] for d, v in src_agg.items() if len(v) == 1} # ---- 口径③ 脚本声明的 SITE_NAME 常量（最精准：记录的 site_name 就是写它的脚本声明的）---- # ⚠️ 必须用 ast.literal_eval 解出真实字符串：不少脚本把中文写成 \uXXXX 转义 # （如 SITE_NAME="\u5ba3\u5a01\u5e02-\u516c\u793a\u516c\u544a"），拿原始文本比中文永远比不中。 sn_const = collections.defaultdict(set) _SELF = os.path.basename(__file__) _pat = re.compile(r"""^\s*(?:SITE_NAME|SITE)\s*=\s*(.+?)\s*(?:#.*)?$""") for p in glob.glob(os.path.join(DIR, "*.py")): name = os.path.basename(p) if name == _SELF: # ⚠️ 必须排除自己：janitor 源码/注释里会出现站点名（如举例「长兴县-公告公示」）， # 否则会把自己当成第二个「写该站点的脚本」→ 撞成歧义 → 永远弃权（2026-09-21 踩过）。 continue try: _src = open(p, encoding="utf-8", errors="ignore").read() except Exception: continue for line in _src.splitlines(): m = _pat.match(line) if not m: continue try: val = ast.literal_eval(m.group(1).strip()) except Exception: continue if isinstance(val, str) and val.strip(): sn_const[val.strip()].add(name) const_map = {sn: list(v)[0] for sn, v in sn_const.items() if len(v) == 1} # ---- 口径②b 「多站点字典」里的 site 值 ---- # 形如 COLUMNS = {"gggs": {"site": "长兴县-公告公示", ...}, ...}，模块级没有 SITE_NAME， # 单靠 SITE_NAME 常量会整类漏掉（crawl_zjcx_hp.py 就是：一个脚本抓 2 个栏目）。 for p in glob.glob(os.path.join(DIR, "*.py")): name = os.path.basename(p) if name == _SELF: continue try: _src = open(p, encoding="utf-8", errors="ignore").read() except Exception: continue for m in re.finditer(r"""['"]site(?:_name)?['"]\s*:\s*(.+?)\s*[,}\n]""", _src): try: val = ast.literal_eval(m.group(1).strip()) except Exception: continue if isinstance(val, str) and val.strip(): sn_const[val.strip()].add(name) const_map = {sn: list(v)[0] for sn, v in sn_const.items() if len(v) == 1} log("映射表: site_name %d | 域名(源码) %d | SITE_NAME常量 %d" % (len(site_map), len(src_map), len(const_map))) # ---- 逐条归属 ---- # ⚠️ 必须先把待处理行物化成 list：若在 SELECT 迭代过程中对同一连接执行 UPDATE/commit， # 游标会被重置，循环提前结束（2026-09-21 实测只填了 500 条就停）。 todo = list(c.execute("""SELECT id, COALESCE(site_name,''), COALESCE(page_url,'') FROM gov_raw WHERE script_name IS NULL OR script_name = ''""")) filled = collections.Counter() still = 0 batch = [] for rid, sn, url in todo: sc = site_map.get(sn) src = "①site" if not sc: sc = site_map_major.get(sn) src = "①'major" if not sc: sc = const_map.get(sn) src = "②const" if not sc: sc = src_map.get(norm_dom(url)) src = "③src" if sc: filled[src] += 1 if not DRY: batch.append((sc, rid)) else: still += 1 if len(batch) >= 500: c.executemany("UPDATE gov_raw SET script_name=? WHERE id=?", batch) db.commit() batch = [] if batch and not DRY: c.executemany("UPDATE gov_raw SET script_name=? WHERE id=?", batch) db.commit() log("回填: ①site %d | ①'major %d | ②const %d | ③src %d | 仍无法归属 %d%s" % (filled["①site"], filled["①'major"], filled["②const"], filled["③src"], still, " [DRY-RUN 未写库]" if DRY else "")) if still: log("仍无法归属的 TOP 8:") for sn, n in c.execute("""SELECT COALESCE(NULLIF(site_name,''),'(空)'), COUNT(*) FROM gov_raw WHERE script_name IS NULL OR script_name='' GROUP BY 1 ORDER BY COUNT(*) DESC LIMIT 8"""): log(" %-36s %d" % (sn[:36], n)) log("剩余空值 %d，用时 %.1fs" % (c.execute("SELECT COUNT(*) FROM gov_raw WHERE script_name IS NULL OR script_name=''").fetchone()[0], time.time() - t0)) db.close() if __name__ == "__main__": main()