guanlan-src/app_algorithmModel/app_algorithmModel_guanlan/service/fleet_views.py

# -*- coding: utf-8 -*-
r"""整场取数视图（P12）：`fleet_view` 及其**传递闭包**内的全部本地函数与模块级全局，机械搬到算法服务。

★来源：`app_backEnd/app_backEnd_guanlan/serve.py`（原样搬运，未改写逻辑）。
搬运范围由 AST 计算（种子 `fleet_view` → 递归展开被调本地函数 → 带上被引用的模块级全局），避免漏项。
"""
from __future__ import annotations

from pathlib import Path as _P
import sys as _sys, pathlib as _plb
from src import paths as P
import os as _os

import collections
import datetime as _dt
import json
import pathlib
import re
import threading
import time
from typing import Any

from app_common.app_common_guanlan.api import paths as _P
from src.windscada.config import farm

from src import paths as _P                                              # 路径唯一真源 (与 cwd 无关)
from src.windscada import i18n
from src.windscada import taxonomy
from src.windscada.subsys import temp_nbm, hydraulic, yaw as yawmod, pitch as pitchmod
import hashlib, time
import numpy as np, pandas as pd
import sys, json, pathlib, re, threading, urllib.parse, collections
import threading as _thr

# ── 原 serve.py 的模块级全局（逐字搬） ──
CFG = farm(); ST = pathlib.Path(CFG['store'])

LOCK = threading.Lock()

CFG = farm(); ST = pathlib.Path(CFG['store'])

TS = {'报警': 0, '危险': 0, '良好': 1, '不可判': 2, '优秀': 3}

_CACHE = {}

_HEAVY_LOCK = _thr.Lock()

_STAMP = {'val': None, 'at': 0.0}

_STAMP_TTL = 2.0          # 秒: 指纹有效期 (stat ~150 个文件 ≈ 1~3 ms, 不值得每请求都做)

_WIN_BUSY: dict = {'sysmx': set(), 'curves': set(), 'm9': set()}

_WIN_CACHE: dict = {'sysmx': {}, 'curves': {}, 'm9': {}}

_WIN_ERR: dict = {'sysmx': {}, 'curves': {}, 'm9': {}}

_WIN_LOCK = _thr.Lock()


def _product_files():
    """页面取数依赖的产物文件清单 (顺序稳定: 供指纹; 只列**产物**, 不含 reference/ 随包契约)。

    覆盖 _load() 直读的件 + 它经 taxonomy/temp_nbm/hydraulic/fusion 间接读的件:
      windscada/*.parquet|csv|json  · ontology/*.json  · pitch/*.parquet
      m5_cms_tcm/{handoff_vibration_v2,component_history,baseline_38}.json  · windcms/报告_CMS*.md
    """
    try:
        from src.windcms.config import cms_out as _cms_out     # 与写侧同一解析口 (WINDCMS_OUT)
        _cms_dir = _cms_out()
    except Exception:
        _cms_dir = _P.cms()
    pats = ((ST, ('*.parquet', '*.csv', '*.json')),
            (_P.ont(), ('*.json',)),
            (_P.pitch(), ('*.parquet',)),
            (_P.m5(), ('handoff_vibration_v2.json', 'component_history.json', 'baseline_38.json')),
            (_cms_dir, ('报告_CMS振动状态评估报告_*.md',)))
    files = []
    for d, ps in pats:
        for pat in ps:
            files.extend(sorted(d.glob(pat)))
    return files


def products_stamp(force=False):
    """产物指纹 (sha1 前 16 位)。force=True 时忽略 TTL 立即重算。"""
    now = time.time()
    if not force and _STAMP['val'] is not None and (now - _STAMP['at']) < _STAMP_TTL:
        return _STAMP['val']
    h = hashlib.sha1()
    for p in _product_files():
        try:
            st = p.stat()
            h.update(f'{_P.rel(p)}|{st.st_mtime_ns}|{st.st_size}\n'.encode('utf-8'))
        except OSError:
            h.update(f'{_P.rel(p)}|MISSING\n'.encode('utf-8'))
    _STAMP.update(val=h.hexdigest()[:16], at=now)
    return _STAMP['val']


def _load():
    with LOCK:
        stamp = products_stamp()
        if _CACHE.get('__loaded'):
            if _CACHE.get('__stamp') == stamp:
                return
            # 产物变了 (典型: 刚跑完重算) → 重载。**先建后换**: 下面任何一步抛异常都不会破坏旧缓存,
            # 请求照旧能用旧数 (降级但不空白), 同时日志留痕。
            print(f'[reload] 产物指纹变化 {_CACHE.get("__stamp")} → {stamp}, 重载', flush=True)
        tmp = {}
        try:
            for key, name in (('tm', 'temp_monthly.parquet'), ('al', 'alarms.parquet'),
                              ('lm', 'loss_monthly.parquet'), ('bins', 'powercurve_bins.parquet'),
                              ('pcd', 'powercurve_dev.parquet')):
                f = ST / name
                if not f.exists():
                    raise ProductsMissing(name, f)
                tmp[key] = pd.read_parquet(f)
            tmp['al']['month'] = tmp['al']['t_on'].dt.to_period('M').astype(str)
            tmp['pcd'] = tmp['pcd'].set_index('turbine')
            for key, name in (('duty', 'duty_monthly.parquet'),):
                f = ST / name
                tmp[key] = pd.read_parquet(f) if f.exists() else None
            try:
                tmp['sysmx'] = taxonomy.system_matrix()
                tmp['treg'] = temp_nbm.registry()
                tmp['treg'] = tmp['treg'][0] if isinstance(tmp['treg'], tuple) else tmp['treg']
                tmp['hyd'], _ = hydraulic.registry()
                tmp['hyd'] = tmp['hyd'].set_index('turbine')
            except FileNotFoundError as e:            # 这些派生件同样在产物仓里; 缺了就按"无产物"处理
                raise ProductsMissing(getattr(e, 'filename', '派生产物'), getattr(e, 'filename', ST))
            zp = _P.pitch() / 'pitch_zero_monthly.parquet'
            tmp['zero'] = pd.read_parquet(zp) if zp.exists() else None
            tmp['__stamp'] = stamp
            tmp['__loaded'] = True
            _CACHE.update(tmp)                        # 原子提交: 失败时不留下半截缓存
        except Exception:
            if _CACHE.get('__loaded'):
                print('[reload] 重载失败 → 继续用上一份缓存 (页面不会空白, 但数是旧的; 看上面的异常)', flush=True)
            raise


def _month_end(m):
    """'YYYY-MM' → 该月最后一天 'YYYY-MM-DD'。"""
    import calendar
    y, mo = int(m[:4]), int(m[-2:])
    return f'{m}-{calendar.monthrange(y, mo)[1]:02d}'


