guanlan-src/scripts/detail_deps.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
r"""`/detail` 详细分析工作台 · 页面 → 取数接口 → 产物 → 来源 的**可复算**台账 (2026-09-18 用户令)。

用户令: 「http://127.0.0.1:28084/detail 下的各页面, 不能用旧版的产出补, 只能参照旧版产物的呈现
样式内容, 必须基于输入数据重算」。本器把这条令变成一份**每次都能重跑出来的清单**:

    页面(页签)  ←  取数接口(在线 /api/*)  ←  产物件  →  来源(台账) + 生成端(族表) + 缺口处置

证据不是印象, 两头都对着源码核:
  · 页面 → 接口: 在 `src/windscada/ui/app.js` 里逐个搜到该接口 (页签表 TABS 与各 fetch 调用);
  · 接口 → 产物: 声明时给出 `scripts/windscada_serve.py` 里的函数名, 本器逐一核对该符号存在;
  · 产物 → 来源: 读 `outputs/<场>/_provenance.json`(逐件来源台账, 由 products_restore_missing
    --refresh 维护) 与 `scripts/products_reverse_audit.py` 的族表(生成端)。
任一条核不上 → 本器报 `[核不上]` 并 rc=5 —— 页面改了接口而清单没跟, 或产物换了生成端, 当场露出来。

`/detail` 是**网关**(scripts/guanlan_gateway.py, 默认 28084)把请求转到**组件服务**
(scripts/windscada_serve.py, 默认 18033)的 `/v2` 前端; 两边都是同一套 `/api/*`, 所以本清单对
在线页与单文件快照 (src/windscada/ui/snapshot.py) 同时成立。

用法:
    python scripts/detail_deps.py                # 打印清单 (人看)
    python scripts/detail_deps.py --probe        # 额外实测在线接口 (需服务在跑; 只打印, 不写盘)
    python scripts/detail_deps.py --write        # 写 docs/detail_页面依赖与重算台账_v0.1.md
    python scripts/detail_deps.py --check        # 文档是否与现场一致 (rc=5 = 该重写)
"""
from __future__ import annotations

import argparse
import fnmatch
import json
import pathlib
import re
import sys

ROOT = pathlib.Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT))
sys.path.insert(0, str(ROOT / 'scripts'))
from src import paths as P                                              # noqa: E402

APPJS = ROOT / 'src' / 'windscada' / 'ui' / 'app.js'
SERVE = ROOT / 'scripts' / 'windscada_serve.py'
DOC = ROOT / 'docs' / 'detail_页面依赖与重算台账_v0.1.md'
BEGIN, END = '<!-- DETAIL-DEPS:BEGIN -->', '<!-- DETAIL-DEPS:END -->'
PBEGIN, PEND = '<!-- DETAIL-PROBE:BEGIN -->', '<!-- DETAIL-PROBE:END -->'

# ── 页签 → 取数接口 (证据: app.js 里能搜到该接口) ────────────────────────────────
#   ★实测校准 (2026-09-18): 所有页签的**底数**都是 `/api/fleet` (app.js:1029 `load()`), 其余按页签
#     各自取: decision→ont_chain(403) · report→rpt_compose/facts(586) · system→maint_survey(728)
#     · generation→curves(752) · vibration/cms→vibcms(931) · assistant→ask*/facts(425/413)。
#     下钻浮层 (`/api/problem` `/api/turbine`, app.js:995) 从任意页签都能触发, 挂在部件页。
PAGES: list[tuple[str, str, list[str]]] = [
    ('overview', '总览', ['/api/fleet']),
    ('energy', '发电与损失', ['/api/fleet']),
    ('component', '部件与系统', ['/api/fleet', '/api/problem', '/api/turbine']),
    ('vibration', '振动评估(CMS)', ['/api/fleet', '/api/vibcms']),
    ('generation', '特性曲线', ['/api/curves']),
    ('fault', '故障与可用率', ['/api/fleet']),
    ('decision', '决策链', ['/api/fleet', '/api/ont_chain']),
    ('assistant', '问答', ['/api/fleet', '/api/ask', '/api/ask_status', '/api/ask_models',
                           '/api/facts', '/api/facts/claim']),
    ('report', '报告', ['/api/fleet', '/api/facts', '/api/rpt_compose']),
    ('system', '系统自查', ['/api/fleet', '/api/maint_survey']),
]

