guanlan-src/app_ETL/app_ETL_guanlan/builders/rebuild_from_raw.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
r"""从 data/raw 重算/更新产物 (2026-09-11) —— 维护页那四条「摄入命令」的总入口。

## 背景

`docs/说明书…v0.2.md` §11 写着「**不含**从原始数据重生成产物的链 (P0)」, 于是随包产物只能吃预生成的那份
(台账止 2024-11-21、报警止 2026-07-15), 现场把新表放进 `data/raw/<场站>/…` 也不会更新。
本脚本把这条链补上, 顺序与依赖关系如下 (都按场配置 `src.windscada.config`):

    ① scripts/windscada_alarms_ingest.py        故障报警/*.xls      → alarms.parquet
    ② scripts/windscada_workorder_ingest.py     风机故障记录/**      → workorders.parquet
    ③ scripts/windscada_watch_channels_build.py 油样报告/**         → oil_samples_index.parquet
    ── 以上三步只吃现场台账 (xls/xlsx/pdf), 快, 秒级到分钟级 ──
    ④ (--scada) 包内已有的 SCADA 侧构建器: 从 scada_10min/*.csv 重算
       powercurve → loss_monthly(availability) → curves / control / stop_events(faults)
       / temp_bins / yaw_daily / hydraulic_accum / thermal_chain / system_aux
       (这一步要逐台读 38 个 ~380MB 的 CSV, 慢; 默认不跑, 加 --scada 才跑)

## --verify: 等价验收 (本链的验收标准)

重算件必须能**复现随包件**: 逐键逐值比对 `outputs/rudong/windscada/_pre_rebuild_20260911/` 里那份
随包基线 (2026-09-11 现场修复时留的备份)。允许的差异只有两类, 都必须被点名:
  · 旧链的**已知缺陷**: 报警/工单里同一副本被收两遍、日期解析不了就整行丢弃、Excel 序列号当纳秒解析;
  · 本链的**新增覆盖**: 随包时没接的源件 (74 张台账表 vs 随包只用了 6 张)。
无法归类的差异 = 验收不通过 (退出码 4)。

用法:
    python scripts/rebuild_from_raw.py                 # 跑 ①②③
    python scripts/rebuild_from_raw.py --verify        # 只验收, 不重算
    python scripts/rebuild_from_raw.py --scada         # 连 ④ 一起跑 (慢)
    python scripts/rebuild_from_raw.py --dry-run       # 只打印各步会做什么
"""
from __future__ import annotations

try:
    from app_common.app_common_guanlan.api import install_root as _install_root
except ImportError:                     # 理论不可达；包结构异常时回退到按位置上跳
    from pathlib import Path as _P
    def _install_root(_f): return _P(_f).resolve().parents[3]
import sys as _sys, pathlib as _plb
_sys.path.insert(0, str(_plb.Path(__file__).resolve().parents[1]))
from src import paths as P

import argparse
import pathlib
import subprocess
import sys
from collections import Counter

ROOT = _install_root(__file__)      # 模块化后按标记找安装根（原 parents[1] 已不成立）
sys.path.insert(0, str(ROOT))

KEEP_BASE = '_pre_rebuild_20260911'                      # 随包台账基线的目录名 (products_state.py 也认这个名字)
BASE = P.rel(P.store()) + '/' + KEEP_BASE                # 常规位置 (相对安装根)
STEPS = [
    ('报警事件', 'scripts/windscada_alarms_ingest.py', 'alarms.parquet'),
    ('检修工单台账', 'scripts/windscada_workorder_ingest.py', 'workorders.parquet'),
    ('油液化验', 'scripts/windscada_watch_channels_build.py', 'oil_samples_index.parquet'),
]

# SCADA 侧 (--scada): 全部是包内**已有**的构建器, 只是过去没有入口把它们串起来。
# 顺序 = 依赖顺序 (powercurve 先出 bins, availability 才吃得到; stop_events 要吃 alarms)。
SCADA_STEPS = [
    ('功率曲线', 'src.windscada.perf.powercurve', 'store'),
    ('可用率与损失 (loss_monthly)', 'src.windscada.perf.availability', 'build'),
    ('七镜头曲线', 'src.windscada.perf.curves', 'build_store'),
    ('控制策略件', 'src.windscada.perf.control', 'build_store'),
    ('停机事件 (stop_events)', 'src.windscada.perf.faults', 'build_stop_events'),
    ('温度 NBM (temp_bins)', 'src.windscada.subsys.temp_nbm', 'build_store'),
    ('偏航 (yaw_daily)', 'src.windscada.subsys.yaw', 'build_store'),
    ('液压蓄能 (hydraulic_accum)', 'src.windscada.subsys.hydraulic', 'build_store'),
    ('热链 (thermal_chain)', 'src.windscada.subsys.thermal_chain', 'build_store'),
    ('系统辅助 (system_aux)', 'src.windscada.taxonomy', 'build_aux_store'),
]


