guanlan-src/scripts/guanlan_cloud_face.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
r"""观澜云端版 · 脱敏面孔 (部署包 CLAUDE_CODE_部署提示词.md §安全边界 / §C 内容与数据审核; 交接 2026-09-06).

云端包只允许"经批准的脱敏摘要和资料索引", 且共享同一份事实契约 (端口与交接重构任务 §内容与数据架构).
本脚本从 **契约 + 四消费者** (outputs/rudong/guanlan/{facts_contract_v0.json, derived/*}) 派生云端可用内容, 落 outputs/rudong/guanlan/cloud/:
  portal_claims.json · detail_cards.json · qa_refs.json · report_summary.md — 与内部消费者同结构, 文本经 src/windscada/deid_public.scrub_or_die (场名/OEM/业主/供应商/文档名/内部系统名), 内嵌 contract_sha256 不变 (仍指向同一契约, 可追溯)
  claims_public.json — 全 119 条的对外投影 (claim_id / 六枚举 / 脱敏标题 / 时间窗 / 样本层级 / 聚合层级 / 适用范围 / 归宿 / 缺失证据 / 原条 sha256); 云端问答引用与卡片只认它
  cloud_manifest.json + scan_report.md — 扫描规则 / 逐文件命中 (脱敏前→后) / 处置 / 每个产物 SHA-256 / 契约 sha / git head / 未脱敏项 (机组号) 的显式声明
铁律 (deid_public 同款): "替换过了"不算数, 只有回扫零命中算; 回扫表 = deid_public.LEAK_RE + 本文件 EXTRA_RE (绝对路径 / 邮箱 / 手机 / 身份证 / 精确坐标 / 密钥格式), 任一命中 → 不出件 (exit 2).
机组号 (WTG\d\d) 默认**保留** (中文版是给业主本人看的; 是否代号化 = 用户裁决, --unitize 走 windscada_cloud_pack.unitize), manifest 里 unitized 字段显式声明, 不静默.
用法: build [--unitize] | check
"""
import argparse, datetime as dt, hashlib, json, re, subprocess, sys
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / "src")); sys.path.insert(0, str(ROOT / "scripts"))
from windscada import deid_public as DP  # noqa: E402  ★导入失败 = 拒绝运行 (闸不可用时放行比没闸更坏)

GUANLAN = ROOT / "outputs/rudong/guanlan"; CONTRACT = GUANLAN / "facts_contract_v0.json"; DERIVED = GUANLAN / "derived"; CLOUD = GUANLAN / "cloud"
CONSUMERS = ("portal_claims.json", "detail_cards.json", "qa_refs.json", "report_summary.md")
sha = lambda b: hashlib.sha256(b).hexdigest(); J = lambda p: json.loads(Path(p).read_text(encoding="utf-8"))
# 部署提示词 §C 点名的扫描项, deid_public.LEAK_RE 没覆盖的部分. 与替换表分开写 (互为独立发现).
EXTRA_RE = [
    ("本机绝对路径", re.compile(r"/Users/[A-Za-z0-9_.\-]+|/Volumes/[A-Za-z0-9_\-]+|/home/[A-Za-z0-9_.\-]+|[A-Za-z]:\\\\[^\s\"']{1,40}")),
    ("邮箱", re.compile(r"[A-Za-z0-9._%+\-]+@[A-Za-z0-9.\-]+\.[A-Za-z]{2,}")),
    ("手机号", re.compile(r"(?<![0-9A-Fa-f.])1[3-9]\d{9}(?![0-9A-Fa-f.])")),   # 2026-09-07: π 的小数 14159265359 被当手机号 → 前后排除小数点  # 十六进制 sha 里的纯数字段不算 (边界排除 hex 字母)
    ("身份证号", re.compile(r"(?<![0-9A-Fa-f.])[1-8]\d{5}(?:19|20)\d{2}(?:0[1-9]|1[0-2])(?:0[1-9]|[12]\d|3[01])\d{3}[\dXx](?![0-9A-Fa-f])")),   # 2026-09-07: 小数尾巴 0.010668214654282766 被当身份证 → 加 "." 前视 + 地区码首位 1-8 + 出生年月日合法
    # 只认大写方位字母且后面不能再接字母/数字: 小写 e/n/s/w 会把 SVG/CSS 里的 12.3456e-5 (科学计数) 当成东经 (单文件实测假阳 17420 处)
    ("精确坐标", re.compile(r"(?<![\d.])\d{2,3}\.\d{4,}\s*°?\s*[NEWS](?![A-Za-z0-9])"                       # 32.1234N / 121.5678E
                          r"|(?<![\d.,\-])(?:(?:1[89]|[2-5]\d)\.\d{4,},\s*(?:7[3-9]|[89]\d|1[0-3]\d)\.\d{4,}"      # 纬,经 (中国量级)
                          r"|(?:7[3-9]|[89]\d|1[0-3]\d)\.\d{4,},\s*(?:1[89]|[2-5]\d)\.\d{4,})(?![\d.]|\s*[\d,\-LMCZlmcz])")),  # 经,纬; 后面紧跟另一对/路径命令 = SVG path 不算 (单文件实测假阳 17420→12→0)
    ("密钥格式", re.compile(r"sk-[A-Za-z0-9]{16,}|AKIA[0-9A-Z]{16}|(?i:api[_-]?key|secret|token)\s*[:=]\s*['\"]?[A-Za-z0-9\-_]{16,}")),
]
INFO_RE = [("原始机组号", re.compile(r"WTG\d{2}")), ("显式台数", re.compile(r"(?<![\d.])38\s*台"))]  # 只报数不拦: 是否代号化 = 用户裁决