# ── 接口 → 产物 → 生成端 (证据: windscada_serve.py 里的符号名, 本器核对存在) ──────
#   req=True  = 缺了这页就出不来 (接口回 no_products / err);
#   req=False = 接口内有 .exists() 兜底, 缺了只是少一块 (页面仍出, 但内容不全)。
APIS: list[dict] = [
    dict(path='/api/fleet', fn='fleet_view', req=[
        'windscada/temp_monthly.parquet', 'windscada/alarms.parquet', 'windscada/loss_monthly.parquet',
        'windscada/powercurve_bins.parquet', 'windscada/powercurve_dev.parquet'],
        opt=['windscada/*.parquet', 'ontology/turbine_params.parquet',
             'm5_cms_tcm/handoff_vibration_v2.json', 'm5_cms_tcm/fleet_scalar_z.parquet',
             'pitch/pitch_daily.parquet', 'pitch/pitch_zero_monthly.parquet'],
        note='硬缺件在 `_load()` 里直接抛 ProductsMissing ⇒ 接口回 err=no_products'),
    dict(path='/api/curves', fn='curve_view',
        req=['windscada/powercurve_bins.parquet'],
        opt=['windscada/curve_lenses.parquet', 'windscada/curve_liveness.parquet',
             'windscada/control_profile.parquet', 'windscada/hydraulic_accum.parquet'],
        note='曲线镜头读 10min 原始档 + 已重算的派生件'),
    dict(path='/api/problem', fn='single_problem', req=['windscada/alarms.parquet'],
        opt=['ontology/objects.json', 'windscada/workorders.parquet'], note='部件下钻浮层'),
    dict(path='/api/turbine', fn='turbine_problems', req=['windscada/alarms.parquet'],
        opt=['ontology/objects.json', 'windscada/workorders.parquet'], note='逐台下钻浮层'),
    dict(path='/api/vibcms', fn='vibcms_results',
        req=['windcms/报告_CMS振动状态评估报告_*.md'],
        opt=['m5_cms_tcm/handoff_vibration_v2.json', 'windcms/cache/*'],
        note='报告是振动六层链 report 步的出件; 缺它本接口如实回 no_report(不再抛 IndexError)'),
    dict(path='/api/maint_survey', fn='maint_framework_view', req=['ontology/objects.json'], opt=[],
        note='维护面盘点读本体对象库 (/v2 的「系统自查」页签取它)'),
    dict(path='/api/maint_std', fn='maint_framework_view', ui='classic', req=['ontology/objects.json'],
        opt=[], note='四判据+成熟度 (经典页 `/?…` 用, /v2 不取)'),
    dict(path='/api/maint_framework', fn='maint_framework_view', ui='classic', req=['ontology/objects.json'],
        opt=[], note='运维体系三维 (经典页用, /v2 不取)'),
    dict(path='/api/dq_findings', fn='dq_findings_view', ui='classic', req=['ontology/objects.json'], opt=[],
        note='只认 finding/data_quality|caliber|system_coverage|tier_trend 四族对象; 这四族**不在重算链上** '
             '(唯一写方 scripts/ingest_ops_2025.py 读 data/raw/工作库, 该目录不在现场数据里)'),
    dict(path='/api/ont_chain', fn='ontology_view', req=['ontology/objects.json'], opt=[], note='故障链/预防链/故障树'),
    dict(path='/api/ontology', fn='ontology_view', ui='classic', req=['ontology/objects.json'], opt=[],
        note='本体总表 (经典页用, /v2 用 fleet 里的本体投影)'),
    dict(path='/api/facts', fn=None,
        req=['guanlan/facts_contract_v0.json', 'guanlan/derived/detail_cards.json'],
        opt=['guanlan/derived/qa_refs.json', 'guanlan/derived/portal_claims.json'],
        note='契约链: 输入是 sop/findings.json + paradigm_r1 三份**人裁底稿** ⇒ 与 data/raw 隔着不止一跳'),
    dict(path='/api/ask', fn=None, req=[], opt=[], note='本机 Ollama; 无产物依赖'),
    dict(path='/api/ask_status', fn=None, req=[], opt=[], note='本机 Ollama; 无产物依赖'),
    dict(path='/api/ask_models', fn=None, req=[], opt=[], note='本机 Ollama; 无产物依赖'),
    dict(path='/api/rpt_compose', fn=None, req=[], opt=[], note='本机 Ollama; 无产物依赖'),
]


