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

# -*- coding: utf-8 -*-
"""知识库图谱化摄入 (2026-08-28 用户令: 四条检修链 + 所有知识库图谱化).

链一 预防: 数据/表现 → 原理图纸 → 失效模式·部件 → 预防性检修 → WI
链二 抢修: 故障报警 → 原理图纸 & 工单 → 失效模式·部件 → 故障抢修 → WI
链三 故障树 + 技术支持: 失效模式 causedBy 树 / Doc explains / 数据逆向
链四 计划: 按损失·风险排序的排程建议

源 (全部现场资料, 逐个声明出处 — 码义靠猜会翻案, 见 memory cross-vendor-channel-numbering):
  A 故障处理手册.xlsx 海上·SWT-4.0-130 737 条 → AlarmCode + WorkInstruction (分步处置)
  B 广核如海…中英文对译表.xlsx 3175 行 → AlarmCode 中文名 (本场码覆盖 263/322)
  C 如海故障代码表.xlsx 2793 行 → 告警类型/故障原因/统一风机状态
  D 维护作业指导书.xlsx (目视/机械/电气测试) → 预防性 MaintTask + WI
  E 技术资料目录 → Doc 索引 (原理图/电气图/手册/定值/结构树)
纪律: 只收本场实际出现过的码 (alarms.parquet 322 个) + 手册有处置的码; 未覆盖的码显式留白不编造。
跑法: .venv/bin/python -m src.ontology.kb_ingest
"""

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 pathlib, re, warnings
import pandas as pd
from .store import Store

warnings.filterwarnings('ignore')
TECH = P.RAW_ROOT / '西门子4.0技术资料'                        # 机理层资料在 data/raw 根下 (原写死 mac 路径)
ONT = P.ont()
ST = P.store()

# 部件/系统归属 (与 windscada workorder 同一套关键词 — 单源纪律)
# ★系统归属改由**原厂结构树**派生 (2026-09-02 用户裁), 不再手写关键词。
#
# 旧写法两个错叠在一起, 把变桨那 298 条挂错了 108 条:
#   ① 关键词过宽: 「液压/轮毂/叶片」被划进变桨, 而偏航有液压站、舱门有液压站、
#      主液压站是独立的; 轮毂属主轴系; 叶片扰流器是气动件
#   ② 顺序优先且首命中即返回: 变桨那条排第一, 「偏航液压泵过热」里的「液压」先命中,
#      永远轮不到「偏航」那条
#   实逮: 7120 偏航液压泵过热 / 7122 偏航液压油位低 / 64040 主轴承(轮毂)温度过高
#         全部挂在变桨下, 连带把 failmode 的「本场报警 N 条」计数也污染了。
#
# 新写法: 词表取自结构树节点名 (叶子/二级/一级), **最长匹配优先**。
#   结构树的名字本身是专有的 —— 「偏航液压制动系统」「变桨泵站」不会互相误伤,
#   而裸「液压」根本不是任何节点名, 于是不再产生泛词误配。
#   顺序不再决定归属, 特异性决定。
_SYS_TERMS_CACHE = None