def scan(text):
    """回扫 → {类别: (命中数, 样例)}; 空 = 干净. 规则 = deid_public.LEAK_RE ∪ EXTRA_RE."""
    out = {n: (c, ex) for n, c, ex in DP.audit(text)}
    for n, rx in EXTRA_RE:
        m = rx.findall(text)
        if m: out[n] = (len(m), sorted(set(m))[:4])
    return out


def info(text): return {n: len(rx.findall(text)) for n, rx in INFO_RE if rx.findall(text)}


def rules_table():
    return [{"name": n, "pattern": rx.pattern, "source": "src/windscada/deid_public.LEAK_RE", "action": "block"} for n, rx in DP.LEAK_RE] + \
           [{"name": n, "pattern": rx.pattern, "source": "scripts/guanlan_cloud_face.EXTRA_RE", "action": "block"} for n, rx in EXTRA_RE] + \
           [{"name": n, "pattern": rx.pattern, "source": "scripts/guanlan_cloud_face.INFO_RE", "action": "report-only (用户裁)"} for n, rx in INFO_RE]


def _scrub(text, unitize):
    if unitize:
        from windscada_cloud_pack import unitize as _u
        text = _u(text)
    return DP.scrub(text)


def public_claims(con, unitize):
    """契约 → 对外投影. 每条带原条 sha256 (指回内部契约, 不带内部路径), 文本字段逐一脱敏."""
    s = lambda v: _scrub(v, unitize) if isinstance(v, str) else v
    rows = []
    for c in con["claims"]:
        rows.append({"claim_id": c["claim_id"], "verdict": c["verdict"], "title_public": s(c["display"]["title_zh_public"]),
                     "time_window": {"window_days": c["time_window"].get("window_days"), "evidence_granularity": s(c["time_window"].get("evidence_granularity"))},
                     "sample_level": {"n_machines_covered": c["sample_level"].get("n_machines_covered"), "n_machines_named": c["sample_level"].get("n_machines_named"), "claim_class": s(c["sample_level"].get("claim_class"))},
                     "aggregation_level": c["aggregation_level"], "scope": {"systems": c["scope"].get("systems"), "module": c["scope"].get("module")},
                     "closure_status": c["closure"]["status"], "missing_evidence": s(c.get("missing_evidence")) if isinstance(c.get("missing_evidence"), str) else None,
                     "falsifiability": s(c.get("falsifiability")) if isinstance(c.get("falsifiability"), str) else None,
                     "source": {"kind": "internal facts contract", "contract_sha256": con["contract_sha256"], "claim_sha256": c["sha256"], "ref_sha16": c["source_refs"][0]["sha16"]}, "probe": c.get("probe")})
    return {"schema": "guanlan-cloud-claims/v0", "face": "public (脱敏面孔; 由 scripts/guanlan_cloud_face.py 自契约生成, 禁手改)", "contract_sha256": con["contract_sha256"], "contract_version": con["version"],
            "n_claims": len(rows), "by_verdict": con["by_verdict"], "claims": rows}