def _expand_braces(glob) -> list[str]:
    """族 glob (str 或 list) → 展开 `{a,b}` 后的模式列表 (与 products_reverse_audit._files_of 同规则)。"""
    out = []
    for g in ([glob] if isinstance(glob, str) else list(glob)):
        if '{' in g:
            head, tail = g.split('{', 1)
            opts, rest = tail.split('}', 1)
            out += [head + o + rest for o in opts.split(',')]
        else:
            out.append(g)
    return out


def _pat_match(rel: str, glob) -> bool:
    """rel 是否落在族 glob 里 (深度对齐: 模式无 `**` 时 `/` 个数必须相等 —— fnmatch 的 `*` 跨 `/`)。"""
    for pt in _expand_braces(glob):
        if '**' not in pt and rel.count('/') != pt.count('/'):
            continue
        if rel == pt or fnmatch.fnmatch(rel, pt):
            return True
    return False


# ── 环境读取 ────────────────────────────────────────────────────────────────────
def load_env(farm: str | None = None):
    root = P.out_root(farm)
    ledger = {}
    lf = root / '_provenance.json'
    if lf.is_file():
        try:
            ledger = (json.loads(lf.read_text(encoding='utf-8')) or {}).get('files') or {}
        except Exception:
            ledger = {}
    disk = {p.relative_to(root).as_posix(): p.stat().st_size
            for p in root.rglob('*') if p.is_file()}
    try:
        from products_reverse_audit import FAMILIES
    except Exception:
        FAMILIES = []

    def family_of(rel: str):
        for fam in FAMILIES:
            if _pat_match(rel, fam['glob']):
                return fam
        return None

    return dict(farm=farm or P.farm(), root=root, ledger=ledger, disk=disk, family_of=family_of,
                fams=FAMILIES)


def expand(pat: str, env) -> list[str]:
    got = [r for r in env['disk'] if fnmatch.fnmatch(r, pat)]
    return sorted(got)


