#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""qc_tier1.py —— 爬虫数据 QC · Tier 1 全自动体检（只读）

产出：/root/gov_crawler/qc_out/qc_tier1_<date>.md  +  .json
判据口径（2026-09-25 与用户确认）：
  * 配置编号【已弃用】—— 历史遗留（当时调度是「50 个/批分批跑」），现行调度不看它；
    故 QC 一律以 Script_name 为主键，不校验编号。
  * 停滞阈值：>30 天算异常（--stale-days 可调）
  * 先出清单，不自动修
"""
import argparse
import datetime
import difflib
import hashlib
import json
import os
import re
import sqlite3
import sys

D = "/root/gov_crawler/"
OUT = D + "qc_out/"
DB = "/root/search.db"
ap = argparse.ArgumentParser()
ap.add_argument("--stale-days", type=int, default=30)
A = ap.parse_args()
TODAY = datetime.date.today()

os.makedirs(OUT, exist_ok=True)
issues = {}   # 维度 -> list


def add(k, rec):
    issues.setdefault(k, []).append(rec)


# ════════ 读 config ════════
raw = open(D + "daily_crawl_config.json", "rb").read()
cfg = json.loads(raw.decode("utf-8"))
cr = cfg["crawlers"] if isinstance(cfg, dict) and "crawlers" in cfg else cfg
cfg_md5 = hashlib.md5(raw).hexdigest()[:12]

print("=== 读配置：%d 条目 | md5 %s ===" % (len(cr), cfg_md5))

disk = {f for f in os.listdir(D) if f.startswith("crawl_") and f.endswith(".py")}


def norm_args(a):
    """args 可能是 list 也可能是字符串 —— 统一成 token 列表（调度器两种都吃）"""
    if a is None:
        return []
    if isinstance(a, list):
        return [str(x) for x in a]
    return str(a).split()


KNOWN = ("--pages", "--col", "--only", "--max-pages", "--year", "--since", "--limit",
         "--all", "--jsonl", "--force", "--days", "--top", "--unit", "--mode", "--site")

for i, c in enumerate(cr):
    s = c.get("script")
    no = i + 1
    if not s:
        add("D1 config 条目自身问题", {"类型": "script 为空", "条目号": no, "name": c.get("name")})
        continue
    if not os.path.exists(D + s):
        add("D1 config 条目自身问题", {"类型": "磁盘无此脚本", "条目号": no, "script": s, "name": c.get("name")})
    toks = norm_args(c.get("args"))
    for t in toks:
        if t.startswith("-") and not any(t.startswith(k) for k in KNOWN):
            add("D1 config 条目自身问题", {"类型": "未知参数", "script": s, "arg": t, "args": toks})
    # ⚠️ 2026-09-25 校准（读 crawl_scheduler_v2.py 源码后剔除假阳性）：
    #   args 为空 = 无害（调度器 cmd 直接拼 script + args，空串即不带参数跑）
    #   sync_mode = 调度器**零引用**的死字段 → 不参与任何判定
    if not toks:
        pass

# 重复 (script,args)
from collections import Counter
dup = Counter((c.get("script"), json.dumps(norm_args(c.get("args")), ensure_ascii=False)) for c in cr)
for (s, a), n in dup.items():
    if n > 1:
        add("D1 config 条目自身问题", {"类型": "同一脚本+同参数重复 %d 次（会重复跑）" % n, "script": s, "args": a})

# 孤儿脚本
cfg_scripts = set(c.get("script") for c in cr if c.get("script"))
for f in sorted(disk - cfg_scripts):
    add("D2 孤儿脚本（磁盘有、config 无 → 永不更新）", {"script": f})

# D3（重构）：不支持 --pages 且**默认会翻全站** → 日跑必撞 SCRIPT_TIMEOUT=600s 被杀 → 静默永不更新
FULLSITE = ("total_pages", "totalpage", "_nPageCount", "pageCount", "totalPages", "page_count")
nosup = 0
for sc in sorted(cfg_scripts):
    fp = D + sc
    if not os.path.exists(fp):
        continue
    try:
        t = open(fp, encoding="utf-8", errors="ignore").read()
    except Exception:
        continue
    if "--pages" in t or "--max-pages" in t:
        continue
    nosup += 1
    risk = [k for k in FULLSITE if k in t]
    loops = bool(re.search(r"range\([^)\n]{0,60}[Pp]age", t)) or bool(re.search(r"while[^:\n]{0,60}[Pp]age\s*<", t))
    cap = ("MAX_PAGES" in t) or ("max_pages" in t)
    if risk and (loops or not cap):
        add("D3 无 --pages 且有全站翻页风险（日跑撞 600s 超时）",
            {"script": sc, "页码变量": ",".join(risk[:3]), "有翻页循环": loops, "有页数上限": cap})
print("  （不支持 --pages 的脚本共 %d 个，其中判为高风险的见 D3）" % nosup)

# D12（已改口径）：手写 INSERT 缺 script_name 的静态判据【已证伪，撤除】
#   2026-09-25 反验：抽 12 个「缺 script_name 的 INSERT」脚本，其站点行全部填好了脚本名
#   （507/507、827/827、572/572…，空名合计 1）→ crawler_lib 在连接层自动注入 script_name，
#   只看 INSERT 列名会误判。真正口径 = 库内实测哪些站点有空名行（见 D11）。
# ════════ 读库 ════════
print("=== 读库 ===")
db = sqlite3.connect("file:%s?mode=ro" % DB, uri=True, timeout=30)
db.execute("PRAGMA busy_timeout=30000")
n_total = db.execute("SELECT COUNT(*) FROM gov_raw").fetchone()[0]
n_empty = db.execute("SELECT COUNT(*) FROM gov_raw WHERE script_name IS NULL OR TRIM(script_name)=''").fetchone()[0]
print("  gov_raw %d 行 | script_name 空 %d" % (n_total, n_empty))

rows = list(db.execute(
    "SELECT script_name, COUNT(*), MAX(publish_date), MIN(publish_date), "
    "SUM(CASE WHEN attachments IS NULL OR attachments='' THEN 1 ELSE 0 END), "
    "SUM(CASE WHEN content IS NULL OR TRIM(content)='' THEN 1 ELSE 0 END), "
    "MAX(inserted_at) "
    "FROM gov_raw WHERE script_name IS NOT NULL AND TRIM(script_name)<>'' GROUP BY script_name"))
print("  distinct script_name: %d" % len(rows))

db_scripts = {}
for s, c, mx, mn, noatt, noempty, lastins in rows:
    db_scripts[s] = dict(rows=c, maxdate=mx, mindate=mn, noatt=noatt, empty=noempty, lastins=lastins)

# D4 库内脚本名 → 磁盘无文件
all_py = {f for f in os.listdir(D) if f.endswith(".py")} | disk
for s, d in sorted(db_scripts.items(), key=lambda x: -x[1]["rows"]):
    if s in all_py:
        continue
    sug = difflib.get_close_matches(s, list(all_py), n=2, cutoff=0.6)
    add("D4 库内脚本名磁盘查无此文件", {"script_name": s, "行数": d["rows"],
                                     "最新": d["maxdate"], "疑似应为此脚本": sug})

# D5 注册了但库里 0 行
for s in sorted(cfg_scripts - set(db_scripts)):
    add("D5 config 注册但库内 0 行", {"script": s})

# D6 库有数据但不在 config
for s in sorted(set(db_scripts) - cfg_scripts):
    if s in all_py or (s or "").endswith(".py"):
        add("D6 库有数据但不在 config（脚本可能已改名/退役）",
            {"script_name": s, "行数": db_scripts[s]["rows"], "最新": db_scripts[s]["maxdate"]})

# D7 停滞 > N 天
stale = []
for s in sorted(cfg_scripts):
    d = db_scripts.get(s)
    if not d:
        continue
    mx = d["maxdate"] or ""
    try:
        days = (TODAY - datetime.date(*[int(x) for x in mx.split("-")[:3]])).days
    except Exception:
        days = None
    if days is None or days > A.stale_days:
        stale.append({"script": s, "行数": d["rows"], "最新发布日期": mx or "(无)",
                      "停滞天数": days if days is not None else "无有效日期",
                      "最后入库": d["lastins"]})
stale.sort(key=lambda x: -(x["停滞天数"] if isinstance(x["停滞天数"], int) else 99999))
for x in stale:
    add("D7 停滞 >%d 天" % A.stale_days, x)

# D8 行数≥30 但全部无附件（linkbox 类高危池）
zero_att = []
for s, d in db_scripts.items():
    if d["rows"] >= 30 and d["noatt"] == d["rows"]:
        zero_att.append({"script_name": s, "行数": d["rows"], "最新": d["maxdate"]})
zero_att.sort(key=lambda x: -x["行数"])
for x in zero_att:
    add("D8 行数≥30 却全部无附件（高危候选，需页面级核查）", x)

# D9 空正文
for s, d in sorted(db_scripts.items(), key=lambda x: -x[1]["empty"]):
    if d["empty"] > 0:
        add("D9 存在空正文行", {"script_name": s, "空正文": d["empty"], "总行数": d["rows"]})

# D10 FTS
gs = db.execute("SELECT COUNT(*) FROM gov_search").fetchone()[0]
add("D10 FTS 计数（需先确认 gov_search 服务范围）",
    {"gov_raw": n_total, "gov_search": gs, "差": gs - n_total})

# D11 空 script_name 的站点分布（可填充候选）
top_empty = list(db.execute(
    "SELECT site_name, COUNT(*) FROM gov_raw WHERE script_name IS NULL OR TRIM(script_name)='' "
    "GROUP BY site_name ORDER BY 2 DESC LIMIT 25"))
# D11b：为空名行站点找候选脚本（域名出现在脚本源码里 → 候选；多候选则留空，宁缺毋滥）
_scripts_src = {}
for _f in sorted(disk):
    try:
        _scripts_src[_f] = open(D + _f, encoding="utf-8", errors="ignore").read()[:200000]
    except Exception:
        pass
_cand = []
for _site, _c in top_empty[:25]:
    _dom = None
    for _r2 in db.execute("SELECT page_url FROM gov_raw WHERE site_name=? AND script_name IS NULL LIMIT 1", (_site,)):
        _m2 = re.search(r"://(?:www\.)?([^/]+)/", _r2[0] or "")
        _dom = _m2.group(1) if _m2 else None
    if not _dom:
        _cand.append((_site, _c, "无URL", []))
        continue
    _hits = [_f for _f, _s in _scripts_src.items() if _dom in _s]
    _cand.append((_site, _c, _dom, _hits[:3]))
for _site, _c, _dom, _hits in _cand:
    add("D11b 空名行站点 → 候选脚本（可填充）",
        {"site_name": _site, "空名行数": _c, "域名": _dom,
         "候选脚本": ",".join(_hits) if _hits else "无候选（需人工）"})

# ════════ 输出 ════════
day = TODAY.strftime("%Y%m%d")
md = ["# 爬虫数据 QC · Tier 1 自动体检", "",
      "- 运行日期：%s | config md5：%s | 条目数：%d | 停滞阈值：>%d 天" % (TODAY, cfg_md5, len(cr), A.stale_days),
      "- 主键：**Script_name**（配置编号已弃用：历史遗留，当时调度为 50 个/批分批跑，现行调度不用它）",
      "- gov_raw：%d 行 | script_name 空：%d（%.2f%%） | distinct script_name：%d" % (
          n_total, n_empty, 100.0 * n_empty / n_total, len(db_scripts)),
      "- 说明：本表**只列异常**，合格项不出现；确认后再定批量策略。", ""]
tot = 0
for k in sorted(issues):
    v = issues[k]
    tot += len(v)
    md.append("## %s —— %d 条" % (k, len(v)))
    md.append("")
    for r in v[:200]:
        md.append("- " + " ｜ ".join("%s=%s" % (kk, vv) for kk, vv in r.items()))
    if len(v) > 200:
        md.append("- …（共 %d 条，余下见 json）" % len(v))
    md.append("")
md.append("## D11 空 script_name 的站点分布 top25")
md.append("")
for s, c in top_empty:
    md.append("- %-46s %6d 行" % (s or "(空)", c))

open(OUT + "qc_tier1_%s.md" % day, "w", encoding="utf-8").write("\n".join(md))
json.dump({"date": str(TODAY), "cfg_md5": cfg_md5, "cfg_entries": len(cr),
           "gov_raw": n_total, "script_name_empty": n_empty, "issues": issues,
           "empty_scriptname_sites": top_empty},
          open(OUT + "qc_tier1_%s.json" % day, "w", encoding="utf-8"), ensure_ascii=False, indent=1)
print("\n=== Tier 1 汇总（异常项总计 %d）===" % tot)
for k in sorted(issues):
    print("  %-52s %5d" % (k[:52], len(issues[k])))
print("\n产出: %sqc_tier1_%s.md / .json" % (OUT, day))
db.close()