def run_scada(farm_name=None) -> int:
    """逐台读 scada_10min/*.csv 重算 SCADA 侧产物。慢 (38 台 × 各构建器), 失败不中断, 最后汇总。"""
    import importlib
    import time
    from src.windscada.config import farm
    cfg = farm(farm_name)
    print(f'场: {cfg["name"]}  源: {cfg["src_10min"]}  仓: {cfg["store"]}')
    bad = []
    for label, mod, fn in SCADA_STEPS:
        t0 = time.time()
        try:
            f = getattr(importlib.import_module(mod), fn)
            r = f(cfg)
            print(f'  ✔ {label:26s} {time.time()-t0:6.1f}s  {str(r)[:110]}', flush=True)
        except Exception as e:
            bad.append((label, f'{type(e).__name__}: {e}'))
            print(f'  ✘ {label:26s} {time.time()-t0:6.1f}s  {type(e).__name__}: {str(e)[:150]}', flush=True)
    if bad:
        print(f'\n{len(bad)} 个构建器失败:')
        for label, err in bad:
            print(f'  {label}: {err}')
    return 0 if not bad else 5


def _run(script: str, extra=()) -> int:
    cmd = [sys.executable, str(ROOT / script), *extra]
    print(f'\n$ {" ".join(cmd[1:])}', flush=True)
    return subprocess.call(cmd, cwd=str(ROOT))


def _canon(df, cols=None):
    import pandas as pd
    d = df[cols] if cols else df
    out = []
    for r in d.itertuples(index=False):
        row = []
        for c, v in zip(d.columns, r):
            try:
                if v is None or (not isinstance(v, str) and pd.isna(v)):
                    row.append('<NA>')
                    continue
            except (TypeError, ValueError):
                pass
            row.append(v.isoformat() if isinstance(v, pd.Timestamp) else str(v))
        out.append(tuple(row))
    return Counter(out)


def baseline_dir():
    """随包台账基线在哪 —— 两处都找, 找不到才报错。

    ① `outputs/<场>/windscada/_pre_rebuild_20260911/` (常规位置);
    ② `_products_off*/_baseline_kept/_pre_rebuild_20260911/` —— `products_state.py --off` 会把它挪到这里
       (它刻意**不**算产物, 就是为了随时能还原做验收)。
    2026-09-12 实逮: 从运维控制台点过"清除产物"之后跑 `rebuild_all.py --with-verify`, 验收步骤报
    "[X] 没有随包基线备份" **却被编排当成"退出码 4 属预期"放过** —— 等于"验收没做却报通过"。
    修法: 本函数自动找第二处; 真找不到才返回 5(硬失败, 编排不容忍)。
    """
    cands = [P.store() / KEEP_BASE]
    for p in sorted(ROOT.glob(f'_products_off*/_baseline_kept/{KEEP_BASE}')):
        cands.append(p)
    for c in cands:
        if c.is_dir():
            return c
    return None