def months_of(win):
    """窗 → 覆盖到的月份列表（月度类数据按此过滤）。"""
    _load()
    all_m = sorted(_CACHE['tm'].month.unique())
    if win == '全程': return all_m
    if win == '2025H2': return [m for m in all_m if '2025-07' <= m <= '2025-12']
    if win == '2026H1': return [m for m in all_m if '2026-01' <= m <= '2026-06']
    if win == '2026年': return [m for m in all_m if m >= '2026-01']
    import re as _re
    if _re.fullmatch(r'\d{4}-\d{2}', win):        # 单月窗 (2026-08-27 用户令: 按月份过滤和选择)
        return [m for m in all_m if m == win]
    m2 = _re.fullmatch(r'(\d{4}-\d{2})~(\d{4}-\d{2})', win)
    if m2:                                          # 月区间窗 "2025-04~2025-10"
        return [m for m in all_m if m2.group(1) <= m <= m2.group(2)]
    m3 = _re.fullmatch(r'(\d{4}-\d{2}-\d{2})~(\d{4}-\d{2}-\d{2})', win)
    if m3:                                          # ★自定义起止（含）→ 相交的月 (用户令 2026-09-21)
        a, b = sorted((m3.group(1), m3.group(2)))
        return [m for m in all_m if _month_end(m) >= a and f'{m}-01' <= b]
    n = 1 if win == '近30日' else 3
    return all_m[-n:]


def win_range(win):
    """窗 → (起, 止) 日期串，**含两端**（用户令 2026-09-21：时间窗支持自定义起止日期）。

    预设/逐月/月区间窗一律落到"该时间窗覆盖月的月首/月末"，于是与 `months_of()` 同口径；
    日区间窗原样返回。日粒度件（停机事件/报警/日粒度判据）用它做**含端**过滤。
    """
    import re as _re
    if win:
        if _re.fullmatch(r'\d{4}-\d{2}-\d{2}~\d{4}-\d{2}-\d{2}', win):
            a, b = win.split('~')
            return (a, b) if a <= b else (b, a)
        if _re.fullmatch(r'\d{4}-\d{2}', win):
            return (f'{win}-01', _month_end(win))
        m2 = _re.fullmatch(r'(\d{4}-\d{2})~(\d{4}-\d{2})', win)
        if m2:
            return (f'{m2.group(1)}-01', _month_end(m2.group(2)))
        if win == '2025H2':
            return ('2025-07-01', '2025-12-31')
        if win == '2026H1':
            return ('2026-01-01', '2026-06-30')
    ms = months_of(win)
    if not ms:
        return ('0001-01-01', '9999-12-31')
    return (f'{ms[0]}-01', _month_end(ms[-1]))


def span_of(win):
    """窗 → 计算层用的 (起, 止) 元组（含）；判级/曲线等按窗重算的口子都吃这个。"""
    return win_range(win)