def _sys_terms():
    """(词, 项目九系统) 列表, 按词长降序 —— 长词=更专, 先匹配。"""
    global _SYS_TERMS_CACHE
    if _SYS_TERMS_CACHE is not None:
        return _SYS_TERMS_CACHE
    import json as _j, pathlib as _p
    # 结构树二级 → 项目九系统 (九系统是交付分册口径, 保持不动; 见 CLAUDE.md 两层并存)
    L2 = {'叶片': '叶片与叶根', '导流罩模块': '叶片与叶根',
          '转子系统': '变桨', '变桨液压': '变桨', '变桨系统': '变桨',
          '传动结构': '齿轮箱', '齿轮箱系统': '齿轮箱',
          '发电机': '发电机', '发电机系统': '发电机', '集中润滑': '主轴承',
          '偏航器': '偏航', '偏航系统': '偏航',
          '塔筒': '塔架与基础', '基础环': '塔架与基础', '塔筒爬梯装置': '塔架与基础',
          '变压器和开关设备': '变流器', '变频系统': '变流器', '动力装置': '变流器',
          '主控系统': '主控与传感网', '控制和保护系统': '主控与传感网',
          '塔基控制柜': '主控与传感网',
          '机舱罩': None, '机架': None, '机舱吊车系统': None, '灭火设备': None,
          '急救装置': None, '等电位连接': None, '液压系统': None}   # None = 不强判
    terms = []
    # path 必须相对**仓库根**而非 cwd -- 服务侧 (mcp_server) 的 cwd 与入库侧不同,
    #   读不到会静默退化成半张词表 (只剩九系统名 + 同义词), 分类质量无声下降。
    #   读不到就响亮打印, 不假装成功。
    _root = _install_root(__file__)
    _f = P.objects_json()
    try:
        db_ = _j.loads(_f.read_text(encoding='utf-8'))
    except Exception as _e:
        import sys as _s
        print('WARN _sys_terms 读不到本体 %s (%s) -- 词表退化, 分类将不完整' % (_f, _e), file=_s.stderr)
        db_ = {}
    # ① RDS-PP 叶子: 最专, 经其所属二级映射到九系统
    for k, v in db_.items():
        if not k.startswith('maintsystem/rdspp/'):
            continue
        nm = str((v.get('props') or {}).get('name_zh') or '')
        if not nm or not any('\u4e00' <= c <= '\u9fff' for c in nm):
            continue
        # 叶子名里含哪个二级词就归哪个二级 (如「偏航驱动」「偏航泵站阀块」→ 偏航器)
        sysn = next((L2[t] for t in sorted(L2, key=len, reverse=True)
                     if t in nm and L2.get(t)), None)
        if sysn is None:
            for a, b in (('偏航', '偏航'), ('变桨', '变桨'), ('叶片', '叶片与叶根'),
                         ('主轴', '主轴承'), ('齿轮', '齿轮箱'), ('发电机', '发电机')):
                if a in nm:
                    sysn = b
                    break
        if sysn:
            terms.append((nm, sysn))
    # ② 结构树二级/一级
    for k, v in db_.items():
        if not k.startswith('maintsystem/tree/'):
            continue
        nm = str((v.get('props') or {}).get('name_zh') or '')
        if L2.get(nm):
            terms.append((nm, L2[nm]))
    # ③ 兜底: 九系统自身的名字 (专有, 不含泛词)
    for nm in ('变桨', '偏航', '齿轮箱', '发电机', '主轴承', '变流器', '塔架与基础',
               '主控与传感网', '叶片与叶根'):
        terms.append((nm, nm))
    # ④ 结构树词表覆盖不到但确属专有的同义词 (**不含任何泛词**)
    # ★这批是旧 SYSKW 里**本来就对**的专有词。我换结构树词表时把它们连同泛词一起删了,
    #   dry-run 逮到 228 条撤边里混着「风向标阻塞」「通讯模块错误」「网侧DI4信号错误」——
    #   全是判得出的。**过度纠正也是错**: 该删的是共享泛词 (液压/轮毂), 不是专有词。
    terms += [('风向标', '主控与传感网'), ('风速仪', '主控与传感网'), ('风速传感', '主控与传感网'),
              ('PLC', '主控与传感网'), ('UPS', '主控与传感网'), ('光纤', '主控与传感网'),
              ('RMIO', '主控与传感网'), ('看门狗', '主控与传感网'),
              ('网侧', '变流器'), ('机侧', '变流器'), ('ISU', '变流器'), ('INU', '变流器'),
              ('直流母线', '变流器'), ('水冷', '变流器'), ('并网', '变流器')]
    # 齿轮箱族叶子词: 结构树只到「传动结构/齿轮箱系统」二级, 叶子名在手册里另有叫法。
    #   `齿轮油`/`变速箱`/`齿轮支架`/`高速齿轮` 都是**专有复合词**, 不是共享泛词 ——
    #   注意不能加裸 `齿轮`: 偏航齿轮也含它, 且与 `偏航` 同为 2 字, 最长匹配分不开。
    #   (dry-run 逮: 4100~4315 九条齿轮油告警 + 七条变速箱维护作业全被撤边。)
    terms += [('齿轮油', '齿轮箱'), ('变速箱', '齿轮箱'), ('齿轮支架', '齿轮箱'),
              ('高速齿轮', '齿轮箱'), ('中速齿轮', '齿轮箱'), ('低速齿轮', '齿轮箱'),
              ('气象站', '主控与传感网'),
              # 现场码表写的是「风速计」不是「风速仪」-- 只收后者会把 8108/8000/8186/
              # 8184/1005 等 296 条本场报警判成无归属 (dry-run 实逮的倒退向)。
              ('风速计', '主控与传感网'), ('风速', '主控与传感网'), ('风向', '主控与传感网')]
    terms += [('变浆', '变桨'), ('桨距', '变桨'), ('桨叶', '变桨'),
              ('解缆', '偏航'), ('扭缆', '偏航'), ('对风', '偏航'),
              ('齿箱', '齿轮箱'), ('定子', '发电机'), ('绕组', '发电机'),
              ('滑环', '发电机'), ('励磁', '发电机'), ('主轴', '主轴承'),
              ('变频', '变流器'), ('变流', '变流器'), ('逆变', '变流器'),
              ('IGBT', '变流器'), ('塔筒', '塔架与基础'), ('塔架', '塔架与基础'),
              ('叶根', '叶片与叶根'), ('叶片', '叶片与叶根')]
    # ★第二梯队: **角色泛名**, 只在第一梯队全不命中时才用。
    #   最长匹配在这里选错轴 —— 「偏航位置传感器错误」里 `传感器`(3字) 压过 `偏航`(2字),
    #   把它判成主控与传感网。传感器属于它所服务的系统, 除非文本没点名任何系统。
    #   (memory alarm-same-sensor-not-independent: 温度与同部件保护告警常同测点。)

    seen, out = set(), []
    for t, sysn in sorted(terms, key=lambda x: -len(x[0])):
        if t not in seen:
            seen.add(t)
            out.append((t, sysn))
    out += [(t, sysn) for t, sysn in _TIER_TAIL if t not in seen]
    _SYS_TERMS_CACHE = out
    return out
