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

# -*- coding: utf-8 -*-
"""fleet_query.py — 跨场健康查询层 (运维 agent 柱3「智能助手」grounded 底座).

把已 onboard 各场 outputs/<farm>/sop/findings.json 汇成 fleet 健康视图 + 结构化查询。
NL 前端(LLM)挂本层上 → 助手回答"哪些台是齿箱候选/某场状态/某台所有 finding", 但**事实全来自本层**,
本层只报 findings 里有的, 未覆盖场明说"未分析"(不幻觉=柱3 相对通用数字员工的分界)。纯函数, 无副作用。

verdict 六枚举 (docs/SOP.md §0.2), rank 高→低: 定论>准定论·预警>候选>参考>INSUFFICIENT>撤回。
system 分类复用 cases.system_of (同一映射, 单源)。
"""

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]
import json
from pathlib import Path

from src.sop.cases import system_of
from src.sop.schemas import norm_verdict, VERDICTS
# 2026-07-09 合并: verdict rank + system 匹配 抽共享核 (与 assistant 单源, 见 agent_query_common)
from src.sop.agent_query_common import VERDICT_RANK, rank as _rank, system_family_match  # noqa: F401


def load_fleet(outputs_root):
    """glob outputs/<farm>/sop/findings.json → {farm: doc}. 坏/缺文件跳过并记 _errors。"""
    root = Path(outputs_root)
    fleet, errors = {}, []
    for fj in sorted(root.glob('*/sop/findings.json')):
        farm = fj.parent.parent.name
        try:
            doc = json.loads(fj.read_text(encoding='utf-8'))
            if isinstance(doc, dict) and isinstance(doc.get('findings'), list):
                fleet[farm] = doc
            else:
                errors.append(f'{farm}: findings.json 结构异常')
        except (json.JSONDecodeError, OSError) as e:
            errors.append(f'{farm}: {type(e).__name__}')
    fleet['_errors'] = errors
    return fleet


def load_cases(outputs_root):
    """glob outputs/<farm>/sop/cases.json → {farm: [cases]}. 供板 surface 深挖层(case.analysis)。"""
    root = Path(outputs_root)
    out = {}
    for cj in sorted(root.glob('*/sop/cases.json')):
        farm = cj.parent.parent.name
        try:
            d = json.loads(cj.read_text(encoding='utf-8'))
            if isinstance(d, list):
                out[farm] = d
        except (json.JSONDecodeError, OSError):
            pass
    return out


def analysis_priority(cases_fleet, min_verdict='准定论·预警'):
    """从 case.analysis(Phase-2 深挖+独立审) 抽 verdict≥min 的优先项 (grounded)。
    深挖层比自动 fleet-z 更权威 → 板优先关注须含, 否则同 assistant 修前盲点(D29 被埋没)。"""
    floor = _rank(min_verdict)
    out = []
    for farm, cases in (cases_fleet or {}).items():
        for c in cases or []:
            an = c.get('analysis')
            if isinstance(an, dict) and _rank(an.get('verdict')) >= floor:
                out.append({'farm': farm, 'turbine': c.get('turbine'), 'system': c.get('system'),
                            'verdict': norm_verdict(an.get('verdict')), 'mechanism': an.get('mechanism', ''),
                            'ref': an.get('ref')})
    out.sort(key=lambda r: (-_rank(r['verdict']), r['farm'], str(r['turbine'])))
    return out


def flagged(fleet, system=None, min_verdict='候选', farm=None):
    """跨场 flagged 台清单 (grounded): 返回 [{farm,turbine,title,system,verdict,problem_nature}].

    只收 affected_machines 有台 且 verdict rank≥min_verdict 的 finding。
    system: 规范 system key 过滤 (如 'gearbox'); farm: 限定场; 均可 None。
    """
    floor = _rank(min_verdict)
    out = []
    for fm, doc in fleet.items():
        if fm == '_errors' or (farm and fm != farm):
            continue
        for it in doc.get('findings', []):
            if not isinstance(it, dict):
                continue
            if _rank(it.get('verdict')) < floor:
                continue
            am = it.get('affected_machines')
            if isinstance(am, str):
                am = [am]
            if not am:
                continue
            sysk = system_of(it)
            if system and not system_family_match(sysk, system):   # 家族前缀匹配 (共享核, 单源)
                continue
            for t in am:
                out.append({'farm': fm, 'turbine': str(t), 'title': it.get('title'),
                            'system': sysk, 'verdict': norm_verdict(it.get('verdict')),
                            'problem_nature': it.get('problem_nature')})
    out.sort(key=lambda r: (-_rank(r['verdict']), r['farm'], r['turbine']))
    return out


