guanlan-src/app_ontology/app_ontology_guanlan/ontology/audit.py

# -*- coding: utf-8 -*-
"""本体机器审计器 (2026-08-28 用户令: ontology 深度优化和审核).

六项: ①悬空引用 ②孤儿对象 ③链接密度(WorkOrder.resolves/OilSample→部件) ④字段覆盖
⑤判级对拍(抽样 vs L1 实时) ⑥facet 分布。输出问题清单, 不静默PASS。
跑法: .venv/bin/python -m src.ontology.audit"""
from src import paths as P
import json, pathlib, random
from collections import Counter

ONT = P.objects_json()


def audit(sample_turbines=3, seed=7):
    d = json.loads(ONT.read_text(encoding='utf-8'))
    issues, stats = [], {}

    # ① 悬空引用
    dangling = [(o['id'], rel, x) for o in d.values()
                for rel, ids in o.get('links', {}).items() for x in ids if x not in d]
    stats['悬空引用'] = len(dangling)
    if dangling:
        issues.append(f"①悬空引用 {len(dangling)} 条 (links 指向不存在 id), 样例: {dangling[:5]}")

    # ② 孤儿对象 (无出链且无入链; Turbine/元数据类豁免)
    linked = set()
    for o in d.values():
        for ids in o.get('links', {}).values():
            linked.update(ids)
    exempt = {'Turbine', 'MaintenanceWindow'}
    orphans = [o['id'] for o in d.values()
               if o['type'] not in exempt and not o.get('links') and o['id'] not in linked
               and not o['id'].startswith(('meta/', 'decision/user', 'decision/pending'))]
    stats['孤儿对象'] = len(orphans)
    if orphans:
        issues.append(f"②孤儿对象 {len(orphans)} 个 (无链接, 查询游走不可达), 类型分布: "
                      f"{dict(Counter(d[i]['type'] for i in orphans))}, 样例: {orphans[:5]}")

    # ③ 链接密度
    wo = [o for o in d.values() if o['type'] == 'WorkOrder']
    wo_resolve = sum(1 for o in wo if o.get('links', {}).get('resolves'))
    stats['工单resolves率'] = f'{wo_resolve}/{len(wo)}'
    if wo_resolve == 0:
        issues.append(f"③工单 {len(wo)} 条零 resolves 链接 — 换件工单与机制/部件未连, '检修收益'能力断链")
    oil = [o for o in d.values() if o['type'] == 'OilSample']
    oil_comp = sum(1 for o in oil if any('component/' in x for x in o.get('links', {}).get('about', [])))
    stats['油样→部件链接率'] = f'{oil_comp}/{len(oil)}'
    if oil_comp == 0:
        issues.append(f"③油样 {len(oil)} 份只挂机组未挂部件 — 部件履历(29#主轴后序列类)本体不可查")

    # ④ 字段覆盖
    ev = [o for o in d.values() if o['type'] == 'Evidence']
    cc = sum(1 for o in ev if o['props'].get('crosscheck'))
    stats['Evidence.crosscheck覆盖'] = f'{cc}/{len(ev)}'
    if cc / max(len(ev), 1) < 0.2:
        issues.append(f"④Evidence crosscheck(三分判读)覆盖仅 {cc}/{len(ev)} — 转录证据未定支持/排除/不可判")
    vd = [o for o in d.values() if o['type'] == 'Verdict']
    vw = sum(1 for o in vd if o['props'].get('claim_window'))
    stats['Verdict.claim_window覆盖'] = f'{vw}/{len(vd)}'
    if vw < len(vd):
        issues.append(f"④Verdict claim_window 覆盖 {vw}/{len(vd)}")

    # ⑤ 判级对拍 (抽样台 vs L1 实时)
    import sys
    sys.path.insert(0, '.')
    from src.windscada import taxonomy
    mx = taxonomy.system_matrix()
    rng = random.Random(seed)
    ts = rng.sample(sorted(mx), sample_turbines)
    mism = []
    for t in ts:
        for sysn, v in mx[t].items():
            vid = f'verdict/L1/{t}/{sysn}'
            if vid not in d:
                mism.append((vid, 'MISSING'))
            elif d[vid]['props']['level'] != v['状态']:
                mism.append((vid, f"{d[vid]['props']['level']}≠{v['状态']}"))
    stats['判级对拍'] = f'{sample_turbines}台×九系统 {"全一致" if not mism else mism}'
    if mism:
        issues.append(f"⑤判级对拍不一致: {mism} — 本体快照落后于 L1, 须重跑 populate")

    # ⑥ facet 分布
    fc = Counter(o['props'].get('facet') for o in ev)
    stats['facet分布'] = dict(fc)

    return dict(issues=issues, stats=stats, n=len(d))


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
    r = audit()
    print(f"本体审计: {r['n']} 对象, 问题 {len(r['issues'])} 项")
    for i in r['issues']:
        print(' ✗', i)
    print('\n统计:', json.dumps(r['stats'], ensure_ascii=False, indent=1))