def row_of(pat: str, env) -> dict:
    """一条产物声明 → 在位情况 + 来源 + 生成端。"""
    hits = expand(pat, env)
    if hits:
        srcs = sorted({(env['ledger'].get(r) or {}).get('source', '(未登记)') for r in hits})
        fams = []
        for r in hits[:40]:
            f = env['family_of'](r)
            if f and f['id'] not in fams:
                fams.append(f['id'])
        gen = next((env['family_of'](r) for r in hits if env['family_of'](r)), None)
        return dict(pat=pat, n=len(hits), bytes=sum(env['disk'][r] for r in hits),
                    src='·'.join(srcs), builder=(gen or {}).get('gen') or '—',
                    family='·'.join(fams) or '—', ok=True)
    f = env['family_of'](pat) if '*' not in pat else None
    return dict(pat=pat, n=0, bytes=0, src='—', builder='—',
                family=(f or {}).get('id') or '—', ok=False)


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument('--probe', action='store_true', help='实测在线接口 (需服务在跑; 只打印)')
    ap.add_argument('--write', action='store_true', help='写回文档')
    ap.add_argument('--check', action='store_true', help='核对文档与现场一致 (rc=5 = 该重写)')
    ap.add_argument('--farm', default=None)
    a = ap.parse_args()
    env = load_env(a.farm)
    appjs = APPJS.read_text(encoding='utf-8')
    serve = SERVE.read_text(encoding='utf-8')

    # ① 证据核对: 页面端接口在 app.js 里找得到吗; 接口函数在 serve 里找得到吗
    bad = []
    for pid, name, eps in PAGES:
        if f"'{pid}'" not in appjs:
            bad.append(f'页签 {pid} 不在 app.js 的 TABS 里')
        for ep in eps:
            if f"'{ep}'" not in appjs and f"`{ep}" not in appjs and ep not in appjs:
                bad.append(f'页签 {pid} 声明的 {ep} 在 app.js 里搜不到')
    for it in APIS:
        if it.get('ui') == 'classic':
            # 经典页 (/ 老前端) 的接口: 出处是 serve 里的路由分支, 不是 app.js
            if f"'{it['path']}'" not in serve:
                bad.append(f"经典页接口 {it['path']} 在 windscada_serve.py 的路由里搜不到")
        elif f"'{it['path']}'" not in appjs and it['path'] not in appjs:
            bad.append(f"{it['path']} 声明给 /v2 用, 但 app.js 里搜不到它")
        if it['fn'] and f"def {it['fn']}(" not in serve:
            bad.append(f"{it['path']} 声明的实现函数 {it['fn']}() 在 windscada_serve.py 里不存在")

    probe = {}
    if a.probe:
        probe = {it['path']: _probe(it['path']) for it in APIS}

    L = []
    L.append(f'# /detail 页面依赖与重算台账 (自动生成, 禁手改)')
    L.append('')
    L.append(f'> 生成端: `python scripts/detail_deps.py --write`｜场站 `{env["farm"]}`｜'
             f'产物仓 `{P.rel(env["root"])}`｜盘上 {len(env["disk"]):,} 件')
    L.append('>')
    L.append('> 用户令 (2026-09-18): **/detail 各页面不许用旧版产出补**, 只能参照旧版产物的呈现样式与内容, '
             '数值必须由 `data/raw` 重算。本表把每页的取数接口与产物件逐条列清, 缺的写明怎么补。')
    L.append('')
    L.append('## 1 页面 → 取数接口')
    L.append('')
    L.append('| 页签 | 页面 | 取数接口 |')
    L.append('|---|---|---|')
    for pid, name, eps in PAGES:
        for i, ep in enumerate(eps):
            L.append(f'| {pid if i == 0 else ""} | {name if i == 0 else ""} | `{ep}` |')
    L.append('')
    L.append('注: `/api/maint_std` `/api/maint_framework` `/api/dq_findings` `/api/ontology` 属**经典页** '
             '(`/?…`, 组件根), `/v2` 工作台不取它们 —— 但同一批本体产物 (ontology/objects.json) 仍然决定着'
             '它们能不能出数, 故一并列在 §2。')
    L.append('')
    L.append('## 2 接口 → 产物 → 来源 / 生成端')
    L.append('')
    L.append('| 接口 | 产物 | 件 | 大小 | 台账来源 | 生成端(族) | 判定 |')
    L.append('|---|---|---:|---:|---|---|---|')
    gaps = []
    for it in APIS:
        for kind, label in (('req', '✅必需'), ('opt', '○可选')):
            for pat in it[kind]:
                r = row_of(pat, env)
                verdict = ('在位' if r['ok'] else ('**缺**' if kind == 'req' else '缺(页面降级)'))
                if not r['ok']:
                    gaps.append((it['path'], pat, kind, r['family']))
                L.append(f"| {it['path'] if pat == it[kind][0] else ''} | `{pat}` {label} | {r['n']} | "
                         f"{r['bytes'] / 1e6:.1f} MB | {r['src']} | {r['builder']} ({r['family']}) | {verdict} |")
    L.append('')
    L.append('## 3 缺口与处置 (缺的件谁生成、能不能从 data/raw 重算)')
    L.append('')
    if not gaps:
        L.append('无缺口: 表内每件产物都在位且来源已登记。')
    else:
        L.append('| 接口 | 缺件 | 必需 | 生成端 | 能否由 data/raw 重算 |')
        L.append('|---|---|---|---|---|')
        for ep, pat, kind, fam in gaps:
            g = _gen_of(env, pat)
            L.append(f"| `{ep}` | `{pat}` | {'是' if kind == 'req' else '否'} | {g['gen']} | {g['how']} |")
    L.append('')
    L.append('## 4 与输入数据的对应 (重算口径)')
    L.append('')
    L.append('| 产物族 | 输入 (data/raw/<场站>/…) | 生成端 |')
    L.append('|---|---|---|')
    seen = set()
    for it in APIS:
        for pat in it['req'] + it['opt']:
            f = env['family_of'](pat) if '*' not in pat else None
            if not f or f['id'] in seen:
                continue
            seen.add(f['id'])
            L.append(f"| {f['id']} | {f.get('input') or '—'} | `{f.get('gen') or '—'}` |")
    L.append('')
    L.append('## 5 复现')
    L.append('')
    L.append('```')
    L.append('python scripts/detail_deps.py --probe        # 实测在线接口 (页面同路径: 网关 → 组件 /v2)')
    L.append('python scripts/products_restore_missing.py --refresh   # 重算后维护逐件来源台账')
    L.append('python scripts/rebuild_all.py               # 全量重算 (链上含 ⑤a 台账维护 → ⑤ 反向呼应审计)')
    L.append('```')
    body = '\n'.join(L)
    text = f'{BEGIN}\n{body}\n{END}\n'

    # ── 实测段: 与"清单"分开标记 —— 它的数随服务在场与否变, 不该让 --check 抖动 ─────────
    ptext = ''
    if probe:
        P_ = ['## P 实测 (本次运行, 走网关 28084 → 组件 /v2)', '',
              '| 页签 | 接口 | 判定 | 实测 |', '|---|---|---|---|']
        for pid, name, eps in PAGES:
            for i, ep in enumerate(eps):
                r = probe.get(ep) or {}
                st = (f"HTTP {r['http']}" if r.get('http') else '—') + \
                     (f" · {r['err'][:70]}" if r.get('err') else (f" · {r['n']:,} B" if r.get('n') else ''))
                P_.append(f"| {pid if i == 0 else ''} | `{ep}` | {r.get('verdict', '—')} | {st} |")
        P_ += ['', '判读口径: 只认响应**顶层**的 `err`/`error`/`no_products`/`no_report`; '
                   '`/api/ask*` `/api/rpt_compose` 依赖本机模型, 未启动时记「不适用」不算缺件。']
        ptext = f'{PBEGIN}\n' + '\n'.join(P_) + f'\n{PEND}\n'

    if bad:
        print('证据核对不通过 —— 清单与源码/现场脱节:')
        for b in bad:
            print('   [核不上] ' + b)
    print(f'清单: {len(PAGES)} 页签 · {len(APIS)} 接口 · 缺口 {len(gaps)} 件 · '
          f'产物仓 {len(env["disk"]):,} 件')
    if a.write:
        old = DOC.read_text(encoding='utf-8') if DOC.is_file() else ''
        if PBEGIN in old and PEND in old:                 # 旧实测段先整段摘掉, 再按需重写
            old = old[:old.index(PBEGIN)].rstrip() + '\n' + old[old.index(PEND) + len(PEND):].lstrip('\n')
        if BEGIN in old and END in old:
            new = old[:old.index(BEGIN)] + text + old[old.index(END) + len(END):].lstrip('\n')
        else:
            new = (old.rstrip() + '\n\n' if old.strip() else
                   '# /detail 页面依赖与重算台账 v0.1\n\n'
                   '用户令 (2026-09-18): `/detail` 各页面不许用旧版产出补, 必须由 `data/raw` 重算;\n'
                   '旧版产物只可参照**呈现样式与内容**。逐页依赖见下表 (自动生成段)。\n\n') + text
        new = new.rstrip() + '\n\n' + ptext if ptext else new
        DOC.write_text(new, encoding='utf-8')
        print(f'已写 {P.rel(DOC)}' + ('' if ptext else ' (未带实测段; 加 --probe 可一并写入)'))
    if a.check:
        cur = DOC.read_text(encoding='utf-8') if DOC.is_file() else ''
        seg = cur[cur.index(BEGIN):cur.index(END) + len(END)] if (BEGIN in cur and END in cur) else ''
        if seg.strip() != text.strip():
            print('[X] 文档与现场不一致 ⇒ 跑 --write 重写')
            return 5
        print('[OK] 文档与现场一致 (清单段; 实测段随服务状态变, 不参与比对)')
    return 5 if bad else 0


