guanlan-src/app_ontology/app_ontology_guanlan/sop/schema_guard.py

"""schema_guard.py — A1 schema铁律 + A2 落盘自证 的**机器位** (削减读取/落盘侧幻觉).

背景 (memory `tool-hallucination-git-authoritative`): 工具不稳时"读文件成功"的整段输出可能是
幻觉 (海勒斯 2026-07-15: 假报 turbine_10min.parquet shape=60480×65+中文列, 真只有 turbine_sec.parquet
英文列)。本模块把"读前必核 + 落盘自证"从纪律变成**函数**——分析码调 verified_read 而非裸 pd.read_*,
schema铁律 by construction; 落盘后调 attest 拿 sha256 自证。

- A1 verified_read: 存在断言(响亮列出真目录) → 真读拿真列名/shape → require_cols 逐列 assert。
- A2 attest: 落盘/引用文件返回 {sha256, size, rows} 内容寻址自证 (对齐 CL-2 fingerprint 的 sha256, 非md5)。
- 权威源只认: Path.exists()退出码 + 真读到的 df.columns + sha256内容哈希 + git ls-files (不信"读成功"回执)。

用法:
    from src.sop.schema_guard import verified_read, assert_cols, attest
    df, man = verified_read("outputs/x/cleaned/t.parquet", require_cols=["active_power","风速"])
    # ... 落盘后 ...
    print(attest("outputs/x/sop/findings.json"))   # {'exists':True,'sha256':'...','size':...}
"""
from __future__ import annotations

import hashlib
import pathlib
import subprocess

__all__ = ["sha256", "git_tracked", "attest", "verified_read", "assert_cols", "SchemaGuardError",
           "three_source_aligned", "three_source_lint", "THREE_SOURCE_NOTE",
           "INDEPENDENCE_LEVELS", "independence_lint"]

# B5 温度候选独立性分级 (正本定义: outputs/_meta/温度judgment外部验证率_元审计_2026-07-15.md §独立性分级)
# 关键戒: 同组多手段 ≠ 第三方独立 (memory scada-residual / anchor-validation)。
INDEPENDENCE_LEVELS = {
    "L0": "纯温度自证 — 只由该部件温度NBM残差判定, 无任何非温度信号",
    "L1": "同域多通道互证 (swap配对/DE−油温差/六相绕组/多通道同热) — 仍温度, 抓接线/传感器有效, 但非外部真值(冷却系统级偏差会多通道齐偏)",
    "L2": "跨域独立信号 (振动/CMS/油样金属屑/事件故障日志/现场取振开箱) — 唯一破循环的外部锚; L2命中还须超几何期望+组间检验(base-rate)",
}

# A3 三源对齐: 任一 claim 须 (源文件真读 × 独立重跑 × 内容哈希) 三对齐, 单源 → 应 INSUFFICIENT。
# 承载约定 (SOP字段): finding.provenance{src_doc} (源) + finding.reproduced[{claim,cmd}] (独立重跑)
#   + finding.basis.attest{sha256,path} (内容哈希, 用 attest() 生成)。前两腿已在 schemas.validate_review;
#   本模块补第三腿(哈希)并提供 report-ship 交付前 lint (不接 always-on sop_check, 防炸存量场)。
THREE_SOURCE_NOTE = "A3: claim 须 源真读×独立重跑×内容哈希 三对齐; 承载 = provenance.src_doc + reproduced[] + basis.attest.sha256"


class SchemaGuardError(AssertionError):
    """读前核 / 列断言失败 = 响亮 (不静默继续用幻觉 schema)。"""


def sha256(path: str | pathlib.Path, buf: int = 1 << 20) -> str:
    """内容哈希 (对齐 contract_gate._sha256 的 CL-2 指纹约定)。"""
    h = hashlib.sha256()
    with open(path, "rb") as f:
        while chunk := f.read(buf):
            h.update(chunk)
    return h.hexdigest()


