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

# -*- coding: utf-8 -*-
"""ontology MCP 服务 (P2: 本体经 MCP 暴露给 dsh; stdio 协议壳照 windscada/mcp_server).

分工铁律 (用户裁): 本体查询/约束/动作 = 结构化工具 (本文件), 模型只做语义编排.
全部工具只读 (L1) — 模型零写权; 判级零生成 — 只转述对象里的 verdict 原文与审级;
最终回答必过 claim_check 接地闸 (台号/数字越界即拒).
用法: python -m src.ontology.mcp_server"""

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 sys, json, argparse, traceback, re
from pathlib import Path
sys.path.insert(0, str(_install_root(__file__)))
from src.windscada.audit import grounding  # noqa: E402

PROTOCOL = '2025-06-18'
OBJ_PATH = P.objects_json()
FACTS = []
FACTS_MAX = 200          # F11: 原无界 — 长会话下 FACTS 无限增长, 接地闸比对成本与内存同步涨
_DB = None
_DB_MTIME = None


def db():
    """读本体库. F11 修 (Critic 逮): 原缓存只认 None, populate/kb_ingest 重写库后进程内仍读旧快照
    (判级已变而工具还在答旧值 = 静默 stale). 改按文件 mtime 失效重载."""
    global _DB, _DB_MTIME
    if not OBJ_PATH.exists():
        raise FileNotFoundError(f'本体库缺失: {OBJ_PATH} (先跑 populate) — 不可静默降级')
    mt = OBJ_PATH.stat().st_mtime
    if _DB is None or _DB_MTIME != mt:
        _DB = json.loads(OBJ_PATH.read_text(encoding='utf-8'))
        _DB_MTIME = mt
    return _DB


def _fact(txt):
    """记事实并封顶 (F11): 超上限丢最旧, 保留最近的取证."""
    FACTS.append(txt)
    if len(FACTS) > FACTS_MAX:
        del FACTS[:len(FACTS) - FACTS_MAX]


def _brief(o, with_props=False):
    b = dict(id=o['id'], type=o['type'])
    p = o.get('props', {})
    for k in ('name', 'level', 'system', 'component', 'review_tier', 'claim_window', 'status'):
        if k in p:
            b[k] = p[k]
    if with_props:
        b['props'] = p
        b['links'] = o.get('links', {})
    return b


# ── 工具实现 (纯函数, 判断在本体, 本文件只查) ──

def _tid(turbine):
    """台号解析容错 (2026-08-28 实测): 模型问"全场损失"时会传 turbine='全场',
    原式 int('全场') 直接抛异常 → 工具返回"工具错误", 该轮取证白跑。
    非数字一律当"不筛台", 而不是让工具崩掉。"""
    n = re.sub(r'\D', '', str(turbine or ''))
    return f'WTG{int(n):02d}' if n else ''


def ont_stats():
    from collections import Counter
    d = db()
    cnt = dict(Counter(v['type'] for v in d.values()))
    return dict(text=(f"本体库 {len(d)} 对象: {cnt}。判读纪律(写入闸, 已在库保证): "
                      "C1证据必带claim_window+观测面 / C2判级必带审级 / C3独立源数按观测面去重禁按条数 / "
                      "C4判读三分(支持|排除|不可判, 不可判≠阴性) / C5状态措辞必带窗 / C6动作必带可逆级。"
                      "回答纪律: 判级只转述对象原文并注明对象id与审级; 证据窗外该证据不背书; 未列≠正常。"), data=cnt)


def _turbine_scope(d, tid):
    """台的语义邻域 (F5: 通过 links 真实游走, 不硬编码台号).
    turbine → Component(of) → Mechanism(affects 挂台或挂部件) → Verdict(about 挂台/部件/机制)
    → Decision/Action (basedOn/resolves/acceptedBy/informs 链入邻域, 层内传播至收敛).
    修前只按 about∋turbine 收 Verdict, 挂在机制上的关键判级 (如 verdict/WTG29/TCM-16) 游走不到."""
    comps = {k for k, o in d.items() if o['type'] == 'Component'
             and tid in o.get('links', {}).get('of', [])}
    anchors = {tid} | comps
    mechs = {k for k, o in d.items() if o['type'] == 'FailureMechanism'
             and anchors & set(o.get('links', {}).get('affects', []))}
    verds = {k for k, o in d.items() if o['type'] == 'Verdict'
             and (anchors | mechs) & set(o.get('links', {}).get('about', []))}
    core = anchors | mechs | verds
    scen = set()
    for _ in range(4):                     # decision→scenario→decision 链, 层内传播至收敛
        grew = False
        for k, o in d.items():
            if k in scen or o['type'] not in ('Decision', 'Action'):
                continue
            flat = {x for ids in o.get('links', {}).values() for x in ids}
            if flat & (core | scen):
                scen.add(k); grew = True
        if not grew:
            break
    return anchors, mechs, verds, scen


def ont_turbine_brief(turbine=''):
    t = _tid(turbine)
    d = db()
    tid = f'turbine/{t}'
    if tid not in d:
        return dict(text=f'机组不存在: {t}', error=True)
    anchors, mech_ids, verd_ids, scen_ids = _turbine_scope(d, tid)
    verd = [_brief(d[k]) for k in sorted(verd_ids)]
    mech = [_brief(d[k]) for k in sorted(mech_ids)]
    wo = [o for o in d.values() if o['type'] == 'WorkOrder' and o['props'].get('turbine') == t]
    oil = [o['props'].get('report_date', '') for o in d.values() if o['type'] == 'OilSample' and o['props'].get('turbine') == t]
    scen = [_brief(d[k]) for k in sorted(scen_ids)]
    alarm_lv = [v for v in verd if v.get('level') in ('报警', '定论') or '预警' in str(v.get('level', ''))]
    txt = (f"{t} 台级汇总: 判级 {len(verd)} 条 (报警/定论/预警级 {len(alarm_lv)} 条: "
           f"{[(v['id'], v.get('level'), v.get('review_tier')) for v in alarm_lv][:6]}); "
           f"机制 {len(mech)} 个: {[m['id'] for m in mech]}; 工单 {len(wo)} 条; "
           f"油样 {len(oil)} 份(末样 {max(oil) if oil else '无'}); 决策/动作 {[s['id'] for s in scen]}。"
           f"深入用 ont_get(id) / ont_links(id)。")
    return dict(text=txt, data=dict(verdicts=verd, mechanisms=mech, n_workorders=len(wo),
                                    oil_last=max(oil) if oil else None, decisions=scen))


def ont_get(id=''):
    d = db()
    if id not in d:
        near = [k for k in d if str(id) in k][:8]
        return dict(text=f'对象不存在: {id}; 相近id: {near}', error=True)
    return dict(text=json.dumps(_brief(d[id], with_props=True), ensure_ascii=False, default=str)[:5000],
                data=_brief(d[id], with_props=True))


def ont_query(type='', turbine='', contains='', exact='', limit=20):
    """F13: contains 数字串按词边界匹配 ('29' 命中 WTG29 不命中 2029/1729);
    exact= 全等匹配 (id / props 标量值 / links 目标id); 台号类查询走 turbine= (精确) 勿用 contains."""
    import re as _re
    d = db()
    t = _tid(turbine)
    out = []
    for o in d.values():
        if type and o['type'] != type:
            continue
        if t and t not in json.dumps(o.get('links', {}), ensure_ascii=False) and o['props'].get('turbine') != t and f'/{t}/' not in o['id']:
            continue
        if exact:
            vals = {o['id']} | {str(v) for v in o.get('props', {}).values()
                                if not isinstance(v, (dict, list))}
            vals |= {x for ids in o.get('links', {}).values() for x in ids}
            if str(exact) not in vals:
                continue
        if contains:
            blob = json.dumps(o, ensure_ascii=False)
            c = str(contains)
            if c.isdigit():                # 纯数字: 词边界感知, 防 '29' 假阳命中 '2029'
                if not _re.search(rf'(?<!\d){_re.escape(c)}(?!\d)', blob):
                    continue
            elif c not in blob:
                continue
        out.append(_brief(o))
        if len(out) >= int(limit):
            break
    return dict(text=f'{len(out)} 个对象: ' + json.dumps(out[:int(limit)], ensure_ascii=False, default=str)[:4500], data=out)


