#!/usr/bin/env python3
# -*- coding: utf-8 -*-
r"""补齐"包内没有生成端"的产物 —— 只补缺件, 不动 raw 重算件; 并落一份**来源台账** (2026-09-12)。
## 为什么需要它
`rebuild_from_raw.py` 能从 `data/raw` 重算出一部分产物 (L0 仓 16 件 + 本体 3 件); 但随包里另有
一批产物**全库只有读取方、0 处写入方** —— 没有生成端, 从零重算就是拿不到 (工作台主数据
`/api/fleet` 一缺 `pitch/pitch_daily.parquet` / `temp_monthly.parquet` 就整个回 `err=no_products`,
页面看着像"没数据")。
用户令 (2026-09-12): **既要"从零重算", 又要"页面不缺"**。两者只能这样同时成立:
· 能证明是"从 raw 重算出来的"→ 用重算件 (并且逐值对齐随包件才敢用);
· 证明不了的 (没有生成端 / 规则复现不出) → 用随包件补齐, **但必须标明它不是重算件**。
本脚本就是后者的执行者, 并把每一件的来源写进 `outputs/<场>/_provenance.json`。
## 规则 (保守优先)
对随包产物目录里的每个文件:
· 目标**已存在** → 跳过 (那是 raw 重算出来的, 不许被随包件覆盖);
· 目标不存在 → 从随包件拷过去, 记为 `shipped`;
另: 只处理"产物目录"(outputs/<场>/ 下的仓), 不碰 release/ 与 data/。
## 用法
python scripts/products_restore_missing.py --dry-run # 只报要补什么
python scripts/products_restore_missing.py # 真补 + 写台账
python scripts/products_restore_missing.py --stash
# 指定随包件所在目录
"""
from __future__ import annotations
import argparse
import json
import pathlib
import shutil
import sys
import time
ROOT = pathlib.Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT))
from src import paths as P # noqa: E402
# 由 data/raw 重算出来的 (重建器 + 验证依据) —— 这些**永不被随包件覆盖**
RAW_DERIVED = {
# 三门台账 (rebuild_from_raw.py; --verify 与随包基线逐值比对过)
'windscada/alarms.parquet': 'scripts/windscada_alarms_ingest.py',
'windscada/workorders.parquet': 'scripts/windscada_workorder_ingest.py',
'windscada/oil_samples_index.parquet': 'scripts/windscada_watch_channels_build.py',
# SCADA 侧 10 个构建器 (rebuild_from_raw.py --scada)
'windscada/powercurve_dev.parquet': 'src.windscada.perf.powercurve',
'windscada/powercurve_bins.parquet': 'src.windscada.perf.powercurve',
'windscada/loss_monthly.parquet': 'src.windscada.perf.availability',
'windscada/curve_lenses.parquet': 'src.windscada.perf.curves',
'windscada/curve_liveness.parquet': 'src.windscada.perf.curves',
'windscada/control_profile.parquet': 'src.windscada.perf.control',
'windscada/control_schedule.parquet': 'src.windscada.perf.control',
'windscada/stop_events.parquet': 'src.windscada.perf.faults',
'windscada/temp_bins.parquet': 'src.windscada.subsys.temp_nbm',
'windscada/yaw_daily.parquet': 'src.windscada.subsys.yaw',
'windscada/hydraulic_accum.parquet': 'src.windscada.subsys.hydraulic',
'windscada/thermal_chain.parquet': 'src.windscada.subsys.thermal_chain',
'windscada/system_aux.parquet': 'src.windscada.taxonomy',
# 月度派生件 (本轮新增; 逐值对齐随包件后才算数)
'windscada/temp_monthly.parquet': 'scripts/windscada_monthly_build.py (19494/19494 逐值一致)',
# 本体层 (kb_ingest → populate → chain_ingest → trend_ingest; 全链跑完后 audit 0 问题)
'ontology/objects.json': 'python -m src.ontology.kb_ingest + populate + chain_ingest + trend_ingest (audit 0 问题)',
'ontology/retrieval_index.json': 'src.ontology.retrieval.build',
'ontology/turbine_params.parquet': 'src.ontology.maintenance.refresh_params',
}
def stash_dir(explicit=None) -> pathlib.Path:
"""随包产物暂存目录 (products_state --off 挪走后的地方)。"""
if explicit:
return pathlib.Path(explicit)
for p in sorted((ROOT / '_products_off').rglob('rudong')):
if p.is_dir():
return p
raise SystemExit('找不到随包产物暂存目录 (_products_off/**/rudong); 用 --stash 指定')
def main() -> int:
ap = argparse.ArgumentParser()
ap.add_argument('--dry-run', action='store_true')
ap.add_argument('--stash', default=None)
a = ap.parse_args()
stash = stash_dir(a.stash)
dest_root = P.STORE if hasattr(P, 'STORE') else ROOT / 'outputs' / 'rudong'
print(f'随包件: {stash.relative_to(ROOT)}')
print(f'产物仓: {dest_root.relative_to(ROOT)}\n')
prov = {}
for src in sorted(stash.rglob('*')):
if not src.is_file():
continue
rel = src.relative_to(stash).as_posix()
dst = dest_root / rel
if dst.exists():
prov[rel] = dict(source='raw-derived', builder=RAW_DERIVED.get(rel, '(早期重算, 未登记)'))
continue
prov[rel] = dict(source='shipped', why='包内无生成端 / 规则未复现 → 用随包件补齐')
if not a.dry_run:
dst.parent.mkdir(parents=True, exist_ok=True)
shutil.copy2(src, dst)
by_store = {}
for rel, m in prov.items():
store = rel.split('/')[0]
d = by_store.setdefault(store, {'raw-derived': 0, 'shipped': 0})
d[m['source']] += 1
print(f'{"仓":26s} {"raw 重算":>9s} {"随包补齐":>9s}')
for store, d in sorted(by_store.items()):
print(f' {store:24s} {d["raw-derived"]:9d} {d["shipped"]:9d}')
tot = {k: sum(d[k] for d in by_store.values()) for k in ('raw-derived', 'shipped')}
print(f' {"合计":24s} {tot["raw-derived"]:9d} {tot["shipped"]:9d}')
if a.dry_run:
print('\n(dry-run, 未写盘)')
return 0
out = dest_root / '_provenance.json'
out.write_text(json.dumps(dict(
at=time.strftime('%Y-%m-%d %H:%M:%S'),
note='逐件来源台账: raw-derived = 由 data/raw 重算(含验证依据); shipped = 包内无生成端, 用随包件补齐',
counts=tot, files=prov), ensure_ascii=False, indent=1), encoding='utf-8')
print(f'\n已写来源台账 {P.rel(out)}')
return 0
if __name__ == '__main__':
# 控制台可能是 GBK: print 里的非 GBK 字形编不出来会抛异常, 干成了却退出码 1
for _s in (sys.stdout, sys.stderr):
try: _s.reconfigure(errors='replace')
except Exception: pass
sys.exit(main())