guanlan-src/scripts/set_baseline.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
r"""重建/查看"台账等价验收"的**基线**（2026-09-18 现场问题驱动）。

## 为什么需要它

`rebuild_all.py` 最后一步跑 `rebuild_from_raw.py --verify`：拿当前产物与**基线**逐值比对，
证明"重算能复现同一批数字"。基线是 `outputs/<场>/windscada/_pre_rebuild_20260911/` 下的 4 件
（`alarms` / `workorders` / `oil_samples_index` / `temp_monthly`）。

**2026-09-17 用户令「清除产物」把它一并真删了**（该命令不留备份）⇒ 之后每次重算最后一步都报
`rc=5 找不到随包基线 ⇒ 无法做等价验收（"没验收"不是"通过"）`，门户上就显示成"重算失败"。

本器把"重建基线"变成一条命令：在当前产物**已核实无误**的前提下快照这 4 件，并写清来历。

## 用法

    python scripts/set_baseline.py --check      # 看基线在不在（不看盘面以外的推断）
    python scripts/set_baseline.py              # 用**当前在位产物**快照成基线（同路径, 覆盖）
退出码: 0 成功/在位 · 2 缺源件
"""
from __future__ import annotations

import argparse
import datetime as dt
import json
import pathlib
import shutil
import sys

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

FILES = ('alarms.parquet', 'workorders.parquet', 'oil_samples_index.parquet', 'temp_monthly.parquet')
BASE_NAME = '_pre_rebuild_20260911'


def base_dir(farm: str | None = None) -> pathlib.Path:
    return P.store(farm) / BASE_NAME


def main() -> int:
    ap = argparse.ArgumentParser(description='重建/查看 台账等价验收基线')
    ap.add_argument('--check', action='store_true', help='只看基线在不在')
    ap.add_argument('--farm', default=None)
    a = ap.parse_args()
    farm = a.farm or P.farm()
    store = P.store(farm)
    b = base_dir(farm)
    have = {n: (b / n) for n in FILES}
    print(f'== 等价验收基线 · {P.rel(b)} ==')
    for n, p in have.items():
        print(f'   [{"OK " if p.is_file() else "缺 "}] {n:28s}'
              + (f' {p.stat().st_size:>12,} B' if p.is_file() else ''))
    if a.check:
        ok = all(p.is_file() for p in have.values())
        print(f'   结论: {"基线在位, 重算最后一步可做等价验收" if ok else "基线缺失 ⇒ 重算最后一步会 rc=5（没验收）; 跑本命令(不带 --check)重建"}')
        return 0 if ok else 2
    miss = [n for n, p in have.items() if not (store / n).is_file()]
    if miss:
        print(f'   [X] 当前产物里缺 {miss} —— 先重算产出它们, 再建基线')
        return 2
    b.mkdir(parents=True, exist_ok=True)
    for n in FILES:
        shutil.copy2(store / n, b / n)
    (b / '_来源说明.json').write_text(json.dumps(dict(
        why='rebuild_all 最后一步「台账等价验收」要一份标准答案做逐值比对; 基线目录被「清除产物」真删后, '
            '该步 rc=5「没验收」⇒ 门户上显示重算失败。',
        what='本目录 4 件 = 快照时**在位且已核实**的产物（alarms/workorders/oil_samples_index/temp_monthly）。',
        nature='★ 这是**回归基线**（证明将来的重算能复现同一批数字），不是历史交付基线。',
        built_at=dt.datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
        rebuilt_by='scripts/set_baseline.py',
        verify_by='python scripts/rebuild_from_raw.py --verify（全部通过即基线有效）',
    ), ensure_ascii=False, indent=1) + '\n', encoding='utf-8')
    print(f'   已重建基线: {P.rel(b)}（4 件 + _来源说明.json）')
    print('   复验: python scripts/rebuild_from_raw.py --verify')
    return 0


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