def ont_links(id=''):
    d = db()
    if id not in d:
        return dict(text=f'对象不存在: {id}', error=True)
    fwd = d[id].get('links', {})
    back = [dict(id=o['id'], rel=rel) for o in d.values() for rel, ids in o.get('links', {}).items() if id in ids]
    return dict(text=f'正向: {fwd}; 反向被引: {back[:20]}', data=dict(forward=fwd, backlinks=back[:50]))


def ont_mechanism_search(keyword=''):
    d = db()
    hits = [_brief(o, with_props=True) for o in d.values() if o['type'] == 'FailureMechanism'
            and (not keyword or str(keyword) in json.dumps(o, ensure_ascii=False))]
    return dict(text=f'机制库命中 {len(hits)}: ' + json.dumps(hits, ensure_ascii=False, default=str)[:4500], data=hits)


def ont_scenario(turbine=''):
    t = _tid(turbine)
    d = db()
    hits = [o for o in d.values() if o['id'].startswith('scenario/') and (not t or f'/{t}/' in o['id'])]
    if not hits:
        return dict(text=f'无沙盘对象 (turbine={t or "全部"})', error=True)
    return dict(text=json.dumps([dict(id=o['id'], **o['props']) for o in hits], ensure_ascii=False, default=str)[:6000],
                data=[dict(id=o['id'], **o['props']) for o in hits])


def ont_loss(turbine='', months=3):
    """损失分析: 近 N 月损失结构 (loss_monthly.parquet, L1 仓; 电量 kWh, 金额须另配电价区间口径)."""
    import pandas as pd
    lm = pd.read_parquet(P.store() / 'loss_monthly.parquet')
    ms = sorted(lm.month.unique())[-int(months):]
    d = lm[lm.month.isin(ms)]
    t = _tid(turbine)
    if t:
        d = d[d.turbine == t]
    by_state = d[d.state != '正常发电'].groupby('state').loss.sum().sort_values(ascending=False)
    by_turb = d[d.state != '正常发电'].groupby('turbine').loss.sum().sort_values(ascending=False).head(8)
    gen = d[d.state == '正常发电'].act.sum()
    txt = (f"损失结构 (窗={ms[0]}~{ms[-1]}{', ' + t if t else ', 全场'}): "
           f"分态损失 kWh = {by_state.round(0).to_dict()}; "
           f"台级损失 top = {by_turb.round(0).to_dict()}; 同窗正常发电量 {gen:,.0f} kWh。"
           f"口径: 损失=按状态归因的少发电量估计 (kWh); 折金额须带电价区间 0.391~0.85 元/kWh (合同价未知); "
           f"限电类损失≠设备责任 (口径纪律)。数据窗至 {sorted(lm.month.unique())[-1]}。 "
           # ★给出可引用对象 id (2026-08-28): 答案有效性闸要求每个结论句带 [对象id];
           # 本工具原来只回文字没有 id, 模型只好自造 [全场汇总]/[WTG19] 这类假 id ⇒ 闸判"有效=False"。
           f"来源对象: [loss/fleet/{ms[0]}~{ms[-1]}] (引用本结论时用此 id)。")
    return dict(text=txt, data=dict(months=ms, by_state=by_state.to_dict(), by_turbine=by_turb.to_dict()))


def ont_risk():
    """风险跟踪面: 联动监视读数 + 机制状态 + 盲区纪律 (安全风险类问题的入口)."""
    from src.windscada.subsys import temp_nbm
    wl = temp_nbm.watchlist_check()
    d = db()
    mech = [(o['props'].get('name'), o['props'].get('status'), o.get('links', {}).get('affects', []))
            for o in d.values() if o['type'] == 'FailureMechanism']
    acc = d.get('acceptance/WTG29/state', {}).get('props', {})
    txt = ('联动监视 (温度watch): ' + ' | '.join(wl) +
           f" ;; 机制库状态: {[(n, s) for n, s, _ in mech]}"
           f" ;; 29# 验收状态机: {acc.get('state')} ({acc.get('note', '')[:80]})"
           ' ;; 盲区纪律: 振动未列≠正常 (盲区四条见 handoff detectability), 不可判≠阴性,'
           ' 风险升级判据=自身纵向机制链 (同期+先后+剂量), 单月横比不给判级。'
           # ★可引用 id (2026-08-28, 与 ont_loss 同一处修法): 答案有效性闸要求每个结论句带
           # [对象id]; 本工具原来只回文字没有 id, 模型只好不带引用 ⇒ 闸判"有效=False"。
           # 机制对象本身就是可引的, 直接把它们的 id 给出来。
           + ' ;; 可引用对象: ' + ', '.join(
               f"[{o['id']}]" for o in d.values() if o['type'] == 'FailureMechanism')[:600]
           + ' [risk/fleet/watchlist] (联动监视读数引用此 id)。')
    return dict(text=txt, data=dict(watchlist=wl, mechanisms=[(n, s) for n, s, _ in mech], acceptance=acc))


def ont_brief_pack(turbine=''):
    """一站式取证包 (提速: 台级问题一轮取齐, 替代 stats→brief→get→scenario→loss 多轮).
    合并: 台级汇总+机制详情+沙盘+验收状态+近3月损失+纪律."""
    b = ont_turbine_brief(turbine)
    if b.get('error'):
        return b
    t = turbine if str(turbine).startswith('WTG') else f'WTG{int(turbine):02d}'
    d = db()
    _, mech_ids, verd_ids, _ = _turbine_scope(d, f'turbine/{t}')
    mech_full = [dict(id=k, **d[k]['props']) for k in sorted(mech_ids)]
    # 关键判级全文 (F5: 挂机制的高审级判级如 TCM-16, 台级包里给全文非仅列名)
    key_verd = [dict(id=k, **{kk: d[k]['props'][kk] for kk in ('level', 'title', 'review_tier', 'claim_window')
                              if kk in d[k]['props']})
                for k in sorted(verd_ids)
                if d[k]['props'].get('review_tier') in ('双模审', 'AdvCI-PR', '用户裁')]
    scen = ont_scenario(t)
    loss = ont_loss(t, 3)
    acc = d.get('acceptance/WTG29/state', {}).get('props', {}) if t == 'WTG29' else {}
    txt = (b['text'] + ' ;; 关键判级(高审级)全文: ' + json.dumps(key_verd, ensure_ascii=False, default=str)[:1400]
           + ' ;; 机制详情: ' + json.dumps(mech_full, ensure_ascii=False, default=str)[:2200]
           + ' ;; 沙盘: ' + (scen.get('text', '无')[:2600] if not scen.get('error') else '无')
           + ' ;; 验收状态机: ' + json.dumps(acc, ensure_ascii=False, default=str)[:600]
           + ' ;; ' + loss['text'][:800]
           + ' ;; 纪律: 判级只转述原文带审级与对象id; 观测面≠源数; 未列≠正常; 不可判≠阴性; 状态判断带窗。')
    return dict(text=txt[:9000], data=None)


