guanlan-src/scripts/products_reverse_audit.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
r"""**反向**呼应审计：自输出 → 功能与算法 → 输入（用户令 2026-09-17）。

## 与 `inventory_products.py` 的区别（两个方向，各管一件事）

· `inventory_products.py` 是**正向**：从 `data/raw/<场>/` 的每一类输入出发，找它喂出来的产物，
  比跨度/条数（"输入到了 2026-07，产物只到 2026-04 → 未重算"）。它只覆盖 5 类输入、十几件产物。
· 本器是**反向**：从 `outputs/<场>/` 的**每一件产物**出发，问三个问题：
    ① 它是谁算出来的？（功能与算法 = 生成端）
    ② 它从哪份输入算出来的？（`data/raw/<场>/` 的哪一类）
    ③ 这条"输出 ↔ 输入"的呼应关系**成立吗**？（生成端在位 ∧ 输入在位 ∧ 可机检的对应判据通过）
  逐件判定，最后给出"成立 / 不成立（无生成端）/ 无法验证"三类账。

## 为什么必须做反向

正向只查"我关心的输入有没有被算成产物"，查不出**"盘上这件产物到底有没有来路"**。
2026-09-17 现场就是栽在这里：清了产物之后，页面缺的 `windscada/index.html`、
`m5_cms_tcm/handoff_vibration_v2.json` 这类件**根本没有生成端**（全库 0 处写入方），
正向那几条跨度判据一条都不会报 —— 它们压根不在 `INPUTS` 的 `feeds` 里。
反向一对账就清楚了：**有来路的件**（生成端 + 输入 + 判据都过）与**没来路的件**（只能从交付包补）。

## 判定口径（每条都写进结果里，不含糊）

    ✓ 呼应成立          生成端在位 ∧ 输入在位 ∧ 判据通过（跨度 ⊆ 输入 / 键集 ⊆ 输入 / 计数一致 / 行数 > 0）
    ~ 输入不在位         生成端知道，但 `data/raw/<场>/<类>` 不在 ⇒ **现在无法验证**（放数据后复跑本器）
    ✗ 无生成端           全库 0 处写入方 ⇒ **输出↔输入的呼应在原理上不成立**：这件产物无法由输入推导出来，
                        **无法由重算生出来**；按用户令 2026-09-17 #1 运行期也不从交付包补齐 ⇒ 只能由研发补生成端（本器 --feasibility 给出逐族可逆性）
    ? 未归类             既不在族表里、台账里也没有 —— 需要人工认领（本器把它当缺口报出来）

## 用法

    python scripts/products_reverse_audit.py            # 全量反向审计（人看）
    python scripts/products_reverse_audit.py --check    # 只出结论（有 ✗/? 时退出码 5）
    python scripts/products_reverse_audit.py --write-doc # 把族表写进 docs/系统设计说明.md §13.6

退出码: 0 全部成立（允许 ✗，只要它们都在台账里如实标了 shipped）· 5 有未归类件或判据失败
"""
from __future__ import annotations

import argparse
import fnmatch
import json
import pathlib
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
import inventory_products as INV                                        # noqa: E402

DOC_BEGIN, DOC_END = '<!-- REVERSE-AUDIT:BEGIN -->', '<!-- REVERSE-AUDIT:END -->'

