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

# -*- coding: utf-8 -*-
"""数据维护端口 (2026-08-28 用户令: 设计系统的维护截面, 主要是数据的输入补充;
按 ontology 构建情况与既有数据格式, 建立能随后期运营数据更新的端口; 机理层相对稳定).

分两层 (更新频率与责任人不同, 混在一起会让"机理被数据覆盖"):
  【数据层】随运营滚动 — SCADA 月度/报警/工单/油样/CMS 报告/振动 handoff。
           更新 = 放新文件到约定目录 → 跑对应摄入 → 本体自动刷新。
  【机理层】相对稳定 — 失效模式与原因树/WI 作业指导/图纸索引/实机参数/报警码表。
           更新 = 换厂商资料或现场反馈修正时才动, 变更需过审 (它是判据的地基)。

每个源声明: 层级 / 当前覆盖 / 末次更新 / 约定位置 / 摄入命令 / 更新频率 / 责任方。
跑法: .venv/bin/python -m src.ontology.maintenance [--json]
"""
from src import paths as P
import re
import json, os, pathlib, sys, datetime as _dt

ROOT = P.ROOT                     # 2026-09-16 统一: 原为 `parents[2]` 自推一份 (与本模块的 P.ROOT 同值,
                                  # 但属"影子真源"; 同一文件里还并存自写的 disp()/_RAW_STR —— 一并归到 src/paths.py)
ST = P.store()
ONT = P.ont()
CMS = P.cms()
# 原始件目录: 唯一真源就是 src/paths.py 的 RAW_ROOT (env WINDSCADA_RUDONG_SRC / configs/serve.json
# 的 raw_dir 最终都经它生效)。2026-09-08 xzy 测试逮: 原为开发机绝对路径 /Users/yuanying/rudong/…,
# 在 Windows 上既非法也不存在, 页面照着抄必然补不了数据。
RAW = P.RAW_ROOT
# ★2026-09-11 用户令 A2 + 场站扫描辨识: 数据层源在 data/raw/<场站名称>/ 下。场站目录名不再写死在
#   代码里 —— config.farm() 扫 data/raw 的下一级目录辨识 (辨识依据见 station_note), 这里取结果用。
try:
    from src.windscada.config import farm as _farm, raw_station_dir as _station_dir
    _CFG = _farm()
    STATION = _station_dir()
    _SCAN_NOTE = str(_CFG.get('station_note') or '')
    _N_STATIONS = int(_CFG.get('n_stations') or 0)
except Exception as _e:                                    # 配置读不了也别让整页崩
    STATION = RAW / '如东'
    _SCAN_NOTE = f'场配置/扫描不可用: {type(_e).__name__}: {_e}'
    _N_STATIONS = 0
# 机理层: 技术资料仍在原始件根下 (A2 只点名数据层那四项, 机理层不动)
TECH = RAW / '西门子4.0技术资料'


def disp(p) -> str:
    """给人看的路径 (安装目录内 → `<安装目录>/…`) —— 2026-09-16 起直接转发 src/paths.py 的同名函数。

    原来本模块自己写了一份 (语义相同但少一步 resolve), 与 P.disp 并存 = 同一份"显示规则"两个实现;
    2026-09-09 用户看页面反馈: 直接摆一串本机绝对路径, 客户读不出该往哪放。"""
    return P.disp(p)


def _mtime(p):
    p = pathlib.Path(p)
    return _dt.datetime.fromtimestamp(p.stat().st_mtime).strftime('%Y-%m-%d %H:%M') if p.exists() else None


def _span(parquet, col):
    try:
        import pandas as pd
        d = pd.read_parquet(parquet)
        v = d[col].dropna()
        if hasattr(v, 'dt'):
            v = v[v.dt.year > 2000]
        return f'{v.min()} ~ {v.max()}'.replace(' 00:00:00', ''), len(d)
    except Exception as e:
        return f'读取失败 {str(e)[:30]}', 0


