| 12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697 |
- # -*- coding: utf-8 -*-
- """决策链进度入图谱 (2026-08-28, 与 trend_ingest 同一个缺口).
- 链盘算的是每台"从发现异常到修完验收走到第几步、卡在哪", 是这套系统最核心的结论之一,
- 但它只活在 windscada 的运行期内存里 —— 图谱侧没有节点, 于是"29号卡在哪一步"
- "哪些机组的机制还没定性"这类问法检索不到, 模型只能靠通用工具兜。
- 每台一个 Decision 对象 decision/chain/<台>, 记六步状态、卡点、下一步动作。
- 用 Decision 而非 Evidence: 它不是观测到的事实, 是系统按判据推出的处置状态。
- 跑法: python -m src.ontology.chain_ingest (需 windscada 服务在跑, 从 /api/fleet 取链盘)
- """
- 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 json, pathlib, sys, time, urllib.request
- ROOT = _install_root(__file__)
- ONT = P.ont()
- API = 'http://localhost:18033/api/fleet?win=2026%E5%B9%B4'
- # 逻辑时钟: 本模块吃的是**运行期链盘快照**(API 里没有业务日期), 所以"快照日期"就是它的时钟 ——
- # 这是事实而不是编造时间。★必须显式传: 原代码两个 put 都没给 when, 于是只要某台的六步状态
- # **真的有变化**, Store 的 F8 闸就会抛「force 覆盖 Decision 必须显式传 when」把整个摄入打断
- # (2026-09-12 修 fleet 那条孤岛链接时实逮; 平时只改 props 才会触发, 所以一直没暴露)。
- WHEN = time.strftime('%Y-%m-%d')
- def build(dry=False, api=API):
- from .store import Store
- try:
- with urllib.request.urlopen(api, timeout=300) as r:
- fl = json.loads(r.read())
- except Exception as e:
- print(f'取链盘失败 (windscada 服务未起?): {str(e)[:90]}')
- return {}
- bd = ((fl.get('fus') or {}).get('链盘')) or {}
- rows = bd.get('rows') or []
- if not rows:
- print('链盘为空'); return {}
- s = Store(ONT / 'objects.json')
- n = 0
- for r in rows:
- t = r.get('t')
- if not t:
- continue
- steps = r.get('steps') or []
- seq = ' → '.join(f"{x['k']}({'✔' if x['st'] == 'done' else ('?' if x['st'] == 'unknown' else '×')})"
- for x in steps)
- stuck, kind = r.get('stuck'), r.get('stuck_kind')
- kind_cn = {'open': '未达', 'unknown': '数据不足不可判'}.get(kind, kind or '')
- s.put(dict(id=f'decision/chain/{t}', type='Decision', props=dict(
- name=f"{t[-2:]}# 决策链 {r.get('done')}/6 · 停滞于「{stuck or '全通'}」",
- turbine=t, part=r.get('part'), device_state=r.get('ostate'), evidence_state=r.get('estate'),
- progress=f"{r.get('done')}/6", stuck=stuck, stuck_kind=kind_cn,
- next_action=r.get('next'), evidence_sources='/'.join(r.get('srcs') or []),
- steps=seq,
- decision_kind='状态跟踪',
- reversibility='L1', # 本对象只记录状态, 不触发任何动作
- summary=(f"{t[-2:]}# {r.get('part') or ''} 六步决策链走到 {r.get('done')}/6, "
- f"停滞于「{stuck or '全通'}」({kind_cn})。设备状态 {r.get('ostate')}, "
- f"证据状态 {r.get('estate')}, 证据源 {'/'.join(r.get('srcs') or []) or '—'}。"
- f"下一步: {r.get('next') or '—'}"),
- note='六步=证据/机制/判级/排期/动作/验收; 动作一列因工单台账止 2024-11 一律不可判'),
- # ★links 的值必须是**列表**: 写成字符串 `f'turbine/{t}'` 会被逐字符当目标,
- # 于是 403 条链接指向 't'/'u'/'r'/'b'/'i' 这种单字符 (2026-09-12 跑
- # `python -m src.ontology.audit` 逮到: 悬空引用 403 条)。
- links=dict(about=[f'turbine/{t}'])), force=True, when=WHEN)
- n += 1
- # 全场瓶颈汇总 — 单台看不出"24 台卡在同一步"这件事
- stk = bd.get('stuck') or {}
- if stk:
- top = sorted(stk.items(), key=lambda x: -x[1])
- s.put(dict(id='decision/chain/fleet', type='Decision', props=dict(
- name=f"全场决策链瓶颈 · {top[0][1]} 台停滞于「{top[0][0]}」",
- turbine=None, progress=f"{len(rows)} 台需跟踪 / 38",
- stuck=top[0][0], decision_kind='状态跟踪', reversibility='L1',
- summary=(f"需跟踪 {len(rows)} 台 / 38。停滞分布: " +
- ', '.join(f"{v} 台在「{k}」" for k, v in top) +
- f"。{'瓶颈集中在单一环节, 不是继续取数能解的。' if top[0][1] > len(rows) * 0.5 else ''}"),
- note=bd.get('wo_bound', '')),
- # 全场汇总也要连边: 原先 links={} → 审计报"孤岛对象(无任何边, 查询走不到)"。
- # 它汇总的就是这些台, 挂 about 到它们才符合语义。★注意 links 属于对象顶层, 必须写在
- # props=dict(...) 之外 (写进去就变成 props 里的一个键, 对象本身仍然没有边)。
- links=dict(about=[f'turbine/{r.get("t")}' for r in rows if r.get('t')])), force=True, when=WHEN)
- n += 1
- if not dry:
- s.save()
- print(f'chain_ingest: {n} 个决策链对象')
- return dict(chain=n)
- if __name__ == '__main__':
- build(dry='--dry' in sys.argv)
|