# ── 族表: 输出(glob) → 功能 / 算法(生成端) / 输入类 / 判据 ────────────────────────────
#   pred: span=产物时间跨度必须落在输入跨度内; turbines=产物机组集 ⊆ 输入机组集;
#         rows=行数>0; count=件数/计数与说明一致; manifest=与清单自记一致
FAMILIES: list[dict] = [
    dict(id='vib_window_index', glob='m5_cms_tcm/windows/*/index.parquet', kind='raw-derived',
         func='振动摄入 · 窗索引', algo='scripts/rudong_tcm_index.py（逐 decode JSON 解析 54 列：turbine/'
              'sensor_name/meas_name/trigger_time/rpm/condition_key/alarm_type）',
         gen='scripts/rudong_tcm_index.py', input='windcms', pred=('span', 'turbines', 'rows'),
         time_col='trigger_time'),
    dict(id='vib_spectra', glob=['m5_cms_tcm/windows/*/spectra/**', 'm5_cms_tcm/windows/*/spectra_meta.parquet'],
         kind='raw-derived',
         func='振动摄入 · 谱库', algo='scripts/rudong_tcm_spectra.py（FFT_ 测量 → npz 幅值数组, 分片存 `spectra/p<NN>/`）',
         gen='scripts/rudong_tcm_spectra.py', input='windcms', pred=('npz',)),
    dict(id='vib_raw_manifest', glob='m5_cms_tcm/vib_raw_manifest.json', kind='raw-derived',
         func='振动摄入 · 清单', algo='scripts/vib_raw_build.py（汇总窗/谱件数、时间跨度、missing_chain 缺口）',
         gen='scripts/vib_raw_build.py', input='windcms', pred=('manifest',)),
    dict(id='scada_alarms', glob='windscada/alarms.parquet', kind='raw-derived',
         func='三门台账 · 报警', algo='scripts/windscada_alarms_ingest.py（SpreadsheetML *.xls → 事件表）',
         gen='scripts/windscada_alarms_ingest.py', input='故障报警', pred=('span', 'rows'), time_col='t_on'),
    dict(id='scada_workorders', glob='windscada/workorders.parquet', kind='raw-derived',
         func='三门台账 · 工单', algo='scripts/windscada_workorder_ingest.py（检修台账 → 工单表）',
         gen='scripts/windscada_workorder_ingest.py', input='风机故障记录', pred=('span', 'rows'),
         time_col='t_report'),
    dict(id='scada_oil', glob='windscada/oil_samples_index.parquet', kind='raw-derived',
         func='三门台账 · 油样', algo='scripts/windscada_watch_channels_build.py（油样 PDF → 索引）',
         gen='scripts/windscada_watch_channels_build.py', input='油样报告', pred=('span', 'rows'),
         time_col='date'),
    dict(id='scada_monthly', glob='windscada/temp_monthly.parquet', kind='raw-derived',
         func='月度派生件', algo='scripts/windscada_monthly_build.py（逐值对齐随包件 19,494/19,494）',
         gen='scripts/windscada_monthly_build.py', input='scada_10min', pred=('span', 'rows'), time_col='month'),
    dict(id='scada_derived', glob='windscada/*.parquet', kind='raw-derived',
         func='SCADA 派生分析（功率曲线/损失/曲线/控制/停机/温度/偏航/液压/热链/系统辅助）',
         algo='src/windscada/perf/{powercurve,availability,curves,control,faults} + subsys/{temp_nbm,yaw,'
              'hydraulic,thermal_chain} + taxonomy.py',
         gen='src/windscada/perf/powercurve.py', input='scada_10min', pred=('rows',)),
    dict(id='ontology_core', glob='ontology/{objects.json,retrieval_index.json,turbine_params.parquet}',
         kind='raw-derived',
         func='本体层 · 码表/对象库/检索索引/实机参数',
         algo='python -m src.ontology.kb_ingest → populate → chain_ingest → trend_ingest；'
              'src.ontology.retrieval.build；src.ontology.maintenance.refresh_params',
         gen='src/ontology/kb_ingest.py', input='西门子4.0技术资料', pred=('rows',)),

    dict(id='ontology_aux_build', glob=['ontology/_id_alias.json', 'ontology/retrieval_vec.npy'],
         kind='raw-derived',
         func='本体层 · 别名表与向量索引缓存', algo='src.ontology.kb_ingest（别名）/ '
              'src.ontology.retrieval.build(use_vec=True)（向量）',
         gen='src/ontology/kb_ingest.py', input='西门子4.0技术资料', pred=('rows',)),
    dict(id='ontology_domain', glob=['ontology/oil_2026H2_huabiao.json', 'ontology/parts_cost_public.json',
                                     'ontology/scenario_29_repair.json'],
         kind='shipped',
         func='本体层 · 领域数据件（油样台账/备件价格/场景修复）', algo='（无生成端, 领域正本出件）',
         gen=None, input=None, pred=(), why='领域正本随包发来, 不由 data/raw 推导'),

    # ── 以下族: 全库 0 处写入方 ⇒ 反向呼应**在原理上不成立**（如实记账, 不假装成立）────
    #    ★2026-09-18 用户令补充: 这些件**运行期一律不从交付包补**(用户令 2026-09-17 #1 的延伸),
    #      页面如实为空/降级; 要出件只有两条路 —— ① 研发补生成端; ② 现场正本随原始件进 data/raw。
    #      故本表的 why 里不再写"从交付包补齐"那种话(那是**离线人工补救**, 见 products_restore_missing.py)。
    dict(id='vib_model_run_l6', glob=['m5_cms_tcm/model_run_l6.parquet', 'm5_cms_tcm/model_run_summary.json'],
         kind='raw-derived',
         func='六层链 model_run 步 · L6 过闸谱线表 + 层小结',
         algo='scripts/rudong_model_run.py（候选线=oem_scan_plan 的 11 部件×{BPFI,BPFO,BSF}; 过闸='
              'src/sop/discriminators.py::spectral_line_gates G1–G8; 定级=vib_verdict_and_writeback）',
         gen='scripts/rudong_model_run.py', input='m5_cms_tcm', pred=('rows',),
         why='★2026-09-19 按口径重建（**无标准答案**: 这两件从未随包, 无法逐值对拍）; 峰值拾取仍是本器自定'
             '(目标频率 ±2 bin 取最大), 不得当"复现"用 —— 见 docs/振动六层链_接口规格与缺口_v0.1.md §3'),
    dict(id='vib_fusion_38', glob='m5_cms_tcm/fusion_38.csv', kind='raw-derived',
         func='六层链 fusion 步 · 逐台融合表（报告"融合级"列的来源）',
         algo='scripts/rudong_fusion_run.py::fusion_table（模型侧=model_run_l6 过闸线; CMS 侧=窗索引 '
              'RedMask/YellowMask; 裁决=src/sop/fusion_diag.py::fuse）',
         gen='scripts/rudong_fusion_run.py', input='m5_cms_tcm', pred=('rows',),
         why='★按口径重建（无标准答案）: 台号用 WTGnn 形态（消费端 registry 的键）, 与 l6 的 f"{n}#" 不同'),
    # ★2026-09-19 源代码化（第三批）：三族由观澜自算顶上。★族表是**首个命中生效**，故这几族必须
    #   排在通配族（`vib_handoff_and_scans` = `m5_cms_tcm/*.{json,parquet,csv,md,txt}`、`tcm_replay`、
    #   旧的 `ontology_releases`）**之前**，否则新件会被旧族认领成"无生成端"。
    dict(id='vib_line_energy_share', glob='m5_cms_tcm/line_energy_share.parquet', kind='raw-derived',
         func='六层链 energy_share 步 · 线能量占比',
         algo='scripts/rudong_line_energy_share.py（±2bin 带内能量 ÷ 全谱能量 ×100; 候选线与分母同 oem_scan_plan）',
         gen='scripts/rudong_line_energy_share.py', input='m5_cms_tcm', pred=('rows',),
         why='2026-09-19 补上: 该步脚本此前未随包(chain_gap_check 报"1 个步骤脚本未随包"); 只报占比不定级'),
    dict(id='vib_mask_thresholds', glob=['tcm_compatible_replay/model/mask_thresholds.tsv',
                                         'tcm_compatible_replay/model/mask_thresholds.说明.json'],
         kind='raw-derived', func='TCM 掩码阈值件 (逐台逐测点逐工况的黄/红线)',
         algo='scripts/tcm_mask_thresholds_build.py（中位 + 3/6×MAD 自适应基线, 基线先净化, 样本门 n≥30; '
              '调 src/sop/discriminators.py::adaptive_baseline_threshold 单一实现）',
         gen='scripts/tcm_mask_thresholds_build.py', input='m5_cms_tcm', pred=('rows',),
         why='★**观澜自算**阈值, 不是厂商阈值: 现场导出件里没有掩码定义(25,679 个 decode 只有 NumberOfMasks: 175) '
             '⇒ 厂商阈值无法复原; 口径写在同目录 mask_thresholds.说明.json, 页面据此标注(workbench.py 分支)'),
    dict(id='ontology_releases', glob='ontology/release_*/*', kind='raw-derived',
         func='本体发布层 r1/r2（网关 /release/ 只读暴露）',
         algo='scripts/ontology_release_build.py（对象库投影: r1 契约层 = objects.json 快照 + manifest; '
              'r2 = 同快照 + layering.json 分层覆盖声明, 不合并主库）',
         gen='scripts/ontology_release_build.py', input='西门子4.0技术资料', pred=('rows',),
         why='2026-09-19 由观澜自算顶上: 原为随包件(0 处写入方) ⇒ 新机器 /release/ 两件必 503; '
             'manifest 里写 source.sha16, 谁都能核出两层是同一份对象库的投影'),
    dict(id='vib_fleet_scalar_z', glob='m5_cms_tcm/fleet_scalar_z.parquet', kind='raw-derived',
         func='振动线 · 全场标量 z 值 (fusion 步出件)',
         algo='scripts/rudong_fusion_run.py（逆向工程实现: 窗索引 scalar_value → 同 (传感器,测量,工况) 族中位 '
              '→ 稳健 z=(val−med)/(1.4826·MAD); 与随包样件 8,887 行逐值对拍 val 列 100% 一致）',
         # input 写**原始件目录名**(族表口径: data/raw/<场>/<input>); 真实取数经窗索引一跳,
         # 而窗索引本身是 raw-derived(vib_window_index 族) —— 写 windows/_*.parquet 会让审计判"输入不在位"。
         gen='scripts/rudong_fusion_run.py', input='m5_cms_tcm', pred=('rows',),
         why='2026-09-18 上链: 由 vib_raw_build.py 的 --with-report 分支按当次窗名调用, '
             '使 /api/fleet 的振动标量 z 面不再空白'),
    dict(id='vib_fusion_handoff', glob='m5_cms_tcm/handoff_vibration_v2.json', kind='raw-derived',
         func='融合面 handoff (观澜自算, 同结构于振动线手交件)',
         algo='scripts/rudong_fusion_handoff.py（报告状态等级 + fusion_38 融合裁决 + L6 过闸线 → 同结构, '
              'meta.from 标明"观澜自算"）',
         gen='scripts/rudong_fusion_handoff.py', input='m5_cms_tcm', pred=('rows',),
         why='★2026-09-19 用户令「所有的计算均要形成观澜的源代码」: 这件此前全库 0 处写入方(振动线人写), '
             '新机器装完 `/detail/v2` 的「需要关注/全场状态」必然空白。现由观澜自算出同结构件; '
             '盘上若有**振动线正本**, 生成端不覆盖(正本含人工裁决, 优先)'),
    # ★2026-09-19 用户令「所有的计算均要形成观澜的源代码」: 这两件此前 kind=shipped(振动线人交手交件),
    #   新机器上 ⑦ 的 `src.ontology.trend_ingest` 报"源缺失"、CMS 报告的三层基线卡为空。
    #   现由两个生成端从**窗索引**算: component_history(逐窗标量 x_self → 在升/换件闭环; 窗不足时如实为空
    #   并把数据边界逐条写进 meta) / baseline_38(健康期分段中位 → 自基线 + fleet 基线 + ISO 绝对限)。
    #   ★必须排在通配族 `vib_handoff_and_scans` **之前** (族表首个命中生效), 否则又被认领成"无生成端"。
    dict(id='vib_history_baseline',
         glob=['m5_cms_tcm/component_history.json', 'm5_cms_tcm/baseline_38.json'],
         kind='raw-derived',
         func='振动在升/换件闭环历史 + 三层基线（自 / 机群 / 绝对）',
         algo='scripts/component_history_build.py（逐窗标量 x_self → 末三窗/早三窗在升 + 换件前后闭环; '
              '窗不足时判据不可算则如实空表）+ scripts/baseline_38_build.py（健康期分段中位 + 正常带 + '
              'relMAD → 自基线, 机群同段中位 → fleet 基线, ISO 20816-3 → 绝对限）',
         gen='scripts/component_history_build.py', input='windcms', pred=('rows',),
         why='输入是 CMS 原始导出的**重算产物**(窗索引), 其 raw 祖先是 data/raw/<场>/windcms; '
             '旧随包件是振动线手交件(全库 0 处写入方); 两件都在重算链 ④d/④e, 口径与数据边界写进件内 meta'),
    dict(id='vib_handoff_and_scans', glob='m5_cms_tcm/*.{json,parquet,csv,md,txt}',
         kind='shipped',
         func='振动线出件与专项扫描（handoff / 历史 / 基线 / 各类 freq scan / 判级台账）', algo='（无生成端）',
         gen=None, input=None, pred=(), why='振动六层链四步脚本未随包（oem_frequency_scan 等）⇒ '
                                            '运行期不从交付包补(用户令 2026-09-17); 要出件须研发补生成端, '
                                            '或由现场正本随原始件进 data/raw 后重算'),
    # (2026-09-19: `line_energy_share` / `mask_thresholds` / `ontology_releases` 三族已移到前面按
    #  raw-derived 记账 —— 族表是**首个命中生效**, 必须排在通配族 `vib_handoff_and_scans` / `tcm_replay`
    #  / `ontology_releases(旧)` 之前, 否则新件会被旧族认领成"无生成端")
    dict(id='vib_figs', glob='m5_cms_tcm/figs/*', kind='shipped',
         func='振动线出图（谱图/趋势图）', algo='（无生成端）', gen=None, input=None, pred=(),
         why='随包快照（振动线出件）'),
    # ★2026-09-19 拆族 (用户令「所有的计算均要形成观澜的源代码」): `windscada/index.html` 是
    #   `/detail/v2` 的「全场状态总览」iframe 目标, 此前整族记为"无生成端" ⇒ 新机器上那块必然显示
    #   "产物不在位"。现在它由 `scripts/windscada_overview_build.py` 用**组件同一个快照烘焙器**
    #   (src/windscada/ui/snapshot.py::bake, /v2/snapshot 用的就是它)从重算产物生成 ⇒ 单列一族。
    dict(id='windscada_overview', glob='windscada/index.html', kind='raw-derived',
         func='总览「全场状态」页 (单文件: 接口响应烤进页面, 离线可开)',
         algo='scripts/windscada_overview_build.py（ctx 取数函数与在线 /api/* 同一实现; 经 '
              'src/windscada/ui/snapshot.py::bake 烘焙）',
         gen='scripts/windscada_overview_build.py', input='scada_10min', pred=('rows',),
         why='2026-09-19 用户令后上链: 该页此前是随包件(全库 0 处写入方), 清过产物的机器上 iframe 只能'
             '显示"产物不在位"; 现在由观澜自算, 每次重算随数据更新'),
    dict(id='windscada_pages', glob='windscada/{turbines/*,review/*}', kind='shipped',
         func='逐台静态页 + 评审记录（当前由组件按请求动态出页, 静态件仍属随包快照）', algo='（无生成端）',
         gen=None, input=None, pred=(), why='逐台页在运行期由 /turbine/<台> 动态生成, 静态副本无写入方; '
                                             '运行期不从交付包补(用户令 2026-09-17)'),
    # ★2026-09-18 补: `windcms/cache/*` 是**读侧缓存**(标量按内容哈希落盘), 它不是随包快照 ——
    #   原先 `windcms_pages`(shipped) 先认领了它, 于是重算后新写的缓存被记成"包内无生成端"。
    #   写方在 src/windcms/data.py::load_scalars (窗索引 → 标量), 故单列一族并放在 windcms_pages 之前。
    dict(id='windcms_cache', glob='windcms/cache/*', kind='raw-derived',
         func='CMS 报告侧 · 标量缓存', algo='src/windcms/data.py::load_scalars（窗索引 → 标量数组, '
              '按内容哈希命名落 cache/）', gen='src/windcms/data.py', input='windcms', pred=('rows',)),
    # ★2026-09-18 补 (用户令"页面只能基于输入数据重算"之后): `windcms/**` 整族原先一律按
    #   "无生成端"记账 —— 但**报告两步的生成端就在包内**(scripts/windcms.py report → src/windcms/
    #   report.py + report_std.py), 只要 data/raw/<场>/windcms 里有 decode 导出就出得来。
    #   实测(2026-09-18): 从重算出的窗 w0316 一键跑出标准模版报告 + 38 台逐台页 + 总览, 判级 38 台
    #   (报警 3 / 优秀 35), 缺的只有 model_run/fusion 两步的融合级列(如实写 —)。故拆成两族按 raw-derived 记。
    dict(id='windcms_report_std', glob=['windcms/报告_CMS振动状态评估报告_*.md',
                                        'windcms/报告_CMS振动状态评估报告_*.docx',
                                        'windcms/登记_三轴_38台.csv'], kind='raw-derived',
         func='CMS 标准模版评估报告 (三轴: 状态等级 × 证据状态 × 行动等级建议)',
         algo='src/windcms/report_std.py::build（结构化判级 → 报告 md + docx + 登记 csv; '
              '由 scripts/windcms.py report 调用）',
         gen='src/windcms/report_std.py', input='windcms', pred=('rows',),
         why='★同名件有两种来源: 观澜自算(本族: 从 data/raw/<场>/windcms 的 *_decode.json 算) 与 '
             '厂家报告转录(scripts/vib_reports_build.py, 输入是现场振动分析报告 PDF)。按名字认领不了'
             '这层区别 ⇒ 以"是否由本步生成"为准, 判据看 m5_cms_tcm/vib_raw_manifest.json 的 steps'),
    dict(id='windcms_report_overview', glob=['windcms/report.md', 'windcms/index.html',
                                             'windcms/index_eng.html', 'windcms/overview.html',
                                             'windcms/turbines/*'], kind='raw-derived',
         func='CMS 诊断总览页 + 逐台页 + 结构化层报告',
         algo='src/windcms/report.py::build（逐台 turbines/<台>.html + 总览 overview/index.html + '
              'report.md; 融合级列要 model_run/fusion 出件, 缺则如实写"—"）',
         gen='src/windcms/report.py', input='windcms', pred=('rows',)),
    dict(id='windcms_kb', glob='windcms/kb.json', kind='raw-derived',
         func='CMS 知识库索引 (问答检索用)',
         algo='src/windcms/knowledge.py::build（由 scripts/windcms.py kb 调用; 离线无模型时 0 chunks 也算跑成）',
         gen='src/windcms/knowledge.py', input='windcms', pred=('rows',)),
    dict(id='windcms_pages', glob='windcms/**', kind='shipped',
         func='windcms 里其余件 (厂家报告转录 / 知识库出件等)',
         algo='（无随包生成端）', gen=None, input=None, pred=(),
         why='上述两族与 cache 之外剩下的 windcms 件: 运行期不从交付包补(用户令 2026-09-17), '
             '要出件须研发补生成端'),
    # ★2026-09-19 用户令「要让观澜从重算台账生成 claim」: findings 与三份底稿改由**观澜自算**
    #   (scripts/sop_findings_from_ledger.py 从 alarms/workorders/oil/temp_bins/powercurve_dev/loss_monthly/L6
    #   算 claim, 按契约构建器要的字段结构写) ⇒ 单列一族; 契约四件随之由 scripts/guanlan_facts_contract.py 生成。
    dict(id='ledger_claims', glob=['sop/findings.json', 'paradigm_r1/experiments/*/底稿*.json'],
         kind='raw-derived',
         func='事实契约的 claim 本体与底稿 (观澜自算, 替代人裁底稿)',
         algo='scripts/sop_findings_from_ledger.py（六族 claim: 报警集中/重复检修/停机损失/温度离群/功率曲线偏离/振动过闸线; '
              '每条带 coverage·缺证据·证伪判据, 相对判据封顶「候选」）',
         gen='scripts/sop_findings_from_ledger.py', input='scada_10min', pred=('rows',),
         why='输入是**重算产物**(windscada/*.parquet + m5_cms_tcm/*): 其 raw 祖先是 故障报警/风机故障记录/scada_10min; '
             '本族单列以区别于旧的人裁底稿(同路径旧件无生成端)'),
    dict(id='sop_workspace', glob='sop/**', kind='shipped',
         func='SOP 中间件/评审/事实契约底稿', algo='（无生成端）', gen=None, input=None, pred=(),
         why='评审过程件, 属人工作业留痕'),
    dict(id='tcm_replay', glob='tcm_compatible_replay/**', kind='shipped',
         func='TCM 兼容链回放资产（模型表/掩码阈值/裁决记录）', algo='（无生成端）', gen=None, input=None,
         pred=(), why='随包快照'),
    dict(id='paradigm_r1', glob='paradigm_r1/**', kind='shipped',
         func='范式实验件（E3/E5/E8 底稿）', algo='（无生成端）', gen=None, input=None, pred=(),
         why='实验底稿, 事实契约的输入'),
    # ★2026-09-19 用户令「所有的计算均要形成观澜的源代码」: 变桨面两件原来 kind=shipped(无生成端)
    #   ⇒ 换台机器重算后必然缺件、变桨面恒"不可判"。现由 scripts/pitch_face_build.py 从 data/raw 的
    #   scada_10min(液压/润滑/柱塞/运行状态) + scada_1min(三叶桨距角/压力/有功) 算出, 并进重算链 ④c。
    dict(id='pitch_face', glob='pitch/**', kind='raw-derived',
         func='变桨侧派生件（日粒度液压/润滑/柱塞 + 零位三口径月表）',
         algo='scripts/pitch_face_build.py（10min 开关量 timeon/压力锯齿 + 1min 停机顺桨段/满发段/运行段分档 '
              '三叶桨距角中位 → 相对同月全场中位的零位偏差；口径/阈值/证据见同目录 pitch_face_manifest.json）',
         gen='scripts/pitch_face_build.py', input='scada_10min', pred=('rows',),
         why='输入是 10min 与 1min 两路 SCADA 原始导出件（scada_source: CSV 优先/MDB 回落）；'
             '零位按用户令两口径分列（停机顺桨段/满发段），并据实测反推补上**运行段同工况分档**这一判据轴'
             '（它才能把现场确诊的 19# 单独拎出来）'),
    # (2026-09-19: `ontology_releases` 已移到前面按 raw-derived 记账 —— 现在由
    #  scripts/ontology_release_build.py 从对象库投影生成, 不再是"由研发出件"的随包层)
    # ★2026-09-19 用户令「所有的计算均要形成观澜的源代码」+「从重算台账生成 claim」:
    #   这一族从"无生成端"改成 raw-derived —— 生成端 guanlan_facts_contract.py 本来就在包内,
    #   原来缺的是**它的输入**(人裁底稿); 现在输入由上游族 ledger_claims 自算
    #   (scripts/sop_findings_from_ledger.py) ⇒ 新机器上重算即得, /api/facts 不再 503。
    dict(id='guanlan_contract', glob='guanlan/**', kind='raw-derived',
         func='事实契约与对外派生（门户结论段/取数）',
         algo='scripts/guanlan_facts_contract.py（build→render: 契约 + portal_claims/detail_cards/qa_refs/'
              'report_summary; 输入=上游族 ledger_claims 自算的 sop/findings.json + 三份底稿）',
         gen='scripts/guanlan_facts_contract.py', input='scada_10min', pred=('rows',),
         why='输入与 data/raw 隔着两跳(重算台账 → claims → 契约), 但**两跳的生成端都在包内**; '
             'claim 的判据/缺证据/证伪逐条写在件里, 相对判据封顶「候选」'),
    dict(id='windscada_release_manifest', glob='windscada/release-manifest.json', kind='source-derived',
         func='发布清单快照（`/healthz` 与版本信息读它）',
         algo='scripts/guanlan_baseline_manifest.py（从 src/windscada/ui 源重建 zh 页 → 摘要 '
              '{css_tokens, nav, dom, sha256, source blobs}）',
         gen='scripts/guanlan_baseline_manifest.py', input=None, pred=(),
         why='★ 2026-09-17 修正: 它一度被我按"交证件"错分成非产物; 实际**包内有生成端**'
             '(guanlan_baseline_manifest.py 就写这个路径), 且 guanlan_gateway.py 运行期读它 ⇒ 是产物'),
    dict(id='human_deliverables',
         glob=['windscada/mblub_alarm_cross.json', 'windscada/review_presented.json',
               'windscada/给振动线_温度轴回复_20260826.json', 'windscada/给振动线_温度轴回复_20260826.md',
               'windscada/送审_系统结论与判定_20260828.md'],
         kind='human',
         func='**人工正本/交证件**：与振动线的跨线通知、送审结论、发布评审快照',
         algo='（人写的, 不是算出来的）', gen=None, input=None, pred=(),
         why='这几件是"人对人的交证件/回执", 内容由分析结论抄写与人判断坐实而成; '
             'mblub_alarm_cross.json 还被 src/ontology/populate.py 当作**证据来源**引用 ⇒ '
             '必须随包留在原位, 但**不参与呼应校验**(它本来就没有生成端, 也不该有)'),
    dict(id='scratch_in_products', glob=['**/_*.js', '**/_*audit*.md', '**/*.out', '**/*.err',
                                        '**/*_test_*.parquet', '**/*_test_*.json'],
         kind='not-product',
         func='**非产物**：自审脚本/记录与模型测试输出（躺在产物仓里）', algo='（工具脚本与运行痕迹, 不是产物）',
         gen=None, input=None, pred=(),
         why='2026-09-17: 17 件已移出到 reference/非产物留痕_20260917/ 并留了 _来源说明.json ⇒ '
             '本族现在应为空; 一旦再出现(新脚本往产物仓里写 .out/_audit.js 之类), 本器当场报出来'),
]