SOURCES = [
    dict(层='机理', 名='决策链进度', 产物=ONT / 'objects.json', 时间列=None,
         位置='windscada 链盘 (运行期计算, 经 /api/fleet 取)',
         摄入='python -m src.ontology.chain_ingest   (需 windscada 服务在跑)',
         频率='判级或工单更新后', 责任='windscada → 图谱',
         说明='每台六步进度/停滞环节/下一步 + 全场瓶颈汇总; 补前只在运行期内存里, 图谱检索不到'),
    dict(层='机理', 名='趋势与闭环证据', 产物=ONT / 'objects.json', 时间列=None,
         位置=disp(P.m5() / 'component_history.json') + ' (振动线出件)',
         摄入='python -m src.ontology.trend_ingest',
         频率='振动线每轮出件后', 责任='振动线 → 图谱',
         说明='在升(末三窗/早三窗≥1.3)与换件闭环两类实证入图谱; 补前这类信息只在链盘里, 图谱检索不到'),
    # ── 数据层 (2026-09-11 A2: 四项数据源目录 = data/raw/<场站名称>/{scada_10min,故障报警,风机故障记录,油样报告}) ──
    # 场站目录由 config 扫 data/raw 的下一级目录辨识 (用户令 2026-09-11); 这一行专门把它摆出来 ——
    # 扫不到场站目录时本行 存在=False/覆盖=—, 与下面四行的"无数据"一起构成场景① 的可视证据。
    dict(层='数据', 名='场站数据目录 (扫描辨识)', 产物=STATION, 时间列=None,
         位置=P.disp_dir(RAW) + '<场站名称>' + P.SEP,
         摄入='python scripts/place_raw_data.py  (现场压缩包按约定落位)',
         频率='随现场', 责任='现场/数据线',
         说明=f'data/raw 的下一级目录即场站名称, 取数时扫描辨识; 约定子目录: '
              f'{" / ".join(("scada_10min", "故障报警", "风机故障记录", "油样报告"))}。'
              f'{_SCAN_NOTE}'),
    dict(层='数据', 名='SCADA 10min 月度衍生', 产物=ST / 'loss_monthly.parquet', 时间列='month',
         位置=P.disp_dir(STATION / 'scada_10min'),
         摄入='python scripts/rebuild_from_raw.py --scada  (从 scada_10min/*.csv 重算 loss_monthly 等)',
         频率='月', 责任='数据线',
         说明='逐台平铺 WTG01.csv…WTG38.csv (放目录本身, 不是压缩包); '
              '五态/损失/可用率的底座; 换月必重跑下游判级'),
    dict(层='数据', 名='报警事件', 产物=ST / 'alarms.parquet', 时间列='t_on',
         位置=P.disp_dir(STATION / '故障报警'),
         摄入='python scripts/windscada_alarms_ingest.py',
         频率='月', 责任='数据线',
         说明='码级事件; 注意年度总表 .xls 实为 XML 报警导出非工单; '
              '目录里的「全年/年至今」件是季度件的累计快照, 摄入时会自动跳过 (叠加即翻倍)'),
    dict(层='数据', 名='检修工单台账', 产物=ST / 'workorders.parquet', 时间列='t_report',
         位置=P.disp_dir(STATION / '风机故障记录'),
         摄入='python scripts/windscada_workorder_ingest.py',
         频率='季/年', 责任='现场提供',
         说明='目录内按 {年}年故障记录/ 分年 (2026-09-11 起连同 74 张月/年表一起摄入, '
              '覆盖已推至 2026-07; 跨表内容重复自动归并, 无日期的计划停机行保留)'),
    dict(层='数据', 名='油液化验', 产物=ST / 'oil_samples_index.parquet', 时间列='date',
         位置=P.disp_dir(STATION / '油样报告'),
         摄入='python scripts/windscada_watch_channels_build.py  (产 oil_samples_index)',
         频率='半年', 责任='化验机构',
         说明='SGS + 华标两家, 报告按台号/部件分目录; 文件名自带日期/台号/部件/sample_id; '
              '新一轮到货后重跑, 时效胶囊自动翻绿'),
    dict(层='数据', 名='CMS 振动评估报告', 产物=CMS / '报告_CMS振动状态评估报告_*.md', 时间列=None,
         位置=P.disp_dir(STATION / 'windcms'),
         摄入='python scripts/vib_raw_build.py  (索引→谱; 重生成 CMS 报告用 --with-report, 见 docs 振动数据接入 §3b)',
         频率='月/事件驱动', 责任='振动线',
         说明='本目录两类源件: ①CMS 原始测量导出 (Brande TCM *_decode.json → 索引/谱/六层链, 可重算); '
              '②厂商月度评估报告 (上海电气 12 份用印版 PDF 实测为扫描件无文本层 ⇒ 只作归档证据, 不进判级)。'
              '设备状态五级来源; 观澜自动取最新版不写死日期'),
    dict(层='数据', 名='振动线 handoff', 产物=P.m5() / 'handoff_vibration_v2.json', 时间列=None,
         位置=P.disp_dir(STATION / 'm5_cms_tcm'),
         摄入='振动线出件 → fusion 自动读 (现场正本落本目录则优先采用)',
         频率='事件驱动', 责任='振动线',
         说明='定谳链判级 (候选以上); 与 CMS 报告是两条轴, 取严合并。本目录放 TCM 侧深度分析报告 '
              '(大生科技) 与现场给的接口正本 (handoff_vibration_v2.json / component_history.json); '
              '包内那份 json 目前是 shipped 快照 (振动线分支的产物未随包), 现场给正本后即改由现场件驱动'),
    # ── 机理层 ──
    dict(层='机理', 名='报警码表与处置手册', 产物=ONT / 'objects.json', 时间列=None,
         位置=disp(TECH / '故障处理/故障处理手册.xlsx'), 摄入='python -m src.ontology.kb_ingest',
         频率='换厂商资料时', 责任='厂商/技术', 说明='555 码 + 331 分步处置 WI; 码义变更须过审 (码义靠猜会翻案)'),
    dict(层='机理', 名='预防性维护 WI', 产物=ONT / 'objects.json', 时间列=None,
         位置=disp(TECH / '维护相关/维护作业指导书.xlsx'), 摄入='python -m src.ontology.kb_ingest',
         频率='换版本时', 责任='厂商/技术', 说明='228 项年度任务, 带位置代号与工时'),
    dict(层='机理', 名='实机参数表 (判定规范)', 产物=ONT / 'turbine_params.parquet', 时间列=None,
         位置=disp(TECH / 'Turbine+Parameters.*.xlsx'), 摄入='python -c "from src.ontology.maintenance import refresh_params as f; print(f())"',
         频率='参数变更时', 责任='现场/厂商', 说明='1706 条整定值; ★现导出为 2024-10-21, 若现场改过定值需重导'),
    dict(层='机理', 名='技术资料索引 (图纸/原理)', 产物=ONT / 'objects.json', 时间列=None,
         位置=disp(TECH), 摄入='python -m src.ontology.kb_ingest',
         频率='补充资料时', 责任='技术', 说明='309 份; 只索引文件不解析内容, 新增资料放入目录后重跑即可'),
    dict(层='机理', 名='失效模式与原因树', 产物=ONT / 'objects.json', 时间列=None,
         位置='(派生自码表+工单文本归纳)', 摄入='python -m src.ontology.kb_ingest',
         频率='随码表/工单更新', 责任='本体线', 说明='108 个模式 + 43 条 causedBy; 归纳规则改动 = 判据地基改动, 须过审'),
]