_PARAMS = None
# 元件位号 (西门子 4.0 命名): 206-B5C / 403-B7 / 501-K1 / 453DI5 / 406F5 / 503-W2 / 400-M1
_TAG_RE = re.compile(r'(\d{3}\s?[-–]\s?[A-Z]{1,3}\d{0,3}[A-Z]?|\d{3}[A-Z]{2}\d{1,2})')
# 步骤里的物理量 → 参数表关键词 + 图纸类别 (用于给"检查规范"和"看哪张图")
_QTY = [
    ('压力|油压|高压|低压|bar', 'pressure', '液压图', '压力'),
    ('温度|阻值.*温|℃', 'temperature', '电气图', '温度'),
    ('阻值|电阻|绝缘|Ω', 'resistance', '电气图', '阻值/绝缘'),
    ('电流|电压|供电|回路|接地|短路|断路', 'voltage|current', '电气图', '电气量'),
    ('转速|rpm|超速', 'speed', '原理图', '转速'),
    ('油位|液位|液面', 'level', '液压图', '油位'),
    ('角度|桨距|位置|编码器', 'angle|position', '原理图', '位置/角度'),
]


def _params():
    global _PARAMS
    if _PARAMS is None:
        import pandas as _pd
        f = P.ont() / 'turbine_params.parquet'
        _PARAMS = _pd.read_parquet(f) if f.exists() else None
    return _PARAMS


# 回路 → 应检节点清单 (2026-08-28 用户令: 给出检查的每个节点)。
# 手册步骤常写"检查液压回路是否正常"这类笼统句, 没有位号 → 现场不知道查哪几处。
# 下表按 SWT-4.0-130 的回路构成展开, 每个节点给"查什么"; 位号从同系统手册步骤里实际出现过的抽取补充。
_CIRCUIT_NODES = {
    '液压回路': [('蓄能器', '氮气预充压力、皮囊有无破损'), ('比例阀/电磁阀', '得电动作、阀芯有无卡滞'),
                 ('油缸', '行程到位、杆端密封渗漏'), ('压力传感器', '零点与量程、信号线通断'),
                 ('油管与接头', '外泄、振动磨损、卡箍松动'), ('泵站与滤芯', '压力建立时间、滤芯压差'),
                 ('回油与油位', '油位在标线内、回油是否通畅')],
    '润滑回路': [('润滑泵', '电机绕组与绝缘、出脂压力'), ('分配器', '柱塞往复计数、有无堵塞'),
                 ('主管路与支路', '管路通畅、接头渗漏'), ('轴承注脂口', '出脂是否到位、旧脂颜色'),
                 ('油位/脂位开关', '信号与实物是否一致')],
    '电气回路': [('断路器/熔断器', '是否跳闸、整定值'), ('接触器/继电器', '吸合动作、触点烧蚀'),
                 ('端子排', '紧固力矩、氧化变色'), ('线缆与屏蔽', '通断、绝缘、屏蔽接地'),
                 ('IO 模块/板卡', '指示灯、通道输入输出'), ('电源与UPS', '输出电压、后备时间')],
    '冷却系统': [('冷却风扇/水泵', '启停动作、转向、噪声'), ('散热片/换热器', '清洁度、堵塞'),
                 ('温控开关', '动作温度与整定一致'), ('滤棉/滤网', '积尘、更换周期'),
                 ('管路与液位', '渗漏、冷却液位与浓度')],
    '传感回路': [('传感器本体', '阻值/输出与温度或压力对应'), ('接线与插头', '插头到位、针脚氧化'),
                 ('信号线缆', '屏蔽接地、与动力线分离'), ('采集通道', '同型号通道互换验证')],
    '制动/机械': [('制动器', '闸片厚度、间隙、动作压力'), ('联轴器', '对中、弹性件裂纹'),
                 ('紧固螺栓', '力矩复核、防松标记错位')],
}
_CIRCUIT_HINT = [('液压|油压|蓄能|油缸|阀', '液压回路'), ('润滑|油脂|注脂|分配器|柱塞', '润滑回路'),
                 ('电气|回路|供电|短路|断路|接地|模块|端子|线缆|电压|电流', '电气回路'),
                 ('冷却|散热|风扇|滤棉|水冷|温控', '冷却系统'),
                 ('传感器|编码器|信号|反馈|阻值', '传感回路'),
                 ('制动|刹车|联轴器|螺栓|紧固', '制动/机械')]


def _known_tags(system):
    """该系统手册步骤里实际出现过的元件位号 (给笼统步骤当"可能涉及的位号"参考)."""
    d = db()
    out = {}
    for o in d.values():
        if o['type'] != 'WorkInstruction' or o['props'].get('kind') != '抢修':
            continue
        if system and system not in str(o.get('links', {}).get('appliesTo', [])) and \
           system not in str(o['props'].get('name', '')):
            continue
        for st in o['props'].get('steps', []):
            for t in _TAG_RE.findall(str(st.get('步骤', ''))):
                out[t] = out.get(t, 0) + 1
    return sorted(out.items(), key=lambda kv: -kv[1])[:10]


def _step_enrich(step_text, system, code=''):
    """把一条手册处置步骤补成可执行的检查项 (2026-08-28 用户令:
    "一定要检查系统的原理图纸, 给出具体的检查回路节点部件, 包括检查的规范, 检查内容都要清楚标明").

    产出四段: ①节点部件(元件位号, 来自手册原文, 确定) ②检查内容(手册原文)
    ③判定规范(实机参数表 Min/Max/Current — 按物理量关键词匹配, 非厂商绑定, 须现场以铭牌/图纸复核)
    ④对应图纸(按物理量与系统匹配资料索引)。
    ★纪律: 参数按关键词匹配给的是**候选判据**不是绑定值; 码号与参数 ID 是不同命名空间
    (UP03225 参数 ≠ 告警码 3225, memory 已记), 故不按同号关联。"""
    import re as _re
    d = db()
    txt = str(step_text or '')
    tags = sorted(set(_TAG_RE.findall(txt)))
    qty, pkw, dkind = None, None, None
    for pat, kw, dk, label in _QTY:
        if _re.search(pat, txt):
            qty, pkw, dkind = label, kw, dk
            break
    # 判定规范: 参数表按物理量关键词 + 系统词过滤
    specs = []
    pr = _params()
    if pr is not None and pkw:
        sysen = {'变桨': 'pitch|blade', '偏航': 'yaw', '齿轮箱': 'gear', '发电机': 'generator|gen',
                 '主轴承': 'main bearing|rotor', '变流器': 'converter|inverter|grid'}.get(system, '')
        m = pr.Text.astype(str).str.contains(pkw, case=False, na=False)
        if sysen:
            m &= pr.Text.astype(str).str.contains(sysen, case=False, na=False)
        for _, r in pr[m].head(4).iterrows():
            specs.append(dict(no=str(r['Number']), text=str(r['Text']).split(':', 1)[-1].strip(),
                              lo=(None if r['Min'] != r['Min'] else float(r['Min'])),
                              hi=(None if r['Max'] != r['Max'] else float(r['Max'])),
                              val=(None if r['Value'] != r['Value'] else float(r['Value'])),
                              unit=('' if r['Unit'] != r['Unit'] else str(r['Unit']))))
    # 图纸: 按类别 + 系统
    dwgs = []
    if dkind:
        for o in d.values():
            if o['type'] != 'Doc':
                continue
            pp = o['props']
            if pp.get('kind') == dkind and (not system or pp.get('system') in (None, system)):
                dwgs.append(dict(name=pp['name'], rel=pp.get('rel', '')))
            if len(dwgs) >= 3:
                break
    # 应检节点: 有位号用位号; 没位号则按回路展开清单 (笼统句"检查液压回路"现场不知道查哪几处)
    circuit = next((c for pat, c in _CIRCUIT_HINT if _re.search(pat, txt)), None)
    nodes = _CIRCUIT_NODES.get(circuit, []) if not tags else []
    known = _known_tags(system) if (not tags and circuit) else []
    return dict(tags=tags, qty=qty, specs=specs, drawings=dwgs,
                circuit=circuit, nodes=[dict(name=n, check=c) for n, c in nodes],
                known_tags=[dict(tag=t, n=n) for t, n in known])