# 失效模式分类关键词 (ISO14224 风格)
FMKW = [(r'磨损|磨蚀|点蚀|剥落', '磨损'), (r'疲劳|裂纹|开裂|断裂', '疲劳'),
        (r'润滑|油脂|缺油|油位低|打脂', '润滑失效'), (r'腐蚀|锈|盐雾', '腐蚀'),
        (r'过温|高温|温度高|过热|超温', '过热'), (r'松动|预紧|力矩|紧固|螺栓', '松动/紧固失效'),
        (r'高压|低压|压力|液压|气压|蓄能|油压', '液压/气动失效'),
        (r'污染|异物|杂质|堵塞|滤芯', '污染/异物'), (r'短路|断路|接地|绝缘|电压|电流|供电|电源', '电气失效'),
        (r'通讯|通信|软件|程序|逻辑|超时|看门狗', '控制/软件'), (r'密封|渗漏|泄漏|漏油', '密封失效'),
        (r'传感器|编码器|测量|信号异常|反馈', '测量链失效')]


# 「叶片X」既可能是叶片本体 (气动/结构), 也可能是**变桨执行件** —— 结构树把后者放在
# 轮毂/变桨液压 (变桨阀块) 与 轮毂/转子系统 (叶片轴承/变桨执行机构) 之下。
# 判别靠伴随词: 阀/编码器/驱动/电机/轴承/限位/顺桨 ⇒ 变桨; 扰流器/失速条/尾缘/开裂/雷击 ⇒ 叶片本体。
# (dry-run 实逮: 「3103 轮毂：叶片A阀错误」按裸「叶片」判成叶片与叶根, 实为变桨阀块。)
# ── 梯队尾部: 第一梯队 (结构树专名) 全不命中时, 按此**次序**兜底 ─────────────
# ② 总成: `轮毂` 是共享设施词 (旧 SYSKW 的病根之一, 词表断言明令它不得进第一梯队) ——
#    「轮毂处于服务状态时主轴转速过高」里它会抢走 `主轴`。但结构树把变桨执行与叶片轴承
#    都放在轮毂总成内, 所以文本**只**点了轮毂时 (「轮毂温度高」) 归变桨是结构树支持的。
#    **位置决定它安全: 同一个词在第一梯队是 bug, 在兜底位是信息。**
# ③ 角色泛名: 传感器/通讯/控制器 说的是**角色**不是系统。「偏航位置传感器错误」里
#    `传感器`(3字) 会按最长匹配压过 `偏航`(2字) —— 传感器属于它所服务的系统。
#    总成必须排在角色泛名之前:「轮毂温度传感器错误」轮毂是系统名, 传感器是角色名。
_TIER_TAIL = (('轮毂', '变桨'),
              ('传感器', '主控与传感网'), ('通讯', '主控与传感网'),
              ('通信', '主控与传感网'), ('控制器', '主控与传感网'),
              ('编码器', '主控与传感网'))

_PITCH_ACT = re.compile(r'阀|编码器|驱动|电机|轴承|限位|顺桨|蓄能|油缸|液压缸|变桨|润滑|角度|桨角'
                        r'|油压|压力|高压|低压|桨距|反馈|同步|校准')   # 叶片A低油压/叶片高压 = 变桨液压不是叶片本体
