guanlan-src/scripts/guanlan_facts_contract.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""观澜 · 结构化事实契约 v0 (用户 2026-09-06 步骤 5: 数据/时间窗/结论/图表数据/报告文本只能经契约更新).
契约 = 版本化 JSON: 每条结论 claim_id / 脱敏来源引用 (文件@sha16 + 章节) / 本条 sha256 / 时间窗 / 样本独立层级 / 聚合层级 / 适用范围 / 裁决 (六枚举) / 缺失证据 / 归宿 / 显示文案.
来源 (只读): outputs/rudong/sop/findings.json (119) + E5 底稿 finding_rows (模块) + E3 底稿 rows (归宿提议) + E8 底稿_s0 rows (系统/台号).
消费者 (全部由契约生成, 内嵌 contract_sha256): 门户正文段 portal_claims.json · 详细分析卡片 detail_cards.json · 报告摘要 report_summary.md · 问答引用 qa_refs.json.
探针: probe <claim_id> 在隔离副本改 probe 字段 → 四个消费者输出全部变化 → 恢复 (步骤 5 的验收).
用法: build | render | probe <claim_id> | check"""
import datetime as dt, hashlib, json, os, re, shutil, subprocess, sys, tempfile
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]; OUT = ROOT / "outputs/rudong/guanlan"; DERIVED = OUT / "derived"
CONTRACT = OUT / "facts_contract_v0.json"
SRC = {"findings": ROOT / "outputs/rudong/sop/findings.json", "e5": ROOT / "outputs/rudong/paradigm_r1/experiments/E5_coverage_table/底稿.json",
       "e3": ROOT / "outputs/rudong/paradigm_r1/experiments/E3_candidate_closure/底稿.json", "s0": ROOT / "outputs/rudong/paradigm_r1/experiments/E8_rudong_rebuild/底稿_s0.json"}
sha = lambda b: hashlib.sha256(b).hexdigest(); sha16f = lambda p: sha(Path(p).read_bytes())[:16]; J = lambda p: json.loads(Path(p).read_text(encoding="utf-8"))
SIX = ("定论", "准定论·预警", "候选", "参考", "INSUFFICIENT", "撤回")
BANNED = re.compile(r"中广核|华电|国电|大唐|华能|三峡|龙源|/Users/|[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[a-z]{2,}|1[3-9]\d{9}")   # 业主/集团名 · 本机路径 · 邮箱 · 手机 (对外面孔闸)


def canon(x): return json.dumps(x, sort_keys=True, ensure_ascii=False)


def norm_verdict(v):
    v = str(v or "")
    if v.startswith("【已撤回") or v.startswith("撤回"): return "撤回"
    if v.startswith("候选"): return "候选"
    return v if v in SIX else "INSUFFICIENT"


def build():
    F = J(SRC["findings"])["findings"]; e5 = {r["i"]: r for r in J(SRC["e5"])["finding_rows"]}; e3 = {r["i"]: r for r in J(SRC["e3"])["rows"]}; s0 = {r["i"]: r for r in J(SRC["s0"])["findings_rows"]}
    fsha = sha16f(SRC["findings"]); claims = []
    for i, f in enumerate(F):
        cov = f.get("coverage") if isinstance(f.get("coverage"), dict) else {}
        cid = f.get("id") or f"RD-idx-{i:03d}"; m, c, s = e5.get(i, {}), e3.get(i), s0.get(i, {})
        title = str(f.get("title") or "")
        claim = {"claim_id": cid, "index": i, "verdict": norm_verdict(f.get("verdict")), "verdict_raw": f.get("verdict"),
                 "source_refs": [{"file": "outputs/rudong/sop/findings.json", "sha16": fsha, "index": i, "chapter": f.get("source_chapter"), "module_label": f.get("module")}],
                 "time_window": {"window_days": cov.get("window_days"), "note": cov.get("window_days_note"), "evidence_granularity": cov.get("per_day_evidence")} if cov else {"window_days": None, "note": "coverage 为字符串: " + str(f.get("coverage"))[:80]},
                 "sample_level": {"n_machines_covered": cov.get("n_machines") if cov else None, "n_machines_named": s.get("n_tid"), "named": s.get("tids"), "claim_class": f.get("claim_class")},
                 "aggregation_level": "跨机系统性" if (s.get("n_tid") or 0) >= 3 else ("单台/少数台" if s.get("n_tid") else "场级/口径"),
                 "scope": {"systems": s.get("systems"), "units": s.get("units"), "module": m.get("module"), "module_map_how": m.get("map_how")},
                 "missing_evidence": f.get("mandatory_limitation"), "falsifiability": f.get("falsifiability"), "problem_nature": f.get("problem_nature"),
                 "closure": {"human": (c or {}).get("human") or None, "machine_proposed": (c or {}).get("proposed"), "status": "人裁" if (c or {}).get("human") else ("待裁(机器提议)" if c else "非候选, 未提议")},
                 "display": {"title_zh": title, "title_zh_public": BANNED.sub("[脱敏]", title), "title_en": None, "banned_hits": len(BANNED.findall(title))},   # 对外面孔用 title_zh_public (业主/集团名 → [脱敏])
                 "probe": None}
        claim["sha256"] = sha(canon({k: v for k, v in claim.items() if k != "sha256"}).encode("utf-8")); claims.append(claim)
    git = subprocess.run(["git", "-C", str(ROOT), "rev-parse", "--short", "HEAD"], capture_output=True, text=True).stdout.strip()
    con = {"schema": "guanlan-facts-contract/v0", "version": "v0", "farm": "rudong", "built": dt.datetime.now().isoformat(timespec="minutes"), "git_head": git,
           "inputs": {k: {"path": str(v.relative_to(ROOT)), "sha16": sha16f(v)} for k, v in SRC.items()}, "n_claims": len(claims), "by_verdict": {v: sum(1 for c in claims if c["verdict"] == v) for v in SIX},
           "face": "internal (含场名/台号; 对外面孔须过 redact_check DS-2 + BANNED 零命中)", "banned_hits_total": sum(c["display"]["banned_hits"] for c in claims), "claims": claims}
    con["contract_sha256"] = sha(canon({k: v for k, v in con.items() if k not in ("contract_sha256", "built", "git_head")}).encode("utf-8"))
    OUT.mkdir(parents=True, exist_ok=True); CONTRACT.write_text(json.dumps(con, ensure_ascii=False, indent=1), encoding="utf-8"); return con


def render(con, outdir: Path):
    outdir.mkdir(parents=True, exist_ok=True); cs = con["contract_sha256"]; deliverable = [c for c in con["claims"] if c["verdict"] in ("定论", "准定论·预警")]
    hdr = {"contract_sha256": cs, "contract_version": con["version"], "generated": "由 scripts/guanlan_facts_contract.py render 生成, 禁手改"}
    portal = dict(hdr, paragraphs=[{"claim_id": c["claim_id"], "verdict": c["verdict"], "text": c["display"]["title_zh_public"][:120], "systems": c["scope"]["systems"], "window_days": c["time_window"].get("window_days"), "probe": c["probe"]} for c in deliverable])
    cards = dict(hdr, cards=[{"claim_id": c["claim_id"], "module": c["scope"]["module"], "systems": c["scope"]["systems"], "verdict": c["verdict"], "title": c["display"]["title_zh_public"][:90], "n_named": c["sample_level"]["n_machines_named"], "aggregation": c["aggregation_level"], "closure_status": c["closure"]["status"], "missing": (c["missing_evidence"] or "")[:80] if isinstance(c["missing_evidence"], str) else None, "probe": c["probe"]} for c in con["claims"] if c["verdict"] != "撤回"])
    rep = [f"# 报告摘要 (契约生成; contract {cs[:16]}; 禁手改)\n", f"结论 {con['n_claims']} 条: " + ", ".join(f"{k} {v}" for k, v in con["by_verdict"].items()) + "\n", "## 可交付级 (定论 / 准定论·预警)\n"]
    for c in deliverable: rep.append(f"- **{c['claim_id']}** [{c['verdict']}] {c['display']['title_zh_public'][:100]} — 窗 {c['time_window'].get('window_days')} d; 点名 {c['sample_level']['n_machines_named']} 台; 归宿 {c['closure']['status']}; 缺项: {(c['missing_evidence'] or '—')[:60] if isinstance(c['missing_evidence'], str) else '—'}{' [probe=' + str(c['probe']) + ']' if c['probe'] else ''}")
    qa = dict(hdr, refs=[{"claim_id": c["claim_id"], "cite": f"{c['claim_id']} ({c['verdict']}; {c['source_refs'][0]['file']}@{c['source_refs'][0]['sha16']} #{c['index']}; 窗 {c['time_window'].get('window_days')} d)", "probe": c["probe"]} for c in con["claims"]])
    files = {"portal_claims.json": json.dumps(portal, ensure_ascii=False, indent=1), "detail_cards.json": json.dumps(cards, ensure_ascii=False, indent=1), "report_summary.md": "\n".join(rep) + "\n", "qa_refs.json": json.dumps(qa, ensure_ascii=False, indent=1)}
    for n, s in files.items(): (outdir / n).write_text(s, encoding="utf-8")
    return {n: sha(s.encode("utf-8"))[:16] for n, s in files.items()}


def probe(claim_id: str):
    con = J(CONTRACT); base = render(con, DERIVED)
    iso = json.loads(json.dumps(con)); tgt = next(c for c in iso["claims"] if c["claim_id"] == claim_id); tgt["probe"] = f"PROBE-{dt.datetime.now():%H%M%S}"
    tmp = Path(tempfile.mkdtemp(prefix="facts_probe_")); mut = render(iso, tmp); shutil.rmtree(tmp)
    changed = {n: base[n] != mut[n] for n in base}; restored = render(con, DERIVED)
    return {"claim_id": claim_id, "consumers_changed": changed, "all_changed": all(changed.values()), "restored_equal": restored == base}


def check():
    con = J(CONTRACT); bad = []
    for c in con["claims"]:
        for k in ("claim_id", "verdict", "source_refs", "time_window", "sample_level", "aggregation_level", "scope", "missing_evidence", "closure", "display", "sha256"):
            if k not in c: bad.append(f"{c.get('claim_id')} 缺 {k}")
        if c["verdict"] not in SIX: bad.append(f"{c['claim_id']} verdict 非六枚举: {c['verdict']}")
        if sha(canon({k: v for k, v in c.items() if k != 'sha256'}).encode('utf-8')) != c["sha256"]: bad.append(f"{c['claim_id']} sha 不符 (被手改)")
    for n in ("portal_claims.json", "detail_cards.json", "qa_refs.json"):
        if J(DERIVED / n)["contract_sha256"] != con["contract_sha256"]: bad.append(f"{n} 内嵌 contract sha ≠ 当前契约")
    if con["contract_sha256"][:16] not in (DERIVED / "report_summary.md").read_text(encoding="utf-8"): bad.append("report_summary.md 内嵌 contract sha ≠ 当前契约")
    for n in ("portal_claims.json", "detail_cards.json", "report_summary.md", "qa_refs.json"):
        hits = BANNED.findall((DERIVED / n).read_text(encoding="utf-8"))
        if hits: bad.append(f"{n} 对外面孔含禁词/路径/邮箱/手机: {hits[:3]}")
    return bad


if __name__ == "__main__":
    mode = sys.argv[1] if len(sys.argv) > 1 else "check"
    if mode == "build": con = build(); print("契约", CONTRACT.relative_to(ROOT), con["n_claims"], con["by_verdict"], "sha", con["contract_sha256"][:16], "banned", con["banned_hits_total"]); print("消费者", render(con, DERIVED))
    elif mode == "render": print(render(J(CONTRACT), DERIVED))
    elif mode == "probe": print(json.dumps(probe(sys.argv[2]), ensure_ascii=False))
    else: b = check(); print("PASS" if not b else "FAIL", *b[:10], sep="\n  "); sys.exit(0 if not b else 2)