# 安全前提 (按系统; 通用检修规程, 非本场专用条款 — 现场以业主作业票与厂家手册为准)
_SAFETY = {
    '变桨': ['三叶顺桨至 90° 并确认桨距锁定', '轮毂检修位固定, 转子锁投入', '蓄能器泄压至 0 bar 并验压',
             '轮毂内作业须双人 + 安全带 + 通风', '断开变桨柜动力电并验电挂牌'],
    '齿轮箱': ['机组停机, 转子锁投入', '齿轮油温降至可作业温度 (避免烫伤)', '油路作业前泄压并接油盘',
               '拆检前记录油位与取样 (拆后无法回溯)', '断开润滑泵与冷却电机动力电挂牌'],
    '发电机': ['断开主断路器并验电, 挂牌上锁', '等待定子/绕组残余电压释放', '滑环作业前确认无残压',
               '转子锁投入后方可进入联轴器区', '测绝缘前断开变流器侧连接'],
    '主轴承': ['机组停机, 转子锁投入', '集中润滑泵断电挂牌 (防误动作喷脂)', '取脂样前记录上次加注时间与量',
               '主轴区作业须防坠落措施'],
    '偏航': ['偏航电机断电挂牌', '偏航制动器保压确认 (防机舱自转)', '解缆前确认电缆无扭转应力'],
    '变流器': ['断开网侧与机侧开关, 验电挂牌', '等待直流母线电容放电至安全电压 (按手册等待时间)',
               '水冷系统作业前泄压', '禁止带电插拔功率模块'],
    '主控与传感网': ['断开对应回路电源并验电', 'UPS 侧确认已隔离', '光纤作业防激光直视'],
}
# 记录项 (每次处置必留, 否则下次无法回溯)
_RECORD = ['处置日期与人员', '故障复位前后的读数对比', '更换件的件号与批次', '实测值与判定依据',
           '未解决项与下轮观察点']


def compose_repair_plan(code=''):
    """按本体图谱编制现场处置预案 (2026-08-28 用户令: 自己编制, 对照经典设计, 用 ontology 决策机制与知识图谱).

    与 ont_chain_fault 的区别: 那个是**转录**(码义/手册步骤/工单/图纸原样给出);
    本函数是**编排** — 用图谱把转录素材组织成可执行预案, 每段标明来源性质:
      [转录] 手册与工单原文 (可靠, 出处可查)
      [编排] 我按实证量与检修常识排的顺序 (可质疑, 现场可调)
      [通用] 检修安全规程 (非本场专用条款, 以业主作业票为准)
      [候选] 参数表按物理量匹配的判定值 (非厂商绑定, 以铭牌图纸复核)
    """
    d = db()
    code = str(code).strip()
    a = d.get(f'alarmcode/{code}')
    if not a:
        return dict(text=f'码 {code} 不在本体', error=True)
    pr = a['props']
    sysn = pr.get('system')
    base = ont_chain_fault(code=code).get('data') or {}

    # ① 排查顺序: 按该系统原因节点的实证量排 (手册引用次数 + 工单命中×3) — 先查最常见的
    causes = []
    for o in d.values():
        if o['type'] != 'FailureMode' or o['props'].get('level') != '部件级原因':
            continue
        if sysn and o['props'].get('system') != sysn:
            continue
        cp = o['props']
        causes.append(dict(name=cp['name'].split('·')[-1].strip(),
                           n_ref=cp.get('手册引用次数', 0), n_wo=cp.get('工单命中', 0),
                           score=cp.get('手册引用次数', 0) + cp.get('工单命中', 0) * 3,
                           steps=(cp.get('典型排查') or [])[:2]))
    causes.sort(key=lambda x: -x['score'])

    # ② 备件与工具: 从该系统工单的处置动作与元器件反推 (现场真换过什么)
    parts, acts = {}, {}
    for o in d.values():
        if o['type'] != 'WorkOrder':
            continue
        q = o['props']
        txt = ' '.join(str(q.get(k, '') or '') for k in ('name', 'loc2', 'part'))
        if sysn and not re.search({'变桨': '变桨|桨|叶片|液压|蓄能', '偏航': '偏航|解缆',
                                   '齿轮箱': '齿轮|齿箱|润滑油|滤芯', '发电机': '发电机|定子|滑环',
                                   '主轴承': '主轴', '变流器': '变频|变流|IGBT|水冷',
                                   '主控与传感网': '主控|通讯|传感'}.get(sysn, sysn), txt):
            continue
        if q.get('part'):
            parts[str(q['part'])] = parts.get(str(q['part']), 0) + 1
        if q.get('action'):
            acts[str(q['action'])] = acts.get(str(q['action']), 0) + 1

    # ③ 验收: 复位后观察窗 (按该码本场发生频次定 — 高频码需更长观察)
    n_site = int(pr.get('本场计数', 0))
    watch = '复位后连续观察 7 天' if n_site >= 100 else '复位后连续观察 3 天'
    return dict(
        code=code, name=pr.get('name', ''), system=sysn, n_site=n_site,
        safety=_SAFETY.get(sysn, ['按业主作业票执行停机与隔离']),
        order=[c for c in causes[:5]],
        manual_steps=base.get('steps', []),
        parts=sorted(parts.items(), key=lambda kv: -kv[1])[:6],
        actions=sorted(acts.items(), key=lambda kv: -kv[1])[:5],
        drawings=base.get('docs', []),
        workorders=base.get('workorders', [])[:5],
        accept=[f'故障码 {code} 不再复现 ({watch})',
                '相关信号回到判定范围内 (见各步判定规范)',
                '若为更换件: 记录件号批次并更新台账',
                '未解决项写入下轮观察点'],
        record=_RECORD,
        provenance=dict(转录='上海电气故障处理手册(海上 SWT-4.0-130) + 本场工单台账 2020~2024',
                        编排='排查顺序按本体图谱的原因节点实证量(手册引用+工单命中×3)排序 — 可质疑可调',
                        通用='安全前提为通用检修规程, 非本场专用条款, 以业主作业票与厂家手册为准',
                        候选='判定规范来自实机参数表(2024-10-21), 按物理量匹配 — 非厂商绑定, 现场以铭牌图纸复核'))