# 「蓄能器A/B/C」带**叶片字母后缀** = 每叶片一套的变桨蓄能器 (3411/3412/3413/3216/3218)。
# 裸「蓄能器」不判 -- 主液压站与高速制动器也各有蓄能器, 九系统里没有它们的家。
# (我删裸「蓄能」这个泛词时把带后缀的一并误伤了, dry-run 逮。)
_PITCH_ACC = re.compile(r'蓄能器[^,，。；;]{0,4}[ABC](?![a-zA-Z])|[ABC]\s*蓄能器|叶片\s*[ABC].{0,6}蓄能')
_BLADE_OWN = re.compile(r'扰流器|失速条|尾缘|前缘|开裂|雷击|接闪|防雷|避雷|叶尖|打磨|螺栓|翼板')


# 九系统里**没有家**的系统名 (L2 映射里判 None 的那批: 液压站/灭火/急救/机架/吊车/机舱罩)。
# 它们必须在角色泛名兜底**之前**拦下 —— 否则「液压压力传感器错误」会靠 `传感器` 落进
# 主控与传感网, 那是错的: 它点了系统(液压站), 只是这个分类体系没给它位置。
# **判不出**和**没有家**都返回 None, 但成因不同; 这是分类体系的缺口, 不是判据的漏。
# (与「塔基/变压器和开关设备」同类 —— 用户 2026-09-01 裁「先交付, 回头再改」。)
_HOMELESS = re.compile(r'液压站|液压|灭火|烟雾|消防|急救|应急照明|紧急停止|急停'
                       r'|机舱罩|机架|吊车|刹车|制动器')


def _sys_of(txt):
    """文本 → 九系统。**位置优先**, 同位再比长度; 判不出返回 None (宁可无归属也不错挂)。

    为什么按位置而不是最长匹配: 中文是修饰语在前, **前置的系统词限定整个词组**。
    「偏航齿轮箱油位低」里 `齿轮箱`(3字) 会按最长匹配压过 `偏航`(2字) —— 但那是偏航
    减速箱不是主齿轮箱。同族的错这是第三次 (前两次: 液压压过偏航 / 传感器压过偏航),
    根因都是"用词长当专指度"。位置是更贴合中文构词的专指度代理。

    梯队仍在: 第一梯队 (结构树专名) 全不命中才轮到总成 `轮毂`, 再才是角色泛名。
    """
    t = str(txt)
    if '叶片' in t and not _BLADE_OWN.search(t) and _PITCH_ACT.search(t):
        return '变桨'          # 叶片X阀 / 叶片编码器 = 变桨执行件
    if _PITCH_ACC.search(t):
        return '变桨'          # 蓄能器A/B/C = 每叶片一套的变桨蓄能器
    terms = _sys_terms()
    n_head = len(terms) - len(_TIER_TAIL)      # 第一梯队长度
    hits = [(t.index(w), -len(w), sysn) for w, sysn in terms[:n_head] if w in t]
    if hits:
        return min(hits)[2]
    hm = _HOMELESS.search(t)
    tail = [(t.index(w), sysn) for w, sysn in terms[n_head:] if w in t]
    # 无家闸**也按位置**: 「CAN通讯错误—刹车信号」主语是通讯不是刹车, 位置在前者说了算。
    # 只在无家词出现得**更早**时才判 None, 否则它只是顺带提到。
    if hm and (not tail or hm.start() < min(x[0] for x in tail)):
        return None                            # 点了系统, 但九系统没这个家
    for w, sysn in terms[n_head:]:             # 总成 → 角色泛名, 保持既定次序
        if w in t:
            return sysn
    return None


def _fm_of(txt):
    for pat, c in FMKW:
        if re.search(pat, str(txt)): return c
    return '未定'


def _comp_id(sysname):
    # ★补「叶片与叶根」→ blade: 它是九系统之一, 但此前**没有对应的 component 对象**,
    #   于是任何判成叶片的都静默落进 misc。dry-run 逮到 (3103 轮毂:叶片A阀 → misc)。
    return f'component/fleet/{ {"变桨": "pitch", "偏航": "yaw", "齿轮箱": "gearbox", "发电机": "generator", "主轴承": "main_bearing", "变流器": "converter", "塔架与基础": "tower", "主控与传感网": "control", "叶片与叶根": "blade"}.get(sysname, "misc") }'


