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

"""ON-0 分析前置锁定门 · 焊点1 (SOP §0.8; 2026-07-29 用户裁 "选 a 手动加一行")。

**用法 = 每个分析脚本第一行**::

    from src.sop.analysis_lock import assert_frozen
    LOCK = assert_frozen("changyi")     # 未冻结 / 未过自洽闸 / 找不到本场锁 → 立即 exit 2, 不出数

返回锁 dict, 调用方可直接取窗/口径 (免二次读, 也避免"锁一份、脚本里另写一份"的漂移)。

为什么是"手动加一行"而不是 import 时自动检查 (用户 2026-07-29 二选一裁决):
  显式可读 —— 看脚本头就知道这道门在; 代价 = **新脚本可能忘加**。
  该漏口用 sop_lint 断言在**已声明入口**上兜底 (见 ENTRYPOINTS), 未声明的新脚本仍靠纪律。

**头号设计坑 (本模块存在的主要理由)**: `configs/analysis_lock.yaml` 是**某一场**的锁 (现为花园口)。
若某场没有自己的 `analysis_lock_<farm>.yaml` 就回落到它, 校验会拿**别场的锁**判 PASS = **静默假过**,
恰是本门要防的病的加强版。故: 回落**必须**校验锁内 farm 字段与请求场一致, 不一致 = 拒绝, 不是警告。
"""
from __future__ import annotations

try:
    from app_common.app_common_guanlan.api import install_root as _install_root
except ImportError:                     # 理论不可达；包结构异常时回退到按位置上跳
    from pathlib import Path as _P
    def _install_root(_f): return _P(_f).resolve().parents[4]
from src import paths as P

import os
import sys
from pathlib import Path

_REPO = _install_root(__file__)

# sop_lint 兜底覆盖面 = 这些"当代分析入口"必须调 assert_frozen (漏加即断言 FAIL)。
# 刻意**不含** 1000+ 历史 per-场脚本 (CLAUDE.md R0 "不动骨架": 那批已跑完, 批量改写风险 >> 收益)。
ENTRYPOINTS = (
    "src/sop/farm_pipeline.py",
    "scripts/run_scenario.py",
)


# 焊点2 首付登记: farm → {path, sha, ts, strict}。assert_frozen 写, assert_unchanged 读。
# 进程内内存态 (一次分析 = 一个进程); 跨进程留痕走 outputs/<farm>/sop/on0_lock_trail.jsonl。
_RUNS: dict = {}


def _sha(path: Path) -> str:
    import hashlib
    return hashlib.sha256(Path(path).read_bytes()).hexdigest()


def _trail(root: Path, farm: str, rec: dict) -> None:
    """append-only 留痕 (§0.8 焊点2 "记入日志" 的落点; §9.5 报告级核对仍待实现, 本文件只覆盖 ON-0 这一片)。"""
    import json
    from datetime import datetime
    try:
        d = Path(root) / P.rel(P.sop(farm))
        d.mkdir(parents=True, exist_ok=True)
        rec = {"ts": datetime.now().isoformat(timespec="seconds"), **rec}
        with (d / "on0_lock_trail.jsonl").open("a", encoding="utf-8") as f:
            f.write(json.dumps(rec, ensure_ascii=False) + "\n")
    except Exception as e:                        # 留痕失败不该杀分析, 但必须响亮
        print(f"[ON-0] ⚠ 哈希留痕写失败 ({type(e).__name__}: {e}) — 闭环留痕缺这一条", file=sys.stderr)


def _norm(s) -> str:
    return "".join(str(s or "").lower().split())


def _farm_matches(lock_farm, farm: str) -> bool:
    """锁里的 farm 字段与请求场是否同一场。锁里常写中文全名 (如 '花园口一期(庄源+新风)'),
    请求侧是 slug (如 'huayuankou') → 双向子串 + 别名表。**判不出 = 不匹配** (宁拒勿假过)。"""
    lf, f = _norm(lock_farm), _norm(farm)
    if not lf or not f:
        return False
    if lf == f or f in lf or lf in f:
        return True
    try:                                     # 复用 BF-1 的 slug→中文别名表, 不另维护
        sys.path.insert(0, str(_REPO / "scripts"))
        from sop_check import FARM_ALIASES   # type: ignore
        return any(_norm(a) in lf for a in (FARM_ALIASES.get(farm) or ()))
    except Exception:
        return False


