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 处写入方 ⇒ 反向呼应**在原理上不成立**（如实记账, 不假装成立）────
    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 等），'
                                            '现场正本才可重出；现只能从交付包补齐'),
    dict(id='vib_figs', glob='m5_cms_tcm/figs/*', kind='shipped',
         func='振动线出图（谱图/趋势图）', algo='（无生成端）', gen=None, input=None, pred=(),
         why='随包快照（振动线出件）'),
    dict(id='windscada_pages', glob='windscada/{index.html,turbines/*,review/*}', kind='shipped',
         func='总览「全场状态」静态页 + 逐台页 + 评审记录', algo='（无生成端）',
         gen=None, input=None, pred=(), why='页面件由振动线/历史分析出件，本包无写入方'),
    dict(id='windcms_pages', glob='windcms/**', kind='shipped',
         func='CMS 振动评估报告页/逐台页/缓存', algo='scripts/windcms.py report（本包缺 model_run/fusion '
              '产物, 重生成会掉内容 −97%, 故**按设计跳过**）', gen=None, input=None, pred=(),
         why='生成端在包内但输入不足 ⇒ 现状按随包件冻结'),
    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='实验底稿, 事实契约的输入'),
    dict(id='pitch_shipped', glob='pitch/**', kind='shipped',
         func='变桨侧派生件（零位/日粒度）', algo='（无生成端；rebuild_from_raw --scada 只覆盖其中一部分）',
         gen=None, input=None, pred=(), why='随包快照'),
    dict(id='ontology_releases', glob='ontology/release_*/*', kind='shipped',
         func='本体发布层 r1/r2（只读暴露给网关 /release/）', algo='（无生成端）', gen=None, input=None,
         pred=(), why='发布层清单, 由研发出件'),
    dict(id='guanlan_contract', glob='guanlan/**', kind='shipped',
         func='事实契约与对外派生（门户结论段/取数）', algo='scripts/guanlan_facts_contract.py（读者在包内, '
              '输入是 ontology+m5+sop 三处**产物**）', gen=None, input=None, pred=(),
         why='派生链的中间层：它的"输入"本身是产物而非 data/raw ⇒ 与 raw 之间隔着不止一跳'),
    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 #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]:
        suf = pathlib.Path(rel).suffix.lower()
        if suf in ('.md', '.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())