def _files_of(glob) -> list[str]:
    """族 glob（str 或 list）→ 台账里的相对键（支持 `{a,b}` 花括号与 `*`）。

    ★ 深度必须**显式对齐**（2026-09-17 第一版踩过）: `fnmatch` 的 `*` 是**跨 `/`** 的, 于是
      `windscada/*.parquet` 会把 `windscada/turbines/xx.parquet`、`windscada/_pre_rebuild_*/xx.parquet`
      一起吞进来 —— 反向审计最怕"族把不该管的件认领了", 那样账就假了。
      规则: 模式里没有 `**` 时, 要求的 `/` 个数必须与键的 `/` 个数**相等**(逐条展开后各自判)。
    """
    pats = []
    for g in ([glob] if isinstance(glob, str) else list(glob)):
        if '{' in g:
            head, tail = g.split('{', 1)
            opts, rest = tail.split('}', 1)
            pats += [head + o + rest for o in opts.split(',')]
        else:
            pats.append(g)
    out = []
    for rel in LEDGER:
        for pt in pats:
            if '**' not in pt and rel.count('/') != pt.count('/'):
                continue
            if fnmatch.fnmatch(rel, pt) or rel == pt:
                out.append(rel)
                break
    return sorted(out)


LEDGER: dict[str, dict] = {}