def refresh_params(src=None):
    """重导实机参数表 → turbine_params.parquet (该 parquet 被 gitignore, 换定值后须能一键复现).
    src 缺省取技术资料目录下最新的 Turbine+Parameters.*.xlsx; 表头行自动定位 (原始文件前 6 行是页眉)."""
    import pandas as pd
    f = pathlib.Path(src) if src else None
    if f is None:
        cands = sorted(TECH.glob('Turbine+Parameters*.xlsx'))
        if not cands:
            raise FileNotFoundError(f'未找到参数表: {TECH}/Turbine+Parameters*.xlsx — 判定规范无源, 不静默降级')
        f = cands[-1]
    for hdr in range(3, 14):
        d = pd.read_excel(f, sheet_name='Main', header=hdr)
        if {'Number', 'Text', 'Min', 'Max'} <= set(str(c) for c in d.columns):
            break
    else:
        raise ValueError(f'{f.name}: 未定位到表头行 (期望含 Number/Text/Min/Max)')
    d = d.dropna(subset=['Number', 'Text'])
    d['Number'] = d['Number'].astype(str).str.replace(r'\.0$', '', regex=True)
    out = ONT / 'turbine_params.parquet'
    d.to_parquet(out, index=False)
    return dict(src=str(f), rows=len(d), out=str(out))


def survey():
    rows = []
    for s in SOURCES:
        prod = s['产物']
        if '*' in str(prod):
            fs = sorted(pathlib.Path(str(prod)).parent.glob(pathlib.Path(str(prod)).name))
            prod = fs[-1] if fs else None
        cov, n = ('—', 0)
        if prod and pathlib.Path(prod).exists():
            if s['时间列']:
                cov, n = _span(prod, s['时间列'])
            else:
                cov = pathlib.Path(prod).name
        # 命令可用性自检 (2026-08-28 功能测试逮): 维护端口声明的摄入命令曾有 3 条指向不存在的脚本
        # (windscada_build.py / windscada_alarm_ingest.py 少个 s / windscada_oil_ingest.py),
        # 演示时客户照着敲会直接报错。声明与实现必须对得上, 这一项跟着 survey 一起报。
        _m = re.search(r'scripts/[\w/]+\.py', s['摄入']) or re.search(r'-m\s+([\w.]+)', s['摄入'])
        _cmd_ok = None
        if _m:
            _t = _m.group(0) if _m.group(0).endswith('.py') else _m.group(1).replace('.', '/') + '.py'
            _t = _t.replace('-m ', '')
            _cmd_ok = (P.ROOT / _t.strip()).exists() or (P.SRC / _t.strip()).exists()
        rows.append(dict(层=s['层'], 名=s['名'], 覆盖=cov, 条数=n, 命令可用=_cmd_ok,
                         末次更新=_mtime(prod) if prod else None,
                         存在=bool(prod and pathlib.Path(prod).exists()),
                         位置=s['位置'], 摄入=s['摄入'], 频率=s['频率'], 责任=s['责任'], 说明=s['说明']))
    return rows


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
    if '--refresh-params' in sys.argv:
        print(refresh_params()); sys.exit(0)
    rs = survey()
    if '--json' in sys.argv:
        print(json.dumps(rs, ensure_ascii=False, indent=1))
    else:
        for layer in ('数据', '机理'):
            print(f'\n【{layer}层】')
            for r in [x for x in rs if x['层'] == layer]:
                mark = '✔' if r['存在'] else '✗'
                print(f"  {mark} {r['名']:22s} 覆盖 {str(r['覆盖'])[:34]:36s} 末次 {r['末次更新'] or '—'}")
                print(f"      频率 {r['频率']:8s} 责任 {r['责任']:8s} 摄入: {r['摄入']}")
                print(f"      {r['说明']}")