guanlan-src/app_backEnd/app_backEnd_guanlan/serve.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""windscada serve (M7c): 交互后端 — 工作台(全场)+分台页, 全部带时间窗.
设计: ①台页只渲染"有问题的系统"的证据图(分册两型: A=判据线+摆幅标注, B=本台vs全场四分位带), 无问题不出图;
②台内交互查询: 任一契约通道×时间窗×聚合 即时出图; ③工作台: 系统分类问题(全系统)+故障统计梳理+SOP控制策略;
④时间窗: 月度类数据真窗; 判级类(变桨/偏航/蓄能=近90日, 曲线=2025H2判别窗)固定时间窗如实标注.
端口 8033. 判级=筛查级/候选, P级归综合报告线."""

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[3]
import sys as _sys, pathlib as _plb
_sys.path.insert(0, str(_install_root(__file__)))
from src import paths as P
import sys, json, pathlib, re, threading, urllib.parse, collections
import hashlib, time
sys.path.insert(0, str(_install_root(__file__)))     # 安装根入 sys.path (先于 src.* 导入)
from src import paths as _P                                              # 路径唯一真源 (与 cwd 无关)
import numpy as np, pandas as pd
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

# ── 客户交付脱敏 (2026-08-28 demo 门禁逮出 7 处特征频率 + 21 处判据门槛上屏) ──
# 规则来自 memory client-deliverable-hide-core-parameters: 判据名与实测值可见, 数值门槛与特征频率隐去。
# 分级而非一刀切 — 一刀切会把业主必须知道的口径也糊掉:
#   脱  特征频率 (134.9Hz 是我们解出来的轴承几何, 属核心参数)
#   脱  判据门槛 (≥40kW / |z|≥3 / ≥2.5× — 我们的判别阈值)
#   留  时间口径 (≥1h 段 / ≥3 天 — 业主要据此理解统计范围)
#   留  物理常数 (Cp ≤ 0.593 贝兹极限 — 公开知识, 不是我们的)
#   留  台数与占比 (≥2 个系统报警 — 业主自己数得出来)
# 内部模式: WINDSCADA_INTERNAL=1
import os as _os
INTERNAL = _os.environ.get('WINDSCADA_INTERNAL', '') == '1'
# ★ Hz 后不能用 \b: 中文属于 \w, "597Hz线" 里 z 与 线 之间没有词边界 ⇒ \b 不成立, 实测漏脱 WTG05 的 597Hz。
# ★ 前瞻也不能用 \w: 中文同属 \w, "HS_gen测点274Hz" 里 274 前是"点" ⇒ 断言失败漏脱。
# 只需排除数字与小数点 (防止把 1274 截成 274)。
_RE_HZ = re.compile(r'(?<![\d.])\d+(?:\.\d+)?\s*(?=Hz)')
_KEEP_UNIT = ('h', '小时', '天', '日', '月', '年', '台', '次', '条', '起', '个')
# ★ 判据符号前缀: |z|≥3 / σ≥2 / ×≥2.5 这类无单位阈值必须脱 —
#   否则会被"无单位小整数=计数口径"那条规则放行 (实测 |z|≥3 漏脱)。
_RE_THR = re.compile(r'([≥≤])\s*(\d+(?:\.\d+)?)\s*([a-zA-Z%℃×/·²]*)')
_CRIT_MARK = ('z', 'σ', '×', '倍', 'dev', 'resid', '选择性', '峰', '结构度', 'ratio')


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


# ── 语言投影 (2026-08-31) ────────────────────────────────────────────
# 诊断: 骨架页 HTML 里**一个中文字都没有** (实测 0 字符), 内容全部由 JS 从 API 拉取,
#       而 API 不认 lang —— 所以"英文版"拿到的一直是中文数据。此前三轮翻译 HTML 源码,
#       改的是一个没有内容的文件, 这是英文残留反复修不好的真正原因。
# 修法: 在 API 唯一出口 _jdump 上做文本层投影。**不带 lang 时原样返回**, 中文版行为零变化。
_REQ = threading.local()
try:
    from src.ontology import codes as _codes
except Exception:                                  # 码表缺失不应让服务起不来
    _codes = None


def _lang_of():
    return getattr(_REQ, 'lang', '') or ''


def _enum_fallback(field, value, lang):
    """枚举值映射不出码时的回落 —— **走翻译, 不留中文**。

    根因 (2026-08-31 实测): `part` 这类字段被当成枚举, 但它在工单里其实是**现场手写的
    自由文本** ('齿轮箱高速轴' / 'IGBT、直流保险' / '(振动线未收录)' / '无')。码表只有 18 条,
    覆盖不了; 而 normalize 失败时原实现**原样返回中文**, 于是英文页上直接显示中文。
    把这些硬塞进码表是错的 —— 它们不是枚举语义, 是描述。正确做法是让它们走翻译层。
    查不到才留原文 (不猜译)。
    """
    if lang == 'zh' or not isinstance(value, str) or not value.strip():
        return value
    try:
        from windscada import en_text as _et
    except Exception:
        return value
    for scope in (field, 'part', None):
        try:
            en, _src = _et.lookup(value, scope) if scope else _et.lookup(value)
        except Exception:
            en = None
        if en:
            return en
    return value


# ★这些字段 **不翻译** —— 它们是 JS 的判别值与查表键, 不是给人读的文案。
#   翻了会静默劈叉: 页面上看不出错, 但整段内容消失或整片颜色变灰。
#   2026-09-02 线上实测 (用户逮「有问题的看不见? 只有 good」):
#     · st 被翻成 Alarm  → `rows.filter(r=>r.st==='报警')` 零命中 → **报警 12 台整表不渲染**,
#       而柱状条读的是 sysdist (**键**没被翻) 仍显示 "Alarm 12" ⇒ 数字与明细自相矛盾。
#     · ostate 被翻成 Good/Excellent → `C[c.ostate]` (C 是中文键) 全部 MISS
#       → 融合矩阵**每一格颜色都退成灰**。
#   判据: 凡是 JS 里 `X[obj.字段]` 查中文键表、或 `obj.字段==='中文'` 比较的字段, 一律进本表。
#   显示层 (ui_en.value_script) 负责把渲染出来的中文文本换成英文, 所以人看到的仍是英文。
#   tests/test_keep_zh_discriminators.py 扫源码断言本表齐全, 防以后加字段漏登记。
_KEEP_ZH = frozenset((
    'st', 'ostate', 'wlevel', '状态', '设备状态', '证据状态', '部件', 'liveness',
    'loc',      # 模型本地/云端档位: `v.loc==='本地'` 决定"数据不出本地"这句能不能说
    # ★第三类判别用法, 本闸的扫描器 (tests/test_keep_zh_discriminators.py) 抓不到:
    #   不是比较也不是查表, 而是拼进 URL 路径当 ID (`/problem/${t}/${encodeURIComponent(x)}`)。
    #   syss 被翻成 'Yaw' 后, 浮层请求 sys=Yaw 永远查不到烤的是 sys=偏航 的快照键,
    #   点开必是"本时间窗无记录"——页面不报错, 数据静默消失, 和 st/ostate 那两类同一后果不同机制。
    #   2026-09-03 用户实测逮出 (Watch list → 系统名链接)。显示文字仍由 sysName(x) 客户端译。
    'syss',
))

# ★按**值**豁免 —— 给 `k` 这种一名多用的字段用。
#   `k` 在告警排行里是显示串 (`"7122 偏航液压油位低"`, 要翻),
#   在控制状态与部件配色里是枚举键 (`'停机'` / `'齿轮箱'`, 不能翻)。
#   键名分不开两者, 只能按值。2026-09-02 实测: 我先把 'k' 整个放进 _KEEP_ZH,
#   结果**报警名全部不再翻译** —— 修一个洞挖一个洞, 按值豁免才收得住边界。
#   来源: JS 里 CO / LOSSPLAIN 两张中文键表的键 + `x.k==='停机'` 比较。
_KEEP_ZH_VALS = frozenset((
    '齿轮箱', '发电机', '主轴承', '变桨叶片', '偏航', '测风', '其他',            # CO 配色键
    '限电·绑定', '限电命令·未绑定', '停机(调度令)', '停机', '正常发电', '低风待机',  # LOSSPLAIN 键
))


def _proj(o, lang):
    """把 `key_zh`/`key_en` 投影成裸 `key`。

    英文侧取不到 `_en` 时写 `[translation pending]` 而**不回落中文** ——
    回落会让缺口永远不被发现, 正是英文版残留三轮没被逮住的机制。
    """
    if isinstance(o, list):
        return [_proj(x, lang) for x in o]
    if not isinstance(o, dict):
        return o
    suf = '_' + lang
    other = '_zh' if lang == 'en' else '_en'
    # ★裸 `zh` / `en` 成对 (framework.py 的 SYSTEMS/RECORDS/stages 就是这个写法)。
    #   它们不带下划线, 走不到下面的后缀规则; 而末尾"残留中文一律写 [translation pending]"
    #   的兜底会把 zh 打成 pending —— **旁边的 en 明明就是对的**。实测 145 处 pending
    #   里的大多数是这一种。消费端 JS 读的是 `.zh`, 所以英文版把 en 的值投影到 zh 上。
    if lang != 'zh' and isinstance(o, dict) and 'zh' in o and 'en' in o:
        o = dict(o, zh=o['en'])
    # ★只有 k_en **有值**时才让裸 k 让位 (2026-09-01 parity 闸逮)。
    #   原来无条件让位: 上游给了 action_en=None, 裸 action 就被顶成 None ——
    #   页面上显示 "None", 比留中文糟得多。空的语言副本等于没给, 不该覆盖原值。
    out = {}
    source_fields = {'name', '步骤', '故障', '排查', '原因', '处置', 'check'}
    originals = {key: o.get(key + '_zh', o.get(key)) for key in source_fields} if lang == 'en' else {}
    originals = {key: value for key, value in originals.items() if isinstance(value, str) and _CJK.search(value)}
    langed = {k[:-len(suf)] for k in o if k.endswith(suf) and o[k] not in (None, '', [])}
    for k, v in o.items():
        if k == '_original':
            out[k] = dict(v) if isinstance(v, dict) else v
            continue
        if k.endswith(other):                      # 另一语言的副本不出 API
            continue
        if k.endswith(suf):
            if v not in (None, '', []):
                out[k[:-len(suf)]] = _proj(v, lang)
            continue
        if k in langed:                            # 已有 key_<lang>, 裸 key 让位
            continue
        v2 = _proj(v, lang)
        if k in _KEEP_ZH or (isinstance(v2, str) and v2 in _KEEP_ZH_VALS):
            out[k] = v2                            # 判别值/查表键: 原样留中文 (见 _KEEP_ZH 注释)
            continue
        # 枚举层先查码表 — 这些值有确定译名, 不该退化成 [translation pending]。
        # 列表型枚举 (如 facets) 逐项查。
        if lang != 'zh' and _codes is not None and k in _codes.CODES:
            if isinstance(v2, str):
                c = _codes.normalize(k, v2)
                if c:
                    v2 = _codes.label(k, c, lang)
                else:
                    v2 = _enum_fallback(k, v2, lang)
            elif isinstance(v2, list):
                v2 = [(_codes.label(k, _codes.normalize(k, x), lang)
                       if isinstance(x, str) and _codes.normalize(k, x) else x) for x in v2]
        # 单数字段名对应复数码表 (facet ← facets) 也走一遍
        elif lang != 'zh' and _codes is not None and k.rstrip('s') in _codes.CODES:
            f = k.rstrip('s')
            if isinstance(v2, str):
                c = _codes.normalize(f, v2)
                if c:
                    v2 = _codes.label(f, c, lang)
            elif isinstance(v2, list):
                v2 = [(_codes.label(f, _codes.normalize(f, x), lang)
                       if isinstance(x, str) and _codes.normalize(f, x) else x) for x in v2]
        # ★这里原来只判 isinstance(v2, str) —— **字符串列表整个绕过了英文投影**。
        #   _proj 递归进 list, 但 _proj(str) 原样返回, 所以 ["偏航","变桨"] 这种
        #   一路穿到 API。实测漏掉 5 条路径 (parts[] / syss[] / srcs[] /
        #   open_items[] / blind[]), 是英文页数据层残留中文的唯一来源。
        if lang == 'en':
            # Action counts use display labels as dictionary keys, so _en_deep
            # cannot reach them. Only this display map uses the established
            # work-order action glossary; all other dictionary/URL IDs stay put.
            if k == '动作统计' and isinstance(v2, dict):
                from src.windscada.maint.workorder_en import ACTION
                v2 = {ACTION.get(action) or action: count for action, count in v2.items()}
            v2 = _en_deep(v2, k)
        out[k] = v2
    for key, original in originals.items():
        if key in out and out[key] != original:
            out.setdefault('_original', {})[key] = original
    return out



def _en_deep(v, field='', _depth=0):
    """对字符串与**字符串列表/嵌套列表**一律走 _data_text_en。

    只对 str 生效的旧写法会让 list-of-str 静默穿透 —— 这类漏法的特征是
    "抽查单值全对, 页面上却成片中文", 因为漏的是容器不是值。
    """
    if isinstance(v, str):
        return _data_text_en(v, field) if _CJK.search(v) else v
    if isinstance(v, list) and _depth < 4:
        return [_en_deep(x, field, _depth + 1) for x in v]
    return v


_DATA_EN_CACHE = {}
_DATA_EN_MISS = set()     # 查不到的原文集合 (与命中缓存分开: 未命中的返回值随请求模式而异)


def _data_text_en(v, field=''):
    """数据层中文文本 → 英文。取不到才写 [translation pending]。

    ★为什么要有这一层: 原实现是"残留中文一律 [translation pending]" —— 干净但过于绝对。
    实测 /api/fleet 一次就打出 **2199 处** pending, 而其中大量是厂商报警名、工单原文、
    部件名、判级词, 这些在本项目里**已经有译源**: 厂商中英对照表 3834 条、IPC 部件表
    2113 条、枚举码表、工单名构词表。不查就直接标 pending, 等于把已有的资产扔掉,
    还让英文页看起来像没做完。

    查不到仍写 pending —— **不猜译**。这条不能松: 猜译出来的厂商报警名会误导现场。
    """
    if v in _DATA_EN_CACHE:
        return _DATA_EN_CACHE[v]
    if v in _DATA_EN_MISS:
        return v if getattr(_REQ, 'keep_orig', False) else '[translation pending]'
    out = None
    # ⓪⁻ 数据层**整串**直译表 (融合面盲区/未决项长文)。放在最前: 这些是整段人写的
    #     分析结论, 走模板反解或分层词表都只会碎成片段。
    try:
        from src.windscada import data_tpl_en as _tplN
        out = getattr(_tplN, 'NOTES', {}).get(v)
        if out is None:
            from src.windscada.en_text import standard_term
            out = standard_term(v)
        if out is None:
            out = getattr(_tplN, 'RECORD_EN', {}).get(v)
    except Exception:
        out = None
    if out is not None:
        _DATA_EN_CACHE[v] = out
        return out
    # ⓪ 「码 + 中文描述」→ 本体 AlarmCode 的权威 name_en (查表, 不翻译)
    try:
        from src.windscada import en_text as _et0
        out = _et0.alarm_en(v)
    except Exception:
        out = None
    if out is not None:
        _DATA_EN_CACHE[v] = out
        return out
    # ① 服务端拼串走**模板反解** (精确, 非翻译): 抽数字得形态 → 查英文模板 → 数字按序填回
    try:
        from src.windscada import data_tpl_en as _tpl
        out = _tpl.render(v)
    except ValueError as e:
        print('[_data_text_en] 模板占位符对不上, 已跳过不填: %s' % e, file=sys.stderr)
        out = None
    except Exception:
        out = None
    if out is not None:
        _DATA_EN_CACHE[v] = out
        return out
    try:
        from src.windscada import en_text as _et
        # ① 分层查表 (厂商报警 → IPC 部件 → 枚举/文案)
        for sc in ('fault', 'part', 'ui', None):
            hit, _src = _et.lookup(v, sc) if sc else _et.lookup(v)
            if hit:
                out = hit
                break
        # ② 部件自由文本解析 (复合名/带注解/多部件连写)
        if out is None:
            out = _et.part_en(v)
    except Exception:
        pass
    # ②b 词表直查 —— 部件名/判级词/系统名作为**独立字段**出现时 (fus.表[].部件 / .振动结论 /
    #     rel.表[].部件), 既不是模板串也不在 en_text 分层表里, 而 data_tpl_en 的
    #     CHAN/GRADE2/SYSTEM_EN 早就登记了同一批词 (供模板内部用)。不查就整片留中文
    #     (2026-09-03 实逮 fus.表 328 字)。**同一批词只因出现位置不同就译不出, 是接线漏不是词条漏。**
    if out is None:
        try:
            from src.windscada import data_tpl_en as _t2
            out = _t2.CHAN.get(v) or _t2.SYSTEM_EN.get(v)
            if out is None:
                g = _t2.GRADE2.get(v) or _t2.GRADE.get(v)
                if g:                              # 判级词在表里是小写短语 (模板内嵌用), 独立成字段要首字母大写
                    out = g[0].upper() + g[1:]
        except Exception:
            pass
    if out is None:
        try:
            from src.windscada.maint import workname_en as _wn
            # ★只用 work_name_en (逐段都命中才拼, 命不中返回 None), **不用 category_en**。
            #   category_en 是正则归类兜底: 它对看不懂的文本也会给出一个"像样"的答案。
            #   实测它把化验样注记译成 work item:
            #     '末样2025-01-09 (距振动现窗>12月, stale: 只作检测链回测)'
            #        → 'Condition monitoring — work item'   ← 错, 那不是工单
            #     '末样2026-07-13: 华标判正常(NB/T宽限); 重点值…'
            #        → 'Generator — work item'              ← 错且误导
            #   我在本函数的说明里写了"不猜译"然后把猜译器接了进来。
            #   **看得懂就译, 看不懂就标 pending —— pending 是诚实的, 猜译不是。**
            out = _wn.work_name_en(v)
        except Exception:
            pass
    if out:
        _DATA_EN_CACHE[v] = out
        return out
    # 查不到: 经典页写 [translation pending] (显示层再藏); v2 (keep=1) **留原文** —— 前端按原文
    # 标注为"手册原文", 不丢信息也不冒充英文。命中与未命中分开缓存, 两种模式互不污染。
    _DATA_EN_MISS.add(v)
    return v if getattr(_REQ, 'keep_orig', False) else '[translation pending]'


def maint_framework_view():
    """运维管理体系三维骨架 —— **唯一实现** (在线 API 与快照预烤共用)。

    ★同 dq_findings_view 的教训: 这段原来也是两份, 在线那份我接了 unmapped_top 的
    英文化, 预烤那份没有 ⇒ 在线干净、快照里还是中文, 而上云的是快照。
    抽出来一次, 消掉这类"改了一半"的可能。
    """
    import json as _j, pathlib as _p
    from src.windscada.maint import framework as _fw
    db = _j.loads(_P.objects_json().read_text(encoding='utf-8'))
    wo = [(v.get('props') or {}) for k, v in db.items() if k.startswith('workorder/')]
    r = dict(systems=_fw.system_load(wo)['rows'], tiers=_fw.systems_by_tier(),
             stages=_fw.stage_matrix(), records=_fw.RECORDS,
             gaps=_fw.capability_gaps(wo), coverage=_fw.coverage(wo))
    # unmapped_top = **未映射到系统码的现场手写部件名**, 不是枚举, 码表投影对它无效。
    # 走 part_en 自由文本解析; 空值在这个语境下写 "(not recorded)" 而非破折号。
    lang = _lang_of()
    if lang and lang != 'zh' and (r.get('gaps') or {}).get('unmapped_top'):
        try:
            from src.windscada import en_text as _et
            out = []
            for it in r['gaps']['unmapped_top']:
                if isinstance(it, (list, tuple)) and len(it) == 2:
                    e = _et.part_en(it[0]) or it[0]
                    out.append(['(not recorded)' if e == '\u2014' else e, it[1]])
                else:
                    out.append(it)
            r['gaps'] = dict(r['gaps'], unmapped_top=out)
        except Exception as e:
            print('[maint_framework_view] unmapped_top 英文化失败: %s' % e, file=sys.stderr)
    return r


def dq_findings_view():
    """数据可信度 finding 行 —— **唯一实现**。

    ★2026-08-31: 这段原来存在两份 —— 一份在 /api/dq_findings 处理分支, 一份在快照的
    `_dq()` 预烤闭包里。我给 API 那份补了 sensitivity / age_basis 等字段, 预烤那份没动
    ⇒ **在线看得到, 快照里没有**。而上云交付的恰恰是快照。
    同型教训已在 memory `library-gate-not-wired-to-production`: 落库不等于接上生产路径。
    这里直接消掉重复: 两边都调本函数。
    """
    import json as _j, pathlib as _p
    db = _j.loads(_P.objects_json().read_text(encoding='utf-8'))
    KEEP = ('finding/data_quality/', 'finding/caliber/',
            'finding/system_coverage/', 'finding/tier_trend/')
    rows = []
    for k, o in db.items():
        if not k.startswith(KEEP) or not isinstance(o, dict):
            continue
        p2 = o.get('props') or {}
        rows.append(dict(
            id=k, kind=k.split('/')[1],
            name=p2.get('name_zh') or p2.get('name'),
            name_zh=p2.get('name_zh') or p2.get('name'),
            name_en=p2.get('name_en'),
            grade=p2.get('grade'), claim_window=p2.get('claim_window'),
            evidence=p2.get('evidence_zh') or p2.get('evidence') or [],
            evidence_zh=p2.get('evidence_zh') or [],
            evidence_en=p2.get('evidence_en') or [],
            caveat=p2.get('caveat'), caveat_en=p2.get('caveat_en'),
            impact=p2.get('discriminator_zh') or p2.get('disposition_zh')
                   or p2.get('implication_zh') or p2.get('corrected_comparison_zh'),
            impact_en=p2.get('discriminator_en') or p2.get('disposition_en')
                   or p2.get('implication_en') or p2.get('corrected_comparison_en'),
            got_wrong=p2.get('what_i_got_wrong_zh'),
            got_wrong_en=p2.get('what_i_got_wrong_en'),
            quality_flag=p2.get('quality_flag_zh'), quality_flag_en=p2.get('quality_flag_en'),
            robust=p2.get('robust_zh'), robust_en=p2.get('robust_en'),
            basis=p2.get('basis_zh'), basis_en=p2.get('basis_en'),
            sensitivity=(p2.get('sensitivity') or {}).get('note_zh'),
            sensitivity_en=(p2.get('sensitivity') or {}).get('note_en'),
            age_basis=p2.get('age_basis_zh'), age_basis_en=p2.get('age_basis_en'),
        ))
    rows.sort(key=lambda x: (x['kind'] != 'data_quality', x['id']))
    r = dict(rows=rows, n=len(rows))
    if not rows:
        # ★2026-09-18 用户令("页面只能基于输入数据重算")之后实逮: 清过产物再重算的机器上
        #   `objects.json` 里 **0 条 finding/* 对象** ⇒ 本页只是"空", 不说原因就等于让人以为
        #   "系统自查没问题"。这四族的生成端确实不在重算链上(见族表 guanlan/ontology 说明),
        #   所以如实写清, 并明确"不用旧版产出补"。
        r['note'] = ('本体对象库里没有 finding/* 对象 (四族: data_quality / caliber / system_coverage / '
                     'tier_trend) ⇒ 本页**如实为空**, 不是"自查无发现"。这些 finding 的生成端不在重算链上'
                     '(唯一写方 scripts/ingest_ops_2025.py 的输入 data/raw/工作库 不在现场数据里), '
                     '运行期不从交付包补齐(用户令 2026-09-17); 要出件须研发补生成端。')
    return r


def _en_page(page):
    """lang=en 时把页面文案换成英文。**中文版不经过这里** (lang 空 → 直接返回原页)。

    用的是 ui_en 的**整条短语表**, 不是 i18n_en 的逐词词表 —— 后者会产出
    "Verification 状态机" 这类混杂串 (已实测作废)。缺条目由 ui_en.audit 报出,
    不静默留在页面上。
    """
    lang = _lang_of()
    if not lang or lang == 'zh':
        return page
    try:
        from src.windscada import ui_en
        page = ui_en.apply(page)
        # ★译后语法闸: 保护规则一松一紧都能把 JS 译坏, 而症状是**空白骨架页** ——
        #   此时"可见中文=0", 自查全绿。必须用外部解析器判, 不能靠数引号。
        for _m in re.finditer(r'<script[^>]*>(.*?)</script>', page, re.S):
            _code = _m.group(1)
            if _code.strip() and 'application/json' not in _m.group(0):
                ui_en.check_js(_code, label='英文页内联脚本')
        # 数据值 (报警/定论/工单…) 既做比较又要显示。比较照旧用中文, 显示层在 render
        # 之后替换文本节点 —— 只碰显示不碰逻辑, 所以判断不会失效。
        return page.replace('</body>', ui_en.value_script() + '</body>') \
            if '</body>' in page else page + ui_en.value_script()
    except Exception as e:
        # ★退回中文页比给个坏页安全, 但**必须响亮** —— 静默退回会让"英文版没做完"
        #   看起来像"英文版做完了只是有中文残留"。守护步骤失败必须可见。
        print('[_en_page] 🔴 英文化失败, 已退回中文页: %s: %s' % (type(e).__name__, e),
              file=sys.stderr)
        return page


def _jdump(obj, **kw):
    o = _redact(_jclean(obj))
    lang = _lang_of()
    if lang and lang != 'zh':                      # 中文版走这行以外的路径, 一字不动
        o = _proj(o, lang)
    return json.dumps(o, allow_nan=False, **kw)


def _gap_page(path: str, missing: 'pathlib.Path | None' = None, what: str = '这一块') -> str:
    r"""产物缺失时的兜底页 (2026-09-17 现场问题驱动)。

    原来 `/static/…`、`/turbines/…` 取不到文件就回一个裸 `404` 文本 —— 浏览器里表现为**一块空白**,
    用户看到的现象是"这个板块没有信息", 而不是"缺件"。最典型的就是总览页那个
    `<iframe src="/static/index.html">`(全场状态总览页)。

    ★2026-09-19 更新 (用户令「所有的计算均要形成观澜的源代码」): 该页**已有生成端** ——
    `scripts/windscada_overview_build.py`（用组件同一个快照烘焙器从重算产物生成), 重算链第 ⑦b 步会跑它,
    所以现在"不在位"只意味着**还没重算**, 不再是"包内没有生成端"。本页说明文字已据此改写。

    现在改成结构化说明: 缺哪一件、它归哪条产物线、**怎么补**、去哪儿看全局状态。
    口径与 CMS 的"无产物"页一致 (docs §11.4: 无产物时必须还能打开页面, 且要写明缺件与补齐办法)。
    """
    import html as _h
    rel = ''
    try:
        rel = pathlib.Path(missing).relative_to(pathlib.Path(_P.ROOT)).as_posix() if missing else ''
    except Exception:
        rel = str(missing or '')
    return ('<!doctype html><meta charset="utf-8"><title>产物不在位</title>'
            '<style>body{font:14px/1.7 system-ui,"Microsoft YaHei",sans-serif;color:#183548;background:#f2f6f8;'
            'margin:0;display:flex;align-items:center;justify-content:center;min-height:100vh}'
            '.b{background:#fff;border:1px solid #d4e3e7;border-radius:10px;padding:22px 26px;max-width:800px;'
            'box-shadow:0 6px 20px rgba(14,74,87,.08)}h1{font-size:17px;margin:0 0 10px}'
            'code{background:#eef4f6;border-radius:4px;padding:1px 5px}.s{color:#667f8e;font-size:13px}'
            'a{color:#087f89}li{margin:3px 0}</style>'
            f'<div class="b"><h1>{_h.escape(what)}的数据产物不在位</h1>'
            f'<p class="s">请求路径 <code>{_h.escape(path)}</code></p>'
            + (f'<p>缺的是 <code>{_h.escape(rel)}</code>（属 <code>outputs/&lt;场站&gt;/</code> 下的产物）。</p>'
               if rel else '')
            + '<p>两种情形分开补:</p><ul>'
            '<li><b>有生成端的产物</b>（alarms / workorders / temp_monthly / 振动窗索引谱 / '
            '<code>windscada/index.html</code> …）: 放好 '
            '<code>data/raw/&lt;场站&gt;/</code> 后跑 <code>python scripts/rebuild_all.py</code>, '
            '或点运维控制台的「数据重算」。</li>'
            '<li><b>仍缺生成端的那几族</b>（如 <code>sop/</code>、<code>paradigm_r1/</code> 里的人工底稿、'
            '<code>windcms/</code> 的厂家报告转录）: 按用户令 2026-09-19「所有的计算均要形成观澜的源代码」'
            '正在逐族补（清单见 <code>docs/源代码化清单_v0.1.md</code>）; 在此之前页面如实显示缺件, '
            '且按用户令 2026-09-17 <b>运行期不从交付包补齐</b>。'
            '（2026-09-19 已补上生成端的: 变桨面 <code>pitch/**</code> ← <code>scripts/pitch_face_build.py</code>; '
            '振动在升/闭环 <code>m5_cms_tcm/component_history.json</code>、三层基线 <code>baseline_38.json</code> '
            '← <code>scripts/component_history_build.py</code> / <code>scripts/baseline_38_build.py</code>; '
            '事实契约 <code>guanlan/**</code> ← 重算台账生成 claim; '
            '发布层 <code>ontology/release_r1|r2</code> ← <code>scripts/ontology_release_build.py</code>。）</li></ul>'
            '<p class="s">逐件来源见 <code>outputs/&lt;场站&gt;/_provenance.json</code>；'
            '缺口清单见 <code>docs/系统设计说明.md</code> §7 与 docs/重算缺口与补件清单_v0.1.md。</p>'
            '<p><a href="/v2" target="_top">← 返回工作台</a> · '
            '<a href="http://127.0.0.1:28084/ops" target="_top">运维控制台 · 数据重算</a> · '
            '<a href="http://127.0.0.1:28084/healthz" target="_top">全部模块状态</a></p></div>')

from http.server import ThreadingHTTPServer, BaseHTTPRequestHandler
from src.windscada.config import farm
from src.windscada import taxonomy
from src.windscada.subsys import temp_nbm, hydraulic, yaw as yawmod, pitch as pitchmod
from src.windscada.perf import availability
from src.windscada import i18n

CFG = farm(); ST = pathlib.Path(CFG['store'])
_CJK = re.compile(r'[\u4e00-\u9fff]')


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}')  # 响亮, 不静默空表

LOCK = threading.Lock()
_CACHE = {}


# ── 产物指纹 → **运行中自动重载** (2026-09-16 用户令: "重算后的产物可即时用于页面呈现") ──────
# 背景 (实测): _load() 把产物**一次性**读进内存, 之后只在启动时读一次 ⇒ 重算完成后页面仍显示旧数,
# 必须重启组件服务才更新。判定依据: 改产物后 /detail/api/fleet 等仍回旧值, 重启组件后才变。
# 做法: 取"页面取数依赖的产物文件"的 (路径, mtime_ns, size) 摘要当指纹; _load() 前比一次
# (带 TTL, 免得每请求都 stat 一遍), 指纹变了就重载 ⇒ 重算产物**不再需要重启服务**。
# 也提供 /api/reload 显式重载 (运维控制台重算结束后可点一下)。
_STAMP_TTL = 2.0          # 秒: 指纹有效期 (stat ~150 个文件 ≈ 1~3 ms, 不值得每请求都做)
_STAMP = {'val': None, 'at': 0.0}


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']


class ProductsMissing(RuntimeError):
    """产物缺失 (outputs/<场>/… 被清空或尚未生成)。

    2026-09-11 人工检查空状态时逮: 原来 `temp_monthly.parquet` 一缺, `_load()` 直接抛 FileNotFoundError,
    外层处理器把它当**未知异常**回 500 + traceback —— 人看到的是"服务器错误", 分不清"没算产物"还是"程序坏了"。
    现在这类缺失单独成类, 由请求层回**结构化的无数据**(HTTP 200 + err=no_products), 与 `adhoc_query` 的
    `no_source` 一致。真 bug 仍然照旧 500 并带 traceback (不静默)。"""

    def __init__(self, name, path):
        super().__init__(f'产物缺失: {name} (期望 {path})')
        self.what, self.path = name, path


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


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

WINDOWS = ['近30日', '近90日', '2026年', '2026H1', '2025H2', '全程']
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 in_win(ts, win):
    """按窗过滤**有日粒度**的时间列（含两端）。月度列请用 `months_of()`（闰月/跨月取整不同）。"""
    import pandas as pd
    a, b = win_range(win)
    t = pd.to_datetime(ts, errors='coerce')
    return (t >= pd.Timestamp(a)) & (t <= pd.Timestamp(b) + pd.Timedelta(days=1) - pd.Timedelta(seconds=1))

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


# 九系统的 slug（前端 hash 用）↔ 中文名（服务端取数用）。**必须双向可认**：
# 2026-09-22 用户测试 3/5 实逮 —— v2 的「该问题页」「命中系统」链接过去传的是 slug（`pitch`），
# 而 `/api/problem` 与页面按中文名取数 ⇒ 返回 0 条后页面还写"该系统在判级时间窗内无非常态"（对报警机组是**假陈述**）。
SYS_SLUG = {'变桨': 'pitch', '偏航': 'yaw', '齿轮箱': 'gearbox', '发电机': 'generator',
            '主轴承': 'main_bearing', '变流器': 'converter', '塔架与基础': 'tower',
            '主控与传感网': 'control', '叶片与叶根': 'blade'}
SYS_FROM_SLUG = {v: k for k, v in SYS_SLUG.items()}


def sys_norm(x):
    """系统标识归一：中文名 / 前端 slug / 大小写变体 → 规范中文名；认不出返回 None。"""
    if not x:
        return None
    s = str(x).strip()
    if s in SYS_SLUG:
        return s
    low = s.lower().replace('-', '_')
    if low in SYS_FROM_SLUG:
        return SYS_FROM_SLUG[low]
    for zh, slug in SYS_SLUG.items():                 # 兜底: 去空格/全半角差异后再比
        if s.replace(' ', '') == zh.replace(' ', '') or low == slug.replace('-', '_'):
            return zh
    return None


# ── 按窗重算：后台算 + 缓存（用户令 2026-09-21）─────────────────────────────────
# 判级矩阵 ≈25 s/窗、七镜头 ≈9~44 s/窗 —— 直接同步算会把 HTTP 请求挂几十秒（远端还要过网关，
# 长请求有被掐的风险）。所以统一口径：**缓存命中即回；未命中就起后台线程，本次先回 pending，
# 页面轮询再取**。同一窗只算一次；算完常驻内存（进程内）。
import threading as _thr

_WIN_CACHE: dict = {'sysmx': {}, 'curves': {}, 'm9': {}}
_WIN_BUSY: dict = {'sysmx': set(), 'curves': set(), 'm9': set()}
_WIN_ERR: dict = {'sysmx': {}, 'curves': {}, 'm9': {}}
_WIN_LOCK = _thr.Lock()
# ★重活**串行**（2026-09-21 服务器实逮）: 预热与请求各自起线程 ⇒ 同时跑两个按窗重算（各几百 MB 级
#   中间帧），小内存机器上进程被系统杀掉（日志停在"按窗重算启动"两行、随后网关 503）。
#   现在全部重活排队执行：峰值 = 1 份中间帧。
_HEAVY_LOCK = _thr.Lock()


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 _win_wait(kind, win, timeout=120.0):
    """等某窗的按窗重算落地（预热用）。超时/失败返回 None，绝不抛。"""
    import time as _t
    key = _win_key(win)
    t0 = _t.time()
    while _t.time() - t0 < timeout:
        with _WIN_LOCK:
            if key in _WIN_CACHE[kind]:
                return _WIN_CACHE[kind][key]
        if key in _WIN_ERR[kind]:
            return None
        _t.sleep(1.0)
    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 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 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

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

# ---------- 图规格 (分册两型) ----------
def fig_temp(t, ch, win):
    tm = _CACHE['tm']; ms = months_of(win)
    g = tm[tm.channel == ch]
    piv = g.pivot_table(index='month', columns='turbine', values='med').reindex(ms)
    if t not in piv.columns or piv[t].dropna().empty: return None
    nm = i18n.ch_cn(ch, with_tag=False)
    return dict(kind='band', title=f'{t} {nm} 月度 (发电态P>500kW)', unit='℃', months=ms,
                self=[None if v != v else round(v, 2) for v in piv[t]],
                med=[None if v != v else round(v, 2) for v in piv.median(axis=1)],
                q1=[None if v != v else round(v, 2) for v in piv.quantile(.25, axis=1)],
                q3=[None if v != v else round(v, 2) for v in piv.quantile(.75, axis=1)],
                note='灰带=全场四分位区间, 深线=全场中位, 红线=本台')

def fig_duty(t, ch, title, unit, win, thresholds=None):
    d = _CACHE['duty']
    if d is None: return None
    ms = months_of(win)
    g = d[d.channel == ch]
    piv = g.pivot_table(index='month', columns='turbine', values='val').reindex(ms)
    if t not in piv.columns or piv[t].dropna().empty: return None
    return dict(kind='band', title=f'{t} {title}', unit=unit, months=ms,
                self=[None if v != v else round(v, 2) for v in piv[t]],
                med=[None if v != v else round(v, 2) for v in piv.median(axis=1)],
                q1=[None if v != v else round(v, 2) for v in piv.quantile(.25, axis=1)],
                q3=[None if v != v else round(v, 2) for v in piv.quantile(.75, axis=1)],
                thresholds=thresholds or [], note='灰带=全场四分位区间')

def fig_alarm_bars(t, codes, title, win):
    al = _CACHE['al']; ms = months_of(win)
    g = al[(al.turbine == t) & (al.code.isin(codes)) & (al.month.isin(ms))]
    if not len(g): return None
    cnt = g.groupby('month').size().reindex(ms).fillna(0)
    return dict(kind='bars', title=f'{t} {title} 月度条数', unit='条', months=ms,
                vals=[int(v) for v in cnt], note='时间窗内合计 %d 条' % int(cnt.sum()))

def fig_curve(t):
    b = _CACHE['bins']; g = b[b.turbine == t].sort_values('bin')
    if not len(g): return None
    dv = _CACHE['pcd'].loc[t]
    return dict(kind='xy', title=f'{t} 功率曲线 vs fleet中位 [固定判别窗 2025-H2, 机舱风口径]', unit='kW',
                x=[float(x) for x in g['bin']], self=[round(float(v), 1) for v in g.p],
                med=[round(float(v), 1) for v in g.fleet_p], xlab='风速 m/s',
                note=f"偏差 {dv.dev_w:+.1%} · 风速计偏置 {dv.ws_bias:+.2f} m/s · {dv['判别']}")

def fig_zero(t, win):
    z = _CACHE['zero']
    if z is None: return None
    tt = t.replace('WTG', '')
    g = z[(z.turbine.astype(str).str.contains(tt)) | (z.turbine == t)] if 'turbine' in z.columns else None
    if g is None or not len(g): return None
    ms = months_of(win)
    mcol = [c for c in g.columns if 'month' in c.lower() or '月' in c]
    if not mcol: return None
    g = g[g[mcol[0]].astype(str).isin(ms)]
    # 只画零位偏差本身 (样本数量纲不同, 混画会把判据线压扁 → 移入图注)
    n_med = int(g['n_rows'].median()) if 'n_rows' in g.columns and len(g) else 0
    return dict(kind='multiline', title=f'{t} 桨叶零位偏差 月度', unit='°', months=[str(x) for x in g[mcol[0]]],
                series=[dict(name='零位偏差', vals=[round(float(v), 3) if v == v else None for v in g['zero_dev']])],
                thresholds=[dict(v=0.6, label='报警 0.6°'), dict(v=-0.6, label='')],
                note=f'判据: 变桨分册 §3.2 (报警门 ±0.6°); 月样本中位 {n_med} 行')

SYS_ALARM = {'偏航': yawmod.FAM, '变桨': {'蓄能缸检': ['3411', '3412', '3413'], '泵/过热': ['7106', '7100'], '油位/滤芯': ['7101', '7103', '7104']}}

SYS_CODES = {
    '变桨': ['3130', '3145', '3173', '3224', '3225', '3411', '3412', '3413', '3418', '7100', '7101', '7102', '7103', '7104', '7106', '9306', '9309'],
    '偏航': ['10100', '10104', '10105', '10113', '10125', '10201', '10400', '10401', '7122', '1012'],
    '齿轮箱': ['4101', '4102', '4103'],
    '变流器': ['13902', '13905', '19024', '19170', '19285', '20000', '63036'],
    '主控与传感网': ['8108', '8110', '8171', '8184', '8185', '8186', '8187', '63209', '63004'],
}

def fig_sys_alarms(t, sysname, win):
    """该台该系统报警族 月度条数 (与工单处理史并读)."""
    codes = SYS_CODES.get(sysname)
    if not codes: return None
    ms = months_of(win)
    al = _CACHE['al']
    g = al[(al.turbine == t) & (al.code.isin(codes)) & (al.month.isin(ms))]
    if not len(g): return None
    v = g.groupby('month').size().reindex(ms).fillna(0)
    top = g.groupby('code').size().sort_values(ascending=False).head(3)
    note = '时间窗内合计 %d 条; top: ' % len(g) + '; '.join(f'{c} {i18n.alarm_label(c)[:12]}×{n}' for c, n in top.items())
    return dict(kind='bars', title=f'{t} {sysname}系统 报警月度条数', unit='条', months=ms,
                vals=[int(x) for x in v], note=note)


def turbine_problems(t, win):
    """有问题的系统 → 证据图 (分册式); 无问题不出."""
    _load()
    sysd = _CACHE['sysmx'][t]
    treg = _CACHE['treg']
    out = []
    for s in taxonomy.SYSTEMS:
        st = sysd[s]['状态']
        if st == '优秀': continue
        figs = []
        # 温度证据
        for _, r in treg[(treg.turbine == t)].iterrows():
            ch = r['channel']
            nm = ch.replace('tmp_wtc_', '').replace('_mean', '').replace('_max', '')
            if taxonomy.TEMP2SYS.get(nm) == s and str(r['判']) != '—':
                f = fig_temp(t, ch, win)
                if f: f['title'] = f"{t} {i18n.ch_cn(ch, with_tag=False)} — {r['判']} ({r['dev_K']:+.1f}K)"; figs.append(f)
        # 系统专属证据
        if s == '变桨':
            hv = _CACHE['hyd'].loc[t]
            if hv['判'] != '—':
                f = fig_duty(t, 'HydPress_band', '变桨液压10min摆幅 月度', 'bar', win,
                             thresholds=[dict(v=52, label='正常充压滞环≈52bar')])
                if f: f['title'] = f"{t} 蓄能行为 — {hv['判']}"; figs.append(f)
            for nm, codes in SYS_ALARM['变桨'].items():
                f = fig_alarm_bars(t, codes, f'变桨{nm}报警', win)
                if f: figs.append(f)
            f = fig_zero(t, win)
            if f: figs.append(f)
        if s == '偏航':
            yrow_d = dict(zip(['D1润滑不可能', 'D2过热', 'D2变流器警告', 'D2解缆', 'D3制动液压错误', 'D3油位低', 'D4偏航失败', 'D4风传感'], [0]*8))
            f = fig_duty(t, 'dot_wtc_YawLubPu_timeon', '偏航润滑泵运行时长 月度', '秒/日', win)
            if f: figs.append(f)
            for nm, codes in yawmod.FAM.items():
                f = fig_alarm_bars(t, codes, '偏航·' + nm.lstrip('D1234'), win)
                if f: figs.append(f)
        if s in ('主轴承', '齿轮箱', '发电机'):
            vf = vib_scalar_fig(t, s)
            if vf: figs.append(vf)
            # 振动驱动的问题 (taxonomy 振动映射) → 配套分析曲线: 部件双端温度 + 主轴承柱塞 duty
            if '振动:' in sysd[s]['依据']:
                vib_chs = {'主轴承': ['tmp_wtc_MBearGTm_mean', 'tmp_wtc_MBearHTm_mean'],
                           '发电机': ['tmp_wtc_GenBeRTm_mean', 'tmp_wtc_GenBeGTm_mean'],
                           '齿轮箱': ['tmp_wtc_IMSGenTm_mean', 'tmp_wtc_HSGenTmp_mean']}[s]
                for ch in vib_chs:
                    f = fig_temp(t, ch, win)
                    if f: f['note'] = (f.get('note') or '') + ' ｜ 振动判级交叉轴: 温度=同机制果侧观测(观测面≠源数)'; figs.append(f)
                if s == '主轴承':
                    f = fig_duty(t, 'din_wtc_MBLubPis_timeon', '主轴承润滑柱塞运行时长 月度 (因侧执行链)', '秒/日', win)
                    if f: f['note'] = (f.get('note') or '') + ' ｜ 29#签名: 单次时长塌→回升→2026-07单月突变(泵打不满)'; figs.append(f)
                # 振动摘要配传动链图形 (2026-08-28 用户令: 关联图形填入) — comp=依据里的部件短语, sev=结论级
                _why = sysd[s]['依据']
                _m = re.search(r'振动: ([^ ]+) ([^ (]+)', _why)
                _comp, _lv = (_m.group(1), _m.group(2)) if _m else (s, '')
                _sev = 'var(--bad)' if any(k in _lv for k in ('定论', '报警')) else ('var(--warn)' if '预警' in _lv else 'var(--ok)')
                figs.append(dict(kind='vib', title=f'{t} {s} 振动分析摘要',
                                 comp=_comp, sev=_sev,
                                 text=_why + ' ｜ 谱级深潜与逐窗趋势 → 振动融合板块·振动分析入口 (windcms)'))
            if '配对差' in sysd[s]['依据']:
                pair_chs = {'主轴承': ['tmp_wtc_MBearHTm_mean', 'tmp_wtc_MBearGTm_mean'],
                            '发电机': ['tmp_wtc_GenBeGTm_mean', 'tmp_wtc_GenBeRTm_mean']}.get(s, [])
                for ch in pair_chs:
                    f = fig_temp(t, ch, win)
                    if f: figs.append(f)
            if '润滑' in sysd[s]['依据']:
                ch = {'主轴承': 'din_wtc_MBLubPis_timeon', '齿轮箱': 'din_wtc_GBLubPis_timeon', '发电机': 'dot_wtc_GenLubPu_timeon'}[s]
                f = fig_duty(t, ch, f'{s}润滑 月度', '秒/日', win)
                if f: figs.append(f)
        if s == '主控与传感网':
            if '测风' in sysd[s]['依据'] or '风速计' in sysd[s]['依据']:
                f = fig_alarm_bars(t, yawmod.FAM['D4风传感'], '风传感故障报警', win)
                if f: figs.append(f)
            if '曲线判风速计' in sysd[s]['依据'] or '风速计偏置' in sysd[s]['依据']:
                f = fig_curve(t)
                if f: figs.append(f)
        # ★每个问题配: 该系统报警月度图 + 工单处理史 (用户令)
        try:
            from src.windscada.subsys import workorder as womod
            hist = womod.history(t, s)
        except Exception as e:
            hist = dict(n=0, rows=[], note=f'工单不可用 {e}'[:60], 动作统计={}, 窗='')
        af = fig_sys_alarms(t, s, win)
        if af: figs.append(af)
        # 排版 (2026-08-28 用户令"排版更合理"): ①同标题图去重 (配对差/温度证据两处会重复挂同一通道)
        # ②按阅读顺序排 — 结论定位(传动链) → 证据强度(振动标量) → 时序(温度/润滑/报警) → 曲线
        seen, uniq = set(), []
        for _f in figs:
            k = (_f.get('kind'), _f.get('title'))
            if k in seen: continue
            seen.add(k); uniq.append(_f)
        _ORD = {'vib': 0, 'vibbar': 1, 'band': 2, 'line': 3, 'bar': 4, 'xy': 5, 'text': 6}
        figs = sorted(uniq, key=lambda x: _ORD.get(x.get('kind'), 9))
        out.append(dict(sys=s, 状态=st, 依据=i18n.humanize(sysd[s]['依据']), figs=figs, orders=hist))
    dv = _CACHE['pcd'].loc[t] if t in _CACHE['pcd'].index else None
    if dv is not None and abs(dv.dev_w) > 0.03:
        f = fig_curve(t)
        out.append(dict(sys='功率曲线', 状态='报警' if 'A类' in str(dv['判别']) else '良好',
                        依据=f"偏差 {dv.dev_w:+.1%}, 风速计偏置 {dv.ws_bias:+.2f} m/s → {dv['判别']}", figs=[f] if f else []))
    out.sort(key=lambda x: TS.get(x['状态'], 9))
    return out

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', '—')]

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与低转速混合态档'))


_VIBSENS = {'主轴承': ['Main_bearing_front', 'Main_bearing_rear'],
            '齿轮箱': ['Gear_planet', 'Gear_IMS', 'Gear_HS_rotor_side', 'Gear_HS_generator_side'],
            '发电机': ['Generator_DE', 'Generator_NDE']}
_SENS_CN = {'Main_bearing_front': '主轴承前', 'Main_bearing_rear': '主轴承后', 'Gear_planet': '行星级',
            'Gear_IMS': '中间轴', 'Gear_HS_rotor_side': '高速轴转子侧', 'Gear_HS_generator_side': '高速轴发电机侧',
            'Generator_DE': '发电机DE', 'Generator_NDE': '发电机NDE'}
_VIBMET = [('Peak', '冲击峰值'), ('Kurtosis', '峭度'), ('Rms_Vel', '速度有效值'), ('iso_rms', 'ISO烈度')]
_VIBC = {}


def vib_scalar_fig(t, sysname):
    """TCM 标量: 本台 vs 全场中位 (按测点×指标) + 六窗趋势. 数据=振动线 m5_cms_tcm 产物, 本页只转录不判级."""
    import pandas as _pd
    if 'z' not in _VIBC:
        base = _P.m5()
        try:
            _VIBC['z'] = _pd.read_parquet(base / 'fleet_scalar_z.parquet')
            _VIBC['tr'] = _pd.read_parquet(base / 'scalar_trend_6win.parquet')
        except Exception:
            _VIBC['z'] = _VIBC['tr'] = None
    z, tr = _VIBC['z'], _VIBC['tr']
    if z is None: return None
    sens = _VIBSENS.get(sysname, [])
    if not sens: return None
    g = z[(z.turbine == t) & (z.sensor.isin(sens)) & (z.meas.isin([m for m, _ in _VIBMET]))]
    if not len(g): return None
    # 取最高功率档 (载荷高处信噪比最好); 该档缺则退最近档
    order = sorted(g.bin.unique())
    gb = g[g.bin == order[-1]]
    rows = []
    for sn in sens:
        for mk, mcn in _VIBMET:
            r = gb[(gb.sensor == sn) & (gb.meas == mk)]
            if not len(r): continue
            r = r.iloc[0]
            tv = None
            if tr is not None:
                q = tr[(tr.turbine == t) & (tr.sensor == sn) & (tr.metric == mk)]
                if len(q): tv = float(q.iloc[0]['ratio'])
            rows.append(dict(测点=_SENS_CN.get(sn, sn), 指标=mcn, 本台=round(float(r.val), 3),
                             全场中位=round(float(r.fleet_med), 3), z=round(float(r.z), 2),
                             倍中位=round(float(r.val) / float(r.fleet_med), 2) if r.fleet_med else None,
                             六窗比=round(tv, 2) if tv else None))
    if not rows: return None
    return dict(kind='vibbar', size='L', title=f'{t} {sysname} 振动标量 · 本台 vs 全场 ({str(order[-1]).replace("WPS-ActivePower ", "功率 ").rstrip(",")} kW 档)',
                rows=rows, note='数据=TCM 六窗标量 (振动线 m5_cms_tcm 产物转录); z=fleet 稳健离群度, 六窗比=末窗/首窗 (>1 上升)')


def single_problem(t, sys, win):
    """只回该台该系统这一个问题 (用户令: 点开就是一个问题, 其他不带出来)."""
    return [p for p in turbine_problems(t, win) if p['sys'] == sys]


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 '')))

# ---------- 本体网页面 (ontology 的网页体现: 机制链/决策台/机制库/对象浏览器; 只读, 复用 MCP 纯函数) ----------
def ontology_view():
    from src.ontology import mcp_server as om
    d = om.db()
    from collections import Counter
    stats = dict(Counter(v['type'] for v in d.values()))
    mechs = [dict(id=o['id'], **{k: o['props'].get(k) for k in ('name', 'status', 'facets', 'source')},
                  affects=o.get('links', {}).get('affects', []))
             for o in d.values() if o['type'] == 'FailureMechanism']
    # 29# 机制链 (§2 规划补齐): 机制为中心的分层链
    ev = [dict(id=i, **{k: d[i]['props'].get(k) for k in ('facet', 'claim_window', 'summary', 'source')})
          for i in d['mechanism/WTG29/lub_starvation']['links']['evidencedBy']]
    chain29 = dict(
        mechanism=dict(id='mechanism/WTG29/lub_starvation', **d['mechanism/WTG29/lub_starvation']['props']),
        evidence=ev,
        verdict=dict(id='verdict/WTG29/TCM-16', **{k: d['verdict/WTG29/TCM-16']['props'].get(k)
                                                   for k in ('level', 'title', 'review_tier', 'claim_window')}),
        scenario=dict(id='scenario/WTG29/repair_scheduling_v1',
                      **{k: d['scenario/WTG29/repair_scheduling_v1']['props'].get(k)
                         for k in ('key_insight', 'recommendation', 'scenarios')}),
        action=dict(id='action/WTG29/repair_pump', **d['action/WTG29/repair_pump']['props']),
        acceptance=dict(id='acceptance/WTG29/state', **d.get('acceptance/WTG29/state', {}).get('props', {})),
        tree=dict(id='decision/WTG29/acceptance_tree', **d['decision/WTG29/acceptance_tree']['props']))
    decisions = [dict(id=o['id'], **{k: o['props'].get(k) for k in ('decided_by', 'when', 'decision', 'status', 'outcome')})
                 for o in d.values() if o['id'].startswith('decision/user') or o['id'].startswith('decision/pending')]
    roll = d.get('meta/roll_state', {}).get('props', {})
    constraints = ('C1 证据必带claim_window+观测面+源身份 · C2 判级必带审级+证据链接 · C3 独立源数按观测面去重禁按条数 · '
                   'C4 判读三分(支持|排除|不可判, 不可判≠阴性) · C5 状态措辞窗相关必带窗 · C6 动作必带可逆级(L3须审级)')
    weather = d.get('mwindow/rudong/climatology', {}).get('props', {}).get('monthly', {})
    return dict(stats=stats, constraints=constraints, mechanisms=mechs, chain29=chain29,
                decisions=decisions, roll=roll, weather=weather)


def ontology_list(otype, turbine, limit=60):
    from src.ontology import mcp_server as om
    return om.ont_query(type=otype or '', turbine=turbine or '', limit=int(limit)).get('data', [])


def ontology_obj(oid):
    from src.ontology import mcp_server as om
    r = om.ont_get(id=oid)
    lk = om.ont_links(id=oid).get('data', {})
    return dict(obj=r.get('data'), links=lk, err=r.get('error'))


# ---------- AI 问答 (P2 离线编排的网页入口; dsh + ontology MCP; 串行锁=一次一问, windcms五类错教训) ----------
CUR_WIN = ['2026年']          # 最近一次 fleet 查询的窗, 供报告助手复用
ASK = {'state': 'idle', 'q': '', 'model': '', 'answer': '', 'verify': None, 't0': 0, 'err': '',
       'step': '', 'steps': [], 'review': None, 'xrev': False, 'lang': 'zh'}
# step/steps: 推理进度 (229s 全程无反馈会被读成"给不出结果"); review: 另一模型的交叉审核结果
_ASK_STEP_KEY_EN = {
    '启动': 'Starting',
    '预取': 'Prefetch',
    '推理': 'Reasoning',
    '取证': 'Evidence retrieval',
    '成文': 'Drafting',
    '校闸': 'Grounding gate',
    '交叉审核': 'Cross-review',
    '完成': 'Done',
    '失败': 'Failed',
}


def _ask_step_text(k, d='', lang='zh'):
    k = '' if k is None else str(k)
    d = '' if d is None else str(d)
    if lang != 'en':
        return f'{k}· {d}' if d else k
    key = _ASK_STEP_KEY_EN.get(k, k)
    detail = d.strip()
    m = re.search(r'第\s*(\d+\s*/\s*\d+)\s*轮', detail)
    if m:
        detail = 'round ' + m.group(1).replace(' ', '')
    else:
        m = re.search(r'(\d{1,2})\s*一站式证据包', detail)
        if m:
            alias = 'focus unit'
            try:
                from src.windscada.deid import unitize_answer
                alias = unitize_answer(f"WTG{int(m.group(1)):02d}")
            except Exception:
                pass
            detail = f'{alias} evidence pack'
        elif '闭环' in detail and '证据' in detail:
            n = ''.join(re.findall(r'\d+', detail))
            detail = f'closure and trend evidence: {n} records' if n else 'closure and trend evidence'
        elif '全场' in detail and '判级' in detail:
            n = ''.join(re.findall(r'\d+', detail))
            detail = f'fleet ratings: {n} units' if n else 'fleet ratings'
        elif '换模型' in detail or '复核' in detail:
            detail = 'reviewing with another model'
        elif re.search(r'[\u3400-\u9fff\u3000-\u303f\uff00-\uffef]', detail):
            detail = ''
    if re.search(r'[\u3400-\u9fff\u3000-\u303f\uff00-\uffef]', key):
        key = 'Working'
    if re.search(r'[\u3400-\u9fff\u3000-\u303f\uff00-\uffef]', detail):
        detail = ''
    return f'{key} · {detail}' if detail else key

# 模型档位 (2026-08-28 用户裁: 去掉 qwen3:30b-a3b, 用 27B)。
# 每档标注适用场景与实测耗时 — 让人按"这个问题值不值得等"来选, 而不是凭型号大小猜。
ASK_MODELS = {'qwen3:8b': 'deploy/dsh/model.qwen8b.patch.yml',
              'qwen3.8:27b': 'deploy/dsh/model.qwen27b.patch.yml',
              # 2026-09-06 用户裁 "加档": 部署包写的是 "本机 Ollama 的 DeepSeek 模型", 机上实有 r1 14b/32b 却选不到.
              #   本地档一律走 fast_agent 快路径 (Ollama 直连), 不走 dsh, 故 patch 为 None; r1 不支持 tools 时快路径自动退化为预取直答 (见 fast_agent._chat).
              'deepseek-r1:14b': None, 'deepseek-r1:32b': None,
              'qwen3:30b-a3b': None,   # 2026-09-07: 8B 校闸未过的自动升档目标 (30B 参数 / 3B 激活, 单轮 ~20 s)
              'deepseek-v4-pro': None}
# ★这张表原来只有本地模型时代的三档, **不含实际部署的 deepseek-chat / kimi-fast /
#   qwen-plus** —— 于是 MODEL_SPEC[sel.value] 落空, v.warn 为空, 徽章就显示「零外发」。
#   云端模型对用户说"零外发"是**假保证**, 比时间写错严重得多 (2026-09-01 用户说"太慢"
#   时顺带查出来的)。补全 + 下面 mdlNote 的兜底改为"按外发处理", 宁可说重不可说轻。
MODEL_SPEC = {
    'qwen3:8b':        dict(nm='轻量 8B',  loc='本地', t='约 30 到 60 秒', use='默认档: 快速简答问答 (用户裁 2026-09-07)'),
    'qwen3.8:27b':     dict(nm='标准 27B', loc='本地', t='约 2 到 4 分钟', use='复杂问题档: 多台横向比对与机制推断 (用户裁 2026-09-07)'),
    'qwen3:30b-a3b':   dict(nm='快 30B (a3b)', loc='本地', t='约 20 到 60 秒', use='8B 校闸未过时的自动升档目标; 也可直接选'),
    'deepseek-r1:14b': dict(nm='本机 DeepSeek 14B', loc='本地', t='约 1 到 3 分钟', use='交叉审核审核档 (作答请优先 qwen 27B; 用户裁 2026-09-07)'),
    'deepseek-r1:32b': dict(nm='本机 DeepSeek 32B', loc='本地', t='约 3 到 8 分钟', use='备用: 更强更慢, 引用格式不稳 (约三成概率不出文)'),
    'deepseek-chat':   dict(nm='DeepSeek', loc='云端', t='约 3 到 10 秒',  use='默认档, 状态查询与机制推断',
                            warn='内容会外发到模型厂商, 场名与台号已去标识化'),
    'kimi-fast':       dict(nm='Kimi',     loc='云端', t='约 5 到 20 秒',  use='长文归纳',
                            warn='内容会外发到模型厂商, 场名与台号已去标识化'),
    'qwen-plus':       dict(nm='通义千问',  loc='云端', t='约 5 到 20 秒',  use='备用云端档',
                            warn='内容会外发到模型厂商, 场名与台号已去标识化'),
    'deepseek-v4-pro': dict(nm='云端 DS',  loc='云端', t='约 2 到 6 分钟', use='复杂推理与交叉审核',
                            warn='内容会外发到模型厂商, 场名与台号已去标识化'),
}


def _warm(model):
    # 预热+常驻: 空载请求把模型拉进内存并 keep_alive 30m (与 dsh 冷启并行, 消掉 18GB 级冷加载)
    try:
        import urllib.request
        urllib.request.urlopen(urllib.request.Request(
            'http://127.0.0.1:11434/api/generate', method='POST',
            data=json.dumps(dict(model=model, prompt='', keep_alive='30m')).encode(),
            headers={'Content-Type': 'application/json'}), timeout=120).read()
    except Exception:
        pass


def _ask_worker(q, model, lang='zh'):
    import subprocess, threading, time as _t
    out = _P.ont() / 'ask_last.out'
    try:
        if MODEL_SPEC.get(model, {}).get('loc') == '本地':   # 本地档 (qwen3 / deepseek-r1) 走直连快路径; 原按 'qwen' 前缀硬判, 加 r1 档后改按档表 loc
            # 本地模型走直连快路径 (fast_agent: ollama chat+tools 同进程, 砍 dsh/node/MCP 三层;
            # 实测 8b 116s→19s; 同一套工具与闸, 服务端强制双闸). deepseek 云端仍走 dsh 完整编排.
            threading.Thread(target=_warm, args=(model,), daemon=True).start()
            sys.path.insert(0, str(pathlib.Path('.').resolve()))
            sys.path.insert(0, str(pathlib.Path('scripts').resolve()))
            from src.ontology.fast_agent import answer_escalate as _fast   # 2026-09-07: 8B 拦了自动升 a3b 重答 (用户令)

            def _on_step(k, d=''):
                ASK['step'] = _ask_step_text(k, d, lang)
                ASK.setdefault('steps', []).append(ASK['step'])

            def _on_partial(txt):
                # 先快后准 (用户裁 2026-09-07): 30 s 内先出初答 (契约直出), 状态 partial, 前端继续轮询; 27B 出来后整段替换
                ASK.update(answer=txt, state='partial', partial=True)

            r = _fast(q, model, on_step=_on_step, lang=lang, on_partial=_on_partial)
            _esc = f" · 升档 {r['escalated_from']}→{r['model']}" if r.get('escalated_from') else ''
            ans = r['answer'] + f"\n\n({r['gate']} · {r['rounds']}轮 · 直连快路径{_esc})"
            out.write_text(ans, encoding='utf-8')     # 答案是中文: 缺 encoding 在中文 Windows 上按 GBK 写, 可能崩
            from ontology_p2_verify import verify as _v0
            ASK.update(answer=ans, verify=_v0(ans, lang=lang), partial=False, refined_from=r.get('escalated_from'))
            # 交叉审核: 换一个本地模型复核同一份证据 (同模型自审 = 自己给自己判卷)
            if ASK.get('xrev'):
                _on_step('交叉审核', '换模型复核中')
                from src.ontology.fast_agent import cross_review as _xr
                # 审核档 = DeepSeek r1:14b (用户裁 2026-09-07); 作答本身是 DeepSeek 时换 qwen 27B 审 —— 始终跨家族 (同家族自审 = 自己判卷)
                _rm = 'deepseek-r1:14b' if not model.startswith('deepseek') else 'qwen3.8:27b'
                ASK['review'] = _xr(q, r['answer'], r.get('facts') or [], model=_rm)
            ASK.update(state='done')
            return
        if False:
            q = '/no_think ' + q          # (快路径内已带 /no_think; dsh 路径仅 deepseek)
        cmd = ['deploy/dsh/dsh_run.sh', '--profile', 'headless',
               '--patch', 'deploy/dsh/headless.patch.ontology.yml']
        mp = ASK_MODELS.get(model)
        if mp:
            cmd += ['--patch', mp]
        if lang == 'en':
            cmd.append(q + ' Answer in English only. End every conclusion sentence with [object/id]. Finally run claim_check.')
        else:
            cmd.append(q + ' 每个结论句末尾用[对象id]标注来源。最后调用claim_check。')
        # 2026-09-16: 走 src.proc.run —— 本函数跑在 detail 服务进程里 (无可见控制台), 裸 spawn
        # 会让 Windows 给这条"云端档问答"新建可见控制台窗口 (本地档不走这里)。
        from src import proc as _proc
        r = _proc.run(cmd, capture_output=True, text=True, timeout=600)
        ans = r.stdout.strip() or ('(空输出) stderr: ' + r.stderr[-500:])
        out.write_text(ans, encoding='utf-8')
        sys.path.insert(0, str(pathlib.Path('scripts').resolve()))
        from ontology_p2_verify import verify as _v
        ASK.update(answer=ans, verify=_v(ans, lang=lang), state='done')
    except Exception as e:
        _msg = str(e)
        if 'Connection refused' in _msg or 'Errno 61' in _msg or '11434' in _msg:
            _msg = '本机模型未启动 (Ollama 127.0.0.1:11434 拒连); 离线版不转云, 请先启动 Ollama'   # 2026-09-07 真停实测: 原样抛 urlopen 错误给用户
        ASK.update(state='error', err=_msg[:300])


# ── 报告定制助手 (2026-08-28 用户令: 留一个互动交互窗, 报告按定制, 离线模型支持) ──
# 两件事: ①听懂"给集团看的月报, 重点讲设备问题"这类话 → 选出该放哪些块
#         ②基于**真实数字**写一段总结叙述 — 为图表补充文字结论, 供汇报使用。
# 防编造: 数字只许用下面这份摘要里的, 生成后过 grounding 闸复核, 越界即标注。
RPT_BLOCK_KEYS = ['kpi', 'water', 'emon', 'blame', 'ram', 'parts', 'attrib',
                  'pareto', 'quad', 'frate', 'units', 'chain', 'loop', 'plan', 'caliber']
RPT_SYS = ('你是风电场报告编排助手。根据用户要求做两件事, 严格按格式输出, 不要解释:\n'
           'BLOCKS: 用逗号分隔的块名 (只能从给定清单里选, 按报告顺序排)\n'
           'TEXT: 一段中文总结 (150~300 字, 可分 2~3 句一组)\n'
           '铁律: TEXT 里出现的每个数字都必须来自"可用数据"里, 不许自己算不许估计; '
           '没有依据的话就不写。不要写套话, 不要写"综上所述"。')


def rpt_compose(need, canvas, model='qwen3:8b'):
    import time as _t
    fl = fleet_view(CUR_WIN[0] if CUR_WIN else '2026年')
    E = (fl.get('control') or {}).get('energy') or {}
    R = fl.get('rel') or {}
    M = ((fl.get('m8') or {}).get('mtbf')) or {}
    F = fl.get('faults') or {}
    BD = ((fl.get('fus') or {}).get('链盘')) or {}
    facts = [
        f"统计期 {fl.get('win')}; 38 台 4.0MW",
        f"上网电量 {E.get('act')} MWh; 理论可发 {E.get('theo')}; 损失 {E.get('loss')} MWh; 损失率 {E.get('loss_pct')}%",
        f"等效满发 {E.get('eflh')} h; 时间可用率 {(fl.get('control') or {}).get('avail')}%",
        f"总停机 {R.get('总停机h')} h = 设备类 {R.get('设备类停机h')} + 外部类 {R.get('外部类停机h')}",
        f"平均停机间隔（MTBO）{M.get('MTBF_h')} h; 停机 {M.get('停机事件数')} 起; 单次停机时长（MDT）{M.get('MDT_h')} h",
        f"报警 {F.get('total')} 条 / {F.get('n_codes')} 个码",
        f"需行动机组 {len(BD.get('rows') or [])} 台; 卡点 {BD.get('stuck')}",
        # 占比直接给出: 不给模型就得自己算, 算对了也会被接地闸判为"未接地数字"(实测 31.4/46.6)
        f"损失构成: " + '; '.join(
            f"{x['k']} {x['loss']} MWh (占损失 {round(x['loss'] / max(E.get('loss') or 1, 1) * 100, 1)}%)"
            for x in (E.get('items') or [])),
        f"停机 {R.get('总停机h')} h 中设备类占 {round((R.get('设备类停机h') or 0) / max(R.get('总停机h') or 1, 1) * 100, 1)}%",
        f"端到端闭环 {len(BD.get('closed') or [])} 例",
    ]
    prompt = (f"可用数据:\n" + '\n'.join('- ' + x for x in facts) +
              f"\n\n可选块 (只能从这里选): {', '.join(RPT_BLOCK_KEYS)}"
              f"\n块的含义: kpi=一级指标 water=能量瀑布 emon=逐月发电 blame=损失分账 ram=可靠性象限 "
              f"parts=部件停机 attrib=归因覆盖 pareto=故障帕累托 quad=故障四象限 frate=月度故障率 "
              f"units=重点机组 chain=决策链 loop=闭环实证 plan=下期建议 caliber=口径边界"
              f"\n\n当前画布: {', '.join(canvas) if canvas else '(空)'}"
              f"\n用户要求: {need}")
    import urllib.request as _ur      # 顶部只 import urllib.parse, urllib.request 需显式导入
    body = dict(model=model, messages=[dict(role='system', content=RPT_SYS),
                                       dict(role='user', content='/no_think ' + prompt)],
                stream=False, keep_alive='30m', options=dict(num_ctx=8192, num_predict=800))
    t0 = _t.time()
    try:
        req = _ur.Request('http://127.0.0.1:11434/api/chat', data=json.dumps(body).encode(),
                          headers={'Content-Type': 'application/json'})
        with _ur.urlopen(req, timeout=300) as r:
            txt = (json.loads(r.read())['message'].get('content') or '')
    except Exception as e:
        return dict(err=f'本地模型不可达: {str(e)[:120]}')
    txt = re.sub(r'<think>.*?</think>', '', txt, flags=re.S).strip()
    mb = re.search(r'BLOCKS\s*[:：]\s*(.+)', txt)
    mt = re.search(r'TEXT\s*[:：]\s*(.+)', txt, re.S)
    blocks = [b.strip() for b in re.split(r'[,，\s]+', mb.group(1))] if mb else []
    blocks = [b for b in blocks if b in RPT_BLOCK_KEYS]
    narrative = (mt.group(1).strip() if mt else '').strip()
    # 接地闸: 叙述里的数字必须在 facts 里出现 (同 fast_agent 的 grounding 口径)
    bad = []
    try:
        from src.windscada.audit import grounding as _gr
        ok, b = _gr(narrative, '\n'.join(facts))
        bad = (b or {}).get('numbers') or []
    except Exception:
        pass
    return dict(blocks=blocks, narrative=narrative, model=model,
                secs=round(_t.time() - t0, 1), ungrounded=bad, raw=txt[:900])


ASK_CAT_HINTS = {
    '设备类': '(设备状态类: 从本体的机组档案、对象查询与机制检索三条路取证)',
    '检修排程决策': '(检修排程类: 用 ont_scenario/ont_get 取排期沙盘与验收树; 建议级动作注明待人裁)',
    '损失分析': '(损失分析类: 用 ont_loss/ont_scenario 取证; 电量kWh为主, 金额必带电价区间0.391~0.85口径; 限电类损失≠设备责任)',
    '安全风险预测': '(风险预测类: 用 ont_risk/ont_mechanism_search 取证; 区分升级中/稳定高位/盲区, 未列≠正常, 不可判≠阴性)',
}


ASK_CAT_HINTS_EN = {
    '设备类': '(Equipment state: use unit profiles, object queries and mechanism search for evidence)',
    '检修排程决策': '(Maintenance scheduling: use scenarios and acceptance trees; mark advisory actions as pending human decision)',
    '损失分析': '(Loss analysis: use loss accounts and scenarios; energy in kWh/MWh, monetary value must carry the tariff-range caveat)',
    '安全风险预测': '(Risk prediction: distinguish escalating, stable-high and blind-spot cases; no listing is not proof of normality)',
}


def ask_start(q, model, cat='', xrev=False, lang='zh', display_q=''):
    import threading, time as _t
    if not q.strip():
        return dict(err='问题为空' if lang != 'en' else 'The question is empty')
    if ASK['state'] in ('running', 'partial'):   # partial = 初答已出、27B 复核中, 仍占槽
        return dict(err='已有问题在推理中 (串行锁, 一次一问)', state=ASK['state'])
    if model not in ASK_MODELS:
        return dict(err=f'未知模型 {model}; 可选 {list(ASK_MODELS)}')
    hints = ASK_CAT_HINTS_EN if lang == 'en' else ASK_CAT_HINTS
    q2 = (hints.get(cat, '') + ' ' + q).strip()
    if lang == 'en':
        try:
            from src.windscada.deid import ununitize_question
            q2 = ununitize_question(q2)
        except Exception as e:
            print('[ask] 代号反解失败: %s' % e, file=sys.stderr)
    ASK.update(state='running', q=display_q or q, model=model, answer='', verify=None, err='', t0=_t.time(),
               step='启动' if lang != 'en' else 'Starting', steps=[], review=None, lang=lang)
    ASK['xrev'] = bool(xrev)
    threading.Thread(target=_ask_worker, args=(q2, model, lang), daemon=True).start()
    return dict(state='running', model=model)


def ask_status():
    import time as _t
    d = dict(ASK)
    d['elapsed'] = int(_t.time() - ASK['t0']) if ASK['t0'] else 0
    return d


# ---------- 台内即时查询 ----------
_RAW_CACHE = {}
def adhoc_query(t, ch, win, agg):
    from src.windscada.data import load_10min, contracted_cols
    _load()
    key = t
    with LOCK:
        if key not in _RAW_CACHE:
            if len(_RAW_CACHE) > 6: _RAW_CACHE.clear()
            try:
                _RAW_CACHE[key] = load_10min(t, CFG, groups=['A.功率', 'A.风况', 'A.转速', 'A.状态', 'B.变桨', 'B.温度NBM', 'B.偏航', 'B.润滑液压'])
            except Exception as e:
                # 2026-09-11 场景① 逮: 原始件不在时这里原来直接把异常抛给 HTTP 处理器 → 前端拿到 500,
                # 面板只显示"服务器错误", 人看不出是"没放数据"还是"程序坏了"。改为**结构化的无数据**回答:
                # 与成功路径同一个形状 (kind/title/unit/months/series/note), 前端按正常渲染即可显示原因。
                src = CFG.get('src_10min')
                return dict(kind='multiline', title=f'{t} {i18n.ch_cn(ch)}', unit='', months=[], series=[],
                            err='no_source',
                            note=f'无原始件: 本机应在 {src}\\{t}.csv (即 data\\raw\\<场站名称>\\scada_10min\\)'
                                 f' — 放入后重跑 scripts\\rebuild_from_raw.py --scada')
    d = _RAW_CACHE[key]
    if ch not in d.columns: return dict(err=f'通道 {ch} 不在契约内')
    ms = months_of(win)
    dd = d[d.ts.dt.to_period('M').astype(str).isin(ms)]
    if agg == 'daily':
        g = dd.groupby(dd.ts.dt.date)[ch].median()
        xs = [str(x) for x in g.index]
    else:
        g = dd.groupby(dd.ts.dt.to_period('M').astype(str))[ch].median() if agg == 'med' else \
            dd.groupby([dd.ts.dt.to_period('M').astype(str), dd.ts.dt.date])[ch].sum().groupby(level=0).mean()
        xs = list(g.index)
    agg_cn = {'med': '月中位', 'sum': '月duty(秒/日)', 'daily': '逐日中位'}.get(agg, agg)
    fig = dict(kind='multiline', title=f'{t} {i18n.ch_cn(ch)} ({agg_cn}, {win})', unit='', months=xs,
               series=[dict(name=i18n.ch_cn(ch, with_tag=False), vals=[None if v != v else round(float(v), 3) for v in g])], note='台内即时查询')
    # fleet 带 (若月仓有)
    for store, col in ((_CACHE['tm'], 'med'), (_CACHE['duty'], 'val')):
        if store is None or agg == 'daily': continue
        gg = store[store.channel == ch]
        if len(gg):
            piv = gg.pivot_table(index='month', columns='turbine', values=col).reindex(xs)
            fig = dict(kind='band', title=fig['title'], unit='', months=xs,
                       self=[None if v != v else round(float(v), 3) for v in piv.get(t, pd.Series(index=xs, dtype=float))],
                       med=[None if v != v else round(float(v), 3) for v in piv.median(axis=1)],
                       q1=[None if v != v else round(float(v), 3) for v in piv.quantile(.25, axis=1)],
                       q3=[None if v != v else round(float(v), 3) for v in piv.quantile(.75, axis=1)], note='灰带=全场四分位 (月仓)')
            break
    return fig

def channel_list():
    from src.windscada.data import contract
    c = contract(CFG)
    out = {}
    for g, cols in c['groups'].items():
        out[g] = [dict(code=k, cn=i18n.ch_cn(k, with_tag=False) + ' · ' + k) for k, v in cols.items() if v['liveness'] == '活']
    return out

# ---------- HTML ----------
from src.windscada.design import CSS as BASE_CSS

# 经典页模板（CHART_JS / PAGE_FLEET / PAGE_PROBLEM / PAGE_TURBINE）已随 P5 迁入前端模块；
# 模块间调用只许走 api.py（见 configs/modules.yaml 的 allow）——这里取一次，名字与迁移前不变。
from app_frontEnd.app_frontEnd_guanlan.api import classic_pages as _classic

CHART_JS = _classic.CHART_JS
PAGE_FLEET = _classic.PAGE_FLEET
PAGE_PROBLEM = _classic.PAGE_PROBLEM
PAGE_TURBINE = _classic.PAGE_TURBINE




class H(BaseHTTPRequestHandler):
    def log_message(self, *a): pass
    def _send(self, body, ctype='text/html; charset=utf-8', code=200):
        if 'text/html' in ctype and not self.path.startswith('/v2'):   # 2026-09-06 用户批: /v2 = detail_style_baseline, 不注入 unified-review 样式/导航条 (基线令牌须等于门户)
            from src.windscada.design import GUANLAN_REVIEW_CSS
            text = body.decode('utf-8') if isinstance(body, bytes) else body
            # ★2026-09-22 用户测试 4 实逮: 这里原来写死 `http://127.0.0.1:18084/观澜_如东样板_门户_单文件.html`
            #   —— 用户从别的机器点「返回观澜门户」等于访问**自己机器的 loopback**，且那文件在服务端并不存在
            #   ⇒ 网关回 {"err":"not found","path":"/观澜_…html"}。
            #   修法: 用根相对 `/`（门户就在网关根）+ `data-abs="1"` 让**网关不要按组件前缀改写它**
            #   （不加标记会被改写成 /detail/，永远回不到门户 —— 见 guanlan_gateway.rewrite 的说明）。
            text = text.replace('</header>', '</header><nav class="action-path" aria-label="页面操作">'
                                '<a href="/" data-abs="1" target="_top">← 返回观澜门户</a>'
                                '<a href="/detail/v2" data-abs="1">分析总览 / 重新选择</a></nav>', 1)
            if '</head>' in text:
                body = text.replace('</head>', '<style id="guanlan-unified-review-v1">' + GUANLAN_REVIEW_CSS + '</style></head>', 1)
        b = body.encode('utf-8') if isinstance(body, str) else body
        self.send_response(code)
        self.send_header('Content-Type', ctype)
        self.send_header('Content-Length', str(len(b)))
        self.end_headers()
        self.wfile.write(b)

    def do_POST(self):
        try:
            u = urllib.parse.urlparse(self.path)
            n = int(self.headers.get('Content-Length', 0))
            body = json.loads(self.rfile.read(n) or b'{}')
            if u.path == '/api/ask':
                self._send(_jdump(ask_start(body.get('q', ''), body.get('model', 'qwen3:8b'), body.get('cat', ''),
                                            body.get('xrev', False), body.get('lang', 'zh'), body.get('display_q', '')),
                                  ensure_ascii=False), 'application/json; charset=utf-8')
            else:
                self._send('404', code=404)
        except Exception as e:
            self._send(_jdump(dict(err=str(e)[:300]), ensure_ascii=False), 'application/json; charset=utf-8', 500)

    def do_GET(self):
        try:
            u = urllib.parse.urlparse(self.path)
            # encoding='utf-8' 显式给定 (默认 latin-1 会把中文参数解成乱码 — "齿轮箱"→"齿轮箱",
            # 浏览器 encodeURIComponent 路径侥幸正常, curl/直传中文即崩; 2026-08-28 四链接口实逮)
            q = dict(urllib.parse.parse_qsl(u.query, encoding='utf-8', errors='replace'))
            _REQ.lang = q.get('lang', '')      # 本请求的语言, _jdump 出口据此投影文本层
            _REQ.keep_orig = q.get('keep') == '1'   # v2: 译不出留原文, 不写 [translation pending]
            win = q.get('win', '2026年')
            if u.path == '/':
                _pg = PAGE_FLEET.replace('%CSS%', BASE_CSS).replace('%JS%', CHART_JS)
                self._send(_en_page(_pg))
            elif u.path == '/v2':
                # 新前端 (2026-09-03 用户裁"完全重新做一套, 在原来的基础上"): 同一套 /api/*,
                # 表现层双语原生 —— 不经 _en_page/ui_en 那三层事后翻译。迁移期与 / 并存,
                # 页签逐个搬完后再切默认。
                from src.windscada.ui import build as _v2
                self._send(_v2.render('en' if q.get('lang') == 'en' else 'zh'))
            elif u.path == '/v2/snapshot':
                # 新版单文件快照 (静态站用): 接口响应烤进页面, 拦截器实时优先快照兜底。
                # 取数函数与在线 API 同一实现; 英文走同一套 _redact → _proj 投影 (keep=1 留原文)。
                from src.windscada.ui import snapshot as _snap
                from src.ontology import mcp_server as _omc
                import json as _j3, pathlib as _p3
                def _codes():
                    db = _j3.loads(_p3.Path(_omc.OBJ_PATH).read_text(encoding='utf-8'))
                    return sorted(k.split('/')[-1] for k in db if k.startswith('alarmcode/'))
                _ctx = dict(fleet_view=fleet_view, curve_view=curve_view, vibcms_results=vibcms_results,
                            survey=__import__('src.ontology.maintenance', fromlist=['survey']).survey,
                            ont=_omc, alarm_codes=_codes, turbine_problems=turbine_problems,
                            single_problem=single_problem, _redact=_redact, _jclean=_jclean, _proj=_proj, _REQ=_REQ)
                _v2_lang = 'en' if q.get('lang') == 'en' else 'zh'
                _html, _rep = _snap.bake(_v2_lang, _ctx,
                                          peer=q.get('peer', ''), noindex=q.get('noindex') == '1',
                                          slim=(q.get('slim') != '0' if _v2_lang == 'en' else False))
                print('[v2 snapshot] %s' % _j3.dumps(_rep, ensure_ascii=False)[:600], flush=True)
                self._send(_html)
            elif u.path.startswith('/turbine/'):
                t = u.path.rsplit('/', 1)[1]
                _pg = PAGE_TURBINE.replace('%CSS%', BASE_CSS).replace('%JS%', CHART_JS).replace('%T%', t)
                self._send(_en_page(_pg))
            elif u.path == '/api/maint_survey':
                try:
                    from src.ontology.maintenance import survey as _sv
                    r = dict(rows=_sv())
                except Exception as e:
                    r = dict(rows=[], error=f'维护面盘点失败: {e}'[:200])
                self._send(_jdump(r, ensure_ascii=False), 'application/json; charset=utf-8')
            elif u.path == '/api/ont_chain':
                try:
                    from src.ontology import mcp_server as _om
                    kind = q.get('kind', 'fault')
                    fn = {'fault': lambda: _om.ont_chain_fault(code=q.get('code', '')),
                          'prev': lambda: _om.ont_chain_preventive(system=q.get('system', '')),
                          'tree': lambda: _om.ont_fault_tree(system=q.get('system', '')),
                          'plan': lambda: _om.ont_maint_plan(top=q.get('top', 10)),
                          'doc': lambda: _om.ont_doc_search(keyword=q.get('keyword', ''), kind=q.get('dkind', ''))}
                    r = fn[kind]() if kind in fn else dict(text=f'未知链 {kind}', error=True)
                except Exception as e:
                    r = dict(text=f'本体查询失败: {e}'[:200], error=True)
                self._send(_jdump(r, ensure_ascii=False), 'application/json; charset=utf-8')
            elif u.path == '/api/maint_std':
                # 运维标准四判据 + 成熟度评分 (docs/如东运维标准化设计.md)
                try:
                    import json as _j, pathlib as _p
                    from src.windscada.maint import standards as _st
                    _db = _j.loads(_P.objects_json()
                                   .read_text(encoding='utf-8'))
                    _wo = [(v.get('props') or {}) for k, v in _db.items()
                           if k.startswith('workorder/')]
                    _r = _st.run_all(_wo)
                    _r['score'] = _st.score(_wo, _r)
                except Exception as e:
                    _r = dict(err=f'运维标准判据失败: {e}'[:200])
                self._send(_jdump(_r, ensure_ascii=False), 'application/json; charset=utf-8')
            elif u.path == '/api/maint_framework':
                # 运维管理体系 —— 与快照预烤**同一实现** (maint_framework_view)
                try:
                    _r = maint_framework_view()
                except Exception as e:
                    _r = dict(err=f'运维体系读取失败: {e}'[:200])
                self._send(_jdump(_r, ensure_ascii=False), 'application/json; charset=utf-8')
            elif u.path == '/api/dq_findings':
                # 数据可信度发现 —— 与快照预烤**同一实现** (dq_findings_view)
                try:
                    _r = dq_findings_view()
                except Exception as e:
                    _r = dict(rows=[], err=f'数据可信度读取失败: {e}'[:200])
                self._send(_jdump(_r, ensure_ascii=False), 'application/json; charset=utf-8')
            elif u.path == '/api/vibcms':
                self._send(_jdump(vibcms_results(), ensure_ascii=False), 'application/json; charset=utf-8')
            elif u.path == '/api/curves':
                # ★2026-09-21: 接 win（按所选时间窗重算；未命中就起后台线程 + 回 building，页面轮询）
                self._send(_jdump(curve_view(q.get('win') or '2026年'), ensure_ascii=False),
                           'application/json; charset=utf-8')
            elif u.path == '/api/fleet':
                CUR_WIN[0] = win
                self._send(_jdump(fleet_view(win), ensure_ascii=False), 'application/json; charset=utf-8')
            elif u.path.startswith('/problem/'):
                import urllib.parse as _up
                _, _, t, sysn = u.path.split('/', 3)
                _raw = _up.unquote(sysn)
                _zh = sys_norm(_raw)
                # ★2026-09-22 用户测试 3/5: 认得出就换成规范中文名（页面/接口都按中文名取数），
                #   认不出就把原文交给页面 —— 页面据此**如实说"系统标识无法识别"**，
                #   绝不写成"该系统在判级时间窗内无非常态"（那对报警机组是假陈述）。
                self._send(PAGE_PROBLEM.replace('%CSS%', BASE_CSS).replace('%JS%', CHART_JS)
                           .replace('%T%', t).replace('%SYS%', _zh or _raw)
                           .replace('%SYSRAW%', _raw).replace('%SYSOK%', '1' if _zh else '0'))
            elif u.path == '/api/problem':
                _zh = sys_norm(q['sys']) or q['sys']
                self._send(_jdump(dict(problems=single_problem(q['t'], _zh, win), sys=_zh,
                                       sys_ok=bool(sys_norm(q['sys']))), ensure_ascii=False),
                           'application/json; charset=utf-8')
            elif u.path == '/api/turbine':
                self._send(_jdump(dict(problems=turbine_problems(q['t'], win)), ensure_ascii=False), 'application/json; charset=utf-8')
            elif u.path == '/api/query':
                agg = q.get('agg', 'med')
                self._send(json.dumps(adhoc_query(q['t'], q['ch'], win, 'sum' if agg == 'dutyd' else agg), ensure_ascii=False), 'application/json; charset=utf-8')
            elif u.path == '/api/snapshot':
                _lang = urllib.parse.parse_qs(u.query).get('lang', [''])[0]
                # 单文件快照 (2026-08-28 用户: 我怎么发给别人)。
                # 把各窗 fleet / curves / ontology / 维护盘点全部内联进 HTML, 并在页内改写
                # window.fetch 直接返回内联数据 —— 前端代码一行不改, 收件人双击即开, 不需要
                # 装 Python、不需要连内网、断网也能看。数据按当前受众设置脱敏后再内联。
                try:
                    _wins = ['2026年', '2026H1', '2025H2', '全程']
                    snap, _skip = {}, []

                    def _nk(path, **kw):
                        """快照 key 归一化: 参数**解码后**排序拼接。

                        必须与页内 window.__NK 同语义 —— 直接用 URL 字符串做 key 会被
                        encodeURIComponent 与 urllib quote 的编码差异 (! * ' ( ) 一编一不编)
                        和参数顺序打掉, 而这种失效是**静默**的: 命中不了就退到
                        "快照版不含此功能", 看起来像功能没做, 不像 key 错了。
                        """
                        a = sorted('%s=%s' % (k, v) for k, v in kw.items() if v not in (None, ''))
                        return path + ('?' + '&'.join(a) if a else '')

                    def _bake(key, fn):
                        """烤一个 API 响应。失败**记名**不静默 —— 守护步骤失败必须响亮。

                        ★2026-08-31: 此前直接 _redact(_jclean(...)), **绕过 _jdump 里的语言投影**,
                          于是 ?lang=en 的快照内联的仍是中文数据 (实测 57 万→11 万中文只是
                          页面文案变了, 数据没变)。这里补上同一套投影。
                        """
                        try:
                            # ★顺序是 **_redact 先, _proj 后**, 不要"为了与在线同序"去调它:
                            #   曲线图注等模板的 key 是照**脱敏后**的文本登记的 (含 ▪ 掩码),
                            #   先投影的话字符串里还是真实数字, 模板全部对不上, 整批退回
                            #   [translation pending]。我按"同序"改过一次, 当场被 parity 闸逮住。
                            #   在线路径没有这个问题是因为脱敏在页面层统一做, 不在数据层。
                            _v = _redact(_jclean(fn()))
                            if _lang and _lang != 'zh':
                                _v = _proj(_v, _lang)
                            snap[key] = _v
                            return True
                        except Exception as e:
                            _skip.append('%s: %s' % (key, str(e)[:60]))
                            return False

                    for _w in _wins:
                        _bake(_nk('/api/fleet', win=_w), lambda w=_w: fleet_view(w))
                    _bake('/api/curves', curve_view)
                    _bake('/api/ontology', ontology_view)
                    # ★后加的面板接口也要预烤 —— 漏了会在离线快照里空掉。
                    #   新增接口时必须回到这里补一行, 否则快照静默少一块。
                    def _maint_std():
                        import json as _j2, pathlib as _p2
                        from src.windscada.maint import standards as _st2
                        _db2 = _j2.loads(_P.objects_json()
                                         .read_text(encoding='utf-8'))
                        _wo2 = [(v.get('props') or {}) for k, v in _db2.items()
                                if k.startswith('workorder/')]
                        _r2 = _st2.run_all(_wo2)
                        _r2['score'] = _st2.score(_wo2, _r2)
                        return _r2

                    _maint_fw = maint_framework_view   # ★与在线 API 同一实现
                    _dq = dq_findings_view          # ★与在线 API 同一实现

                    _bake('/api/maint_std', _maint_std)
                    _bake('/api/maint_framework', _maint_fw)
                    _bake('/api/dq_findings', _dq)
                    _bake('/api/maint_survey',
                          lambda: dict(rows=__import__('src.ontology.maintenance',
                                                       fromlist=['survey']).survey()))
                    # ── 扩展预烤 (2026-08-30 用户"完整展示") ───────────────────────
                    # 原先只烤上面 4 个, 另外 7 个接口在离线版里是死的: 部件问题下钻、
                    # 本体对象/清单/链、振动详情点开全是"快照版不含此功能" —— 页面能渲染
                    # 掩盖了功能不可用, 我此前按"十页可交互"汇报是说过头了。
                    # ask / rpt_compose 要现跑模型, 不烤 (页面另给在线提示), 其余全量枚举。
                    _bake('/api/vibcms', vibcms_results)
                    # 离线桩: 快照里没有服务端跑模型, 所以这条**有意**与在线不同。
                    # note 双写 —— 靠翻译层会得到 [translation pending], 而这是给读者看的一句话。
                    _bake('/api/ask_status', lambda: dict(
                        offline=True, models=[],
                        note_zh='离线版: 模型问答需连接系统',
                        note_en='Offline copy: the model Q&A needs a live connection to the system'))
                    # 部件问题: (机组, 系统) 从各窗 systems 展开, 每个窗各烤一份
                    _pairs = set()
                    for _w in _wins:
                        _fv = snap.get(_nk('/api/fleet', win=_w)) or {}
                        for _sysn, _rows in (_fv.get('systems') or {}).items():
                            for _r in (_rows or []):
                                if isinstance(_r, dict) and _r.get('t'):
                                    _pairs.add((_r['t'], _sysn))
                    # ★系统名两种形态都要烤 (2026-09-01 实逮"点开是空浮层")。
                    #   systems 的分组键带**等级后缀** (`塔架与基础·良好`), 而页面上
                    #   `/problem/<t>/<sys>` 链接发出的是**裸系统名** (`塔架与基础`) ——
                    #   于是 key 差一截, 取不到数, 浮层开了但里面是"本时间窗无记录"。
                    #   **这种失效最难发现: 浮层确实弹出来了, 看着像"这台这系统没数据",
                    #   而不像"key 错了"。** 两种形态都烤成同一份数据即可。
                    for _t, _sy in sorted(_pairs):
                        _forms = {_sy}
                        if '·' in _sy:
                            _forms.add(_sy.split('·')[0])
                        for _w in _wins:
                            for _f in sorted(_forms):
                                # ★每种形态烤**它自己**的数据。上一版这里捕获的是带等级的 _sy,
                                #   于是裸名 key 拿到的是带等级那份 —— 实测差出 0 vs 1 条,
                                #   **被对拍闸逮到**。lambda 默认参数必须捕获循环变量 _f 本身。
                                _bake(_nk('/api/problem', t=_t, sys=_f, win=_w),
                                      lambda t=_t, sy=_f, w=_w: dict(problems=single_problem(t, sy, w)))
                    # 整机视图: 页面上机组号链到 /turbine/<t>, 弹层要用
                    for _t in sorted({t for t, _ in _pairs}):
                        for _w in _wins:
                            _bake(_nk('/api/turbine', t=_t, win=_w),
                                  lambda t=_t, w=_w: dict(problems=turbine_problems(t, w)))
                    # 本体清单 + 每个对象详情 (清单里出现的 id 才可能被点开)
                    _oids = set()
                    for _ty in sorted((ontology_view().get('stats') or {})):
                        _k = _nk('/api/ont_list', type=_ty, limit=60)
                        if _bake(_k, lambda ty=_ty: dict(rows=ontology_list(ty, '', 60))):
                            for _row in (snap[_k].get('rows') or []):
                                if isinstance(_row, dict) and _row.get('id'):
                                    _oids.add(_row['id'])
                    for _oid in sorted(_oids):
                        _bake(_nk('/api/ont_obj', id=_oid), lambda i=_oid: ontology_obj(i))
                    # 本体链: kind 五种, 参数取页面下拉的实际取值域
                    try:
                        from src.ontology import mcp_server as _omc
                        _bake(_nk('/api/ont_chain', kind='plan', top='12'),
                              lambda: _omc.ont_maint_plan(top=12))
                        _syslist = sorted({sy for _, sy in _pairs}) or ['齿轮箱']
                        for _sy in _syslist:
                            _bake(_nk('/api/ont_chain', kind='prev', system=_sy),
                                  lambda sy=_sy: _omc.ont_chain_preventive(system=sy))
                            _bake(_nk('/api/ont_chain', kind='tree', system=_sy),
                                  lambda sy=_sy: _omc.ont_fault_tree(system=sy))
                        # ★原来码源是**另一处烤好的** /api/ont_list?limit=60 的行, 于是全场 555 个码
                        #   只进了按码号排序的前 60 个; 而页面上"点一下就查"的示例码
                        #   (3225/64000/64001/4110/13121) 恰好都不在那 60 个里
                        #   ⇒ **页面在推荐自己答不上的码**, 点下去整条抢修链没反应。
                        #   "排序后取前 N"的裁剪最容易把真正要用的挡在外面。
                        # ★改为**直接从本体读**, 不再依赖别处烤没烤、烤了多少 ——
                        #   跨模块借数据是这次截断的根因, 借来的上限管不着。
                        #   (中途踩坑: 我先写成 _omc.ont_list(...), 而 mcp_server 根本没有这个函数,
                        #    AttributeError 被外层 try 吞掉 ⇒ 整段 fault 一条都没烤, 还静默。)
                        import json as _js
                        _db = _js.loads(pathlib.Path(_omc.OBJ_PATH).read_text(encoding='utf-8'))
                        _codes = sorted(k.split('/')[-1] for k in _db
                                        if k.startswith('alarmcode/'))
                        for _c in [c for c in _codes if c]:
                            _bake(_nk('/api/ont_chain', kind='fault', code=_c),
                                  lambda c=_c: _omc.ont_chain_fault(code=c))
                    except Exception as _e:
                        _skip.append('ont_chain: %s' % str(_e)[:60])
                    if _skip:
                        print('[snapshot] 未烤成 %d 项: %s' % (len(_skip), '; '.join(_skip[:6])),
                              file=sys.stderr)
                    print('[snapshot] 预烤 %d 键 (%s)' % (
                        len(snap), ', '.join(sorted({k.split('?')[0] for k in snap}))), file=sys.stderr)
                    if _lang and _lang != 'zh':
                        # ★key 统一在**全部烤完之后**加 lang: 英文页的 fetch 会自动追加
                        #   ?lang=en, 预烤 key 不带就永远对不上 (miss 是静默的, 退到 err 提示,
                        #   看着像功能没做)。
                        #   ★不能在 _bake 内部改 —— 后续代码还要用原 key 从 snap 取值,
                        #     改了就 KeyError (本次实测踩到)。
                        # ★存双 key: 带 lang 与不带。页面首帧的 fetch 可能早于 fetch patch
                        #   生效, 那一发不带 lang —— 只存带 lang 的会 miss 一次 (实测首屏 fleet)。
                        #
                        # ★★ 加 lang 必须走**同一个归一化器**, 不能字符串拼接 (2026-09-01 实逮)。
                        #   __NK 把参数**按字母排序**, 所以正确形态是 `?lang=en&sys=…&t=…&win=…`;
                        #   而拼接得到的是 `?sys=…&t=…&win=…&lang=en` —— 差一个顺序, 全部 miss。
                        #   后果极隐蔽: 十个标签页照常显示 (它们命中的是**不带 lang** 的那份),
                        #   只有**点开下钻**才走带 lang 的 key ⇒ 浮层弹出来却是"本时间窗无记录",
                        #   看着像"这台这系统没数据", 不像 key 错了。用户报的"有些点不开"就是它。
                        def _with_lang(k):
                            base, _, q = k.partition('?')
                            kv = dict(x.split('=', 1) for x in q.split('&') if '=' in x) if q else {}
                            kv['lang'] = _lang
                            return base + '?' + '&'.join(sorted('%s=%s' % (a, b) for a, b in kv.items()))
                        snap = dict(snap, **{_with_lang(k): v for k, v in snap.items()})
                    page = PAGE_FLEET.replace('%CSS%', BASE_CSS).replace('%JS%', CHART_JS)
                    if _lang and _lang != 'zh':
                        # 快照页面本身也走英文化 (文案表 + 显示层脚本), 并改 lang 属性 ——
                        # 此前只投影了 API 路径, 快照页漏了, 导出的 index.html 一直是 lang="zh"
                        page = _en_page(page).replace('<html lang="zh"', '<html lang="%s"' % _lang, 1)
                    inject = ('<script>window.__SNAP__='
                              + json.dumps(snap, ensure_ascii=False, allow_nan=False) + ';\n'
                              '/* 快照模式: 拦截 fetch 从内联数据取。key 走归一化(参数解码后排序)而非 */'
                              '/* 原样 URL —— 编码差异(encodeURIComponent 不编 !*\'() 而 quote 编)与参数 */'
                              '/* 顺序都会让字符串比对落空, 且落空是静默的: 退到 err 提示, 看着像功能没做。 */'
                              'window.__NK=function(u){var s=String(u),i=s.indexOf("?");if(i<0)return s;'
                              'var p=s.slice(0,i),a=s.slice(i+1).split("&").filter(Boolean).map(function(kv){'
                              'var j=kv.indexOf("=");if(j<0)return kv;var k=kv.slice(0,j),v=kv.slice(j+1);'
                              'try{v=decodeURIComponent(v.replace(/\\+/g," "))}catch(e){}'
                              'return v===""?null:k+"="+v}).filter(function(x){return x!==null}).sort();'
                              'return p+(a.length?"?"+a.join("&"):"")};'
                              'window.__HIT={ok:0,miss:[]};window.__RF=window.fetch;'
                              # ★没烤的窗按钮直接摘掉: 让用户点一个必然空白的选项, 比少给几个窗更糟。
                              #   默认窗取**已烤窗里最新的那个**, 供上面的回落候选用。
                              'window.__WINS=(function(){var s={};Object.keys(window.__SNAP__).forEach('
                              'function(k){var m=/[?&]win=([^&]*)/.exec(k);if(m)s[decodeURIComponent(m[1])]=1});'
                              'return Object.keys(s)})();'
                              'window.__DEFWIN=window.__WINS[0]||"";'
                              'window.__WINLABEL=' + json.dumps(
                                  'Latest completed build' if _lang == 'en'
                                  else '最新一次完整分析') + ';'
                              # ★快照里**去掉窗选择器**, 换成固定标签 (2026-09-01 用户令
                              #   "设定不随时间窗动, 就显示最新的")。
                              #   理由: 选择器给约 20 个窗而只烤了 4 个, 选中其余任何一个都空白;
                              #   而这 4 个窗的内容**完全一样** (fus 40 行 / 需跟踪 76 台次 逐窗相同),
                              #   所以留着选择器只会让人以为能切、切了还空。
                              #   ★先前那版是"摘掉没烤的 option" —— 摘过头了: 选项显示文本已英文化
                              #   ("2026 full year"), 而 __WINS 是中文键, 匹配不上就把有效的也摘了。
                              #   **文本匹配在译过的 UI 上天然不可靠**, 所以改为整体替换而非逐项筛。
                              'window.__PINWIN=function(){try{'
                              'var sel=document.querySelector("select#win");if(!sel)return;'
                              'var lab=document.createElement("span");lab.id="winpin";'
                              'lab.style.cssText="font-size:12.5px;color:var(--muted);padding:5px 9px;'
                              'border:1px solid var(--line);border-radius:7px;background:var(--panel)";'
                              'lab.textContent=window.__WINLABEL||"Latest build";'
                              'lab.title="This snapshot is fixed to the latest completed build; '
                              'the panels below do not vary by window.";'
                              'sel.parentNode.replaceChild(lab,sel)}catch(e){}};'
                              'try{new MutationObserver(function(){window.__PINWIN()})'
                              '.observe(document.documentElement,{childList:true,subtree:true})}catch(e){}'
                              'document.addEventListener("DOMContentLoaded",function(){'
                              'setTimeout(window.__PINWIN,300)});'
                              # 问答接口不能被离线拦截: 云上 nginx /api/ 后面有真服务在跑,
                              # 其余接口才走预烤。放行名单写死, 免得日后新增接口误穿。
                              'window.fetch=function(u,o){var k=window.__NK(u);'
                              'if(/^\\/api\\/(ask|ask_status|ask_models|rpt_compose|rpt_export)\\b/.test(String(u)))return window.__RF(u,o);'
                              # ★第 4 个候选: 把 win 换成默认窗 (2026-09-01 用户令"不随时间窗动, 就显示最新的")。
                              #   窗选择器给了约 20 个选项而快照只烤了 4 个 —— 选中其余任何一个
                              #   都会整页空白 (表现为"该系统全场无异常"、振动页空), 看着像没数据,
                              #   实则是没烤。这些面板的内容本就不随窗变 (四个窗 fus 完全一样),
                              #   所以回落到默认窗是**正确答案**而不是遮丑。
                              # ★同 src/windscada/ui/snapshot.py 的坑: 这里不能 encodeURIComponent —— k 与
                              #   __SNAP__ 键都是原文(未编码)中文, 编码后拼出的键谁都不匹配, 兜底静默失效
                              #   (2026-09-03 实逮: 换窗点浮层报"本时间窗无记录", 默认窗其实有数据)。
                              'var c=[k,k.replace(/&?win=[^&]*/,"").replace(/\\?$/,""),String(u).split("?")[0],'
                              'k.replace(/win=[^&]*/,"win="+(window.__DEFWIN||""))];'
                              'for(var i=0;i<c.length;i++){var d=window.__SNAP__[c[i]];'
                              'if(d!==undefined){window.__HIT.ok++;return Promise.resolve({ok:true,'
                              'json:function(){return Promise.resolve(d)}})}}'
                              'if(window.__HIT.miss.indexOf(k)<0)window.__HIT.miss.push(k);'
                              'return Promise.resolve({ok:true,json:function(){return Promise.resolve('
                              '{err:"离线版不含此功能 (模型问答与报告生成需连接系统)"})}})};'
                              '</script>')
                    # ★英文版改走**源码级生成**, 不再注入运行时翻译 hook
                    #   (2026-08-31 用户令: 不要打补丁式翻译, 整个系统重新生成一遍)。
                    #   hook 的本质是给中文系统套一层英文 —— 只译写入 DOM 的那一刻,
                    #   JS 逻辑与 JSON 键仍是中文, 长尾补不完且源码里满是中文。
                    #   改为在页面拼好后整页替换 (见本函数末尾与 i18n_en.to_english_source)。
                    #   ★中文版完全不受影响: 不带 lang=en 时这条分支不执行, 国内照常用。
                    # ★ 单文件版的下钻: 页面上 74 个链接指向 /problem/<t>/<sys> 与 /turbine/<t>,
                    #   那是服务端独立页面 —— 快照里点了就是死链 (404/白页)。此前"十页可交互"
                    #   的说法没覆盖这一层。改法: 拦截这两类链接, 用已预烤的同一份数据在页内
                    #   开浮层, 图表复用共享的 drawFig, 与在线版看到的是同一批图。
                    # ★注入块里的文案要**自己**按语言选: 它加在 _en_page 之后,
                    #   英文化那一遍已经跑完了, 这里的中文字面量永远不会被翻译。
                    #   实逮: 英文页浮层里弹出"本时间窗无记录"。
                    #   ('状态'/'依据' 是 JSON 键, '年' 是窗名数据值 —— 都不能动。)
                    inject += ('<script>(function(){var M=null;'
                               'function esc(s){return String(s==null?"":s).replace(/[&<>]/g,function(c){'
                               'return {"&":"&amp;","<":"&lt;",">":"&gt;"}[c]})}'
                               'window.__CLOSEM=function(){if(M)M.style.display="none"};'
                               'function box(){if(M)return M;M=document.createElement("div");'
                               'M.style.cssText="position:fixed;left:0;top:0;right:0;bottom:0;z-index:9999;'
                               'background:rgba(0,0,0,.6);overflow:auto";'
                               'M.onclick=function(e){if(e.target===M)window.__CLOSEM()};'
                               'document.body.appendChild(M);return M}'
                               'async function open2(t,sys){'
                               'var w=(window.D&&window.D.win)||"2026年";'
                               'var u=sys?("/api/problem?t="+t+"&sys="+encodeURIComponent(sys)+"&win="+encodeURIComponent(w))'
                               ':("/api/turbine?t="+t+"&win="+encodeURIComponent(w));'
                               'var d={};try{d=await (await fetch(u)).json()}catch(e){}'
                               'var ps=(d&&d.problems)||[];'
                               'var h=\'<div style="max-width:1120px;margin:26px auto;background:var(--card,#111a24);\''
                               '+\'border:1px solid var(--line,#22303d);border-radius:10px;padding:18px 20px">\''
                               '+\'<div style="display:flex;justify-content:space-between;align-items:flex-start;gap:12px">\''
                               '+"<h3 style=\'margin:0 0 4px\'>"+esc(t)+(sys?" · "+esc(sys):"")+"</h3>"'
                               '+\'<button onclick="window.__CLOSEM()" style="background:none;border:1px solid \'' 
                               '+\'var(--line,#22303d);color:inherit;border-radius:6px;padding:2px 9px;cursor:pointer">×</button></div>\';'
                               'if(!ps.length)h+=\'<p class="small">' + ('No records in this window' if _lang == 'en' else '本时间窗无记录') + '</p>\';'
                               'ps.forEach(function(p){h+=\'<div class="card mb"><b>\'+esc(p.sys)+\'</b> · \'+esc(p["状态"])'
                               '+\'<p class="small">\'+esc(p["依据"])+\'</p>\';'
                               'if(p.figs&&p.figs.length&&typeof drawFig==="function"){h+=\'<div class="grid">\';'
                               'p.figs.forEach(function(f){try{h+=drawFig(f)}catch(e){}});h+="</div>"}'
                               'h+="</div>"});'
                               'h+="</div>";var b=box();b.innerHTML=h;b.style.display="block";b.scrollTop=0}'
                               'document.addEventListener("click",function(e){'
                               'var a=e.target&&e.target.closest?e.target.closest(\'a[href^="/problem/"],a[href^="/turbine/"]\'):null;'
                               'if(!a)return;e.preventDefault();e.stopPropagation();'
                               'var q=a.getAttribute("href").split("/").filter(Boolean);'
                               'open2(decodeURIComponent(q[1]||""),q[0]==="problem"?decodeURIComponent(q[2]||""):"")},true);'
                               'document.addEventListener("keydown",function(e){if(e.key==="Escape")window.__CLOSEM()});'
                               '})();</script>')
                    page = page.replace('<script>%JS%', inject + '<script>%JS%') if '%JS%' in page else page.replace('</head>', inject + '</head>')
                    if inject not in page:
                        page = page.replace('</head>', inject + '</head>')
                    # ★这里**曾经**调 i18n_en.to_english_source(page) 做逐词替换。
                    #   它跑在 inject 之后, 作用在含 __SNAP__ JSON 的整页上, 把预烤 key 里的
                    #   中文吃掉: win=2026年 → win=2026, win=全程 → win=full window,
                    #   于是页面按原 key 取数全部 miss —— **而 miss 是静默的**, 退到 err 提示,
                    #   看着像功能没做。_PROT 保护不了 JSON 里的 key。
                    #   页面英文化已在前面由 _en_page (整段短语表 + NEVER 数据值保护) 完成,
                    #   这里不再需要第二遍, 已删。
                    # ★成品页语法闸 —— 校验**最终字节**, 不是中间态 (2026-08-31)。
                    #   _en_page 里也有一道, 但它跑在 value_script 与 __SNAP__ 注入**之前**,
                    #   护不住后注入的脚本。而快照是要上云的成品, 坏了就是白屏,
                    #   且白屏时"可见中文=0"会让所有自查指标全绿 —— 最坏的一种失败。
                    try:
                        from src.windscada import ui_en as _uie
                        _nchk = 0
                        for _m in re.finditer(r'<script([^>]*)>(.*?)</script>', page, re.S):
                            if 'application/json' in _m.group(1) or not _m.group(2).strip():
                                continue
                            _uie.check_js(_m.group(2), label='快照脚本#%d' % (_nchk + 1))
                            _nchk += 1
                        print('[snapshot] 成品页语法闸: %d 个脚本全部通过 node --check' % _nchk,
                              file=sys.stderr)
                        if _nchk == 0:
                            print('[snapshot] 🔴 语法闸切出 0 个脚本 —— **本次未校验**, 非通过',
                                  file=sys.stderr)
                    except Exception as _e:
                        print('[snapshot] 🔴 成品页语法闸失败: %s' % _e, file=sys.stderr)
                        raise
                    if _lang == 'en':
                        _n1 = len(re.findall(r'[\u4e00-\u9fff]', page))
                        print('[snapshot] 英文快照: 页面残留中文 %d 字 (含内联数据)' % _n1,
                              file=sys.stderr)
                    # ★脱敏必须是**最后一步**, 且只对上云那一版做 (2026-09-01 用户令)。
                    #   中文快照是内部工作副本, 要保留真实场名/机型, 不脱敏。
                    #   英文快照是要发到公网的, 而它此前带着 rudong×366 / Siemens×194 /
                    #   SWT×89 —— **同时页面上写着 "site, owner and OEM removed"**,
                    #   产物与自己的声明不符, 这比不声明更糟。
                    #   放最后: 对象 id 与引用它的地方必须换成同一个词, 只能整体做。
                    if _lang == 'en':
                        from src.windscada import deid_public as _dp
                        page = _dp.scrub_or_die(page, label='英文快照 (上云件)')
                        print('[snapshot] 脱敏回扫: 干净 (rudong/OEM/机型/省市/业主/坐标 全零)',
                              file=sys.stderr)
                    fn = (f'Guanlan_Snapshot_{CUR_WIN[0]}_EN.html' if _lang == 'en'
                          else f'观澜快照_{CUR_WIN[0]}_{"上报版" if not INTERNAL else "内部版"}.html')
                    data = page.encode('utf-8')
                    self.send_response(200)
                    self.send_header('Content-Type', 'text/html; charset=utf-8')
                    self.send_header('Content-Length', str(len(data)))
                    self.send_header('Content-Disposition',
                                     "attachment; filename*=UTF-8''" + urllib.parse.quote(fn))
                    self.end_headers()
                    self.wfile.write(data)
                except Exception as e:
                    import traceback
                    self._send(_jdump(dict(err=str(e)[:200], tb=traceback.format_exc()[-500:]),
                                      ensure_ascii=False), 'application/json; charset=utf-8', 500)
            elif u.path == '/api/rpt_export':
                # 导出 Word/Excel (2026-08-28 用户令: 导出 demo)。图表块转数据表, 不放位图 —
                # Word 里的位图既不能编辑也无法核对, 上报集团要的是能核的数字。
                q = urllib.parse.parse_qs(u.query)
                fmt = q.get('fmt', ['docx'])[0]
                blocks = [x for x in q.get('blocks', [''])[0].split(',') if x]
                aud = q.get('aud', ['in'])[0]
                w = q.get('win', ['2026年'])[0]
                narr = q.get('narr', [''])[0] or None
                try:
                    from src.windscada import report_export as _rx
                    fl = fleet_view(w)
                    if aud == 'up':          # 上报版走同一套脱敏规则
                        fl = _redact(_jclean(fl))
                    data = (_rx.to_xlsx if fmt == 'xlsx' else _rx.to_docx)(fl, blocks, w, aud, narr)
                    ext = 'xlsx' if fmt == 'xlsx' else 'docx'
                    mime = ('application/vnd.openxmlformats-officedocument.spreadsheetml.sheet' if fmt == 'xlsx'
                            else 'application/vnd.openxmlformats-officedocument.wordprocessingml.document')
                    fn = f'运行分析报告_{w}_{"上报版" if aud == "up" else "内部版"}.{ext}'
                    self.send_response(200)
                    self.send_header('Content-Type', mime)
                    self.send_header('Content-Length', str(len(data)))
                    self.send_header('Content-Disposition',
                                     "attachment; filename*=UTF-8''" + urllib.parse.quote(fn))
                    self.end_headers()
                    self.wfile.write(data)
                except Exception as e:
                    import traceback
                    self._send(_jdump(dict(err=str(e)[:200], tb=traceback.format_exc()[-400:]),
                                      ensure_ascii=False), 'application/json; charset=utf-8', 500)
            elif u.path in ('/api/facts', '/api/facts/claim'):
                # 结构化事实契约 (用户批 2026-09-06 "8033 /api 切到契约生成"): 只读 derived/ 产物, 不算数不改写;
                # 生成器 = scripts/guanlan_facts_contract.py render, 契约 sha 随体返回供前端/CI 对拍.
                import json as _json
                _dd = _P.guanlan() / 'derived'
                try:
                    if u.path == '/api/facts':
                        _body = _dd.joinpath('detail_cards.json').read_text(encoding='utf-8')
                    else:
                        _cid = urllib.parse.parse_qs(u.query).get('id', [''])[0]
                        _qa = _json.loads(_dd.joinpath('qa_refs.json').read_text(encoding='utf-8'))
                        _hit = next((r for r in _qa.get('refs', []) if r.get('claim_id') == _cid), None)
                        _body = _jdump(dict(contract_sha256=_qa.get('contract_sha256'), claim=_hit, err=None if _hit else 'unknown claim_id'), ensure_ascii=False)
                    self._send(_body, 'application/json; charset=utf-8')
                except FileNotFoundError as e:
                    self._send(_jdump(dict(err='facts derived 缺失: %s' % e), ensure_ascii=False), 'application/json; charset=utf-8', 503)
            elif u.path == '/api/rpt_compose':
                q = urllib.parse.parse_qs(u.query)
                if q.get('probe', [''])[0]:
                    # 前端只探"编排接口是否存在" (v2 bindReport probe=1); 原先照跑 rpt_compose('', [], model) 走模型,
                    # 汇报定制页首屏空等 ~23 s (2026-09-06 实测) —— 探针只答存在性, 不算数.
                    self._send(_jdump(dict(ok=True, probe=True), ensure_ascii=False), 'application/json; charset=utf-8'); return
                r = rpt_compose(q.get('need', [''])[0], [x for x in q.get('canvas', [''])[0].split(',') if x],
                                q.get('model', ['qwen3:8b'])[0])
                self._send(_jdump(r, ensure_ascii=False), 'application/json; charset=utf-8')
            elif u.path == '/api/ask_status':
                self._send(_jdump(ask_status(), ensure_ascii=False), 'application/json; charset=utf-8')
            elif u.path == '/api/ask_models':
                # 与云端问答微服务 (windscada_ask_serve.py) 同形: {models: {id: {...}}}。此前本地没有
                # 这条路由, 经典页的模型下拉在本地永远是"无可用模型" —— v2 与经典页同源取清单。
                self._send(_jdump(dict(models={k: {} for k in ASK_MODELS}), ensure_ascii=False),
                           'application/json; charset=utf-8')
            elif u.path == '/api/ontology':
                self._send(_jdump(ontology_view(), ensure_ascii=False), 'application/json; charset=utf-8')
            elif u.path == '/api/ont_list':
                self._send(_jdump(dict(rows=ontology_list(q.get('type', ''), q.get('t', ''), q.get('limit', 60))),
                                  ensure_ascii=False), 'application/json; charset=utf-8')
            elif u.path == '/api/ont_obj':
                self._send(_jdump(ontology_obj(q.get('id', '')), ensure_ascii=False), 'application/json; charset=utf-8')
            elif u.path == '/api/channels':
                self._send(_jdump(channel_list(), ensure_ascii=False), 'application/json; charset=utf-8')
            elif u.path.startswith('/static/'):
                f = ST / u.path[len('/static/'):]
                if f.is_file():
                    self._send(f.read_bytes(), 'text/html; charset=utf-8')
                else:
                    # 裸 404 会让总览 iframe 变成"一块空白"(用户 2026-09-17 报的正是这个现象) ⇒ 结构化缺件页
                    self._send(_gap_page(u.path, f, '总览（全场状态）'), 'text/html; charset=utf-8')
            elif u.path.startswith('/turbines/'):
                f = ST / u.path.lstrip('/')
                if f.is_file():
                    raw = f.read_text(encoding='utf-8', errors='ignore')
                    light = '<style>:root[data-theme="light"]{--ground:#f3f6f9;--panel:#fff;--panel-2:#f8fbfc;--ink:#0c1622;--muted:#5a6b7d;--line:#d8e1e9;--line-2:#eaf0f5;--accent:#0d7183;--accent-soft:#e6f3f5;--accent-ink:#095764;--ok:#137a68;--warn:#8a5a12;--bad:#a32d38;--unknown:#4d5d6e;--info:#0d7183}</style><script>(function(){try{document.documentElement.setAttribute("data-theme","light")}catch(e){}})();</script>'
                    raw = raw.replace('</head>', light + '</head>', 1) if '</head>' in raw else light + raw
                    self._send(raw)
                else:
                    self._send(_gap_page(u.path, f, '机组详情页'), 'text/html; charset=utf-8')
            elif u.path.startswith('/windcms/'):
                # windcms 振动分析本体嵌入 (用户令: 原振动分析板块进入振动融合板块; 静态挂载非复制,
                # 正本仍在 outputs/rudong/windcms 由振动线维护, 本路由只读)
                base = (ST.parent / 'windcms').resolve()
                f = (base / u.path[len('/windcms/'):]).resolve()
                if base in f.parents or f == base:
                    if f.is_dir(): f = f / 'index.html'
                    if f.is_file():
                        ext = f.suffix.lower()
                        mime = {'.html': 'text/html; charset=utf-8', '.json': 'application/json; charset=utf-8',
                                '.png': 'image/png', '.svg': 'image/svg+xml', '.css': 'text/css',
                                '.js': 'application/javascript', '.md': 'text/plain; charset=utf-8'}.get(ext, 'application/octet-stream')
                        self._send(f.read_bytes(), mime)
                    else:
                        self._send('404', code=404)
                else:
                    self._send('403', code=403)
            elif u.path == '/api/reload':
                # 产物重载 (2026-09-16 用户令: 重算产物即时可用于页面呈现)。
                # 平时**不需要**调它 —— _load() 每次请求都会比产物指纹, 变了自动重载;
                # 这个端点给"重算结束想立刻生效"的显式动作, 也便于运维界面/脚本确认当前指纹。
                _old = reload_products(urllib.parse.parse_qs(u.query).get('why', ['manual'])[0])
                self._send(_jdump(dict(ok=True, old_stamp=_old, stamp=products_stamp(force=True),
                                       files=len(_product_files())), ensure_ascii=False),
                           'application/json; charset=utf-8')
            else:
                self._send('404', code=404)
        except ProductsMissing as e:
            # 产物缺失不是"程序坏了": 回 200 + 结构化无数据, 前端按正常渲染显示原因 (2026-09-11 空状态检查)
            self._send(_jdump(dict(kind='multiline', title='无产物', unit='', months=[], series=[],
                                   err='no_products',
                                   note=f'{e.what} 不存在 → {_P.rel(e.path)}; 先放原始件再跑 '
                                        f'scripts/rebuild_from_raw.py (SCADA 侧加 --scada) 或 '
                                        f'scripts/rebuild_all.py 生成产物。'
                                        f'"包内没有生成端"的件按用户令 2026-09-17 **运行期不由交付包补齐**: '
                                        f'要么等研发补生成端(逐族可逆性见 docs/系统设计说明.md §13.6), '
                                        f'要么由人离线补救(products_restore_missing.py --stash <…>, 不在运行链里); '
                                        f'2026-09-16 起清除产物不留备份, 故没有 --on 还原'), ensure_ascii=False),
                       'application/json; charset=utf-8')
        except Exception as e:
            import traceback
            self._send(_jdump(dict(err=str(e), tb=traceback.format_exc()[-500:]), ensure_ascii=False), 'application/json; charset=utf-8', 500)

if __name__ == '__main__':
    # 统一日志口径 (用户令 2): 之后每一行都带 时间戳/级别/组件（详情见 src/logfile.py）
    import pathlib as _pl, sys as _sys0
    _sys0.path.insert(0, str(_install_root(__file__)))
    from src import logfile as _lf; _lf.prefix_stdout('detail')
    # 控制台可能是 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
    ap = argparse.ArgumentParser()
    import os
    ap.add_argument('--port', type=int, default=int(os.environ.get('PORT', 8033)))   # PORT 环境变量优先, 供 preview 分配端口
    ap.add_argument('--host', default='127.0.0.1', help='0.0.0.0 = 对局域网开放 (演示用, 结束请关)')
    ap.add_argument('--farm', default=None, help='场名 (默认 rudong; 可用场见 app_ETL/configs/farms/)')
    a = ap.parse_args()
    print(f'windscada serve :{a.port}', flush=True)
    # --host 0.0.0.0 供局域网演示 (2026-08-28 用户: 怎么发给别人)。
    # 默认仍只绑本机 — 内部数据不该因为一次演示就对整个网段开放, 要开是显式动作。
    if getattr(a, 'farm', None):
        # ★CFG/ST 在模块加载时就绑定了 (第 81 行), 这里必须显式重绑而不是只切 current —
        # 否则 --farm 只改了配置模块里的指针, 服务用的仍是启动时那个场 (静默不生效)。
        # 用模块字典赋值而非 global: main 里更早处已引用过 CFG, global 声明会触发
        # "used prior to global declaration"。改模块属性等效且不受声明位置约束。
        from src.windscada import config as _cfg
        import sys as _sysmod
        _me = _sysmod.modules[__name__]
        _cfg.set_current(a.farm)
        _me.CFG = _cfg.farm()
        _me.ST = pathlib.Path(_me.CFG['store'])
        _CACHE.clear()
        _STAMP['val'] = None            # 换场后产物指纹必须重算, 否则新场的缓存判为"未变化"
        print(f'场: {_me.CFG["name"]} ({a.farm}, {_me.CFG["n_turbines"]} 台, 仓 {_me.ST})')
    else:
        print(f'场: {CFG["name"]} (默认; 可用 {list(__import__("src.windscada.config", fromlist=["available"]).available())})')
    _host = getattr(a, 'host', '127.0.0.1')
    if _host != '127.0.0.1':
        import socket as _sk
        try:
            _ip = _sk.gethostbyname(_sk.gethostname())
        except Exception:
            _ip = _host
        print(f'⚠ 已对局域网开放: http://{_ip}:{a.port}  (同网段任何人可访问, 演示结束请关闭)')
    # ★按窗重算的**预热**（用户令 2026-09-21：判级/曲线随所选时间窗变化）。
    #   判级矩阵 ≈25 s/窗、镜头 ≈10~45 s/窗 —— 不预热的话重启后第一个访客要等 pending 轮询。
    #   这里在**后台守护线程**里顺序把预设窗算进进程缓存（页面照常可用，未就位的窗如实标 pending）。
    def _warm():
        if _os.environ.get('GUANLAN_WIN_WARM', '1') == '0':
            print('[warm] 按窗预热被 GUANLAN_WIN_WARM=0 关掉（按需算，首次看窗会 pending 一轮）', flush=True)
            return
        import time as _t
        n = 0
        for w in list(WINDOWS):
            try:
                sysmx_of(w)                       # 未命中即起后台线程
                if _win_wait('sysmx', w, timeout=180) is not None:
                    n += 1
                _t.sleep(2)                       # 留口气: 小内存机器上连续重算更容易被系统盯上
            except Exception as e:
                print(f'[warm] 判级 {w} 失败: {type(e).__name__}: {e}'[:160], flush=True)
        for w in ('2026年',):
            try:
                curves_of(w)                      # 未命中即起后台线程
                _win_wait('curves', w, timeout=240)
            except Exception as e:
                print(f'[warm] 镜头 {w} 失败: {type(e).__name__}: {e}'[:160], flush=True)
        print(f'[warm] 按窗缓存预热完成: 判级 {n}/{len(WINDOWS)} 窗（可用内存 {_mem_mb()} MB）', flush=True)

    _thr.Thread(target=_warm, name='win-warm', daemon=True).start()
    ThreadingHTTPServer((_host, a.port), H).serve_forever()