def load_ledger(farm: str) -> dict[str, dict]:
    """台账 = `_provenance.json`(逐件来源) + `_derived_manifest.json`(生成端自登记)。"""
    out: dict[str, dict] = {}
    root = P.out_root(farm)
    for fn in ('_provenance.json', '_derived_manifest.json'):
        f = root / fn
        if not f.is_file():
            continue
        try:
            d = json.loads(f.read_text(encoding='utf-8'))
        except Exception:
            continue
        for rel, v in (d.get('files') or {}).items():
            if isinstance(v, dict):
                out.setdefault(rel, dict(source=v.get('source') or 'raw-derived',
                                         builder=v.get('builder') or v.get('by') or '',
                                         why=v.get('why') or '',))
    return out


def _input_home(station: pathlib.Path, sub: str) -> pathlib.Path | None:
    """输入类目录在哪 —— 优先场站目录下, 其次 raw 根下。

    ★ 机理层资料按 A2 约定**不在场站目录下**（`data/raw/西门子4.0技术资料/`, 见 src/windscada/config.py）。
      只查场站目录会把本体层判成"输入不在位"（2026-09-17 第一版就这么误报过 3 件）。
    """
    for base in (station, station.parent):
        d = base / sub
        if d.is_dir():
            return base
    return None


def _parquet_span(p: pathlib.Path, col: str | None):
    import pandas as pd
    try:
        d = pd.read_parquet(p)
    except Exception:
        return None, None, 0
    n = len(d)
    if col and col in d.columns:
        s = pd.to_datetime(d[col], errors='coerce')
        if s.notna().any():
            return str(s.min())[:10], str(s.max())[:10], n
    return None, None, n