def fleet_health(fleet):
    """fleet 健康板: 每场 finding/候选计数 + 跨场 system tally + verdict 分布。grounded。"""
    farms = {k: v for k, v in fleet.items() if k != '_errors'}
    per_farm, sys_tally, verdict_tally = {}, {}, {}
    for fm, doc in farms.items():
        fs = doc.get('findings', [])
        cand = [it for it in fs if isinstance(it, dict) and _rank(it.get('verdict')) >= _rank('候选')
                and it.get('affected_machines')]
        systems = sorted({system_of(it) for it in cand})
        per_farm[fm] = {'n_findings': len(fs), 'n_candidate_findings': len(cand),
                        'n_machines': doc.get('_n_machines'), 'systems_flagged': systems}
        for it in fs:
            if not isinstance(it, dict):
                continue
            v = norm_verdict(it.get('verdict'))
            verdict_tally[v] = verdict_tally.get(v, 0) + 1
            if v in VERDICTS and _rank(v) >= _rank('候选') and it.get('affected_machines'):
                sysk = system_of(it)
                sys_tally[sysk] = sys_tally.get(sysk, 0) + 1
    return {'n_farms': len(farms), 'per_farm': per_farm,
            'system_tally': dict(sorted(sys_tally.items(), key=lambda x: -x[1])),
            'verdict_tally': verdict_tally, 'errors': fleet.get('_errors', [])}


def farm_status(fleet, farm):
    """单场状态 (honest): 未覆盖场明说, 不返回空当健康。"""
    if farm not in fleet or farm == '_errors':
        return {'farm': farm, 'covered': False,
                'note': f'{farm} 无 findings.json (未 onboard/分析); 不等于健康, 是无数据'}
    doc = fleet[farm]
    fs = doc.get('findings', [])
    rows = []
    for it in fs:
        if not isinstance(it, dict):
            continue
        rows.append({'title': it.get('title'), 'verdict': norm_verdict(it.get('verdict')),
                     'system': system_of(it), 'flagged': it.get('affected_machines')})
    rows.sort(key=lambda r: -_rank(r['verdict']))
    return {'farm': farm, 'covered': True, 'n_machines': doc.get('_n_machines'),
            'n_findings': len(fs), 'findings': rows}


def turbine_status(fleet, farm, turbine):
    """单台所有 finding (honest)。"""
    if farm not in fleet or farm == '_errors':
        return {'farm': farm, 'turbine': turbine, 'covered': False, 'note': f'{farm} 未覆盖'}
    hits = []
    for it in fleet[farm].get('findings', []):
        if not isinstance(it, dict):
            continue
        am = it.get('affected_machines')
        am = [am] if isinstance(am, str) else (am or [])
        if str(turbine) in [str(x) for x in am]:
            hits.append({'title': it.get('title'), 'verdict': norm_verdict(it.get('verdict')),
                         'system': system_of(it), 'recommendation': it.get('recommendation')})
    hits.sort(key=lambda r: -_rank(r['verdict']))
    return {'farm': farm, 'turbine': turbine, 'covered': True, 'n_flagged': len(hits),
            'findings': hits or [{'note': '该台无 flagged finding (相对健康 or 未被判别器命中)'}]}


# 规范部件 system 前缀 (区分"部件级" vs 场特有描述性 slug)
_CANON_SYS_PREFIX = ('gearbox', 'gen', 'main_brg', 'main_shaft', 'yaw', 'pitch',
                     'driveline', 'coupling', 'safety_chain', 'underperf', 'converter')