def git_tracked(path: str | pathlib.Path) -> bool:
    """git ls-files 交叉核: 文件是否已被 git 追踪 (A1 '真读到' 的第二道)。"""
    p = pathlib.Path(path)
    try:
        r = subprocess.run(["git", "ls-files", "--error-unmatch", str(p)],
                           capture_output=True, text=True, cwd=p.resolve().parent)
        return r.returncode == 0
    except Exception:
        return False


def attest(path: str | pathlib.Path, rows: bool = True) -> dict:
    """A2 落盘/引用自证: 返回内容寻址证据。不存在 = 响亮 (exists=False, 不假装)。"""
    p = pathlib.Path(path)
    if not p.exists():
        return {"path": str(p), "exists": False, "note": "文件不存在 — 任何'已落盘'声称为幻觉"}
    out = {"path": str(p), "exists": True, "sha256": sha256(p),
           "size": p.stat().st_size, "tracked": git_tracked(p)}
    if rows and p.suffix.lower() in (".parquet", ".csv"):
        try:
            import pandas as pd
            d = pd.read_parquet(p) if p.suffix.lower() == ".parquet" else pd.read_csv(p, nrows=None)
            out["rows"] = int(len(d))
            out["cols"] = int(d.shape[1])
        except Exception as e:
            out["rows_error"] = str(e)[:120]
    return out


def assert_cols(df, cols: list[str], where: str = "") -> None:
    """A1 列断言: 逐列 assert 在 df.columns; 缺 = 响亮列出**真有的**列 (禁用幻觉列名)。"""
    have = list(df.columns)
    missing = [c for c in cols if c not in have]
    if missing:
        raise SchemaGuardError(
            f"schema铁律 [{where or '?'}]: 需要列缺失 {missing}。"
            f" 真实列 ({len(have)}): {have[:40]}{' …' if len(have) > 40 else ''}")


def verified_read(path: str | pathlib.Path, require_cols: list[str] | None = None,
                  columns: list[str] | None = None, kind: str = "auto") -> tuple:
    """A1 读前必核: 存在断言 → 真读 → require_cols 逐列 assert。返回 (df, manifest)。

    幻觉防线: 文件不存在 → 响亮 FileNotFoundError + 列出**真实目录内容** (海勒斯型:
    你以为读 turbine_10min, 真目录只有 turbine_sec → 立刻现形, 不会假读出 schema)。
    columns: 只载这些列 (效率); 断言先于载数据 (parquet 读真 schema 名核, 子集读不失断言)。
    """
    p = pathlib.Path(path)
    if not p.exists():
        sibs = sorted(x.name for x in p.parent.iterdir()) if p.parent.exists() else ["<父目录也不存在>"]
        raise FileNotFoundError(
            f"schema铁律: {p} 不存在。真实同目录内容: {sibs}  (禁信'读成功'—先ls核真名)")
    import pandas as pd
    k = kind
    if k == "auto":
        k = "parquet" if p.suffix.lower() == ".parquet" else "csv"
    need = list(dict.fromkeys((require_cols or []) + (columns or [])))
    if k == "parquet":
        import pyarrow.parquet as _pq
        real_names = list(_pq.ParquetFile(str(p)).schema.names)   # 真 schema, 不载数据
        if need:                                                  # 断言先于载 (子集读也核)
            miss = [c for c in need if c not in real_names]
            if miss:
                raise SchemaGuardError(f"schema铁律 [{p.name}]: 需要列缺失 {miss}。"
                                       f" 真实列 ({len(real_names)}): {real_names[:40]}")
        df = pd.read_parquet(p, columns=columns) if columns else pd.read_parquet(p)
        ncols_total = len(real_names)
    else:
        df = pd.read_csv(p, usecols=columns) if columns else pd.read_csv(p)
        ncols_total = df.shape[1]
        if require_cols:
            assert_cols(df, require_cols, where=p.name)
    man = {"path": str(p), "sha256": sha256(p), "rows": int(len(df)),
           "cols": int(ncols_total), "cols_real": list(df.columns), "tracked": git_tracked(p)}
    return df, man