def _gen_of(env, pat: str) -> dict:
    """缺口件 → 生成端与"能否从 data/raw 重算"的如实判断 (依据: 族表的 kind/algo/why)。"""
    f = next((fam for fam in env['fams'] if _pat_match(pat, fam['glob'])), None)
    if not f:
        return dict(gen='—', how='族表里没有这一族 ⇒ 需人工归口')
    if f.get('gen'):
        return dict(gen=f"`{f['gen']}`", how='能: 放原始件到 data/raw 后 `rebuild_all.py`')
    return dict(gen='(无生成端)', how=f"**不能**: {str(f.get('why') or '')[:60]} ⇒ 需研发补生成端")


def _probe(ep: str) -> dict:
    """实测一个接口 (走网关, 与页面同路径)。判读只认**顶层**的 err/error/no_products/no_report ——
    ★2026-09-18 首版按正则全文搜 `"err":` 判读, 结果把 fleet 响应里某个以 err 结尾的字段的**值**
    (per_turbine) 当成了报错, 页面明明好着却显示"有缺件痕迹"。判读不许靠猜文本。"""
    import urllib.error
    import urllib.request
    base = 'http://127.0.0.1:28084/detail'
    q = {'/api/fleet': '?win=2026%E5%B9%B4',
         '/api/turbine': '?t=WTG01&win=2026%E5%B9%B4',
         '/api/problem': '?t=WTG01&sys=%E5%8F%98%E6%A1%A8%E7%B3%BB%E7%BB%9F&win=2026%E5%B9%B4',
         '/api/ont_chain': '?kind=prev&system=%E5%8F%98%E6%A1%A8%E7%B3%BB%E7%BB%9F',
         '/api/facts/claim': '?id=RD-000'}.get(ep, '')
    try:
        with urllib.request.urlopen(base + ep + q, timeout=60) as r:
            raw, code = r.read(), r.status
    except urllib.error.HTTPError as e:
        raw, code = e.read(), e.code
    except Exception as e:
        return dict(http=None, n=0, err=f'{type(e).__name__}: {e}', verdict='连不上')
    try:
        obj = json.loads(raw.decode('utf-8', 'replace'))
    except Exception:
        return dict(http=code, n=len(raw), err='(非 JSON)', verdict='?')
    if not isinstance(obj, dict):
        return dict(http=code, n=len(raw), err='', verdict='有数')
    msg = next((str(obj[k]) for k in ('err', 'error') if obj.get(k)), '')
    if ep == '/api/ask':      # 只收 POST (页面上是表单提交), GET 探测必 404 —— 不是缺件
        return dict(http=code, n=len(raw), err='', verdict='不适用(POST 接口)')
    # 本机模型 (Ollama) 没起 / 未接: 这不是"重算缺件", 标成不适用 —— 与 `guanlan.py check` 的口径一致
    if ep in ('/api/ask_status', '/api/ask_models', '/api/rpt_compose'):
        return dict(http=code, n=len(raw), err=msg,
                    verdict='不适用(本机模型未启动)' if msg or code >= 400 else '有数')
    if obj.get('no_products') or obj.get('no_report') or (ep == '/api/facts/claim' and code >= 400):
        return dict(http=code, n=len(raw), err=msg or 'no_products', verdict='无产物(如实)')
    if msg:
        return dict(http=code, n=len(raw), err=msg, verdict='报错')
    return dict(http=code, n=len(raw), err='', verdict='有数')


if __name__ == '__main__':
    sys.exit(main())