def _canon_head(v):
    """verdict → 规范六枚举头 (复合 '候选·预警(...)'→'候选'; 供 fleet 汇总归一, 保留 honest)。"""
    nv = norm_verdict(v)
    if nv in VERDICT_RANK:
        return nv
    head = nv.split('·', 1)[0].split(' ', 1)[0].strip()
    return head if head in VERDICT_RANK else nv


def render_board(fleet, cases_fleet=None):
    """机群健康总览 markdown (grounded, 可复现; 非手写)。优先关注=准定论·预警及以上。"""
    h = fleet_health(fleet)
    strong = flagged(fleet, min_verdict='准定论·预警')
    # verdict 归一汇总
    vt = {}
    for v, n in h['verdict_tally'].items():
        vt[_canon_head(v)] = vt.get(_canon_head(v), 0) + n
    canon_sys = {k: v for k, v in h['system_tally'].items() if k.startswith(_CANON_SYS_PREFIX)}
    farm_sys = {k: v for k, v in h['system_tally'].items() if not k.startswith(_CANON_SYS_PREFIX)}

    L = ['# 机群健康总览 v0.1 (grounded · 自动生成)',
         '> 由 `src/sop/fleet_query.py::render_board` 渲染, 勿手改。数据源 = 各场 '
         f'`findings.json` ({h["n_farms"]} 场) + `cases.json` 深挖层(case.analysis)。',
         '> ⚠ verdict 纪律: **候选**=fleet-z 相对排名, 待现场; **准定论·预警**=多证但仍 SCADA; '
         '**定论/确诊**才可对外交付(RV-1)。**未列场 = 未分析 ≠ 健康**。\n']

    # 最强证据拆两类: 部件级(该推现场) vs 场级特征(已确诊分析, 非部件抢修)。verdict≠tier。
    comp = [r for r in strong if str(r['system']).startswith(_CANON_SYS_PREFIX)]
    charac = [r for r in strong if not str(r['system']).startswith(_CANON_SYS_PREFIX)]
    L.append('## 最强证据 finding (定论 / 准定论·预警)')
    L.append('> ⚠ **verdict=证据强度 ≠ tier=行动紧迫度**(正交两轴)。定论≠抢修: 场级特征类(限电/KPI/通讯)'
             '是**已确诊的分析结论**非部件抢修; 部件级才对应现场动作, 且候选级仍须现场坐实。\n')
    # 深挖层(case.analysis Phase-2+独立审) 并入部件级优先: 比自动fleet-z权威, 去重(同键已在comp则跳)
    comp_keys = {(r['farm'], str(r['turbine']), r['system']) for r in comp}
    ana = [a for a in analysis_priority(cases_fleet)
           if str(a['system']).startswith(_CANON_SYS_PREFIX)
           and (a['farm'], str(a['turbine']), a['system']) not in comp_keys]
    L.append(f'### A. 部件级 ({len(comp) + len(ana)} 项 · 对应现场动作)')
    if comp or ana:
        L.append('| 场 | 台 | 系统 | verdict | 来源 | 依据 |')
        L.append('|---|---|---|---|---|---|')
        for r in comp:
            L.append(f"| {r['farm']} | {r['turbine']} | {r['system']} | {r['verdict']} | finding | {r['title']} |")
        for a in ana:
            ref = f" (据 {str(a['ref']).split('/')[-1]})" if a.get('ref') else ''
            L.append(f"| {a['farm']} | {a['turbine']} | {a['system']} | {a['verdict']} | **深挖** | {a['mechanism'][:48]}{ref} |")
    else:
        L.append('_(无部件级达 准定论·预警 · 全fleet 部件类最高仅候选, 见花名册)_')
    L.append(f'\n### B. 场级特征分析 ({len(charac)} 项 · 已确诊, 非部件抢修)')
    if charac:
        seen = set()
        for r in charac:
            key = (r['farm'], r['title'])
            if key in seen:
                continue
            seen.add(key)
            L.append(f"- **{r['farm']}** 【{r['verdict']}】{r['title']}")

    L.append(f'\n## 各场候选花名册 ({h["n_farms"]} 场)')
    L.append('| 场 | 机数 | 候选finding | 总finding | flagged系统 |')
    L.append('|---|--:|--:|--:|---|')
    for fm, s in sorted(h['per_farm'].items(), key=lambda x: -x[1]['n_candidate_findings']):
        if s['n_candidate_findings']:
            L.append(f"| {fm} | {s['n_machines'] or '?'} | {s['n_candidate_findings']} | "
                     f"{s['n_findings']} | {', '.join(s['systems_flagged'])} |")

    L.append('\n## 跨场 system 分布 (候选级 finding 数)')
    L.append('**部件级**: ' + (', '.join(f'{k} ×{v}' for k, v in canon_sys.items()) or '无'))
    if farm_sys:
        L.append('**场特有分析** (描述性, 非标准部件键): '
                 + ', '.join(f'{k[:24]} ×{v}' for k, v in list(farm_sys.items())[:8]))

    L.append('\n## verdict 分布 (全fleet, 归一六枚举)')
    L.append(' · '.join(f'{k} {v}' for k, v in sorted(vt.items(), key=lambda x: -_rank(x[0]))))

    L.append('\n## 诚实边界')
    L.append(f'- {h["n_farms"]} 场已分析入板; 数据盘 45 场 → 余未 onboard (见 `docs/数据盘索引_WINDDATA_DATA2.md`), '
             '未列 ≠ 健康。')
    L.append(f'- **候选级需现场坐实** (fleet-z 相对排名, 禁当确诊); INSUFFICIENT {vt.get("INSUFFICIENT", 0)} 条 = 判不了+缺什么(诚实出口)。')
    if h['errors']:
        L.append(f'- ⚠ 载入异常: {h["errors"]}')
    return '\n'.join(L)


