guanlan-src/scripts/chain_gap_check.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
r"""六层链"缺料"检查器（2026-09-17 用户令 #2 的收口件）。

作用只有一个：**把"还缺什么"变成一条命令**，而不是靠人记着哪四个脚本没随包。

    python scripts/chain_gap_check.py            # 人看
    python scripts/chain_gap_check.py --check    # 只出结论（缺料 → 退出码 5）

判据（全部是"文件在不在"这种可核事实，不含猜测）：

  A. 四步脚本是否在位 —— 壳 `src/windcms/pipeline.py::analyze` 会调它们；
  B. 两件"标准答案"是否在位 —— `model_run_l6.parquet` / `fusion_38.csv`（逐值对拍的前提）；
  C. 台账里这一族还剩多少"✗ 无生成端"（接料后应显著下降）。

取料单：`docs/向振动线取料单_v0.1.md`；接口规格：`docs/振动六层链_接口规格与缺口_v0.1.md`。
"""
from __future__ import annotations

import argparse
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

# 壳里写死的四步脚本（改壳才会变；这里只是把"缺什么"列清楚）
STEPS = [
    ('oem_scan', 'scripts/rudong_tcm_oem_scan.py', '特征频率扫描（轴承/齿轮/保持架/叶片 1P）'),
    ('energy_share', 'scripts/rudong_line_energy_share.py', '线能量占比'),
    ('model_run', 'scripts/rudong_model_run.py', '六层模型出件（model_run_l6.parquet + 摘要）'),
    ('fusion', 'scripts/rudong_fusion_run.py', '融合面出件（fusion_38.csv 等）'),
]
# 逐值对拍用的"标准答案"（随包件里有，但两件关键件从未随过包）
SAMPLES = [
    ('model_run_l6.parquet', 'model_run 步的标准答案'),
    ('fusion_38.csv', 'fusion 步的标准答案'),
]


def main() -> int:
    ap = argparse.ArgumentParser(description='六层链缺料检查')
    ap.add_argument('--check', action='store_true', help='只出结论（缺料 → rc=5）')
    ap.add_argument('--farm', default=None)
    a = ap.parse_args()
    farm = a.farm or P.farm()
    m5 = P.m5(farm)

    missing_scripts = [s for s in STEPS if not (ROOT / s[1]).is_file()]
    partial = []          # 脚本在位、但自报 complete=false ⇒ 只是"部分实现"(分母就位/分子待口径)
    have_fusion_impl = (ROOT / 'scripts/rudong_fusion_run.py').is_file()
    missing_samples = [s for s in SAMPLES if not (m5 / s[0]).is_file()]

    print(f'== 六层链缺料检查 · 场站 {farm} ==')
    print(f'   壳: src/windcms/pipeline.py::analyze（它按固定路径调这四步）')
    print(f'   规格: docs/振动六层链_接口规格与缺口_v0.1.md ｜ 取料单: docs/向振动线取料单_v0.1.md')
    print()
    print('A. 四步脚本:')
    for name, rel, what in STEPS:
        p = ROOT / rel
        ok = p.is_file()
        tag, extra = ('OK ', '') if ok else ('缺 ', '')
        if ok:
            # ★ 脚本在位 ≠ 这一步能出产物: 逆向实现可能是"分母就位、分子待口径"(自报 --status)。
            #   这里调一次 --status 把真实状态问出来, 免得自查器给出"看起来齐了"的假结论。
            import json as _json
            import subprocess as _sp
            try:
                r = _sp.run([sys.executable, str(p), '--status'], capture_output=True, text=True,
                            errors='replace', timeout=120)
                st = _json.loads((r.stdout or '').strip().splitlines()[-1]) if (r.stdout or '').strip() else {}
                if st and st.get('complete') is False:
                    tag = '部分'
                    extra = '（' + st.get('denominator', '') + '；缺: ' + '；'.join(st.get('missing', []) or []) + '）'
            except Exception:
                pass
        if ok and tag == '部分':
            partial.append(name)
        if not ok and name == 'fusion' and have_fusion_impl:
            extra = '（本包已逆向实现其中一件产物: fleet_scalar_z；源码仍缺）'
        print(f'   [{tag}] {name:12s} {rel:44s} {what}{extra}')
    print()
    print('B. 逐值对拍的标准答案:')
    for nm, why in SAMPLES:
        ok = (m5 / nm).is_file()
        print(f'   [{"OK " if ok else "缺 "}] {nm:28s} {why}  ({P.rel(m5 / nm)})')
    # 已完成的那件: 用本器自带 verify 报一句现状（不写盘）
    fr = ROOT / 'scripts/rudong_fusion_run.py'
    if fr.is_file():
        print()
        print('C. 已自研部分（fleet_scalar_z）: 跑 python scripts/rudong_fusion_run.py --verify 看逐值对拍')
    print()
    rc = 5 if (missing_scripts or partial) else 0
    tail = ('四步脚本齐全且都已落地 ⇒ 六层链可跑' if not (missing_scripts or partial)
            else (f'{len(missing_scripts)} 个步骤脚本未随包' if missing_scripts else '')
            + ('；' if (missing_scripts and partial) else '')
            + (f'{len(partial)} 个为"部分实现"（分母就位 / 分子待口径）' if partial else '')
            + ' ⇒ 六层链跑不动（取料单见上）')
    print(f'   结论: {tail}')
    if a.check:
        print(f'[{"OK" if rc == 0 else "X"}] chain_gap_check rc={rc}')
    return rc


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