def ont_chain_fault(code=''):
    """链二 抢修: 报警码 → 码义/系统 → 失效模式 → 处置 WI 分步 (现场手册原文转录, 不产判级)."""
    d = db()
    code = str(code).strip()
    a = d.get(f'alarmcode/{code}')
    if not a:
        return dict(text=f'码 {code} 不在本体 (本场未出现且手册未收录 — 未收录≠无此码)', error=True)
    pr = a['props']
    L = [f"[alarmcode/{code}] {pr.get('name','')} | 系统 {pr.get('system') or '未归属'}"
         f" | 本场 {pr.get('本场计数',0)} 条 | 告警类型 {pr.get('告警类型') or '—'}"]
    if pr.get('故障原因'): L.append(f"码表故障原因: {pr['故障原因']}")
    fms = [o for o in d.values() if o['type'] == 'FailureMode' and f'alarmcode/{code}' in o.get('links', {}).get('indicates', [])]
    for f in fms:
        L.append(f"→ 失效模式 [{f['id']}] {f['props']['name']} (本场该类报警 {f['props'].get('本场报警条数',0)} 条, 工单 {f['props'].get('工单条数',0)} 条)")
    wi = d.get(f'wi/fault/{code}')
    if wi:
        L.append(f"→ 处置 WI [{wi['id']}] ({wi['props']['n_step']} 步, 源: {wi['props']['source']})")
        for st in wi['props']['steps']:
            L.append(f"   {st['序']}. [{st.get('类别','')}] {st['步骤']}")
    else:
        L.append('→ 本码无手册处置条目 (手册覆盖本场 161/322 码) — 需查图纸或厂商支持')
    # 技术依据: 先按码义关键词匹配文件名, 再退回系统级 (原来只按 system → "叶片高压"推荐吊具手册)
    _kws = [w for w in re.findall(r'[\u4e00-\u9fa5]{2,}', str(pr.get('name', ''))) if len(w) >= 2]
    _pri = [o for o in d.values() if o['type'] == 'Doc' and any(w in o['props']['name'] for w in _kws)]
    _sys = [o for o in d.values() if o['type'] == 'Doc' and o['props'].get('system') == pr.get('system')]
    _seen, docs = set(), []
    for o in _pri + _sys:
        if o['id'] in _seen: continue
        _seen.add(o['id']); docs.append(o)
        if len(docs) >= 4: break
    if docs: L.append('技术依据: ' + '; '.join(f"[{o['id']}] {o['props']['name']}" for o in docs))
    # name_en 带出: 本体 AlarmCode/FailureMode 上就存着厂商对译表的权威译名; 不带出来,
    # 出口投影只能拿裸描述去查 (码不在串里 → 查不到 → pending)。经典页与 v2 同受益。
    _nm = pr.get('name', '')
    if str(_nm).strip().lower() in ('nan', 'none', 'null'):   # pandas NaN 漏成字符串 (码表 17 个码)
        _nm = ''
    _data = dict(code=code, name=_nm, name_en=pr.get('name_en'), system=pr.get('system'),
                 n_site=pr.get('本场计数', 0), alarm_type=pr.get('告警类型', ''),
                 cause_hint=pr.get('故障原因', ''),
                 failmodes=[dict(id=f['id'], name=f['props']['name'], name_en=f['props'].get('name_en'),
                                 n_alarm=f['props'].get('本场报警条数', 0), n_wo=f['props'].get('工单条数', 0))
                            for f in fms],
                 steps=[dict(st, **_step_enrich(st.get('步骤', ''), pr.get('system'), code))
                        for st in (wi['props']['steps'] if wi else [])],
                 wi_source=(wi['props']['source'] if wi else None),
                 docs=[dict(id=o['id'], name=o['props']['name'], kind=o['props'].get('kind'),
                            rel=o['props'].get('rel')) for o in docs])
    # 链二第三段: 该系统历史工单 (现场真做过什么 — 手册给"应该怎么修", 工单给"实际怎么修的")
    sysn = pr.get('system')
    if sysn:
        wos = [o for o in d.values() if o['type'] == 'WorkOrder' and _wo_sys(o) == sysn]
        wos.sort(key=lambda o: str(o['props'].get('date', '')), reverse=True)
        if wos:
            from collections import Counter as _C
            acts = _C(o['props'].get('action') for o in wos if o['props'].get('action'))
            L.append(f'历史工单 ({sysn}, 台账 2020~2024): {len(wos)} 条 · 处置 '
                     + ', '.join(f'{k}×{v}' for k, v in acts.most_common(4)))
            for o in wos[:5]:
                q = o['props']
                L.append(f"   {q.get('date','')} {q.get('turbine','')} {str(q.get('name',''))[:26]}"
                         f" → {q.get('action','')} {str(q.get('part','') or '')[:14]}")
            L.append('   (工单窗止 2024-11-21; 2025-26 现场未提供 ⇒ "0 条"是记录缺失非未修过)')
            _nz = lambda v: '' if v in (None, 'None', 'nan') else str(v)
            # ★同时带出 *_en: 工单对象本来就存着 name_en / action_en / part_en (来自厂商
            #   对译表与构词表), 那是权威译名。原来只取中文字段, 英文侧只能事后翻译 ——
            #   而事后翻译对厂商报警名会猜错 (见 _data_text_en 的"不猜译"纪律)。
            #   下游按 key_en → key 投影即可, 中文侧不受影响。
            # 空日期不留空格: 大部件更换里有 7 条**台账源头就没填** (源表 time_on/time_off
            #   与「大部件维修时间」三处皆空), 界面留空会被读成"数据丢了"。显式写出来。
            _dt = lambda p: (_nz(p.get('date'))[:10]
                             or ('台账未填' if p.get('date_note') else '—'))
            _data['workorders'] = [dict(date=_dt(o['props']), date_en=('not recorded in ledger'
                                        if not _nz(o['props'].get('date')) and o['props'].get('date_note')
                                        else None),
                                        turbine=_nz(o['props'].get('turbine')),
                                        name=_nz(o['props'].get('name'))[:40],
                                        name_en=_nz(o['props'].get('name_en'))[:60] or None,
                                        action=_nz(o['props'].get('action')) or '—',
                                        action_en=_nz(o['props'].get('action_en')) or None,
                                        part=_nz(o['props'].get('part'))[:18],
                                        part_en=_nz(o['props'].get('part_en'))[:30] or None)
                                   for o in wos[:8]]
            _data['wo_total'] = len(wos)
    return dict(text='\n'.join(L), data=_data)


# 这里原有一份 _WOKW -- 与 kb_ingest 旧 SYSKW **同样的病**:
#   变桨|..|液压|蓄能|轮毂|叶片|叶根 首匹配即中且变桨在最前, 于是「偏航液压泵过热」
#   先撞上 液压 判成变桨, 根本走不到偏航那条。入库侧 2026-09-02 已改按原厂结构树派生,
#   **服务侧这份拷贝没跟着改** = memory library-gate-not-wired-to-production 的原样复发:
#   判据落库不等于接上生产路径。此处删拷贝, 直接调同一个 _sys_of。
from .kb_ingest import _sys_of as _kb_sys_of


def _wo_sys(o):
    """工单 -> 九系统。与入库侧 appliesTo 走**同一个判据**, 不再各自维护关键词表。"""
    p = o.get('props', {})
    txt = ' '.join(str(p.get(k, '') or '') for k in ('name', 'loc2', 'part'))
    return _kb_sys_of(txt)


def ont_workorder(turbine='', system='', keyword='', limit=20):
    """检修工单查询 (链二"检修工单"段): 按台/系统/关键词; 现场实际做过什么 + 处置分布."""
    d = db()
    t, sysn, kw = str(turbine).strip(), str(system).strip(), str(keyword).strip()
    if t and not t.startswith('WTG'):
        t = f'WTG{int(t):02d}' if t.isdigit() else t
    wos = [o for o in d.values() if o['type'] == 'WorkOrder']
    if t: wos = [o for o in wos if o['props'].get('turbine') == t]
    if sysn: wos = [o for o in wos if _wo_sys(o) == sysn]
    if kw: wos = [o for o in wos if kw in json.dumps(o['props'], ensure_ascii=False)]
    if not wos:
        return dict(text=f'无匹配工单 (台 {t or "全场"} / 系统 {sysn or "全部"} / 关键词 {kw or "无"}); '
                         '台账窗 2020-01~2024-11, 2025-26 现场未提供 ⇒ 空≠未修过')
    from collections import Counter as _C
    acts = _C(o['props'].get('action') for o in wos if o['props'].get('action'))
    parts = _C(o['props'].get('part') for o in wos if o['props'].get('part'))
    wos.sort(key=lambda o: str(o['props'].get('date', '')), reverse=True)
    L = [f'工单 {len(wos)} 条 (台 {t or "全场"} / 系统 {sysn or "全部"}{"/ 关键词 " + kw if kw else ""})',
         '处置分布: ' + ', '.join(f'{k}×{v}' for k, v in acts.most_common(6)),
         '涉及部件: ' + ', '.join(f'{k}×{v}' for k, v in parts.most_common(6)),
         '']
    for o in wos[:int(limit)]:
        q = o['props']
        L.append(f"[{o['id']}] {q.get('date','')} {q.get('turbine','')} · {str(q.get('name',''))[:30]}"
                 f" · {q.get('loc2','')} · {q.get('action','')} {str(q.get('part','') or '')[:16]}")
    L.append('')
    L.append('口径: 台账窗 2020-01-03~2024-11-21 (六类源 data-first 核过: 集团月表 2025-26 全场无记录, '
             '两个"年度总表.xls"实为报警 XML 导出); 电量损失列只登记不聚合 (累计快照直接 sum 虚高)。')
    return dict(text='\n'.join(L))


