guanlan-src/scripts/ontology_p2_verify.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""P2 验收器: dsh 输出的答案 id 提取 → 本体机器对拍 (验收=句句可溯 + 判级词与对象一致, 非人读).
用法: .venv/bin/python scripts/ontology_p2_verify.py <dsh输出文件>"""
import json, pathlib, re, sys

ROOT = pathlib.Path(__file__).resolve().parent.parent
OBJ = json.loads((ROOT / 'outputs/rudong/ontology/objects.json').read_text())
# 事实契约 id (claim/<claim_id>): 离线问答接契约后 (2026-09-06) 答案会引用它, 对象不在本体而在契约里.
_CP = ROOT / 'outputs/rudong/guanlan/facts_contract_v0.json'
CLAIMS = {c['claim_id']: c for c in json.loads(_CP.read_text(encoding='utf-8')).get('claims', [])} if _CP.is_file() else {}

def _unit_aliases():
    try:
        import re as _re
        from src.windscada.deid import unit_map
        ids = sorted(set(_re.findall(r'WTG(\d{2})', '\n'.join(OBJ.keys()))))
        return {alias: 'WTG' + raw for raw, alias in unit_map(ids).items()}
    except Exception:
        return {}


UNIT_ALIAS = _unit_aliases()


def _canonical_id(oid):
    out = oid
    for alias, raw in sorted(UNIT_ALIAS.items(), key=lambda kv: -len(kv[0])):
        out = re.sub(r'(?<![A-Za-z0-9])' + re.escape(alias) + r'(?![A-Za-z0-9])', raw, out)
    out = out.replace('/site/', '/rudong/')
    return out


LEVEL_EN = {
    '优秀': ('Excellent',),
    '良好': ('Good',),
    '报警': ('Alarm',),
    '危险': ('Critical',),
    '不可判': ('Not evaluable', 'Indeterminate', 'Undeterminable'),
    '预警': ('Alert', 'Warning', 'Early warning'),
    '定论': ('Established', 'Confirmed'),
    '候选': ('Candidate',),
    '观察': ('Observation', 'Watch'),
    '监视': ('Watch', 'Monitoring'),
    '销案': ('Closed',),
    '参考': ('Reference',),
    '正样本': ('Healthy reference', 'Positive sample'),
    '仲裁输入': ('Arbitration input',),
}


def _level_tokens(level, lang):
    if lang != 'en':
        return (level,)
    vals = []
    for zh, ens in LEVEL_EN.items():
        if zh in (level or ''):
            vals.extend(ens)
    return tuple(dict.fromkeys(vals)) or ((level,) if level else ())


def verify(text, lang='zh'):
    ids = re.findall(r'\[\s*(?:对象id|对象|来源|出处|id)?\s*[:：]?\s*([a-z]+/[A-Za-z0-9_/.\-]+)\s*\]', text or '')
    ids = [i for i in ids if '/' in i]
    seen = list(dict.fromkeys(ids))
    rows, n_ok = [], 0
    canonical = {}
    for i in seen:
        ci = _canonical_id(i)
        canonical[i] = ci
        if i.startswith('claim/'):
            c = CLAIMS.get(i[6:])
            if not c:
                rows.append((i, 'MISSING', '契约里无此结论编号 — 模型编造 id'))
            else:
                ok = c['verdict'] in (text or '')
                rows.append((i, 'OK' if ok else 'LEVEL_MISMATCH', f"契约裁决='{c['verdict']}' {'出现在答案中' if ok else '未在答案出现 — 判级词疑被改写'}"))
                n_ok += ok
            continue
        if ci not in OBJ:
            rows.append((i, 'MISSING', '对象不存在 — 模型编造 id'))
            continue
        o = OBJ[ci]
        lv = o.get('props', {}).get('level')
        if o['type'] == 'Verdict' and lv:
            toks = _level_tokens(lv, lang)
            ok = (not toks) or any(t in (text or '') for t in toks)
            rows.append((i, 'OK' if ok else 'LEVEL_MISMATCH',
                         f"对象判级='{lv}' {'有对应判级词出现在答案中' if ok else '未在答案出现 — 判级词疑被改写'}"))
            n_ok += ok
        else:
            rows.append((i, 'OK', f"{o['type']} 存在"))
            n_ok += 1
    verdict_cited = any(OBJ.get(canonical.get(i, i), {}).get('type') == 'Verdict' or (i.startswith('claim/') and i[6:] in CLAIMS) for i in seen)
    gate = (('claim_check' in (text or '') and '通过' in (text or '')) or
            ('grounding gate' in (text or '').lower() and 'passed' in (text or '').lower()))
    ok_all = bool(seen) and all(r[1] == 'OK' for r in rows) and verdict_cited
    return dict(引用id数=len(seen), 全通过=ok_all, 引用了判级对象=verdict_cited,
                接地闸出现=gate, 明细=rows)


if __name__ == '__main__':
    p = pathlib.Path(sys.argv[1])
    lang = sys.argv[2] if len(sys.argv) > 2 else 'zh'
    r = verify(p.read_text(), lang=lang)
    print(json.dumps(r, ensure_ascii=False, indent=1, default=str))
    sys.exit(0 if r['全通过'] else 2)