def find_lock(farm: str, root: Path | None = None):
    """→ (path, reason)。per 场锁优先; 默认锁**仅在 farm 字段对得上**时才认。"""
    root = Path(root or _REPO)
    per = P.config(f"analysis_lock_{farm}.yaml")
    if per.is_file():
        return per, "per 场锁"
    dflt = P.config("analysis_lock.yaml")
    if dflt.is_file():
        import yaml
        try:
            doc = yaml.safe_load(dflt.read_text(encoding="utf-8")) or {}
        except Exception:
            return None, "默认锁解析失败"
        if _farm_matches(doc.get("farm"), farm):
            return dflt, "默认锁 (farm 字段匹配)"
        return None, (f"仅有默认锁但它是别场的 (锁内 farm={doc.get('farm')!r} ≠ 请求 {farm!r}) "
                      f"— **拒绝跨场回落**, 否则等于拿别场的锁给本场放行")
    return None, "无任何锁文件"


def assert_frozen(farm: str, *, root=None, strict: bool = True) -> dict:
    """ON-0 焊点1 硬闸。返回锁 dict; 不过闸 → 打印响亮说明并 `sys.exit(2)`。

    strict=False → 只警告不退出 (仅供**探索性**脚本; 出数入报告的一律 strict)。
    可用环境变量 `ON0_SKIP=1` 临时跳过 —— 会响亮打印"已跳过"留痕 (调试用, 不是常规路径)。
    """
    root = Path(root or _REPO)
    bar = "=" * 72

    if os.environ.get("ON0_SKIP") == "1":
        print(f"[ON-0] ⚠⚠ ON0_SKIP=1 → **跳过前置锁定门** (farm={farm})。"
              f"仅限调试; 出数入报告禁用此路。", file=sys.stderr)
        return {}

    path, reason = find_lock(farm, root)
    if path is None:
        msg = (f"\n{bar}\n[ON-0] 🔴 分析前置锁定门未过: farm={farm} — {reason}\n"
               f"按 SOP §0.8: **任何分析出数前必先冻结《分析配置单》**, 未冻结不得出数入报告。\n"
               f"要做的事: 建 configs/analysis_lock_{farm}.yaml, 冻结六项 (L1 分析窗 / L2 额定装机 /\n"
               f"  L3 数据源权威表 / L4 指标口径字典 / L5 物理可达基准 / L6 可复现契约),\n"
               f"  并由**用户裁决**填 frozen + frozen_by (Claude 自冻不算, §0.8 A 闸)。\n"
               f"根因实证: 窗/口径/源/容量/基准未锁 → 同数据两次结果不一 "
               f"(利用小时 1330 vs 2389 / 改造前后 +0.3 vs +23.6)。\n{bar}")
        if strict:
            print(msg, file=sys.stderr)
            sys.exit(2)
        print(msg + "\n[ON-0] strict=False → 仅警告, 继续 (出数入报告禁用)", file=sys.stderr)
        return {}

    import yaml
    try:
        doc = yaml.safe_load(path.read_text(encoding="utf-8")) or {}
    except Exception as e:                   # 坏 YAML 必响亮 (实测 guojiadian 锁语法就是坏的, 一直没人发现)
        print(f"\n{bar}\n[ON-0] 🔴 锁文件 YAML 解析失败: {path.relative_to(root)}\n    {type(e).__name__}: {e}\n"
              f"锁读不出 = 等于没锁 (§0.8 未冻结不得出数)。修 YAML 后重跑。\n{bar}", file=sys.stderr)
        if strict:
            sys.exit(2)
        return {}

    sys.path.insert(0, str(root))
    from src.sop import schemas as S
    fails = list(S.validate_analysis_lock(doc))

    frozen, frozen_by = doc.get("frozen"), str(doc.get("frozen_by") or "")
    if not frozen:
        fails.append("frozen 空 (未冻结 = 不得出数, §0.8)")
    elif "用户裁决" not in frozen_by:
        fails.append(f"frozen 有值但 frozen_by={frozen_by!r} 不含『用户裁决』"
                     f" — Claude 自冻自锁不算权威 (§0.8 A 闸 frozen_by 门)")

    if fails:
        print(f"\n{bar}\n[ON-0] 🔴 锁自洽闸未过: {path.relative_to(root)} ({len(fails)} 条)", file=sys.stderr)
        for f in fails:
            print(f"    ✗ {f}", file=sys.stderr)
        print(f"修锁后重跑。{bar}", file=sys.stderr)
        if strict:
            sys.exit(2)
        return doc

    sha = _sha(path)
    _RUNS[farm] = {"path": str(path), "sha": sha, "strict": strict}
    _trail(root, farm, {"event": "start", "lock": str(path.relative_to(root)),
                        "sha256": sha, "frozen": str(frozen), "strict": strict})
    print(f"[ON-0] ✅ 前置锁定门过 · farm={farm} · {reason} {path.relative_to(root)} · "
          f"frozen={frozen} · by={frozen_by[:28]} · sha256:{sha[:16]}")
    return doc