def ont_chain_preventive(system='', component=''):
    """链一 预防: 部件/系统 → 该系统失效模式 → 预防性任务 → WI 步骤与工时."""
    d = db()
    key = str(system or component).strip()
    fms = [o for o in d.values() if o['type'] == 'FailureMode' and key and key in o['props'].get('system', '')]
    # 部件 id 是英文 (component/fleet/gearbox), key 是中文 → 先经映射再比 (原直接 in 比对恒 False)
    _CID = {'变桨': 'pitch', '偏航': 'yaw', '齿轮箱': 'gearbox', '发电机': 'generator', '主轴承': 'main_bearing',
            '变流器': 'converter', '塔架与基础': 'tower', '主控与传感网': 'control',
            '叶片与叶根': 'blade'}     # 2026-09-02 结构树重分类后 component/fleet/blade 已存在
    _cid = f"component/fleet/{_CID.get(key, '')}" if key in _CID else None
    tasks = [o for o in d.values() if o['type'] == 'MaintTask' and o['props'].get('task_kind') == '预防性'
             and _cid and _cid in (o.get('links', {}).get('appliesTo') or [])]
    if not key:
        return dict(text='需给 system/component (如 齿轮箱/变桨/主轴承)', error=True)
    fms_sorted = sorted(fms, key=lambda x: -(x['props'].get('本场报警条数') or 0))
    L = [f'== {key} 预防链 ==', f'失效模式 {len(fms)} 类 · 预防性任务 {len(tasks)} 项']
    for f in fms_sorted[:6]:
        pr = f['props']
        L.append(f"· [{f['id']}] {pr['name']} — 本场报警 {pr.get('本场报警条数',0)} 条 / 工单 {pr.get('工单条数',0)} 条")
    for t in tasks[:10]:
        wi = (t.get('links', {}).get('hasInstruction') or [None])[0]
        w = d.get(wi) if wi else None
        L.append(f"· 任务 [{t['id']}] {t['props']['name']} (周期 {t['props'].get('interval')})"
                 + (f" → WI {w['props'].get('步骤号')} {len(w['props']['steps'])} 步" if w else ''))
    if not tasks: L.append('(该系统无匹配的年度维护条目 — 维护作业指导书按位置代号编排, 未必逐系统对齐)')
    # ★结构化 data (2026-09-03 新前端): 原来只回 text 拼串, 英文页只能整块显示中文 —— 文本 blob
    #   无法按字段投影 name_en。data 带 *_en 字段, 出口 _jdump 按 lang 投影, 前端按 key 取文案。
    def _fm(f):
        pr = f['props']
        return dict(id=f['id'], name=pr.get('name'), name_en=pr.get('name_en'),
                    n_alarm=pr.get('本场报警条数') or 0, n_wo=pr.get('工单条数') or 0)
    def _task(t):
        wi = (t.get('links', {}).get('hasInstruction') or [None])[0]
        w = d.get(wi) if wi else None
        return dict(id=t['id'], name=t['props'].get('name'), name_en=t['props'].get('name_en'),
                    interval=t['props'].get('interval'),
                    wi=(dict(id=w['id'], n_step=len(w['props'].get('steps') or []),
                             steps=w['props'].get('steps') or []) if w else None))
    data = dict(system=key, n_fm=len(fms), n_task=len(tasks),
                failmodes=[_fm(f) for f in fms_sorted], tasks=[_task(t) for t in tasks])
    return dict(text='\n'.join(L), data=data)


def ont_fault_tree(system=''):
    """链三 故障树: 系统 → 失效模式(按实证量排) → 各自的征兆码与处置; 附本场证据量."""
    d = db()
    key = str(system).strip()
    fms = [o for o in d.values() if o['type'] == 'FailureMode' and (not key or key in o['props'].get('system', ''))]
    fms.sort(key=lambda x: -(x['props'].get('本场报警条数', 0) + x['props'].get('工单条数', 0) * 10))
    tops = [f for f in fms if f['props'].get('level') != '部件级原因']
    L = [f"== 故障树 {key or '全场'} ({len(tops)} 个系统级失效模式, 按实证量排) =="]
    for f in tops[:8]:
        pr = f['props']
        L.append(f"[{f['id']}] {pr['name']} · 报警 {pr.get('本场报警条数',0)} 条 / 工单 {pr.get('工单条数',0)} 条 / 码 {pr.get('n_codes',0)} 个")
        if pr.get('示例码'): L.append(f"    征兆码: {', '.join(pr['示例码'][:5])}")
        # 二层: causedBy 部件级原因 (手册排查方向 + 工单实证计数)
        for cid in (f.get('links', {}).get('causedBy') or []):
            c = d.get(cid)
            if not c: continue
            cp = c['props']
            L.append(f"    ← 原因 [{cid}] {cp['name'].split('·')[-1].strip()}"
                     f" · 手册引用 {cp.get('手册引用次数',0)} 次 / 工单命中 {cp.get('工单命中',0)} 次 / 涉码 {cp.get('n_codes',0)}")
            for st in (cp.get('典型排查') or [])[:2]:
                L.append(f"        · {st}")
    L.append('口径: 一层=系统×失效类(归纳自本场码表+工单); 二层=手册 solutionName 排查方向归一 + 工单原因文本计数。')
    L.append('      因果来自手册与工单原文, 非我方推断; 树深两层不外推; 现场定位仍需图纸与实测。')
    _data = dict(system=key or '全场', nodes=[
        dict(id=f['id'], name=f['props']['name'].split('·')[-1].strip(),
             fm_class=f['props'].get('fm_class'),
             n_alarm=f['props'].get('本场报警条数', 0), n_wo=f['props'].get('工单条数', 0),
             codes=(f['props'].get('示例码') or [])[:5],
             causes=[dict(id=c, name=(d.get(c, {}).get('props', {}).get('name', c)).split('·')[-1].strip(),
                          n_ref=d.get(c, {}).get('props', {}).get('手册引用次数', 0),
                          n_wo=d.get(c, {}).get('props', {}).get('工单命中', 0),
                          n_codes=d.get(c, {}).get('props', {}).get('n_codes', 0),
                          steps=(d.get(c, {}).get('props', {}).get('典型排查') or [])[:3])
                     for c in (f.get('links', {}).get('causedBy') or []) if c in d])
        for f in tops[:8]])
    return dict(text='\n'.join(L), data=_data)