def build(unitize=False):
    con = J(CONTRACT); CLOUD.mkdir(parents=True, exist_ok=True)
    files, before, after, infos = {}, {}, {}, {}
    for n in CONSUMERS:
        raw = (DERIVED / n).read_text(encoding="utf-8"); before[n] = scan(raw)
        files[n] = _scrub(raw, unitize)
    files["claims_public.json"] = json.dumps(public_claims(con, unitize), ensure_ascii=False, indent=1)
    before["claims_public.json"] = scan(json.dumps({k: v for k, v in con.items() if k != "claims"}, ensure_ascii=False) + "".join(c["display"]["title_zh_public"] for c in con["claims"]))
    for n, t in files.items():
        after[n] = scan(t); infos[n] = info(t)
        for cn in ("portal_claims.json", "detail_cards.json", "qa_refs.json", "claims_public.json"):
            if n == cn: assert json.loads(t)["contract_sha256"] == con["contract_sha256"], n  # 脱敏不得动契约 sha (可追溯性)
    leaks = {n: a for n, a in after.items() if a}
    git = subprocess.run(["git", "-C", str(ROOT), "rev-parse", "--short", "HEAD"], capture_output=True, text=True).stdout.strip()
    man = {"schema": "guanlan-cloud-face/v0", "mode": "cloud", "built": dt.datetime.now().isoformat(timespec="minutes"), "git_head": git, "unitized": bool(unitize),
           "contract": {"path": "outputs/rudong/guanlan/facts_contract_v0.json", "contract_sha256": con["contract_sha256"], "version": con["version"], "n_claims": con["n_claims"], "by_verdict": con["by_verdict"]},
           "inputs": {n: sha((DERIVED / n).read_bytes()) for n in CONSUMERS},
           "files": {n: {"sha256": sha(t.encode("utf-8")), "bytes": len(t.encode("utf-8"))} for n, t in files.items()},
           "scan": {"rules": rules_table(), "hits_before": {n: {k: v[0] for k, v in h.items()} for n, h in before.items()}, "hits_after": {n: {k: v[0] for k, v in h.items()} for n, h in after.items()},
                    "report_only": infos, "verdict": "FAIL" if leaks else "PASS"},
           "not_included": ["原始 SCADA / CMS 波形", "findings.json 原件与内部路径", "业主名 / 精确场址", "个人信息", "密钥"],
           "decisions_pending": ["机组号是否代号化 (--unitize; 现 unitized=%s)" % bool(unitize), "中文版上云路线: 本脱敏面孔 vs 2026-09-01 裁决的不可猜路径+禁索引内部件 (windscada_ecs_deploy.py --internal-zh)"]}
    rep = ["# 云端脱敏面孔 · 扫描报告 (自动生成, 禁手改)", "", f"契约 {con['contract_sha256'][:16]} · git {git} · unitized={bool(unitize)} · 判定 **{man['scan']['verdict']}**", "",
           "## 规则", "", "| 名称 | 来源 | 处置 |", "|---|---|---|"] + [f"| {r['name']} | {r['source']} | {r['action']} |" for r in rules_table()] + ["", "## 逐文件命中 (脱敏前 → 后)", "", "| 文件 | 脱敏前 | 脱敏后 | 只报不拦 |", "|---|---|---|---|"]
    for n in files:
        b = ", ".join(f"{k}×{v[0]}" for k, v in before[n].items()) or "0"; a = ", ".join(f"{k}×{v[0]}" for k, v in after[n].items()) or "0"; i = ", ".join(f"{k}×{v}" for k, v in infos[n].items()) or "—"
        rep.append(f"| {n} | {b} | {a} | {i} |")
    rep += ["", "## 处置", "", "- 脱敏前命中全部由 deid_public.scrub 替换 (全局一致重命名: 如东→观澜 / 西门子→the OEM / 上海电气→OEM / 业主集团→[owner] / 文档名→[technical document] / findings.json→findings store)。",
            "- 脱敏后任一 block 类命中 → 本脚本 exit 2 不出件; 本次 " + ("**有残留, 未出件**" if leaks else "零命中, 出件"), "- 只报不拦项 (机组号 / 台数) 是否代号化 = 用户裁决 (TASKS.md 观澜上云演示 · 待用户裁决 §云端版)。", "",
            "## 产物 SHA-256", ""] + [f"- {n}: {v['sha256']} ({v['bytes']} B)" for n, v in man["files"].items()]
    if leaks:
        print("✗ 脱敏后仍有残留, 拒绝出件:", json.dumps(leaks, ensure_ascii=False)); return 2
    for n, t in files.items(): (CLOUD / n).write_text(t, encoding="utf-8")
    (CLOUD / "cloud_manifest.json").write_text(json.dumps(man, ensure_ascii=False, indent=1), encoding="utf-8")
    (CLOUD / "scan_report.md").write_text("\n".join(rep) + "\n", encoding="utf-8")
    print("云端面孔", CLOUD.relative_to(ROOT), "文件", len(files), "契约", con["contract_sha256"][:16], "回扫", man["scan"]["verdict"], "只报", infos.get("detail_cards.json"))
    return 0