def assert_unchanged(farm: str, *, root=None, strict: bool | None = None, stage: str = "出数前") -> bool:
    """ON-0 焊点2 = 哈希"结束再核"闭环 (SOP §0.8; 2026-07-29 用户点名收尾)。

    与焊点1 配对: `assert_frozen` 开跑时记锁的 sha256 ("首付"), 本函数在**产物落盘前**重算比对。
    不一致 = 锁在分析过程中被改 → **算出来的数已不对应声明的配置**, 落盘即造成
    "报告说按 A 口径算, 锁里写的是 B" 的静默错配。

    **放在写产物之前, 不是之后** —— 之后再发现, 错的产物已经在盘上了。

    诚实边界 (承 §9 威胁模型): 本闸防的是**中途改锁** (手滑 / 并发会话改同一锁 / 为让数好看而调锁);
    **不防**开跑前就把锁改好, 也不防同时伪造锁与结果 —— 一致性校验证「自洽」非「为真」。

    strict=None → 沿用 assert_frozen 当时的 strict (开跑 warn 的别在收尾突然硬杀)。
    返回 True=一致 / False=不一致或无法核 (strict 时不返回, 直接 exit 2)。
    """
    root = Path(root or _REPO)
    bar = "=" * 72
    rec = _RUNS.get(farm)
    eff_strict = rec["strict"] if (strict is None and rec) else bool(strict)

    if os.environ.get("ON0_SKIP") == "1":
        print(f"[ON-0] ⚠⚠ ON0_SKIP=1 → 跳过{stage}哈希再核 (farm={farm})", file=sys.stderr)
        return True

    if not rec:
        msg = (f"[ON-0] 🔴 {stage}哈希再核无从进行: farm={farm} 没有首付记录 "
               f"(assert_frozen 未调 ∨ 未过闸) — **闭环缺前半**")
        print(msg, file=sys.stderr)
        _trail(root, farm, {"event": "recheck", "stage": stage, "ok": False, "why": "无首付记录"})
        if eff_strict:
            sys.exit(2)
        return False

    p = Path(rec["path"])
    if not p.is_file():
        print(f"\n{bar}\n[ON-0] 🔴 {stage}哈希再核失败: 锁文件在分析过程中**消失** {p}\n{bar}", file=sys.stderr)
        _trail(root, farm, {"event": "recheck", "stage": stage, "ok": False, "why": "锁文件消失"})
        if eff_strict:
            sys.exit(2)
        return False

    now = _sha(p)
    ok = (now == rec["sha"])
    _trail(root, farm, {"event": "recheck", "stage": stage, "lock": str(p.relative_to(root)),
                        "sha256_start": rec["sha"], "sha256_now": now, "ok": ok})
    if ok:
        print(f"[ON-0] ✅ {stage}哈希再核一致 · farm={farm} · sha256:{now[:16]} (锁全程未变)")
        return True

    print(f"\n{bar}\n[ON-0] 🔴 {stage}哈希再核**不一致** — 锁在分析过程中被改!\n"
          f"    锁文件: {p.relative_to(root)}\n"
          f"    开跑时 sha256: {rec['sha'][:32]}…\n"
          f"    现在   sha256: {now[:32]}…\n"
          f"⇒ 本次算出的数**已不对应声明的配置**; 落盘会造成『报告说按 A 口径算、锁里是 B』的静默错配。\n"
          f"处置二选一: ①锁改是对的 → 用新锁**重跑**分析 ②锁被误改 → 还原锁再跑。\n"
          f"(留痕已入 {P.rel(P.sop(farm))}/on0_lock_trail.jsonl)\n{bar}", file=sys.stderr)
    if eff_strict:
        sys.exit(2)
    return False