def ont_maint_plan(top=10):
    """链四 检修计划: 双轴排序 (风险 × 损失) 的排程建议.
    风险 = 本体 Verdict 判级 (定论/危险 > 报警 > 预警) × 命中系统数;
    损失 = 该台窗内停机损失 MWh (SCADA 实测, 非台账 — 台账止 2024-11 且有物理上限问题);
    两轴都给出, 不合成单一分数 (合成会掩盖"高风险低损失"与"低风险高损失"两类不同处置)。"""
    d = db()
    RANK = {'定论': 4, '危险': 4, '报警': 3, '预警': 2, '候选': 1}
    agg = {}
    for o in d.values():
        if o['type'] != 'Verdict': continue
        lv = o['props'].get('level')
        if lv not in RANK: continue
        if o['props'].get('consumable') is False:      # F3: 非部件级消费行不进排程 (证据在, 判级不下游)
            continue
        about = ' '.join(o.get('links', {}).get('about') or []) + ' ' + o['id']
        t = next((x[:6] for x in re.findall(r'WTG\d{2}', about)), None)
        if not t: continue
        sysn = o['props'].get('system') or (o['props'].get('title', '') or o['id'].split('/')[-1])[:28]
        a = agg.setdefault(t, dict(turbine=t, risk=0, n=0, items=[], tiers=set()))
        a['risk'] = max(a['risk'], RANK[lv]); a['n'] += 1
        a['items'].append(f"{sysn}[{lv}]"); a['tiers'].add(o['props'].get('review_tier', ''))
    # 损失轴: SCADA 实测停机损失 (窗内)
    import pandas as _pd, pathlib as _pl
    lp = P.store() / 'loss_monthly.parquet'
    loss = {}
    if lp.exists():
        lm = _pd.read_parquet(lp)
        lm = lm[(lm.month.astype(str) >= '2026-01') & (lm.state == '停机')]
        loss = (lm.groupby('turbine')['loss'].sum() / 1000).round(0).to_dict()
    rows = sorted(agg.values(), key=lambda x: (-x['risk'], -loss.get(x['turbine'], 0), -x['n']))[:int(top)]
    L = ['== 检修排程建议 (双轴: 风险级 × 实测停机损失) ==']
    L.append(f"{'机组':<8}{'风险':<6}{'命中':<5}{'停机损失MWh':<13}命中项")
    for a in rows:
        lv = {4: '定论/危险', 3: '报警', 2: '预警', 1: '候选'}[a['risk']]
        L.append(f"{a['turbine']:<8}{lv:<6}{a['n']:<5}{loss.get(a['turbine'], 0):<13,.0f}{'; '.join(a['items'][:4])}")
    hi_loss = sorted(loss.items(), key=lambda kv: -kv[1])[:3]
    L.append('')
    L.append('损失轴独立看 (高损失但未必高风险级, 需查是否限电/外部): ' +
             ', '.join(f'{t} {v:,.0f} MWh' for t, v in hi_loss))
    L.append('口径: 风险=Verdict 判级(筛查级/候选, 非定谳) × 命中系统数; 损失=2026 窗内"停机"态 SCADA 实测')
    L.append('      (不含限电与调度令; 工单台账损失止 2024-11 且需过物理上限核, 未用)。')
    L.append('★ 本表只排序不定案: 执行前过 SOP §6.1 审级 + 天气窗 (ont_scenario) + 备件与船期。')
    return dict(text='\n'.join(L))


def ont_doc_search(keyword='', kind=''):
    """链三 技术支持: 图纸/原理/手册检索 (按名称与类别; 返回路径供现场调阅)."""
    d = db()
    kw, kd = str(keyword).strip(), str(kind).strip()
    hits = [o for o in d.values() if o['type'] == 'Doc'
            and (not kw or kw in o['props']['name'] or kw in str(o['props'].get('rel', '')))
            and (not kd or kd == o['props'].get('kind'))]
    if not hits:
        kinds = sorted({o['props'].get('kind') for o in d.values() if o['type'] == 'Doc'})
        return dict(text=f'未命中 (关键词 {kw!r} 类别 {kd!r}); 可用类别: {kinds}')
    L = [f'技术资料命中 {len(hits)} 份:']
    for o in hits[:15]:
        pr = o['props']
        L.append(f"[{o['id']}] {pr['name']} · {pr['kind']}" + (f" · {pr['system']}" if pr.get('system') else '') + f" · {pr['rel']}")
    return dict(text='\n'.join(L))


def ont_fleet_verdicts(level='', tier=''):
    """全场判级速查 — 紧凑表, 一次给全 (2026-08-28 功能测试逮).

    原来问"全场哪些机组有报警级"只能走 ont_query(type=Verdict), 它回 371 个对象的完整 JSON;
    进上下文时被截断到 2000 字符, 模型只看到前几条, 于是自己写"(需补充, 当前数据截断)" —
    答案不是错而是**残**。列举类问题要的是覆盖不是细节, 所以这里只回一行一条的紧凑表:
    台号 | 部件/系统 | 级别 | 审级。38 台全量约 1.5KB, 一轮取齐, 不触截断。

    level/tier 可选过滤 (如 level='报警' 或 '定论')。
    """
    d = db()
    rows = []
    for k, v in d.items():
        if v.get('type') != 'Verdict':
            continue
        pr = v.get('props') or {}
        lv = str(pr.get('level') or pr.get('判级') or '')
        tr = str(pr.get('review_tier') or pr.get('审级') or '')
        if level and level not in lv:
            continue
        if tier and tier not in tr:
            continue
        # 台号: id 里的 WTGnn, 取不到就跳过 (L1/vib 等非台级 verdict)
        m = re.search(r'WTG(\d{2})', k)
        if not m:
            continue
        part = k.split('/')[-1]
        rows.append((m.group(1), part, lv, tr, k))
    if not rows:
        return dict(text=f'无匹配判级 (level={level or "全部"}, tier={tier or "全部"})')
    # ★按台聚合而非逐条列 (2026-08-28): 逐条 371 行 = 17KB, 仍会触上下文截断;
    # 一台一行 (该台各级别下的系统清单 + 一个代表 id 供引用) 全量约 2KB, 一轮取齐。
    ORD = {'定论': 0, '危险': 1, '报警': 2, '预警': 3, '候选': 4}
    # 正常级不逐系统列 (占全量 4.2KB 的一多半, 且不是提问要的东西): 压成一句计数。
    NORMAL = ('优秀', '良好', '正样本', '销案')
    by_t = {}
    for tid, part, lv, tr, oid in rows:
        e = by_t.setdefault(tid, dict(lv={}, worst=9, rep=oid, tier=tr))
        e['lv'].setdefault(f'{lv}({tr or "—"})', []).append(part)
        r = ORD.get(lv, 9)
        if r < e['worst']:
            e['worst'], e['rep'], e['tier'] = r, oid, tr
    lines = []
    for tid in sorted(by_t, key=lambda t: (by_t[t]['worst'], t)):
        e = by_t[tid]
        ab, nm = {}, 0
        for k, v in e['lv'].items():
            (nm := nm + len(v)) if k.split('(')[0] in NORMAL else ab.setdefault(k, v)
        seg = '; '.join(f"{k} {','.join(v[:6])}{'…' if len(v) > 6 else ''}" for k, v in
                        sorted(ab.items(), key=lambda kv: ORD.get(kv[0].split('(')[0], 9)))
        if nm:
            seg = (seg + '; ' if seg else '') + f'其余 {nm} 项优秀/良好'
        lines.append(f"{tid}# {seg} [{e['rep']}]")
    head = (f'全场判级按台汇总 {len(by_t)} 台 / {len(rows)} 条' +
            (f' (筛 level={level})' if level else '') + ':')
    tail = '\n注: 一台可有多条(不同系统各自判级), 上表按台合并并按最严级别排序; 未列台不等于正常。'
    return dict(text=head + '\n' + '\n'.join(lines) + tail, n=len(rows), n_turbine=len(by_t))