def check():
    """云端目录回扫 + manifest sha 对账 + 契约 sha 一致. 任一坏 → 列出; 空 = PASS."""
    bad = []
    if not (CLOUD / "cloud_manifest.json").is_file(): return ["cloud_manifest.json 不存在 (先 build)"]
    man = J(CLOUD / "cloud_manifest.json"); con = J(CONTRACT)
    if man["contract"]["contract_sha256"] != con["contract_sha256"]: bad.append("manifest 契约 sha ≠ 当前契约 (契约已变, 须重 build)")
    for n, v in man["files"].items():
        p = CLOUD / n
        if not p.is_file(): bad.append(f"{n} 缺失"); continue
        t = p.read_text(encoding="utf-8")
        if sha(t.encode("utf-8")) != v["sha256"]: bad.append(f"{n} sha ≠ manifest (被手改)")
        for k, (c, ex) in scan(t).items(): bad.append(f"{n} 残留 {k}×{c} {ex}")
        if n.endswith(".json") and n != "cloud_manifest.json" and json.loads(t).get("contract_sha256") != con["contract_sha256"]: bad.append(f"{n} 内嵌契约 sha ≠ 当前契约")
    return bad


if __name__ == "__main__":
    # 控制台可能是 GBK(中文 Windows 代码页 936): 正文里的 ✔ ✗ ✅ ⚠ 这类字符编不出来会抛
    # UnicodeEncodeError, 脚本干成了事却以退出码 1 结束(同类坑见 src/console.py)。降级为 '?' 而不是崩;
    # 不用 import 是为了兼顾 python -m 与直接当脚本跑两种启动方式。
    import sys as _sys
    for _s in (_sys.stdout, _sys.stderr):
        try: _s.reconfigure(errors='replace')
        except Exception: pass
    ap = argparse.ArgumentParser(); ap.add_argument("mode", nargs="?", default="check", choices=["build", "check"]); ap.add_argument("--unitize", action="store_true", help="机组号代号化 (用户裁后再开)")
    a = ap.parse_args()
    if a.mode == "build": sys.exit(build(a.unitize))
    b = check(); print("PASS" if not b else "FAIL", *b[:12], sep="\n  "); sys.exit(0 if not b else 2)