# ---------------------------------------------------------------------------
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
    import argparse
    ROOT = _install_root(__file__)
    ap = argparse.ArgumentParser(description='跨场健康查询 (柱3 grounded 底座)')
    ap.add_argument('--summary', action='store_true', help='fleet 健康板 (终端)')
    ap.add_argument('--board', action='store_true', help='渲染机群健康总览 md → outputs/_fleet/健康总览_v0.1.md')
    ap.add_argument('--system', help='按 system 过滤 flagged 台 (如 gearbox/gen_de_brg/yaw)')
    ap.add_argument('--farm', help='单场状态')
    ap.add_argument('--turbine', help='单台 (配 --farm)')
    ap.add_argument('--min-verdict', default='候选')
    a = ap.parse_args()
    fleet = load_fleet(ROOT / 'outputs')

    if a.board:
        out = ROOT / 'outputs' / '_fleet' / '健康总览_v0.1.md'
        out.parent.mkdir(parents=True, exist_ok=True)
        out.write_text(render_board(fleet, load_cases(ROOT / 'outputs')), encoding='utf-8')
        print(f'✅ 机群健康总览 → {out.relative_to(ROOT)}')
    elif a.turbine and a.farm:
        r = turbine_status(fleet, a.farm, a.turbine)
        print(json.dumps(r, ensure_ascii=False, indent=2))
    elif a.farm:
        print(json.dumps(farm_status(fleet, a.farm), ensure_ascii=False, indent=2))
    elif a.system or a.min_verdict != '候选':
        rows = flagged(fleet, system=a.system, min_verdict=a.min_verdict)
        print(f'flagged 台 {len(rows)} (system={a.system}, ≥{a.min_verdict}):')
        for r in rows:
            print(f"  [{r['verdict']}] {r['farm']}/{r['turbine']} · {r['system']} · {r['title']}")
    else:
        h = fleet_health(fleet)
        print(f"=== Fleet 健康板 ({h['n_farms']} 场) ===")
        print(f"跨场 system 候选 tally: {h['system_tally']}")
        print(f"verdict 分布: {h['verdict_tally']}")
        print("每场候选 finding 数 (有候选的场):")
        for fm, s in sorted(h['per_farm'].items(), key=lambda x: -x[1]['n_candidate_findings']):
            if s['n_candidate_findings']:
                print(f"  {fm:20s} {s['n_candidate_findings']:2d}候选/{s['n_findings']}总 · {s['systems_flagged']}")
        if h['errors']:
            print(f"⚠ 坏/缺: {h['errors']}")