def _npz_ok(p: pathlib.Path) -> bool:
    """振动谱件: npz 能读出来且非空（列数不是"行", 不能拿 parquet 的行数判它）。"""
    try:
        import numpy as np
        with np.load(p, allow_pickle=False) as z:
            return any(z[k].size > 0 for k in z.files)
    except Exception:
        return False


# 已知"含人工裁决/校准"的随包快照（按用户令 2026-09-17 #3 显式登记为人工件）。依据是文档里已经写明的性质:
# docs §7「handoff_vibration_v2.json / component_history.json 是随包快照, 该件含人工裁决/校准更新,
# 不是测量数据的函数」; baseline_38.json 的三层基线 self/absolute 两层由人工坐实; findings.json 是判级发现台账。
HUMAN_SNAPSHOTS = {
    'm5_cms_tcm/component_history.json',
    'm5_cms_tcm/baseline_38.json',
    'm5_cms_tcm/handoff_vibration_v1.json',
    'm5_cms_tcm/findings.json',
}

# ── 逆向工程可行性 (用户令 2026-09-17 #2) ────────────────────────────────────────────
# 问的是: "这批没有生成端的产物, 能不能从原始件/在包生成端**推**出来?"
# 判定不靠印象, 靠两条机器证据:
#   ① 生成端在不在包内（在 → 只是没接上, 跑一次就有）;
#   ② 产物正文里有没有**人工判断**的痕迹（"裁决/审核/校准/评审/经验"这类字段）——
#      含人工判断的件不是任何输入的纯函数, 逆向工程推不出来, 只能把那一步人工工作重做。
import re as _re                                                         # noqa: E402
JUDGE_PAT = _re.compile(r'人工|裁决|审核|复核|校准|评审|经验值|专家|judg|review|manual|calibrat|verdict|sign[_ ]?off',
                        _re.I)
# ★ 只认"**字段**叫这个名字"（`"裁决": …` / `裁决,`），不认正文里顺嘴提一句 ——
#   2026-09-17 第一版拿整篇子串匹配, 结果把**页面**里的展示标签(`index.html` 里的"评审/人工")当成
#   人工判断, 于是把"页面"整族误判成不可逆。这属于"判据太糙 → 结论假"。
JUDGE_KEY_PAT = _re.compile(r'["\']?[^"\',:]{0,24}(人工|裁决|审核|复核|校准|评审|经验值|专家|'
                            r'judg|review|manual|calibrat|verdict|sign[_ ]?off)[^"\',:]{0,24}["\']?\s*[:=]',
                            _re.I)