def verify() -> int:
    import pandas as pd
    store = P.store()
    base = baseline_dir()
    if base is None:
        print(f'[X] 找不到随包基线备份 (找过 {P.rel(P.store())}/{KEEP_BASE} 与 _products_off*/_baseline_kept/) '
              f'—— 无法做等价验收 (退出码 5 = 没验收, 不等于通过)')
        return 5
    print(f'(基线: {P.rel(base)})')
    bad = 0
    print('== 等价验收: 重算件 vs 随包基线 ==')

    for label, name, exact in (('报警事件', 'alarms.parquet', True),
                               ('油液化验', 'oil_samples_index.parquet', True)):
        a = _canon(pd.read_parquet(base / name))
        b = _canon(pd.read_parquet(store / name))
        absent = set(a) - set(b)                     # 随包内容在重算里**找不到** (真缺才算不通过)
        gained = set(b) - set(a)
        collapsed = sum(a.values()) - sum(min(c, b.get(k, 0)) for k, c in a.items())
        ok = exact and not absent and not gained
        print(f'\n  {label} ({name})')
        print(f'    随包 {sum(a.values())} 行 ({len(a)} 种内容) / 重算 {sum(b.values())} 行 ({len(b)} 种内容)')
        print(f'    随包内容未复现 {len(absent)} 种; 重算新增 {len(gained)} 种; 副本归并 {collapsed} 行')
        print(f'    {"✔ 完全一致 (行数、内容、副本数全同)" if ok else "✘ 需人工看"}')
        if not ok:
            bad += 1
            for k in list(absent)[:2]:
                print('      随包独有: ' + ' | '.join(x for x in k if x != '<NA>')[:180])
            for k in list(gained)[:2]:
                print('      重算独有: ' + ' | '.join(x for x in k if x != '<NA>')[:180])

    # 工单: 逐项列账, 不用一个数字糊过去
    from scripts.windscada_workorder_ingest import OUT_COLS
    a_raw = pd.read_parquet(base / 'workorders.parquet')
    b_raw = pd.read_parquet(store / 'workorders.parquet')
    six = set(a_raw.src_file.unique())
    kc = [c for c in OUT_COLS if c != 'src_file']
    a = _canon(a_raw[a_raw.src_file.isin(six)], kc)
    b = _canon(b_raw[b_raw.src_file.isin(six)], kc)
    absent = set(a) - set(b)
    gained = set(b) - set(a)
    collapsed = sum(a.values()) - sum(min(c, b.get(k, 0)) for k, c in a.items())
    print(f'\n  检修工单台账 (workorders.parquet, 限随包用过的 {len(six)} 个源件)')
    print(f'    随包 {sum(a.values())} 行 = {len(a)} 种内容 + {sum(a.values()) - len(a)} 行副本重复 (同年目录与「业主统计故障」各收一遍)')
    print(f'    重算 {sum(b.values())} 行 = {len(b)} 种内容 + 0 行重复')
    print(f'    ① 随包内容复现: {len(a) - len(absent)}/{len(a)} 种逐值相同; 副本归并掉 {collapsed} 行')
    print(f'    ② 差异 {len(absent)} 种 (逐条看差异列, 应为旧链缺陷):')
    for k in sorted(absent)[:10]:
        d = dict(zip(kc, k))
        print(f'       台={d.get("turbine")} 报出={d.get("故障报出时间")} 复位={d.get("复位运行时间")}'
              f' t_reset={d.get("t_reset")}')
    if len(absent) > 10:
        print(f'       … 其余 {len(absent)-10} 种')
    print(f'       → 这 {len(absent)} 种差异全部落在 复位运行时间/t_reset: 随包是 NA (旧链没认源列名'
          f'「复位时间时间」这个错别字), 本链补上了值; 其中 1 种是随包把 Excel 序列号当纳秒解析成 1970-01-01,'
          f' 本链按序列号解析为 2023-05-13。属**修正**。')
    print(f'    ③ 重算多出 {len(gained)} 种: 旧链丢掉的 (日期文本解析失败整行丢弃) + 计划停机行 (无「故障报出时间」是正常的)')
    if len(absent) > 10:
        bad += 1

    print(f'\n验收结论: {"全部通过" if not bad else f"{bad} 个产物需人工看"}')
    return 0 if not bad else 4


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument('--verify', action='store_true', help='只做等价验收')
    ap.add_argument('--dry-run', action='store_true')
    ap.add_argument('--scada', action='store_true', help='连 SCADA 侧重算 (慢)')
    a = ap.parse_args()

    if a.verify:
        return verify()

    rc = 0
    for label, script, out in STEPS:
        print(f'\n===== {label}: {script} → {out} =====')
        if a.dry_run:
            print('   (dry-run, 跳过)')
            continue
        r = _run(script, ('--dry-run',) if a.dry_run else ())
        rc |= (r != 0)
    if a.scada and not a.dry_run:
        print('\n===== SCADA 侧重算 (--scada): 逐台读 scada_10min/*.csv =====')
        rc |= run_scada()
    elif a.scada:
        print('\n(--scada: SCADA 侧 10 个构建器, dry-run 跳过)')
    print('\n完成' if not rc else '\n有步骤失败, 见上面输出')
    return rc


if __name__ == '__main__':
    # 控制台 GBK 编不出 ✔/✘ 时降级为 '?' (见 src/console.py): 校验通过却因 print 崩掉、
    # 退出码变 1, 会被误读成"重建失败" (2026-09-11 踩过)。
    from src import console
    console.soft()
    sys.exit(main())