def _mem_mb():
    """本进程可用内存（MB）；拿不到就返回 None（不因为这些诊断信息把重算搞挂）。"""
    try:
        import ctypes

        class _MS(ctypes.Structure):
            _fields_ = [('dwLength', ctypes.c_ulong), ('dwMemoryLoad', ctypes.c_ulong),
                        ('ullTotalPhys', ctypes.c_ulonglong), ('ullAvailPhys', ctypes.c_ulonglong),
                        ('ullTotalPageFile', ctypes.c_ulonglong), ('ullAvailPageFile', ctypes.c_ulonglong),
                        ('ullTotalVirtual', ctypes.c_ulonglong), ('ullAvailVirtual', ctypes.c_ulonglong),
                        ('ullAvailExtendedVirtual', ctypes.c_ulonglong)]
        st = _MS()
        st.dwLength = ctypes.sizeof(_MS)
        ctypes.windll.kernel32.GlobalMemoryStatusEx(ctypes.byref(st))
        return int(st.ullAvailPhys // (1024 * 1024))
    except Exception:
        return None


def _win_key(win):
    a, b = span_of(win)
    return f'{a}~{b}'


def _win_get(kind, win, fn):
    """→ 值 或 None（None = 正在算/刚起算）。失败**记名**（`_WIN_ERR`）而不是静默当"没数据"。"""
    key = _win_key(win)
    with _WIN_LOCK:
        if key in _WIN_CACHE[kind]:
            return _WIN_CACHE[kind][key]
        if key not in _WIN_BUSY[kind]:
            _WIN_BUSY[kind].add(key)

            def _run():
                try:
                    with _HEAVY_LOCK:                       # 重活串行: 同一时刻只算一份
                        m0 = _mem_mb()
                        print(f'[win] {kind} 按窗重算开算 {key}（可用内存 {m0} MB）', flush=True)
                        v = fn()
                        print(f'[win] {kind} 按窗重算完成 {key}（可用内存 {_mem_mb()} MB）', flush=True)
                except Exception as e:                      # 守护失败必响亮
                    _WIN_ERR[kind][key] = f'{type(e).__name__}: {e}'[:200]
                    print(f'[win] {kind} 按窗重算失败 {key}: {_WIN_ERR[kind][key]}', flush=True)
                    v = None
                with _WIN_LOCK:
                    if v is not None:
                        _WIN_CACHE[kind][key] = v
                    _WIN_BUSY[kind].discard(key)
            _thr.Thread(target=_run, name=f'win-{kind}-{key}', daemon=True).start()
            print(f'[win] {kind} 按窗重算启动 {key}（首次约数十秒，页面会先出 pending）', flush=True)
        return None


def sysmx_of(win):
    """→ (判级矩阵, pending)。`pending=True` 时返回的是"另一口径"的旧矩阵，页面必须如实标注。"""
    from src.windscada import taxonomy
    m = _win_get('sysmx', win, lambda: taxonomy.system_matrix(CFG, span=span_of(win)))
    if m is None:
        return _CACHE['sysmx'], True
    return m, False


def m9_of(win):
    """→ (控制参数一致性, pending)。按窗重算（走窄仓 ≈5 s/窗）；干净窗直接用正式产物。"""
    from src.windscada.perf import control as _cm
    a, b = span_of(win)
    if (a, b) == (_cm.WIN[0], '2025-12-31'):
        return _cm.registry(CFG), False
    v = _win_get('m9', win, lambda: _cm.registry(CFG, span=(a, b)))
    if v is None:
        return _cm.registry(CFG), True          # 未就位: 先给上一口径, 页面按 pending 标注
    return v, False


def fleet_view(win):
    _load()
    ms = months_of(win)
    # ★2026-09-21 用户令「判级也按所选时间窗重算」: 判级矩阵按窗算（后台+缓存，首次 pending）。
    sysmx, sysmx_pending = sysmx_of(win)
    # ① 系统分类问题 (全系统)
    systems, sysdist = {}, {}
    for s in taxonomy.SYSTEMS:
        rows = [dict(t=t, st=sysmx[t][s]['状态'], why=i18n.humanize(sysmx[t][s]['依据']))
                for t in CFG['turbines'] if sysmx[t][s]['状态'] in ('报警', '不可判')]   # 良好不出 (用户令: 只显示有问题的)
        rows.sort(key=lambda r: TS.get(r['st'], 9))
        systems[s] = rows
        # 全场分布 (系统入口卡用: rows 只含问题台, 不能当全场分母 — 会显示"共12台")
        from collections import Counter as _C
        sysdist[s] = dict(_C(sysmx[t][s]['状态'] for t in CFG['turbines']))
        # 良好台清单单列 (系统详情页第三档)
        systems[s + '·良好'] = [dict(t=t, st='良好', why=i18n.humanize(sysmx[t][s]['依据']))
                                for t in CFG['turbines'] if sysmx[t][s]['状态'] == '良好']
    # ①b 关注清单: ≥2 系统报警 (逐台独立台页)
    watch = []
    for t in CFG['turbines']:
        al_sys = [x for x in taxonomy.SYSTEMS if sysmx[t][x]['状态'] == '报警']
        if len(al_sys) >= 2:
            watch.append(dict(t=t, n=len(al_sys), syss=al_sys,
                              # ★不截断: 截到 70 字会把依据切成半句, 而模板反解要完整形态
                              #   ⇒ 英文侧整条落回中文 (2026-09-03 实逮, 总览 7 卡 4 张中文)。
                              #   显示长度归前端 CSS 管。
                              why='；'.join(f"{x}: {i18n.humanize(sysmx[t][x]['依据'])}" for x in al_sys)))
    watch.sort(key=lambda r: -r['n'])
    n_alarm_t = sum(1 for t in CFG['turbines'] if any(sysmx[t][x]['状态'] == '报警' for x in taxonomy.SYSTEMS))
    # ② 故障统计梳理 (真时间窗)
    al = _CACHE['al']; a = al[al.month.isin(ms)]
    top_txt = a.groupby(['code', 'text']).agg(n=('code', 'size'), dur_h=('dur_s', lambda x: x.sum() / 3600)).reset_index()
    pareto_n = top_txt.sort_values('n', ascending=False).head(12)
    pareto_d = top_txt.sort_values('dur_h', ascending=False).head(12)
    monthly = a.groupby('month').size().reindex(ms).fillna(0)
    per_t = a.groupby('turbine').size().sort_values(ascending=False).head(10)
    # 经典故障分析: 条数与时长必须配对看 (2026-08-28)。两个独立排行榜分不出
    # "高频短时(信号抖动或重复触发)" 与 "低频长时(硬故障)" — 处置方向相反, 混在一起会误派工。
    # 第三维=影响台数, 分离"单台刷屏"与"全场批次共性"。
    # 残月识别 (2026-08-28 审核逮): 2026-07 只有 6 天数据(覆盖 19.4%), 却被当整月
    # 并进"2026年"的占比与月度趋势 → 末柱"下降"是假象。占比类分母也因此不完整。
    import calendar as _cal
    _lmc = _CACHE['lm']; _lmc = _lmc[_lmc.month.astype(str).isin(ms)]
    _cov = {}
    for _mm in ms:
        _sub = _lmc[_lmc.month.astype(str) == _mm]
        if not len(_sub):
            _cov[_mm] = 0.0; continue
        _y, _mo = int(_mm[:4]), int(_mm[-2:])
        _cal_h = _cal.monthrange(_y, _mo)[1] * 24
        _cov[_mm] = round(float(_sub['rows_'].sum() / 6 / max(_sub['turbine'].nunique(), 1) / _cal_h), 3)
    _nt = a.groupby(['code', 'text'])['turbine'].nunique().rename('nt')
    _med = a.groupby(['code', 'text'])['dur_s'].median().rename('med_s')
    qd = top_txt.set_index(['code', 'text']).join([_nt, _med]).reset_index()
    qd = qd.sort_values('n', ascending=False).head(30)
    faults = dict(
        pareto_n=[dict(k=f"{r.code} {i18n.alarm_label(r.code, r.text)[:18]}", v=int(r.n)) for _, r in pareto_n.iterrows()],
        pareto_d=[dict(k=f"{r.code} {i18n.alarm_label(r.code, r.text)[:18]}", v=round(float(r.dur_h), 1)) for _, r in pareto_d.iterrows()],
        monthly=dict(months=ms, vals=[int(v) for v in monthly],
                     cov=[_cov.get(x, 0.0) for x in ms],
                     per_day=[round(float(monthly[x]) / max(_cov.get(x, 0.0) * _cal.monthrange(int(x[:4]), int(x[-2:]))[1], 1e-9), 1)
                              if _cov.get(x, 0) > 0.02 else None for x in ms]),
        cov=_cov, cov_min=min(_cov.values()) if _cov else 1.0,
        partial=[x for x in ms if _cov.get(x, 1) < 0.5],
        quad=[dict(code=str(r.code), k=i18n.alarm_label(r.code, r.text)[:20], n=int(r.n),
                   h=round(float(r.dur_h), 1), nt=int(r.nt), med=round(float(r.med_s), 1))
              for _, r in qd.iterrows()],
        n_codes=int(len(top_txt)),
        per_t=[dict(k=k, v=int(v)) for k, v in per_t.items()], total=int(len(a)))
    # ③ SOP 控制策略
    lm = _CACHE['lm']; l = lm[lm.month.astype(str).isin(ms)]
    hrs = l.groupby('state')['rows_'].sum() / 6
    loss = l.groupby('state')['loss'].sum() / 1000
    hrs_tot = max(hrs.sum(), 1)
    stop_h = hrs.get('停机', 0) + hrs.get('停机(调度令)', 0)
    disp_h = hrs.get('停机(调度令)', 0)
    lw_h = hrs.get('低风待机', 0)
    # 口径与 availability.summary 单源一致: 1 − 停机/(总 − 调度令 − 低风待机)
    avail = 1 - (stop_h - disp_h) / max(hrs_tot - disp_h - lw_h, 1)
    pcd = _CACHE['pcd']
    # 能量账闭合 (2026-08-28 经典分析): 四行工况表看不出"电量流向"。
    # 理论可发 = 实际上网 + Σ各态损失; 过物理上限核 (不可超 38台×4.0MW×窗时长)。
    _act = l.groupby('state')['act'].sum() / 1000
    _wf = [dict(k=str(k), loss=round(float(loss.get(k, 0)), 0), act=round(float(_act.get(k, 0)), 0),
                h=round(float(hrs.get(k, 0)), 0)) for k in hrs.index]
    _A = float(_act.sum()); _L = float(loss.sum())
    _cap = len(CFG['turbines']) * 4.0 * float(hrs_tot) / max(len(CFG['turbines']), 1)
    # 逐月能量 (2026-08-28 报表端口): 汇报要环比与趋势, 全时间窗聚合给不出。
    # 残月覆盖率一并带出 — 报表里拿残月和整月比环比会读反 (故障月度图已踩过一次)。
    _em = []
    for _mm in ms:
        _sub = l[l.month.astype(str) == _mm]
        if not len(_sub):
            continue
        _a1 = float(_sub['act'].sum()) / 1000
        _l1 = float(_sub['loss'].sum()) / 1000
        _em.append(dict(m=_mm, act=round(_a1, 0), loss=round(_l1, 0), theo=round(_a1 + _l1, 0),
                        loss_pct=round(_l1 / max(_a1 + _l1, 1) * 100, 1),
                        cov=_cov.get(_mm, 1.0),
                        eflh=round(_a1 / max(len(CFG['turbines']) * 4.0, 1), 0)))
    energy = dict(monthly=_em, act=round(_A, 0), loss=round(_L, 0), theo=round(_A + _L, 0),
                  loss_pct=round(_L / max(_A + _L, 1) * 100, 1),
                  eflh=round(_A / max(len(CFG['turbines']) * 4.0, 1), 0),
                  cap=round(_cap, 0), cap_ok=bool(_A + _L < _cap),
                  items=sorted(_wf, key=lambda r: -r['loss']))
    control = dict(
        energy=energy,
        states=[dict(k=k, h=round(float(v), 0), pct=round(float(v / hrs_tot * 100), 1), loss=round(float(loss.get(k, 0)), 0)) for k, v in hrs.items()],
        avail=round(float(avail * 100), 1),
        curve=dict(sigma=round(float(pcd.dev_w.std() * 100), 2),
                   cands=[dict(t=i, dev=round(float(r.dev_w * 100), 2), 判=r['判别']) for i, r in pcd.iterrows() if r['判别'] != '—'],
                   note='固定判别窗 2025-H2'),
        anchors='限电命令面占比 2025≈0.197 → 2026≈0.769 (专项锚)')
    m8 = None
    try:
        from src.windscada.perf import faults as fmod
        if (ST / 'stop_events.parquet').exists():
            m8 = dict(mtbf=fmod.mtbf_summary(ms), stop_pareto=fmod.stop_pareto(ms), seasonal=fmod.seasonal())
    except Exception as e:
        m8 = dict(err=str(e)[:120])
    m9 = None
    try:
        from src.windscada.perf import control as cmod
        if (ST / 'control_profile.parquet').exists():
            m9, m9_pending = m9_of(win)          # ★按窗重算（用户令 2026-09-21）
    except Exception as e:
        m9 = dict(err=str(e)[:120])
        m9_pending = False
    try:
        from src.windscada.perf import reliability as relmod
        rel = relmod.overview(CFG, span=span_of(win))
    except Exception as e:
        rel = dict(err=str(e)[:120])
    try:
        from src.windscada.subsys import fusion as fusmod
        ftab, fmeta = fusmod.fusion_table(CFG, all_turbines=True)
        mrows, mheads = fusmod.matrix(CFG, ms)   # 整体时间窗: 矩阵现窗/记忆轨迹随全局窗选择器
        cap = fusmod.capability()
        ev, ev_bounds = fusmod.drivetrain_events(ms, CFG)
        lvcnt = {}
        for r in mrows:
            lv = r['振动']['level']; lvcnt[lv] = lvcnt.get(lv, 0) + 1
        fkpi = dict(定论=lvcnt.get('bad', 0), 预警=lvcnt.get('warn', 0), 候选观察=lvcnt.get('note', 0),
                    销案正常=lvcnt.get('ok', 0), 未列=lvcnt.get('unlisted', 0),
                    收录台=len(CFG['turbines']) - lvcnt.get('unlisted', 0), 全场=len(CFG['turbines']),
                    机制链=sum(1 for r in mrows if '机制链' in r['润滑'].get('note', '')),
                    系数=dict(wear=cap['classes'].get('wear_progressive', {}).get('coef'),
                            thermal=cap['classes'].get('thermal_acute', {}).get('coef'),
                            unknown=cap['classes'].get('unknown', {}).get('coef')))
        # 矩阵补 报警/工单 两列 (2026-08-28 用户令) — 报警=时间窗内真窗计数+族分布; 工单=历史台账(窗见列头)
        # 报警/工单两列只统计传动链三部件 (2026-08-28 用户令: 只说发电机/齿轮箱/主轴) —
        # 本矩阵每行是"该台传动链状态", 全场报警计数会把变桨/偏航/主控的量混进来 (316 条里绝大多数与传动链无关)
        _DT_PAT = '齿轮|齿箱|润滑油|油冷|滤芯|发电机|定子|绕组|滑环|主轴承|主轴|轴承'
        _a = _CACHE['al']; _aw = _a[_a.month.isin(ms)]
        _aw = _aw[_aw.text.str.contains(_DT_PAT, na=False)]
        _acnt = _aw.groupby('turbine').size()
        _arank = _acnt.rank(ascending=False, method='min')
        _FAM = [('齿轮箱', '齿轮|齿箱|润滑油|油冷|滤芯'), ('发电机', '发电机|定子|绕组|滑环'),
                ('主轴承', '主轴承|主轴')]
        try:
            from src.windscada.subsys import workorder as _wo
            _wod = _wo.load(CFG)
        except Exception:
            _wod = None
        for _r in mrows:
            _t = _r['turbine']; _g = _aw[_aw.turbine == _t]
            _fam, _taken = [], set()
            _CODEFAM = {'3225': '变桨叶片'}       # 文本为空的码走码号归族 (3225=变桨液压, 现场西门子资料对译表)
            for _nm, _pat in _FAM:
                _byc = _g.code.astype(str).map(_CODEFAM) == _nm
                _sub = _g[~_g.index.isin(_taken) & (_g.text.str.contains(_pat, na=False) | _byc)]
                _taken |= set(_sub.index)
                if len(_sub): _fam.append(dict(k=_nm, n=int(len(_sub))))
            _oth = len(_g) - sum(f['n'] for f in _fam)
            if _oth > 0: _fam.append(dict(k='其他', n=int(_oth)))
            # 刷屏闸: 单码占比过半 → 条数不代表"问题多"而是一个码在抖 (08# 4609/4918=93.7% 码3225 集中2026-01)
            _burst = None
            if len(_g):
                _bc = _g.groupby('code').size().sort_values(ascending=False)
                _share = float(_bc.iloc[0]) / len(_g)
                if _share >= 0.5:
                    _sub2 = _g[_g.code == _bc.index[0]]
                    _burst = dict(code=str(_bc.index[0]), share=round(_share, 3), n=int(_bc.iloc[0]),
                                  months=sorted({str(m)[:7] for m in _sub2.t_on.dt.to_period('M').astype(str)}),
                                  med_s=float(_sub2.dur_s.median()) if 'dur_s' in _sub2 else None)
            # 明细 (用户令"浮窗说明之前发生问题"): top 码 + 首末时间
            _top = []
            if len(_g):
                for (_c2, _tx), _sub3 in _g.groupby(['code', 'text']):
                    _top.append(dict(code=str(_c2), text=str(_tx)[:34], n=int(len(_sub3)),
                                     first=str(_sub3.t_on.min())[:10], last=str(_sub3.t_on.max())[:10]))
                _top.sort(key=lambda x: -x['n'])
            _r['报警'] = dict(n=int(_acnt.get(_t, 0)), rank=int(_arank.get(_t, 0)) if _t in _arank else None,
                             tot=len(CFG['turbines']), fam=_fam, burst=_burst, items=_top[:6])
            if _wod is not None and len(_wod):
                _w = _wod[_wod.turbine == _t]
                _wtxt = (_w.get('故障名称', '').astype(str) + ' ' + _w.get('故障位置二级', '').astype(str)
                         + ' ' + _w.get('维修对象', '').astype(str) + ' ' + _w.get('元器件名称', '').astype(str))
                _w = _w[_wtxt.str.contains(_DT_PAT, na=False)]
                _acts = _w['维修动作'].replace('', pd.NA).dropna().value_counts() if len(_w) else None
                _wi = []
                for _, _wr in _w.sort_values('t_report', ascending=False).head(6).iterrows():
                    _wi.append(dict(date=str(_wr.get('t_report'))[:10], name=str(_wr.get('故障名称') or '')[:30],
                                    act=str(_wr.get('维修动作') or ''), part=str(_wr.get('维修对象') or _wr.get('元器件名称') or '')[:14]))
                _r['工单'] = dict(n=int(len(_w)),
                                  acts=[dict(k=str(k), n=int(v)) for k, v in (_acts.head(3).items() if _acts is not None else [])],
                                  items=_wi,
                                  last=(lambda v: str(v.max())[:10] if len(v) else None)(_w.t_report.dropna()[_w.t_report.dropna().dt.year > 2000]) if len(_w) else None)
            else:
                _r['工单'] = dict(n=0, acts=[], last=None)
        _wyr = ''
        if _wod is not None and len(_wod):
            _tv = _wod.t_report.dropna()
            _tv = _tv[_tv.dt.year > 2000]        # 空日期落 1970 epoch, 剔除后再报窗 (否则列头写"1970~")
            if len(_tv): _wyr = f"{_tv.min():%Y}~{_tv.max():%Y}"
        mheads = dict(mheads, 报警=f"传动链三部件 · {ms[0]}~{ms[-1]}" if ms else '传动链三部件',
                      工单=f"传动链三部件 · 台账 {_wyr}" if _wyr else '台账不可用')
        # 决策链进度盘 (2026-08-28 用户问"这个到底能怎么用"): 原来只有 29# 一台的静态六格,
        # 是展示牌不是工作面 — 打开它做不了任何决定。改为对全部需跟踪台报"走到第几步、卡在哪、下一步做什么"。
        # 判定规则单源在此, 前端只渲染不判断。
        _CLOOP = {}
        try:
            _ch = json.loads((ST.parent / 'm5_cms_tcm' / 'component_history.json').read_text(encoding='utf-8'))
            for _k, _v in _ch.get('summary', {}).items():
                if _k.startswith('★换件闭环') and isinstance(_v, list):
                    for _r in _v:
                        _CLOOP[_r['turbine']] = _r
                if _k.startswith('★当前在升') and isinstance(_v, list):
                    for _r in _v:
                        _CLOOP.setdefault(_r['turbine'], {}).update(rising=_r)
        except Exception:
            pass
        _HIT = {'bad', 'warn', 'note'}
        # ★一台可能有多行 (29#/17# 各 2 行: 齿轮箱 + 主轴承)。用 {turbine: row} 直接建字典会被后一行覆盖,
        # 实测 29# 因此取到"齿轮箱二级行星内齿圈·监视"那行, 而它的决策链讲的是"主轴承+集中润滑泵·定论"
        # ⇒ 浮窗里部件、温度通道、油样全是另一个部件的 (memory: 画图须确认各维度来自同一个体)。
        _SEV = {'bad': 0, 'warn': 1, 'note': 2, 'stale': 3, 'ok': 4, 'unlisted': 5}
        _tbi = {}
        for _row in ftab.to_dict('records'):
            _k = _row['turbine']
            if _k not in _tbi or _SEV.get(_row.get('级'), 9) < _SEV.get(_tbi[_k].get('级'), 9):
                _tbi[_k] = _row
        board = []
        for _r in mrows:
            t = _r['turbine']
            srcs = [k for k in ('振动', '温度', '润滑', '油液') if (_r.get(k) or {}).get('level') in _HIT]
            lvl = (_r.get('振动') or {}).get('level')
            tb = _tbi.get(t, {})
            mech = ('机制链' in ((_r.get('润滑') or {}).get('note') or '')
                    or '机制链' in (tb.get('振动结论') or ''))
            n_wo = (_r.get('工单') or {}).get('n') or 0
            cl = _CLOOP.get(t) or {}
            closed = bool(cl.get('verdict') == '恢复')
            rising = cl.get('rising')
            if not (lvl in _HIT or (srcs and lvl != 'ok')):
                continue
            # 六步: done=已达 / open=未达 / unknown=数据不足不可判
            steps = [
                dict(k='证据', st='done' if srcs else 'open', v=f"{len(srcs)} 源", d='/'.join(srcs)),
                dict(k='机制', st='done' if mech else 'open',
                     v='已定性' if mech else '未定性', d='根因定到部件与失效模式'),
                dict(k='判级', st='done' if lvl in _HIT else 'open',
                     v=tb.get('证据状态') or lvl or '—', d=f"设备状态 {tb.get('设备状态') or '—'}"),
                dict(k='排期', st='done' if t == 'WTG29' else 'open',
                     v='情景A 2026-09' if t == 'WTG29' else '未排',
                     d='进检修排程沙盘 (风险×损失双轴)'),
                # ★这一列一律"不可判", 不是偷懒: 工单台账止 2024-11 而判级时间窗在 2026,
                # 台账里的历史工单不可能是针对本次问题的 ⇒ 用 n>0 判"已派工"会造出假的完成态。
                dict(k='动作', st='unknown',
                     v=(f"台账 {n_wo} 单" if n_wo else '台账无'),
                     d='工单台账 2020~2024-11, 25-26 现场未提供 — 历史单不对应本次问题, 须向现场核实'),
                # ★历史闭环不占本格: 05#/20# 2024~2025 换过齿轮箱且已验证恢复, 但本轮又有新的候选级证据,
                # 把历史闭环填进第六格会出现"验收已达而机制未定"的自相矛盾行。历史闭环走 hist_loop 单独标记。
                dict(k='验收', st='open', v='未验',
                     d='本轮动作执行后复测同一判据是否回落' + (
                        f" (该台 {cl.get('replace_date')} 有过一次闭环: {cl.get('component')} 降至 {cl.get('ratio')}×)" if closed else '')),
            ]
            _first = next((x for x in steps if x['st'] != 'done'), None)
            stuck = _first['k'] if _first else None
            stuck_kind = _first['st'] if _first else None
            NEXT = {'证据': '补测: 四源全为常态, 先确认监测面有效',
                    '机制': '定性: 现场/取样把根因定到部件与失效模式 — 机制不清则排期与动作都无依据',
                    '判级': '送审: 按 RV-1 触发相应审级',
                    '排期': '排程: 进沙盘比情景, 出建议窗口',
                    '动作': '核实: 向现场调取 2025~2026 工单 — 台账止于 2024-11, 系统无法判定是否已派工',
                    '验收': '复测: 动作已执行, 复测同一判据是否回落'}
            board.append(dict(t=t, lvl=lvl, ostate=tb.get('设备状态'), estate=tb.get('证据状态'),
                              part=tb.get('部件'), steps=steps, stuck=stuck, stuck_kind=stuck_kind,
                              next=NEXT.get(stuck, ''), done=sum(1 for x in steps if x['st'] == 'done'),
                              rising=rising, hist_loop=(cl if closed else None), srcs=srcs))
        _ORD = {'bad': 0, 'warn': 1, 'note': 2}
        board.sort(key=lambda r: (_ORD.get(r['lvl'], 3), -r['done']))
        _cl_rows = [v for v in _CLOOP.values() if v.get('verdict') == '恢复']
        fus = dict(链盘=dict(rows=board, closed=_cl_rows,
                            stuck=dict(collections.Counter(r['stuck'] for r in board if r['stuck'])),
                            wo_bound='工单台账 2020~2024-11 (现场未提供 25-26)'),
                   表=ftab.to_dict('records'),
                   矩阵=mrows, 列窗=mheads, kpi=fkpi,
                   事件=ev, 事件边界=ev_bounds,
                   能力=dict(classes=cap['classes'], blind=[str(b) for b in cap['blind']], loop=cap.get('loop', {})),
                   窗=dict(振动=fmeta.get('振动窗', '—'), SCADA=fmeta.get('SCADA窗', ''), 油样=fmeta.get('油样窗', '')),
                   色标=fmeta.get('色标', {}), 数据时点=fmeta.get('数据时点', ''), handoff日期=fmeta.get('handoff日期', ''),
                   证据窗末=fmeta.get('证据窗末', ''),
                   纪律=str(fmeta.get('纪律', '')), 盲区=[str(b) for b in fmeta.get('盲区', [])],
                   open_items=[str(o) for o in fmeta.get('open_items', [])], gap=fmeta.get('gap'))
    except Exception as e:
        fus = dict(err=str(e)[:160])
    kpi = dict(报警台=n_alarm_t, 关注台=len(watch), 全场=len(CFG['turbines']))
    return dict(win=win, months=ms, all_months=sorted(_CACHE['tm'].month.unique()),
                systems=systems, sysdist=sysdist, watch=watch, kpi=kpi, rel=rel, fus=fus, faults=faults, control=control, m8=m8, m9=m9,
                # ★2026-09-21: 判级轴现在**按所选窗重算**；首次是后台算，这一份仍是旧口径 ⇒ 如实标 pending
                sysmx_pending=bool(sysmx_pending),
                m9_pending=bool(m9_pending),
                win_pending=bool(sysmx_pending or m9_pending),
                sysmx_span=list(span_of(win)),
                note=('判级轴(变桨/偏航/蓄能/温度)按所选时间窗重算；曲线按所选时间窗重算；'
                      '故障统计/五态/温度月轨迹/停机台账=所选窗真窗'
                      + ('（⚠ 判级矩阵正在按所选时间窗重算，下面系统卡暂为上一份口径，稍后自动刷新）' if sysmx_pending else '')))


def fleet(win: str = '2026年') -> dict[str, Any]:
    """详情层口径的薄包装：`win = q.get('win', '2026年')`（serve.py 的默认窗）。

    ★不改写搬运来的 `fleet_view`，只补"默认窗"这一层 —— 实逮：不加它，缺参调用会
      `TypeError: fleet_view() missing 1 required positional argument: 'win'`（算法服务 500）。
    """
    return fleet_view(win)
from src import paths as _P                                              # 路径唯一真源 (与 cwd 无关)
from src.windscada import taxonomy
from src.windscada.subsys import temp_nbm, hydraulic, yaw as yawmod, pitch as pitchmod
import hashlib, time
import numpy as np, pandas as pd
import threading as _thr

# ── curves 视图所需（同法逐字搬） ──
LENSES = [('wsb', 'grd_wtc_ActPower_mean', 'L1 功率曲线 (风速-功率)', '风速 m/s (机舱风)', 'kW'),
          ('wsb', 'cp', 'L6 风能利用系数 Cp (风速-Cp)', '风速 m/s', 'Cp'),
          ('wsb', 'lam', 'L5 叶尖速比 λ (风速-λ)', '风速 m/s', 'λ'),
          ('grb', 'tq', 'L4 转矩-转速 (发电机侧)', '发电机转速 rpm', 'kNm'),
          ('pwb', 'tur_wtc_PitcPosA_mean', 'L2 功率-桨距 (控制律/削峰)', '功率 kW', '°'),
          ('pwb', 'tur_wtc_GenRpm_mean', 'L3 功率-发电机转速 (饱和点)', '功率 kW', 'rpm'),
          ('pwb', 'p3', 'L2b 三叶桨距极差 (集距/不平衡)', '功率 kW', '°'),
          ('pwb', 'ratio', 'L7 转速比 gen/rot (传动链)', '功率 kW', '—')]

CFG = farm(); ST = pathlib.Path(CFG['store'])

_WIN_BUSY: dict = {'sysmx': set(), 'curves': set(), 'm9': set()}

_WIN_CACHE: dict = {'sysmx': {}, 'curves': {}, 'm9': {}}

_WIN_ERR: dict = {'sysmx': {}, 'curves': {}, 'm9': {}}



def curves_of(win):
    """→ (七镜头分箱件 或 None, pending)。按窗重算（干净窗就是正式产物本身）。"""
    from src.windscada.perf import curves as _cv
    a, b = span_of(win)
    if (a, b) == (_cv.WIN[0], '2025-12-31'):     # 与正式产物的判别窗一致 ⇒ 直接用产物, 不重算
        import pandas as _pd
        return _pd.read_parquet(ST / 'curve_lenses.parquet'), False
    st = _win_get('curves', win, lambda: _cv.build_store(CFG, span=(a, b), write=False))
    return st, st is None


def curve_view(win='2025H2'):
    """特性曲线多镜头 (SOP §4.6c): fleet 中位+四分位带 + 过绝对锚的离群台曲线.

    ★2026-09-21 用户令「发电性能随时间窗变化」: 七镜头**按所选时间窗重算**（首次后台算，先回 pending），
    7 张月度时序图按所选窗的月份过滤。原口径（2025H2 干净判别窗）仍可作为窗之一被选中，
    选到它时直接用正式产物，不重算。
    """
    import pandas as pd
    from src.windscada.perf import curves as cv
    a, b = span_of(win)
    ms = months_of(win)
    store, pending = curves_of(win)
    if pending:
        return dict(building=True, win=win, span=[a, b], months=ms,
                    note=f'正在按所选时间窗 {a} ~ {b} 重算七镜头曲线（首次约 10~45 秒），完成后自动刷新。'
                         f'期间下方月度时序图已按所选窗过滤，可直接看。')
    wlabel = f'{a} ~ {b}'
    figs = []
    for xc, yc, title, xlab, unit in LENSES:
        r = cv.lens(xc, yc, store_df=store, win_label=wlabel)
        if not r: continue
        outs = sorted(r['离群'].items(), key=lambda kv: -abs(kv[1]['z']))[:3]
        a = r['anchor']
        series = [dict(name='全场中位', vals=r['fleet'])]
        series += [dict(name=f"{t.replace('WTG','')}# {v['resid']:+.4g}{a['单位']}", vals=r['per_t'][t]) for t, v in outs]
        figs.append(dict(kind='multiline', title=title, unit=unit, months=[f"{x:g}" for x in r['x']],
                         series=series, xlab=xlab,
                         note=(f"显著门 |残差|≥{a['门']}{a['单位']} ({a['说明']}) ∧ |z|≥3 → 离群 "
                               f"{'/'.join(t.replace('WTG','')+'#' for t,_ in outs) if outs else '无'}"
                               f" ｜ 灰域=全场四分位, 逐档中位聚合, 已剥限电(只正常发电态)"
                               f" ｜ 样本 {r['n_total']:,} 个10min点 / {r['n_turbines']} 台 / 最小档 {r['n_min_bin']:,}"
                               f" ｜ 时间窗 {r['win']}（随所选时间窗）"),
                         band=dict(q1=r['q1'], q3=r['q3'])))
        # 残差视图 (§4.6c④ per机必算同型残差): 绝对量尺度上四分位带只有几个像素, 残差面才看得见
        if outs:
            rs = [dict(name=f"{t.replace('WTG','')}# {v['resid']:+.4g}{a['单位']}",
                       vals=[None if (r['per_t'][t][i] is None or r['fleet'][i] is None) else round(r['per_t'][t][i] - r['fleet'][i], 4)
                             for i in range(len(r['x']))]) for t, v in outs]
            q1d = [round(r['q1'][i] - r['fleet'][i], 4) for i in range(len(r['x']))]
            q3d = [round(r['q3'][i] - r['fleet'][i], 4) for i in range(len(r['x']))]
            figs.append(dict(kind='multiline', title=title.split(' (')[0] + ' · 同型机群残差 (本台−全场中位)', unit=unit,
                             months=[f"{x:g}" for x in r['x']], series=rs, xlab=xlab,
                             thresholds=[dict(v=a['门'], label=f"显著门 +{a['门']}{a['单位']}"), dict(v=-a['门'], label='')],
                             band=dict(q1=q1d, q3=q3d),
                             note=(f"虚线=物理绝对锚 ±{a['门']}{a['单位']} ({a['说明']}); 灰域=全场四分位残差带; 出带且过锚才算离群"
                                   f" ｜ 样本 {r['n_total']:,} 点 / {r['n_turbines']} 台 ｜ 时间窗 {r['win']}")))
    # ---- 时序件: 控制参数月度 (M9b) + 偏航动态月度 (M12) ----
    def _ts(path, val, title, unit, xlab, picks=None, note='', thr=None):
        f = ST / path
        if not f.exists(): return None
        d = pd.read_parquet(f)
        if val not in d.columns: return None
        piv = d.pivot_table(index='month', columns='turbine', values=val)
        # ★2026-09-21 用户令: 月度时序图按**所选时间窗**的月份过滤（月度件无法按日切，如实写"按月取整"）
        if ms:
            piv = piv.reindex([m for m in [str(x) for x in piv.index] if m in set(ms)])
        if not len(piv): return None
        msx = [str(x) for x in piv.index]
        med = piv.median(axis=1)
        dev = (piv.sub(med, axis=0)).abs().mean()
        picks = picks or list(dev.sort_values(ascending=False).head(3).index)
        series = [dict(name='全场中位', vals=[None if v != v else round(float(v), 3) for v in med])]
        series += [dict(name=t.replace('WTG', '') + '#', vals=[None if v != v else round(float(v), 3) for v in piv[t]])
                   for t in picks if t in piv.columns]
        # 样本量与窗随图走 (2026-08-28 门禁 F1): 7 张时序图原来只有结论没有依据规模,
        # 违反第一性原理②"Sample size declared"。逐月×台的非空格数即样本量。
        _n = int(piv.notna().to_numpy().sum())
        _nt = int(piv.shape[1])
        note2 = (note + f" ｜ 样本 {_n:,} 个月×台 / {_nt} 台 / {len(msx)} 个月"
                        f" ｜ 时间窗 {msx[0]}~{msx[-1]}（月度聚合，随所选时间窗·按月取整）") if msx else note
        return dict(kind='multiline', title=title, unit=unit, months=msx, series=series, xlab=xlab,
                    note=note2, thresholds=thr or [],
                    band=dict(q1=[None if v != v else round(float(v), 3) for v in piv.quantile(0.25, axis=1)],
                              q3=[None if v != v else round(float(v), 3) for v in piv.quantile(0.75, axis=1)]))
    for args in (
        ('control_monthly.parquet', 'p_cap', '时序① 满发功率封顶 月度 (取每月高位稳定值; 阶跃=配置变更)', 'kW', '月', None,
         '灰域=全场四分位; 参数级分组见发电性能页 (4175/4200/4225 三组); 阶跃=配置变更, 平移=工况'),
        ('control_monthly.parquet', 'w_cap', '时序② 转速封顶 月度 (取每月高位稳定值)', 'rpm', '月', None, '两轴独立分组 (1672/1680)'),
        ('control_monthly.parquet', 'pitch_rated', '时序③ 额定段桨距角 月度 (标定漂移)', '°', '月',
         ['WTG03', 'WTG34', 'WTG19'], '离群台=桨距调度同档差 top3'),
        ('yaw_dynamic_monthly.parquet', 'err_sd', '时序④ 对风散布 σ 月度 (原始, 未清洗)', '°', '月',
         ['WTG08', 'WTG16', 'WTG04'], '三台19个月全程高 → 非发作型; 清洗后降至5-7°, 判为机舱位置通道拖偏(A类数据质量)非对风故障'),
        ('yaw_dynamic_monthly.parquet', 'travel_day', '时序⑤ 偏航行程 月度 (抗采样物理量)', '°/日', '月', None,
         '行程=活动量代理, 非真磨损量; 与状态位次数互核'),
        ('yaw_dynamic_monthly.parquet', 'twist_span', '时序⑥ 扭缆角月跨度 (ScYawPos 累计位置)', '°', '月', None, '解缆动作后回零'),
        ('yaw_dynamic_monthly.parquet', 'pump_med', '时序⑦ 偏航泵压 月度 (小站泵压)', 'bar', '月', None, '全场极齐 → 无泵压异常'),
    ):
        f = _ts(*args)
        if f: figs.append(f)
    return dict(figs=figs, physics=cv.physics_check(),
                note=('镜头口径 (SOP §4.6c): X轴优先功率(直测干净量); 机舱风 self-ref → 只判形状/相对, '
                      '绝对达成率须现场测风; per机异常必算同型机群残差(禁眼估); 工况段筛选=剔除0kW与低转速混合态档'))



def curves(win: str = '2026年') -> dict[str, Any]:
    """详情层口径的薄包装（`curve_view(q.get('win') or '2026年')`）。"""
    return curve_view(win)
from src import paths as _P                                              # 路径唯一真源 (与 cwd 无关)
import hashlib, time

# ── vibcms/reload 视图所需（逐字搬） ──
CFG = farm(); ST = pathlib.Path(CFG['store'])



def vibcms_results():
    """windcms 评估报告 → 结果层转录 (2026-08-28 用户令: 生接改融合·只显示结果·模型与计算隐藏).
    取最新一期报告md; '融合级(模型)/CMS红黄'两列=模型输出, 不出结果层; 分析功能留独立cms.

    ★2026-09-18 用户令"页面只能基于输入数据重算、不许用旧版产出补"之后实逮: 清过产物再重算的机器上
    `windcms/` 里没有 `报告_CMS振动状态评估报告_*.md`, 于是 `sorted(glob)[-1]` 抛
    **IndexError: list index out of range**, 页面看到的是"windcms 报告解析失败: list index out of range"
    —— 与 `ProductsMissing` 的教训同一个病: **缺产物被报成了程序坏了**。现在按缺产物如实回结构化的
    `no_report`, 并写清它由谁生成。★2026-09-19 再修: 报告步的生成端(`scripts/windcms.py report`)与它的
    上游(`rudong_model_run.py`/`rudong_fusion_run.py`)都已随包并上链, 缺件只剩"还没重算到那一步"这一个原因,
    故这里再附一句"重算是否正在跑"(读 run/ops_job.json), 免得把"正在跑"读成"系统坏了"。
    """
    reps = sorted(ST.parent.glob('windcms/报告_CMS振动状态评估报告_*.md'))
    if not reps:
        # ★2026-09-19 修: 原话还写着"本包六层链的 model_run/fusion 两步脚本未随包" —— 那两步已在
        #   2026-09-19 按口径重建并上链(scripts/rudong_model_run.py / rudong_fusion_run.py), 话说反了。
        #   真实原因只有两种: ① 重算还没走到 ④b 的报告步(清过产物后这一步要跑很久); ② windcms 产物被清且未重算。
        try:
            from src import opsjob as _oj
            _run = (' ' + _oj.running_text()) if _oj.running_text() else ''
        except Exception:
            _run = ''
        return dict(date=None, overall='', action=[], grades=[], diff=dict(only_cms=[], only_handoff=[]),
                    no_report=True,
                    note_zh='本机没有 CMS 振动状态评估报告 (由重算链 ④b 的 report 步生成: '
                            'python scripts/windcms.py report)。如实为空, 不用旧版产出补。' + _run,
                    err='无产物: windcms/报告_CMS振动状态评估报告_*.md 不在位')
    try:
        rp = reps[-1]
        md = rp.read_text(encoding='utf-8')       # 不写 encoding 会按系统 locale(cp936) 读 → 中文报告乱码
        date = rp.stem.rsplit('_', 1)[-1]
        overall = ''
        if '### 3.1' in md:
            seg = md.split('### 3.1')[1].split('###')[0]
            overall = next((l.strip() for l in seg.splitlines() if l.strip().startswith('全场')), '')
        def _tab(sec):
            return [[c.strip() for c in l.strip('|').split('|')] for l in sec.splitlines() if l.startswith('| WTG')]
        action = _tab(md.split('### 3.2')[1].split('## ')[0]) if '### 3.2' in md else []
        grades = []
        if '## 附录 A' in md:
            for r in _tab(md.split('## 附录 A')[1]):
                # 列: 机组/前/后/齿/发/综合/融合级(模型·隐)/CMS红黄(隐)/行动
                grades.append(dict(t=r[0], 主轴承前=r[1], 主轴承后=r[2], 齿轮箱=r[3], 发电机=r[4],
                                   综合=r[5], 行动=r[8] if len(r) > 8 else ''))
        # 与融合矩阵(handoff 定谳链)口径差异台
        from src.windscada.subsys import fusion as _f
        ft, _ = _f.fusion_table()
        hset = {r.turbine for _, r in ft.iterrows() if _f.verdict_class(str(r.振动结论))[1] in ('bad', 'warn')}
        cset = {r[0] for r in action}
        diff = dict(only_cms=sorted(cset - hset), only_handoff=sorted(hset - cset))
        # ★2026-09-19: 把**这个窗真实覆盖的日期区间**一并回给页面。原来只回 date(出件日) 与表格,
        #   页面上看不到"数据从哪天到哪天" ⇒ 用户拿 2026-03~04 的导出却只看到 4 月的谱时,
        #   无法判断是"数据没呈现"还是"源件本身就那么点"(实测源件最早只到 2026-03-16, 逐台更晚)。
        _span, _win = '', ''
        try:
            from src.windcms import config as _wcfg, data as _wdata
            _w2 = _wcfg.farm(str(CFG.get('key') or 'rudong'))
            _span = _wdata.span_text(_w2) or ''
            _win = ','.join(_wdata.windows(_w2))
        except Exception:
            _span = _win = ''
        return dict(date=date, overall=overall, action=action, grades=grades, diff=diff,
                    时间范围=_span, 窗=_win)
    except Exception as e:
        return dict(error=f'windcms 报告解析失败: {e}')  # 响亮, 不静默空表


def reload_products(reason=''):
    """显式清缓存 (下一次请求重载)。返回清前的指纹, 供日志/接口回显。"""
    with LOCK:
        old = _CACHE.get('__stamp')
        _CACHE.clear()
    print(f'[reload] 清产物缓存 ({reason or "手动"}) 旧指纹={old}', flush=True)
    return old



def vibcms() -> Any:
    """详情层 `/api/vibcms`：`vibcms_results()` 原样返回。"""
    return vibcms_results()


def reload(why: str = 'manual') -> dict[str, Any]:
    """详情层 `/api/reload`：`dict(ok, old_stamp, stamp=products_stamp(force=True), files=len(_product_files()))`。"""
    old = reload_products(why or 'manual')
    return dict(ok=True, old_stamp=old, stamp=products_stamp(force=True), files=len(_product_files()))

# ── 脱敏/清洗（rpt_export 等需要；逐字搬，勿手改） ──
INTERNAL = _os.environ.get('WINDSCADA_INTERNAL', '') == '1'
_CRIT_MARK = ('z', 'σ', '×', '倍', 'dev', 'resid', '选择性', '峰', '结构度', 'ratio')
_KEEP_UNIT = ('h', '小时', '天', '日', '月', '年', '台', '次', '条', '起', '个')
_RE_HZ = re.compile(r'(?<![\d.])\d+(?:\.\d+)?\s*(?=Hz)')
_RE_THR = re.compile(r'([≥≤])\s*(\d+(?:\.\d+)?)\s*([a-zA-Z%℃×/·²]*)')
import math

def _jclean(o):
    """NaN/Inf → null (非法 JSON 防线; pd.DataFrame 会把 None 变 NaN)."""
    if isinstance(o, float) and (math.isnan(o) or math.isinf(o)): return None
    if isinstance(o, dict): return {k: _jclean(v) for k, v in o.items()}
    if isinstance(o, (list, tuple)): return [_jclean(x) for x in o]
    return o


def _redact_text(t):
    if INTERNAL or not isinstance(t, str) or not t:
        return t
    if '0.593' in t or '贝兹' in t or 'Betz' in t:      # 物理常数不动
        return _RE_HZ.sub('▪▪', t)
    t = _RE_HZ.sub('▪▪', t)

    def _th(m):
        sign, num, unit = m.group(1), m.group(2), m.group(3) or ''
        pre = t[max(0, m.start() - 8):m.start()]
        if any(c in pre for c in _CRIT_MARK):          # |z|≥3 / ×≥2.5 = 判据阈值, 必脱
            return f'{sign}▪{unit}'
        if unit in _KEEP_UNIT or (not unit and float(num) <= 10 and float(num) == int(float(num))):
            return m.group(0)          # 时间/计数口径保留 (≥1h段 / ≥2 个系统)
        return f'{sign}▪{unit}'
    return _RE_THR.sub(_th, t)


def _redact(o):
    if INTERNAL:
        return o
    if isinstance(o, str):
        return _redact_text(o)
    if isinstance(o, dict):
        return {k: _redact(v) for k, v in o.items()}
    if isinstance(o, list):
        return [_redact(v) for v in o]
    return o