def ont_search(q='', top=8):
    """语义+术语混合检索 — 不知道该调哪个工具时的入口 (2026-08-28 检索层)。

    此前模型只能靠 ont_query 的字段过滤找对象, 遇到"哪些机组风险在上升"这类
    没有确定入口的问法就得多轮试探。本工具直接按问句召回最相关的对象并给出 id 与摘要,
    模型拿到 id 后再用 ont_get 取全文即可, 省掉试探。
    """
    from src.ontology.retrieval import search as _s
    hits = _s(q, top=int(top))
    if not hits:
        return dict(text=f'未检索到与「{q}」相关的对象')
    d = db()
    lines = []
    for oid, sc, src in hits:
        o = d.get(oid) or {}
        pr = o.get('props') or {}
        # 摘要取值顺序 (2026-08-28 实测): verdict 的 props 没有 name 字段, 原来只显示
        # "Verdict  level=预警" —— 模型看不出这条是什么, 只能靠 id 猜, 于是拿"齿轮箱高速轴预警"
        # 去回答"换件后振动降下来了吗", 得出与事实相反的结论。摘要必须带实质内容。
        nm = (pr.get('name') or pr.get('summary') or pr.get('conclusion') or pr.get('note')
              or pr.get('component') or '')
        extra = ' '.join(f'{k}={pr[k]}' for k in
                         ('turbine', 'level', 'status', 'direction', 'ratio', 'pre', 'post',
                          'replace_date', 'review_tier', 'date')
                         if pr.get(k) not in (None, ''))
        lines.append(f"[{oid}] {o.get('type','')} {str(nm)[:70]} {extra}".rstrip())
    return dict(text=f'检索「{q}」命中 {len(hits)} 个对象 ({hits[0][2]}):\n' + '\n'.join(lines) +
                     '\n用 ont_get 取任一 id 的全文。', n=len(hits))


TOOLS = {
    'ont_search': (ont_search, {'q': '哪些机组风险在上升', 'top': 8},
                   '★混合检索 (术语+语义): 不确定该调哪个工具、或问法没有明确入口时先用它召回相关对象 id'),
    'ont_fleet_verdicts': (ont_fleet_verdicts, {'level': '报警', 'tier': ''},
                           '★全场判级速查 (全场级列举问题首选: 一次给全 38 台的 台号|部件|级别|审级 紧凑表, 不会被截断)'),
    'ont_chain_fault': (ont_chain_fault, dict(code='3225'), '链二抢修: 报警码→码义/系统→失效模式→手册分步处置WI'),
    'ont_chain_preventive': (ont_chain_preventive, dict(system='齿轮箱'), '链一预防: 系统/部件→失效模式→预防性任务→WI步骤'),
    'ont_fault_tree': (ont_fault_tree, dict(system='变桨'), '链三故障树: 系统失效模式按实证量排序+征兆码'),
    'ont_maint_plan': (ont_maint_plan, dict(top=10), '链四计划: 按风险级与命中数排序的检修排程建议'),
    'ont_workorder': (ont_workorder, dict(turbine='29', system='主轴承'), '检修工单查询: 按台/系统/关键词, 现场实际做过什么+处置分布'),
    'ont_doc_search': (ont_doc_search, dict(keyword='齿轮箱', kind='原理图'), '技术支持: 图纸/原理/手册检索, 返回现场可调阅路径'),
    'ont_brief_pack': (ont_brief_pack, {'turbine': '29'}, '★一站式取证包 (台级问题首选, 一轮取齐: 判级/机制/沙盘/验收/损失/纪律)'),
    'ont_stats': (ont_stats, {}, '本体库概览+判读纪律 (全场级问题开局)'),
    'ont_turbine_brief': (ont_turbine_brief, {'turbine': '29'}, '台级汇总: 判级(带审级)/机制/工单/油样/决策 — 台级问题的入口'),
    'ont_get': (ont_get, {'id': 'verdict/WTG29/TCM-16'}, '按 id 取对象全文 (props+links)'),
    'ont_query': (ont_query, {'type': 'Verdict', 'turbine': '29', 'contains': '', 'exact': '', 'limit': 20},
                  '按类型/台/关键词过滤查询 (台号用 turbine= 精确参数; contains 数字按词边界; exact= 全等)'),
    'ont_links': (ont_links, {'id': 'mechanism/WTG29/lub_starvation'}, '对象的正向链接与反向被引'),
    'ont_mechanism_search': (ont_mechanism_search, {'keyword': '润滑'}, '失效机制库检索'),
    'ont_scenario': (ont_scenario, {'turbine': '29'}, '排期沙盘/Scenario 推演结果 (建议级, L3 人裁)'),
    'ont_loss': (ont_loss, {'turbine': '', 'months': 3}, '损失分析: 近N月分态/台级损失结构 (kWh; 金额须带电价区间口径; 限电≠设备责任)'),
    'ont_risk': (ont_risk, {}, '风险跟踪面: 联动监视读数+机制状态+验收状态机+盲区纪律 (安全风险类入口)'),
}

CLAIM_CHECK = dict(name='claim_check', description='接地闸: 最终回答前必须调用; 台号/数字须来自本会话工具事实, 越界即拒.',
                   inputSchema={'type': 'object', 'properties': {'answer': {'type': 'string'}}, 'required': ['answer']})


def _schema(params):
    props = {k: {'type': ('number' if isinstance(v, (int, float)) and not isinstance(v, bool) else 'string'),
                 'description': f'例: {v}'} for k, v in (params or {}).items()}
    return {'type': 'object', 'properties': props}


def tools_list():
    return [dict(name=n, description=desc, inputSchema=_schema(ex)) for n, (fn, ex, desc) in TOOLS.items()] + [CLAIM_CHECK]


def tools_call(name, args):
    if name == 'claim_check':
        if not FACTS:
            return {'content': [{'type': 'text', 'text': '接地闸: 尚无工具事实, 先取证。'}], 'isError': True}
        ok, bad = grounding((args or {}).get('answer', ''), '\n'.join(FACTS))
        txt = '接地闸 通过' if ok else f"接地闸 不通过: 越界台号 {bad['turbines']} / 数字 {bad['numbers']} — 只用工具返回的量改写后再查"
        return {'content': [{'type': 'text', 'text': txt}], 'isError': not ok}
    if name not in TOOLS:
        return {'content': [{'type': 'text', 'text': f'未知工具 {name}'}], 'isError': True}
    r = TOOLS[name][0](**(args or {}))
    _fact(f"[{name}] {str(r.get('text', ''))[:4000]}")
    return {'content': [{'type': 'text', 'text': str(r.get('text', ''))}], 'isError': bool(r.get('error'))}


def serve():
    out = sys.stdout
    for line in sys.stdin:
        line = line.strip()
        if not line:
            continue
        try:
            msg = json.loads(line)
        except Exception:
            continue
        mid, method, params = msg.get('id'), msg.get('method', ''), msg.get('params') or {}
        resp = None
        try:
            if method == 'initialize':
                resp = {'protocolVersion': params.get('protocolVersion', PROTOCOL),
                        'capabilities': {'tools': {'listChanged': False}},
                        'serverInfo': {'name': 'ontology', 'version': '0.1.0'},
                        'instructions': ('如东风场语义本体 (对象/判级/机制/沙盘)。规则: ①台级问题**首选 ont_brief_pack 一轮取齐**, '
                                         '全场级问题先 ont_stats ②判级只转述对象原文并注明对象id与审级, 禁自创判级 '
                                         '③证据窗外不背书, 未列≠正常, 不可判≠阴性 ④最终回答前必过 claim_check ⑤取证够即答, 勿重复查询。')}
            elif method == 'ping':
                resp = {}
            elif method == 'tools/list':
                resp = {'tools': tools_list()}
            elif method == 'tools/call':
                resp = tools_call(params.get('name'), params.get('arguments'))
            elif method.startswith('notifications/'):
                continue
            else:
                if mid is not None:
                    out.write(json.dumps({'jsonrpc': '2.0', 'id': mid, 'error': {'code': -32601, 'message': method}}) + '\n'); out.flush()
                continue
        except Exception:
            if mid is not None:
                out.write(json.dumps({'jsonrpc': '2.0', 'id': mid, 'error': {'code': -32000, 'message': traceback.format_exc()[-600:]}}) + '\n'); out.flush()
            continue
        if mid is not None:
            out.write(json.dumps({'jsonrpc': '2.0', 'id': mid, 'result': resp}, ensure_ascii=False, default=str) + '\n'); out.flush()


if __name__ == '__main__':
    argparse.ArgumentParser().parse_args()
    serve()