def judgement_evidence(farm: str, rels: list[str], cap: int = 8) -> tuple[int, list[str], int]:
    """→ (含人工判断字段的抽样比例%, 样例, 被检查件数)。

    只查**数据/文本件**(.json/.csv/.md/.txt); **页面(.html/.js)不算** —— 页面是产物的渲染,
    里面出现"评审/人工"是展示标签, 不代表这份件内含判断。
    """
    hit, samples, checked, pages = 0, [], 0, 0
    for rel in rels[:cap]:
        p = P.out_root(farm) / rel
        suf = p.suffix.lower()
        if suf in ('.html', '.js', '.css', '.htm'):
            pages += 1
            continue
        try:
            if suf == '.json':
                txt = p.read_text(encoding='utf-8', errors='replace')[:200000]
                pat = JUDGE_KEY_PAT
            elif suf in ('.csv', '.txt'):
                txt = p.read_text(encoding='utf-8', errors='replace')[:50000]
                pat = JUDGE_KEY_PAT
            elif suf == '.md':
                txt = p.read_text(encoding='utf-8', errors='replace')[:50000]
                pat = JUDGE_PAT
            else:
                continue
        except Exception:
            continue
        checked += 1
        if pat.search(txt):
            hit += 1
            samples.append(rel)
    return (100 * hit // checked if checked else 0), samples, pages


def feasibility(fam: dict, rels: list[str], gen_ok: bool, in_ok: bool,
                judge_pct: int, judge_samples: list[str], pages: int = 0) -> tuple[str, str]:
    """→ (可逆性判定, 依据)。四类: 可逆(直接) / 可逆(需反推口径) / 不可逆(人工判断) / 应移出产物仓。"""
    if fam['kind'] == 'not-product':
        return '应移出产物仓', '它本就不是产物(工具脚本/交证件/测试输出) ⇒ 该从产物仓移走, 而不是"补生成端"'
    if gen_ok and in_ok:
        return '可逆(直接)', f"生成端在包内({fam['gen']})且输入在位 ⇒ 接上/重跑即可"
    if pages and pages >= max(3, len(rels[:8]) // 2):
        return '可逆(页面可再生)', ('这一族主要是**页面**(.html): 页面是产物的渲染, 不是独立数据 ⇒ '
                                    '接上在包的页面生成端重出即可(未必逐字节同, 能力与数据一致)')
    m_n = sum(1 for r in rels if r.endswith(('.parquet', '.csv', '.npz')))
    t_n = len(rels) - m_n
    split = f'（{m_n} 件测量形态 / {t_n} 件文本·判断件）' if (m_n and t_n) else ''
    # 两类都占相当比重时, 结论必须是"混合"而不是挑一类代表全族 —— 否则一句判定就把另一半骗过去了
    if m_n and t_n and min(m_n, t_n) * 4 >= len(rels):
        return '混合(测量件可反推 / 文本件需人定)', split + \
               '测量形态的件可由 data/raw 反推口径 + 逐值对拍; 文本/判断件(评审、裁决、回复)推不出来'
    if judge_pct >= 50 and judge_samples:
        return '不可逆(人工判断)', (f"抽样 {len(judge_samples)} 件里都写着人工判断字段(如 {judge_samples[0]}) "
                                    f"⇒ 不是输入的纯函数, 逆向工程推不出来")
    if m_n and m_n * 2 >= len(rels):
        return '可逆(需反推口径)', split + '测量形态的件是测量数据的函数 ⇒ 可从 data/raw 反推口径 + 逐值对拍' \
                                        '(做法同 temp_monthly: 反推 → 逐值一致才敢用)'
    if gen_ok:
        return '半可逆(生成端在包, 输入不足)', '生成端在包内, 缺的是上游输入 ⇒ 上游补齐后即可重出'
    if judge_pct > 0:
        return '半可逆(需人工裁定)', f'抽样里 {judge_pct}% 的件含人工判断字段 ⇒ 机器部分可推、判断部分要人定'
    return '需人工裁定', '既无生成端、形态也不是测量函数 ⇒ 需研发给口径或领域正本'


def _turbines_of(p: pathlib.Path, col: str = 'turbine', cap: int = 200000):
    import pandas as pd
    try:
        d = pd.read_parquet(p, columns=[col]) if col else pd.read_parquet(p)
    except Exception:
        return None
    try:
        return {str(x).upper() for x in d[col].dropna().unique()} if col in d.columns else None
    except Exception:
        return None


def _input_turbines(station: pathlib.Path, sub: str):
    """输入侧机组集合: 逐台 CSV 文件名 / 目录名里的 WTG 号。"""
    d = station / sub
    if not d.is_dir():
        return None
    out = set()
    for p in d.rglob('*'):
        if p.is_file():
            for m in __import__('re').finditer(r'(WTG\s?\d{1,2})', p.name.upper()):
                out.add(m.group(1).replace(' ', ''))
    return out or None


def classify_human(farm: str, rels: list[str], cap: int = 600) -> set[str]:
    """逐件判"是不是人工件"（用户令 2026-09-17 #3：人工件不再按自动产物参与呼应校验）。

    判据：**非测量形态**的件（.md/.json/.csv/.txt）里出现人工判断字段（按字段名匹配）。
    测量形态件（.parquet/.csv 数值、.npz）一律不算人工件 —— 它们是数据的函数，属"可反推"那一路。
    """
    human: set[str] = set()
    for rel in rels[:cap]:
        if rel in HUMAN_SNAPSHOTS:            # ① 文档里已写明含人工裁决/校准的随包快照
            human.add(rel)
            continue
        suf = pathlib.Path(rel).suffix.lower()
        if suf == '.md':                      # ② 产物仓里的 .md 一律是报告/交证/评审记录(人写的)
            human.add(rel)
            continue
        if suf == '.txt':
            pat = JUDGE_PAT
        elif suf in ('.json', '.csv'):
            pat = JUDGE_KEY_PAT
        else:
            continue
        p = P.out_root(farm) / rel
        try:
            txt = p.read_text(encoding='utf-8', errors='replace')[:120000]
        except Exception:
            continue
        if pat.search(txt):                   # ③ 正文里有"人工判断字段"
            human.add(rel)
    return human


def write_human_manifest(farm: str | None = None) -> int:
    """把人工件逐件落成 `outputs/<场>/_human_artifacts.json`（机器台账，供自检与审计引用）。"""
    farm = farm or P.farm()
    global LEDGER
    LEDGER = load_ledger(farm)
    entries: dict[str, dict] = {}
    for fam in FAMILIES:
        if fam['kind'] not in ('shipped', 'human', 'not-product'):
            continue
        rels = [r for r in _files_of(fam['glob']) if r not in entries]
        if not rels:
            continue
        if fam['kind'] == 'human':
            for r in rels:
                entries[r] = dict(kind='human', family=fam['id'], evidence='族内显式登记为人工正本/交证件')
            continue
        for r in classify_human(farm, rels):
            if r not in entries:
                entries[r] = dict(kind='human', family=fam['id'],
                                  evidence='件内含人工判断字段（按字段名匹配），不是输入的纯函数')
    out = P.out_root(farm) / '_human_artifacts.json'
    out.write_text(json.dumps(dict(
        at=__import__('time').strftime('%Y-%m-%d %H:%M:%S'),
        note='人工件台账（用户令 2026-09-17 #3）：这些件由人写成/坐实，不是任何输入的纯函数 ⇒ '
             '不参与"输出↔输入呼应"校验；随包留在原位，重算不会也不该重写它们。'
             '本文件由 scripts/products_reverse_audit.py --write-human-manifest 生成。',
        count=len(entries), files=entries), ensure_ascii=False, indent=1), encoding='utf-8')
    by_fam: dict[str, int] = {}
    for v in entries.values():
        by_fam[v['family']] = by_fam.get(v['family'], 0) + 1
    print(f'人工件台账: {len(entries)} 件 → {P.rel(out)}')
    for k, v in sorted(by_fam.items(), key=lambda kv: -kv[1]):
        print(f'   {k:24s} {v:4d} 件')
    return 0


def audit(farm: str | None = None, verbose: bool = True):
    farm = farm or P.farm()
    global LEDGER
    LEDGER = load_ledger(farm)
    # ★ 场站原始件目录用唯一取用口 (场站名可与 farm 键不同: 键 = rudong, 目录 = data/raw/如东)
    from src.windscada.config import raw_station_dir
    station = pathlib.Path(raw_station_dir(farm))
    verdict: dict[str, dict] = {}
    fam_rows = []
    fails: list[str] = []

    for fam in FAMILIES:
        # ★ 族按**声明顺序**认领, 先声明的先拿 (否则 `m5_cms_tcm/*` 这种宽通配会把 windows/ 下
        #   1704 件也吞进来 —— fnmatch 的 `*` 是跨 `/` 的, 2026-09-17 第一版就这么误判过)
        rels = [r for r in _files_of(fam['glob']) if r not in verdict]
        rels = [r for r in rels if not r.lower().endswith(('.log', '.jsonl'))]
        if not rels:
            continue
        row = dict(id=fam['id'], n=len(rels), func=fam['func'], algo=fam['algo'], input=fam['input'],
                   pred='+'.join(fam['pred']) or '—', verdict='', note='')
        if fam['kind'] == 'not-product':
            # ★ 不是产物, 却躺在产物仓里: 反向审计对它的结论是"它压根不该按产物管" ——
            #   既不需要生成端, 也谈不上与输入呼应（这一类最该被人看见, 故单独一类, 不当失败计）。
            jp, js, pg = judgement_evidence(farm, rels)
            fv, fw = feasibility(fam, rels, False, False, jp, js, pg)
            row['verdict'] = '✗ 非产物'
            row['note'] = fam.get('why', '')
            row['rev'], row['rev_why'] = fv, fw
            for r in rels:
                verdict[r] = dict(fam=fam['id'], verdict='✗',
                                  why='非产物（过程留痕/工具脚本躺在产物仓里）: ' + fam.get('why', ''))
        elif fam['kind'] == 'human':
            # ★ 人工正本/交证件 (用户令 2026-09-17 #3): 由人写、被运行期当证据引用,
            #   **不参与呼应校验**——它本来就没有、也不该有生成端; 单独计数, 不混进"无生成端"的失败堆里。
            row['verdict'] = '◆ 人工件'
            row['note'] = fam.get('why', '')
            row['rev'], row['rev_why'] = '人工件(正本/交证)', '人写的, 不参与呼应校验; 随包留在原位'
            for r in rels:
                verdict[r] = dict(fam=fam['id'], verdict='◆',
                                  why='人工件（人工正本/交证件）: ' + fam.get('why', ''))
        elif fam['kind'] == 'shipped':
            jp, js, pg = judgement_evidence(farm, rels)
            gen_file = ROOT / fam['gen'] if fam.get('gen') else None
            gen_ok0 = bool(gen_file and gen_file.is_file())
            # ★ 逐件分人工件 (用户令 2026-09-17 #3): 含人工判断字段的件**不按自动产物参与呼应校验**,
            #   单独计 ◆; 余下才是真正的"无生成端"。
            human = classify_human(farm, rels)
            rest = [r for r in rels if r not in human]
            fv, fw = feasibility(fam, rest or rels, gen_ok0, False, jp, js, pg)
            row['verdict'] = '✗ 无生成端' if not human else f'✗ 无生成端 + ◆ 人工件 {len(human)}'
            row['n'] = len(rest)
            row['note'] = fam.get('why', '')
            row['rev'], row['rev_why'] = fv, fw
            if jp:
                row['rev_why'] += f'；人工判断字段抽样命中 {jp}%' + (f'（如 {js[0]}）' if js else '')
            for r in rest:
                verdict[r] = dict(fam=fam['id'], verdict='✗', why='无生成端（全库 0 处写入方）: ' + fam['why'])
            for r in human:
                verdict[r] = dict(fam=fam['id'], verdict='◆',
                                  why='人工件（含人工判断字段）: 不参与呼应校验, 随包留在原位')
        else:
            # ① 生成端在位
            gen = ROOT / fam['gen'] if fam.get('gen') else None
            gen_ok = bool(gen and gen.is_file())
            # ② 输入在位（场站目录优先, 机理层资料在 raw 根下 —— 见 _input_home）
            spec = INV.INPUTS.get(fam['input'] or '')
            i_s = i_e = None
            i_n = 0
            i_gran = '年'
            home = _input_home(station, fam['input']) if fam['input'] else None
            if home and spec:
                i_s, i_e, i_n, _note, i_gran = INV.input_span(home, fam['input'], spec)
            elif home:
                # 机理层资料这类"没有跨度判据"的输入: 按件数清点即算在位（INV.INPUTS 里没有它们的
                # 跨度口径 —— 技术资料是文档而不是时序数据, 拿跨度比毫无意义）
                i_n = sum(1 for p in (home / fam['input']).rglob('*') if p.is_file())
                i_gran = '年'
            elif fam['kind'] == 'source-derived':
                # ★ 输入是**包内源码**而不是 data/raw（如发布清单快照: 从 src/windscada/ui 重建页再摘要）。
                #   生成端在位 = 输入在位, 这类产物的"呼应"是对源码的, 不是对原始件的。
                i_n = 1
                i_s = i_e = '包内源码'
            in_ok = i_n > 0
            passed, notes = [], []
            if fam['pred'] and in_ok:
                if 'span' in fam['pred']:
                    p_s, p_e, n = _parquet_span(P.out_root(farm) / rels[0], fam.get('time_col'))
                    if p_s and i_s and (p_s[:7] < i_s[:7]):
                        # ★ 年粒度输入（按文件名年份推的）不判失败: 与正向检查同一条教训 ——
                        #   台账 xls 里可能含比文件名年份更早的历史记录, 拿年粒度当"输入起点"会造假缺口
                        #   (2026-09-17 实测: 工单产物起点 2020-01 而输入文件名最早 2021 ⇒ 不是数据串了)。
                        msg = f'产物起点 {p_s} 早于输入起点 {i_s}'
                        if i_gran in ('日', '月'):
                            fails.append(f"{fam['id']}: {msg}（跨窗混入 / 键错）")
                            notes.append('⚠ ' + msg)
                        else:
                            notes.append(f'{msg}（输入为年粒度, 只作参考: 台账可能含更早的历史行）')
                    passed.append(f'跨度 {p_s}~{p_e} ⊆ 输入 {i_s}~{i_e}')
                if 'turbines' in fam['pred']:
                    pt = _turbines_of(P.out_root(farm) / rels[0])
                    it = _input_turbines(home, fam['input']) if home else None
                    if pt and it:
                        extra = pt - it
                        if extra:
                            fails.append(f"{fam['id']}: 产物含输入里没有的机组 {sorted(extra)[:5]}")
                        passed.append(f'机组 {len(pt)} 台 ⊆ 输入 {len(it)} 台')
                if 'rows' in fam['pred']:
                    tot = 0
                    for r in rels[:400]:
                        _s, _e, n = _parquet_span(P.out_root(farm) / r, None) if r.endswith('.parquet') \
                            else (None, None, 1)
                        tot += n
                    if tot <= 0:
                        fails.append(f"{fam['id']}: 产物行数合计为 0")
                    passed.append(f'件内数据行合计 {tot:,}' + ('（抽样 400 件）' if len(rels) > 400 else ''))
                if 'npz' in fam['pred']:
                    sample = rels[:5]
                    bad = [r for r in sample if not _npz_ok(P.out_root(farm) / r)]
                    if bad:
                        fails.append(f"{fam['id']}: 抽样 {len(bad)}/{len(sample)} 件 npz 读不出或为空")
                    passed.append(f'{len(rels)} 件谱（抽样 {len(sample)} 件 npz 均可读非空）· 输入源 {i_n} 件')
                if 'manifest' in fam['pred']:
                    try:
                        mf = json.loads((P.out_root(farm) / rels[0]).read_text(encoding='utf-8'))
                        passed.append('清单自记: ' + ', '.join(f'{k}={v}' for k, v in list(mf.items())[:3]))
                    except Exception as e:
                        fails.append(f"{fam['id']}: 清单读不出来 ({type(e).__name__})")
            if not gen_ok:
                row['verdict'] = '✗ 生成端不在位'
                row['note'] = f"缺 {fam['gen']}"
                fails.append(f"{fam['id']}: 生成端不在位 {fam['gen']}")
            elif not in_ok:
                row['verdict'] = '~ 输入不在位'
                row['note'] = (f"缺 {(home / fam['input']) if home else station / str(fam['input'])}"
                               ' ⇒ 现在无法验证（放数据后复跑本器）')
            else:
                row['verdict'] = '✓ 呼应成立'
                row['note'] = '；'.join(notes + passed)
                row['rev'], row['rev_why'] = '已在重算链上', '生成端在位、输入在位、判据通过 —— 无需逆向工程'
            for r in rels:
                verdict[r] = dict(fam=fam['id'], verdict=row['verdict'][0],
                                  why=row['note'] if row['verdict'][0] != '✓' else row['note'][:120])
        fam_rows.append(row)

    # 未归类件 (台账里有, 但没有任何族接住)
    unclassified = []
    for rel, v in LEDGER.items():
        if rel.lower().endswith(('.log', '.jsonl')):
            continue
        if rel in verdict:
            continue
        # 产物走通配没接住的, 也按台账来源判定
        unclassified.append(rel)
        verdict[rel] = dict(fam='(未归类)', verdict='?', why='既不在族表、也无法从台账判定来路')

    counts = {}
    for v in verdict.values():
        counts[v['verdict']] = counts.get(v['verdict'], 0) + 1
    if verbose:
        print(f'== 反向呼应审计 · 场站 {farm} · 台账 {len(LEDGER)} 件 · 判定 {len(verdict)} 件 ==')
        print(f'   输入根: {P.rel(station)}')
        print()
        print(f'{"族":26s} {"件数":>6s} {"判定":14s} {"输入类":12s} {"功能 / 算法 / 依据"}')
        for r in fam_rows:
            print(f'  {r["id"]:24s} {r["n"]:6d} {r["verdict"]:14s} {str(r["input"] or "—"):12s} '
                  f'{r["func"][:34]}')
            if r['note']:
                print(f'      └─ {r["note"][:150]}')
        if unclassified:
            print(f'\n  【未归类 {len(unclassified)} 件】')
            for r in unclassified[:20]:
                print(f'      ? {r}')
        print('\n  判定汇总: ' + ' · '.join(f'{k}={v}' for k, v in sorted(counts.items())))
        print(f'  判据失败 {len(fails)} 条' + ('：' + '；'.join(fails[:3]) if fails else ''))
        ok = counts.get('✓', 0)
        print(f'  结论: 输出↔输入呼应**成立** {ok} 件 / **不成立(无生成端)** {counts.get("✗", 0)} 件 / '
              f'**人工件** {counts.get("◆", 0)} 件 / 无法验证 {counts.get("~", 0)} 件 / 未归类 {counts.get("?", 0)} 件')
        # ── 逆向工程可行性 (用户令 #2): 没有呼应关系的那些, 能不能推出来 ──
        import collections as _c
        by_rev = _c.Counter()
        for r in fam_rows:
            by_rev[r.get('rev', '?')] += r['n']
        print()
        print('== 反向「可逆性」矩阵（用户令 2026-09-17 #2: 无生成端的件能否从原件推出来）==')
        print(f'{"族":26s} {"件数":>6s} {"可逆性":22s} 依据')
        for r in fam_rows:
            if r['verdict'][0] == '✓':
                continue
            print(f'  {r["id"]:24s} {r["n"]:6d} {r.get("rev", "?"):22s} {r.get("rev_why", "")[:96]}')
        print('  按件数汇总: ' + ' · '.join(f'{k}={v}' for k, v in by_rev.most_common()))
    rc = 5 if (fails or unclassified) else 0
    return fam_rows, verdict, counts, fails, unclassified, rc


def doc_block(farm: str) -> str:
    fam_rows, verdict, counts, fails, unclassified, _rc = audit(farm, verbose=False)
    L = [DOC_BEGIN, '### 13.6 反向呼应审计：自输出 → 功能与算法 → 输入（自动生成，勿手改）', '',
         f'台账 {len(verdict)} 件产物逐件回溯：**呼应成立 {counts.get("✓", 0)} 件** · '
         f'**不成立（无生成端）{counts.get("✗", 0)} 件** · 无法验证 {counts.get("~", 0)} 件 · '
         f'未归类 {counts.get("?", 0)} 件。'
         '判据：① 生成端在位 ② 输入在位 ③ 跨度 ⊆ 输入 / 机组集 ⊆ 输入 / 行数 > 0 / 计数与清单一致。', '',
         '| 输出族（产物 glob） | 件数 | 反向判定 | 功能 | 算法 / 生成端 | 输入（data/raw/<场>/） | 判据 | 说明 |',
         '|---|---:|---|---|---|---|---|---|']
    for r in fam_rows:
        fam = next(f for f in FAMILIES if f['id'] == r['id'])
        L.append(f'| `{fam["glob"]}` | {r["n"]} | {r["verdict"]} | {r["func"]} | {r["algo"]} | '
                 f'{r["input"] or "—（无生成端）"} | {r["pred"]} | {r["note"][:160]} |')
    if unclassified:
        L += ['', f'**未归类 {len(unclassified)} 件**（需人工认领）：' + '、'.join(f'`{x}`' for x in unclassified[:12])]
    L += ['', '#### 反向「可逆性」：没有呼应关系的那批，能否从原始件推出来（用户令 2026-09-17 #2）', '',
          '判据两条机器证据：① 生成端在不在包内 ② 产物正文里有没有**人工判断**字段'
          '（`人工/裁决/审核/校准/评审/经验/judge/review/calibrat/verdict`，抽样读件统计命中率）。'
          '含人工判断的件不是任何输入的纯函数 ⇒ 逆向工程推不出来，只能把那一步人工工作重做。', '',
          '| 输出族 | 件数 | 可逆性 | 依据 |', '|---|---:|---|---|']
    for r in fam_rows:
        if r['verdict'][0] == '✓':
            continue
        L.append(f'| `{r["id"]}` | {r["n"]} | {r.get("rev", "?")} | {r.get("rev_why", "")} |')
    import collections as _c
    by_rev = _c.Counter()
    for r in fam_rows:
        by_rev[r.get('rev', '?')] += r['n']
    L += ['', '**按件数汇总**：' + ' · '.join(f'{k} = {v} 件' for k, v in by_rev.most_common()), '',
          '**人工件**：含人工判断字段的件已逐件登记在 outputs/&lt;场&gt;/_human_artifacts.json（用户令 2026-09-17 #3），'
          '它们**不参与**呼应校验（人写成/坐实的东西不是任何输入的纯函数）；生成方式：'
          'python scripts/products_reverse_audit.py --write-human-manifest。', '',
          '> 结论口径：`可逆(直接)` = 包内已有生成端，接上即可；`可逆(需反推口径)` = 内容是测量数据的函数，'
          '照 `temp_monthly` 的办法反推 + 逐值对拍；`不可逆(人工判断)` = 逆向工程推不出来，'
          '要么由研发补生成端、要么承认它是人工件（不该按产物管）。', DOC_END]
    return '\n'.join(L)


def write_doc(farm: str) -> int:
    doc = ROOT / 'docs' / '系统设计说明.md'
    txt = doc.read_text(encoding='utf-8')
    blk = doc_block(farm)
    i, j = txt.find(DOC_BEGIN), txt.find(DOC_END)
    if i >= 0 and j > i:
        txt = txt[:i] + blk + txt[j + len(DOC_END):]
    else:                                             # 首次: 追加到文末
        txt = txt.rstrip('\n') + '\n\n' + blk + '\n'
    doc.write_text(txt, encoding='utf-8', newline='\n')
    print(f'已写入 {P.rel(doc)} (§13.6 反向呼应审计块)')
    return 0


def main() -> int:
    ap = argparse.ArgumentParser(description='反向呼应审计: 输出 → 功能与算法 → 输入')
    ap.add_argument('--farm', default=None)
    ap.add_argument('--check', action='store_true', help='只出结论 (有未归类/判据失败 → rc=5)')
    ap.add_argument('--write-doc', action='store_true', help='把族表写进 docs/系统设计说明.md §13.6')
    ap.add_argument('--write-human-manifest', action='store_true',
                    help='生成人工件台账 outputs/<场>/_human_artifacts.json (用户令 2026-09-17 #3)')
    a = ap.parse_args()
    if a.write_human_manifest:
        return write_human_manifest(a.farm)
    if a.write_doc:
        return write_doc(a.farm or P.farm())
    _rows, _v, counts, fails, uncl, rc = audit(a.farm, verbose=True)
    if a.check:
        print(f'[{"OK" if rc == 0 else "X"}] 反向呼应审计 rc={rc}'
              + (f'（未归类 {len(uncl)} 件 / 判据失败 {len(fails)} 条）' if rc else '（未归类 0 件、判据无失败）'))
    return rc


if __name__ == '__main__':
    for _s in (sys.stdout, sys.stderr):
        try:
            _s.reconfigure(errors='replace')
        except Exception:
            pass
    sys.exit(main())