def load_sources():
    """四源装载 + 覆盖率报数 (缺哪块响亮, 不静默降级)."""
    out, miss = {}, []
    def rd(key, path, **kw):
        f = TECH / path
        if not f.exists(): miss.append(str(path)); return None
        try:
            out[key] = pd.read_excel(f, **kw); return out[key]
        except Exception as e:
            miss.append(f'{path} 读失败 {str(e)[:40]}'); return None
    rd('hb', '故障处理/故障处理手册.xlsx', sheet_name='海上')
    rd('tr', '广核如海风电场西门子风机故障代码中英文对译表.xlsx', sheet_name='Sheet1')
    rd('rh', '如海故障代码表.xlsx', sheet_name='Sheet1')
    for sh in ('目视部分', '机械部分', '电气测试'):
        rd(f'wi_{sh}', '维护相关/维护作业指导书.xlsx', sheet_name=sh)
    if miss:
        print('⚠ 源缺失/读失败:', miss)
    return out


def build(dry=False):
    src = load_sources()
    s = Store(ONT / 'objects.json')
    al = pd.read_parquet(ST / 'alarms.parquet')
    live_codes = set(al.code.astype(str).str.strip())
    live_cnt = al.code.astype(str).str.strip().value_counts().to_dict()
    stat = dict(AlarmCode=0, WorkInstruction=0, FailureMode=0, MaintTask=0, Doc=0, Component=0)

    # 部件对象 (fleet 级, 供失效模式挂靠)
    # ★九系统全集: 原先这里只有 8 个, 漏了「叶片与叶根」—— 而 `_comp_id()` 的映射表里**有**它,
    #   于是 458 条 alarmcode 的 about 指向 `component/fleet/blade` 全是悬空引用。
    #   2026-09-12 跑 `python -m src.ontology.audit` 逮到(悬空 458 条), 与 `_comp_id` 注释里
    #   "补「叶片与叶根」→ blade" 那次修改是同一件事漏了后半截。
    for sysname in ('变桨', '偏航', '齿轮箱', '发电机', '主轴承', '变流器', '塔架与基础', '主控与传感网', '叶片与叶根'):
        cid = _comp_id(sysname)
        if cid not in s.objects:
            s.put(dict(id=cid, type='Component', props=dict(name=sysname, level='系统', scope='fleet'), links={}))
            stat['Component'] += 1

    # ---- 链二: 报警码 → 处置 WI ----
    hb, tr, rh = src.get('hb'), src.get('tr'), src.get('rh')
    cn = {}
    if tr is not None:
        for _, r in tr.iterrows():
            cn[str(r['故障代码']).strip()] = str(r.get('中文名称', '') or '').strip()
    rhmap = {}
    if rh is not None and '状态编码' in rh.columns:
        for _, r in rh.iterrows():
            k = str(r['状态编码']).strip()
            rhmap[k] = dict(告警类型=str(r.get('告警类型', '') or ''), 故障原因=str(r.get('故障原因', '') or ''),
                            统一状态=str(r.get('统一风机状态', '') or ''))
    hbmap = {}
    if hb is not None:
        hb4 = hb[hb['机型'].astype(str).str.contains('4.0-130', na=False)]
        for _, r in hb4.iterrows():
            code = str(r['故障代码']).strip()
            steps = []
            for i in range(10):
                intro = r.get(f'solutions.{i}.introduction')
                name = r.get(f'solutions.{i}.solutionName')
                if pd.notna(intro) and str(intro).strip():
                    steps.append(dict(序=i + 1, 类别=str(name or '').strip(), 步骤=str(intro).strip()))
            hbmap[code] = dict(desc=str(r['故障描述']).strip(), steps=steps)

    for code in sorted(live_codes | set(hbmap)):
        h = hbmap.get(code, {})
        name = h.get('desc') or cn.get(code) or ''
        meta = rhmap.get(code, {})
        sysname = _sys_of(name + ' ' + meta.get('故障原因', ''))
        links = {}
        if sysname: links['about'] = [_comp_id(sysname)]
        s.put(dict(id=f'alarmcode/{code}', type='AlarmCode',
                   props=dict(code=code, name=name or '(码表未收录)', name_cn=cn.get(code, ''),
                              system=sysname, 告警类型=meta.get('告警类型', ''), 故障原因=meta.get('故障原因', ''),
                              本场计数=int(live_cnt.get(code, 0)), 本场出现=code in live_codes,
                              有处置=bool(h.get('steps')),
                              source='故障处理手册.xlsx 海上SWT-4.0-130 + 对译表 + 如海故障代码表'),
                   links=links))
        stat['AlarmCode'] += 1
        if h.get('steps'):
            wid = f'wi/fault/{code}'
            s.put(dict(id=wid, type='WorkInstruction',
                       props=dict(name=f'{code} {h["desc"]} — 抢修处置', kind='抢修',
                                  steps=h['steps'], n_step=len(h['steps']),
                                  source='上海电气 故障处理手册.xlsx (海上 SWT-4.0-130)'),
                       links=dict(about=[f'alarmcode/{code}'], **({'appliesTo': [_comp_id(sysname)]} if sysname else {}))))
            stat['WorkInstruction'] += 1
            # 抢修任务 (事件驱动)
            s.put(dict(id=f'task/fault/{code}', type='MaintTask',
                       props=dict(name=f'{code} {h["desc"]} 抢修', task_kind='抢修',
                                  trigger=f'报警码 {code}', source='故障处理手册'),
                       links=dict(hasInstruction=[wid], about=[f'alarmcode/{code}'],
                                  **({'appliesTo': [_comp_id(sysname)]} if sysname else {}))))
            stat['MaintTask'] += 1

    # ---- 链一: 预防性检修 WI + 任务 (维护作业指导书 三部分) ----
    for sh, kind in (('目视部分', '目视检查'), ('机械部分', '机械作业'), ('电气测试', '电气测试')):
        d = src.get(f'wi_{sh}')
        if d is None: continue
        d = d.copy(); d.columns = [str(c).strip() for c in d.columns]
        itemcol = '操作项目' if '操作项目' in d.columns else d.columns[1]
        for _, r in d.iterrows():
            step, item, std = r.get('步骤'), r.get(itemcol), r.get('检查标准')
            if pd.isna(step) or pd.isna(item): continue
            step, item = str(step).strip(), str(item).strip()
            if not step or not item: continue
            loc = (re.match(r'^([A-Z]\d{1,2})', item) or [None, ''])[1]
            sysname = _sys_of(item + ' ' + str(std or ''))
            wid = f'wi/maint/{step}'
            steps = [dict(序=i + 1, 步骤=x.strip()) for i, x in enumerate(re.split(r'[\n；]', str(std or ''))) if x.strip()]
            s.put(dict(id=wid, type='WorkInstruction',
                       props=dict(name=item[:80], kind=kind, 步骤号=step, 位置代号=loc,
                                  steps=steps or [dict(序=1, 步骤='(手册未给分步, 见原文)')],
                                  工时h=(float(r['作业时间']) if pd.notna(r.get('作业时间')) and str(r.get('作业时间')).replace('.','',1).isdigit() else None),
                                  技术人员=str(r.get('技术人员') or '') or None,
                                  source='SWT-4.0-130 维护作业指导书.xlsx · ' + sh),
                       links={'appliesTo': [_comp_id(sysname)]} if sysname else {}))
            stat['WorkInstruction'] += 1
            s.put(dict(id=f'task/maint/{step}', type='MaintTask',
                       props=dict(name=item[:80], task_kind='预防性', interval='年度',
                                  位置代号=loc, source='年度维护检查表 (A9B1039475) + 维护作业指导书'),
                       links=dict(hasInstruction=[wid], **({'appliesTo': [_comp_id(sysname)]} if sysname else {}))))
            stat['MaintTask'] += 1

    wof = ST / 'workorders.parquet'
    wo = pd.read_parquet(wof) if wof.exists() else pd.DataFrame()

    # ---- 失效模式: 从实证归纳 (报警码描述 + 工单), 不编造; 每个模式记来源与证据量 ----
    fm_acc = {}
    for oid, o in list(s.objects.items()):
        if o['type'] != 'AlarmCode': continue
        pr = o['props']; sysname = pr.get('system')
        if not sysname: continue
        cls = _fm_of(str(pr.get('name', '')) + ' ' + str(pr.get('故障原因', '')))
        if cls == '未定': continue
        k = (sysname, cls)
        a = fm_acc.setdefault(k, dict(codes=[], n_alarm=0, wo=0))
        a['codes'].append(pr['code']); a['n_alarm'] += int(pr.get('本场计数', 0))
    if wof.exists():
        for _, r in wo.iterrows():
            txt = ' '.join(str(r.get(c, '') or '') for c in ('故障名称', '故障描述', '维修对象', '故障位置二级'))
            sysname, cls = _sys_of(txt), _fm_of(txt)
            if not sysname or cls == '未定': continue
            fm_acc.setdefault((sysname, cls), dict(codes=[], n_alarm=0, wo=0))['wo'] += 1
    for (sysname, cls), a in sorted(fm_acc.items()):
        fid = f'failmode/{sysname}/{cls}'
        s.put(dict(id=fid, type='FailureMode',
                   props=dict(name=f'{sysname} · {cls}', fm_class=cls, system=sysname,
                              n_codes=len(a['codes']), 本场报警条数=a['n_alarm'], 工单条数=a['wo'],
                              示例码=a['codes'][:8],
                              source='归纳自本场报警码表(有系统归属者)+工单文本, 非厂商FMEA — 粒度=系统×失效类'),
                   links=dict(affects=[_comp_id(sysname)],
                              indicates=[f'alarmcode/{c}' for c in a['codes'][:20]],
                              mitigatedBy=[f'task/fault/{c}' for c in a['codes'][:20] if f'task/fault/{c}' in s.objects])))
        stat['FailureMode'] += 1

    # ---- 链三 故障树: 系统级失效模式 causedBy 部件级原因节点 ----
    # 原因节点 = 手册 solutionName (排查方向, 448 种) 归一 + 工单"故障原因"实证计数。
    # 纪律: 原因来自手册与工单原文, 不是我推的因果; 每个节点记 n_codes/n_ref/工单命中, 树深两层不外推。
    CAUSE_NORM = [
        (r'线路|电缆|线缆|端子|接线|插头|插件|连接器', '线路·接线', '电气失效'),
        (r'温度传感器|压力传感器|传感器|编码器|探头|开关信号|限位', '传感器·编码器', '测量链失效'),
        (r'IO模块|IO板|模块|板卡|主板|电网模块|CPU|通讯板', '模块·板卡', '电气失效'),
        (r'风扇|散热|滤棉|滤芯|冷却|换热|水泵', '散热·通风·冷却', '过热'),
        (r'接触器|继电器|断路器|保险|熔断|开关电源|UPS|电源', '开关·供电元件', '电气失效'),
        (r'参数|设定|软件|程序|逻辑|复位|重启', '参数·软件', '控制/软件'),
        (r'液压|油泵|阀|油管|蓄能|油缸|压力', '液压元件', '液压/气动失效'),
        (r'电机|马达|驱动器|变频|变流', '驱动·变流元件', '电气失效'),
        (r'螺栓|紧固|力矩|预紧', '螺栓·紧固', '松动/紧固失效'),
        (r'油脂|润滑|注脂|油位', '润滑供给', '润滑失效'),
        (r'轴承|齿轮|齿面|联轴器', '机械传动件', '磨损'),
        (r'密封|渗漏|漏油|油封', '密封件', '密封失效'),
    ]

    def _cause_of(name):
        for pat, node, cls in CAUSE_NORM:
            if re.search(pat, str(name)): return node, cls
        return None, None

    # a) 从手册: 码 → 排查方向; 聚合成原因节点并连到该码所属的系统级失效模式
    cause_acc = {}
    for code, h in hbmap.items():
        a = s.objects.get(f'alarmcode/{code}')
        if not a: continue
        sysname = a['props'].get('system')
        if not sysname: continue
        for st in h.get('steps', []):
            node, cls = _cause_of(st.get('类别', '') + ' ' + st.get('步骤', ''))
            if not node: continue
            k = (sysname, node)
            acc = cause_acc.setdefault(k, dict(cls=cls, codes=set(), n_ref=0, wo=0, steps=[]))
            acc['codes'].add(code); acc['n_ref'] += 1
            if len(acc['steps']) < 6: acc['steps'].append(f"[{code}] {st.get('步骤','')[:70]}")
    # b) 从工单实证: 故障原因/排查项目文本命中同一批原因节点 → 计数 (真实发生过多少次)
    if wof.exists():
        for _, r in wo.iterrows():
            txt = ' '.join(str(r.get(c, '') or '') for c in ('故障原因', '排查项目', '元器件名称', '维修对象'))
            sysname, _ = _sys_of(txt), None
            node, cls = _cause_of(txt)
            if not sysname or not node: continue
            acc = cause_acc.setdefault((sysname, node), dict(cls=cls, codes=set(), n_ref=0, wo=0, steps=[]))
            acc['wo'] += 1
    for (sysname, node), a in sorted(cause_acc.items()):
        cid = f'failmode/{sysname}/原因/{node}'
        parent = f'failmode/{sysname}/{a["cls"]}'
        s.put(dict(id=cid, type='FailureMode',
                   props=dict(name=f'{sysname} · {node}', fm_class=a['cls'], system=sysname, level='部件级原因',
                              n_codes=len(a['codes']), 手册引用次数=a['n_ref'], 工单命中=a['wo'],
                              示例码=sorted(a['codes'])[:8], 典型排查=a['steps'],
                              source='手册 solutionName 归一 + 工单故障原因/排查项目文本计数; 树深两层, 不外推更深因果'),
                   links=dict(affects=[_comp_id(sysname)],
                              indicates=[f'alarmcode/{c}' for c in sorted(a['codes'])[:20]])))
        stat['FailureMode'] += 1
        stat.setdefault('causedBy', 0)
        # 系统级失效模式 ← causedBy ← 部件级原因 (父存在才连, 不造父)
        if parent in s.objects:
            pa = s.objects[parent]
            lk = pa.setdefault('links', {}).setdefault('causedBy', [])
            if cid not in lk:
                lk.append(cid); stat['causedBy'] += 1
            s.put(pa, force=True, when='2026-08-28')

    # ---- 技术资料索引 (Doc): 图纸/原理/手册/定值可检索 ----
    KIND = [(r'原理图|原理', '原理图'), (r'电气图|接线', '电气图'), (r'液压图', '液压图'), (r'爆炸图', '爆炸图'),
            (r'结构树', '结构树'), (r'定值|参数', '定值参数'), (r'维护手册|安装维护|维护', '维护手册'),
            (r'故障处理|故障逻辑|故障代码', '故障处置'), (r'培训|课件', '培训'), (r'说明书', '说明书')]
    if TECH.exists():
        for f in sorted(TECH.rglob('*')):
            if not f.is_file() or f.name.startswith('~$') or f.suffix.lower() not in ('.pdf', '.xlsx', '.xls', '.doc', '.docx', '.pptx'):
                continue
            nm = f.name
            kind = next((k for pat, k in KIND if re.search(pat, nm)), '其他')
            sysname = _sys_of(nm)
            s.put(dict(id='doc/' + re.sub(r'[^0-9A-Za-z\u4e00-\u9fa5]+', '_', str(f.relative_to(TECH)))[:110],
                       type='Doc',
                       props=dict(name=nm[:90], kind=kind, path=str(f), rel=str(f.relative_to(TECH)),
                                  system=sysname, size_kb=round(f.stat().st_size / 1024)),
                       links={'explains': [_comp_id(sysname)]} if sysname else {}))
            stat['Doc'] += 1

    # ---- 孤儿消化 (审计 ②): 不硬塞归属, 而是给"未归属"一个明确的家 ----
    s.put(dict(id='catalog/alarmcode', type='Doc',
               props=dict(name='报警码本 (本场出现 ∪ 手册收录)', kind='码本',
                          path='(派生自 故障处理手册 + 对译表 + 如海故障代码表 + 本场 alarms)',
                          note='未归属系统的码 = 码表无部件线索, 非"无关"; 现场定位需查图纸或厂商'),
               links={}))
    n_orph_code = 0
    for oid, o in list(s.objects.items()):
        if o['type'] != 'AlarmCode' or o.get('links'): continue
        o.setdefault('links', {})['partOf'] = ['catalog/alarmcode']
        s.put(o, force=True, when='2026-08-28'); n_orph_code += 1
    # 文档: 无系统归属者按顶层目录归类 (目录名本身就是分类信息)
    n_orph_doc = 0
    for oid, o in list(s.objects.items()):
        if o['type'] != 'Doc' or o.get('links') or oid == 'catalog/alarmcode': continue
        top = str(o['props'].get('rel', '')).split('/')[0][:40] or '根目录'
        cat = 'catalog/doc/' + re.sub(r'[^0-9A-Za-z\u4e00-\u9fa5]+', '_', top)[:60]
        if cat not in s.objects:
            s.put(dict(id=cat, type='Doc', props=dict(name=f'资料目录 · {top}', kind='目录',
                                                      path=str(TECH / top)), links={}))
        o.setdefault('links', {})['partOf'] = [cat]
        s.put(o, force=True, when='2026-08-28'); n_orph_doc += 1
    stat['孤儿归位'] = dict(码=n_orph_code, 文档=n_orph_doc)

    if dry:
        print('DRY:', stat); return stat
    s.save()
    print('kb_ingest:', stat)
    return stat


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 sys
    build(dry='--dry' in sys.argv)