def three_source_aligned(finding: dict) -> bool:
    """A3: (源真读 × 独立重跑 × 内容哈希) 三对齐? 单源/缺腿 → False (该 claim 应 INSUFFICIENT)。"""
    prov = finding.get("provenance") or {}
    if isinstance(prov, list):  # provenance 可为条目列表 (schemas.validate_review 口径)
        prov = prov[0] if prov else {}
    basis = finding.get("basis") or {}
    has_source = bool(prov.get("src_doc") or prov.get("source_read"))
    rep = finding.get("reproduced")
    has_rerun = bool(rep and all(isinstance(r, dict) and r.get("claim") and r.get("cmd") for r in rep)) \
        or bool(prov.get("independent_rerun"))
    has_hash = bool((basis.get("attest") or {}).get("sha256") or prov.get("content_hash"))
    return has_source and has_rerun and has_hash


def three_source_lint(findings, high_tiers=("定论", "准定论")) -> list[dict]:
    """A3 交付前 lint: 高阶结论(定论/准定论族)但三源未对齐 → flag(应降 INSUFFICIENT)。
    report-ship §6 用; 返回 [{id, verdict, missing}]。空 = 全部三源对齐或全低阶。"""
    out = []
    for it in findings or []:
        v = str(it.get("verdict", ""))
        if any(t in v for t in high_tiers) and not three_source_aligned(it):
            prov = it.get("provenance") or {}
            prov = (prov[0] if prov else {}) if isinstance(prov, list) else prov
            basis = it.get("basis") or {}
            missing = []
            if not (prov.get("src_doc") or prov.get("source_read")):
                missing.append("源真读(provenance.src_doc)")
            rep = it.get("reproduced")
            if not (rep and all(isinstance(r, dict) and r.get("claim") and r.get("cmd") for r in rep)):
                missing.append("独立重跑(reproduced[claim,cmd])")
            if not ((basis.get("attest") or {}).get("sha256") or prov.get("content_hash")):
                missing.append("内容哈希(basis.attest.sha256)")
            out.append({"id": it.get("id"), "verdict": v, "missing": missing,
                        "issue": "A3 三源未对齐 → 应降 INSUFFICIENT"})
    return out


def independence_lint(findings, require_ids=None, high_tiers=("定论", "准定论")) -> list[dict]:
    """B5 温度候选独立性 lint (报告/交付前调)。flag 三类:
      ① independence_level 非法 (∉ L0/L1/L2);
      ② L0/L1(未跨域外部验证) 但结论高阶(定论/准定论族) 且未显式标 external_unvalidated → 须标或降级;
      ③ require_ids 指定的温度域候选缺 independence_level (强制带分级)。
    返回 [{id, issue, ...}]; 空 = 全合规。"""
    out = []
    req = set(require_ids or [])
    for it in findings or []:
        fid = it.get("id")
        lvl = it.get("independence_level")
        v = str(it.get("verdict", ""))
        if fid in req and not lvl:
            out.append({"id": fid, "issue": "温度域候选缺 independence_level (须标 L0/L1/L2)"})
            continue
        if lvl is not None and lvl not in INDEPENDENCE_LEVELS:
            out.append({"id": fid, "level": lvl,
                        "issue": f"independence_level '{lvl}' ∉ {list(INDEPENDENCE_LEVELS)}"})
            continue
        if lvl in ("L0", "L1") and any(t in v for t in high_tiers) and not it.get("external_unvalidated"):
            out.append({"id": fid, "level": lvl, "verdict": v,
                        "issue": f"{lvl}(未跨域外部验证) 但结论 {v} → 须显式标 external_unvalidated=True "
                                 f"或降级 (B5: 同域多通道≠外部真值, L2才破循环)"})
    return out


if __name__ == "__main__":
    import json
    import sys
    if len(sys.argv) > 1:
        print(json.dumps(attest(sys.argv[1]), ensure_ascii=False, indent=1))