#!/usr/bin/env python3 # -*- coding: utf-8 -*- r"""产物开关 —— 把 outputs\<场>\ 下的产物**直接清掉** / 看当前状态 (人工检查页面空状态用)。 ## 2026-09-16 用户令: 清除产物**不留备份** 原设计是"移动到 `_products_off/<场>/` + 清单留痕, 一条命令还原"。用户令改为**不备份**: 清掉就是清掉。于是本脚本: · `--off --yes` 真删除 (rmtree), 打印删了什么; **不再产生 `_products_off*` 目录**; · `--status` 看产物在位/已清 (清掉后就是"已清", 没有备份清单可列); · `--on` 已**移除**: 没有备份就没有"挪回"。恢复缺失的**随包件**按用户令 2026-09-17 **不由交付包补齐**: 放原始件到 `data/raw/<场站>/` 后 `python scripts/rebuild_all.py`; 没有生成端的族由研发补生成端 (逐族可逆性见 `docs/系统设计说明.md` §13.6)。 (离线人工补救的口子仍在 `scripts/products_restore_missing.py --stash <…>`, 但它**不在运行链里**, 也不该被运行期依赖。) ## 保留不动的 - `data\raw\` 现场数据 (检查的是"产物没了会怎样", 不是"数据没了会怎样") - `release\` 门户与三维/交付件 (那是外壳, 不是数据产物) - `logs\` `run\` 运行期文件 - ★ `outputs\<场>\windscada\_pre_rebuild_20260911\` **不再特殊保留**: 它曾是"随包基线备份"而在清除时豁免, 现在按"不留备份"一并删掉 —— 等价验收因此暂时没有标准答案(如实报 rc=5, 不假装通过); 要恢复验收能力, 就用**当前在位且已核验的产物**重新落一份基线(见 docs §7 那条)。 用法: python scripts/products_state.py --status # 看当前产物状态 python scripts/products_state.py --off --yes # 清掉全部产物 (真删除, 不可恢复; --yes 是防手滑) """ from __future__ import annotations import argparse 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 LEGACY_OFF = ROOT / '_products_off' # 旧版暂存区 (只用于提示, 不再写入) def farms() -> list[pathlib.Path]: out = ROOT / 'outputs' return sorted(p for p in out.iterdir() if p.is_dir()) if out.is_dir() else [] def plan() -> list[pathlib.Path]: """→ 待清除的产物目录列表 (产物仓本身)。""" return [d for fam in farms() for d in sorted(fam.iterdir())] def count(p: pathlib.Path) -> tuple[int, float]: if p.is_file(): return 1, p.stat().st_size / 1024 / 1024 fs = [f for f in p.rglob('*') if f.is_file()] return len(fs), sum(f.stat().st_size for f in fs) / 1024 / 1024 def do_off(assume_yes=False) -> int: items = [p for p in plan() if p.exists()] if not items: print('产物已经是空的 (用 --status 看)') return 0 total_n = total_mb = 0 rows = [] for src in items: n, mb = count(src) rows.append((P.rel(src), n, mb)) total_n += n total_mb += mb print('即将**删除**以下产物 (按用户令: 不留备份, 删除后不可恢复):') for rel, n, mb in rows: print(f' {rel:34s} {n:5d} 件 {mb:8.1f} MB') print(f' 合计 {len(rows)} 项 / {total_n} 件 / {total_mb:.1f} MB') print(' 恢复办法: 放原始件到 data/raw/<场站>/ 后 python scripts/rebuild_all.py; 缺生成端的族由研发补 (docs §13.6)') if not assume_yes: print('\n[X] 未执行: 这是不可恢复操作, 请显式加 --yes (运维控制台里点按钮时已由前端确认过一次)') return 2 for src in items: shutil.rmtree(src, ignore_errors=True) if src.is_dir() else src.unlink(missing_ok=True) if LEGACY_OFF.exists(): print(f' [!] 检测到旧版暂存区 {P.rel(LEGACY_OFF)} —— 按"不留备份"的口径, 本次**未删除**它; ' f'确认不需要后请手工删掉 (它是旧设计留下的产物备份)') print(f'\n已删除 {len(rows)} 项 / {total_n} 件 / {total_mb:.1f} MB (无备份)') return 0 def do_status() -> int: items = [p for p in plan() if p.exists()] if not items: print('当前: 产物**已清空** (页面应呈空状态)') print(' 恢复: python scripts/rebuild_all.py (缺生成端的族由研发补, 见 docs §13.6)') return 0 print(f'当前: 产物**在位** ({len(items)} 项)') for src in items: n, mb = count(src) print(f' {P.rel(src):34s} {n:5d} 件 {mb:8.1f} MB') print(' 说明: 按 2026-09-16 用户令, 清除产物不留备份 (没有 _products_off/ 可还原)') return 0 if __name__ == '__main__': ap = argparse.ArgumentParser() g = ap.add_mutually_exclusive_group(required=True) g.add_argument('--off', action='store_true', help='清掉全部产物 (**直接删除, 不留备份**)') g.add_argument('--status', action='store_true', help='看当前状态') g.add_argument('--on', action='store_true', help='[已移除] 旧版的"挪回" —— 见文件头说明') ap.add_argument('--yes', action='store_true', help='确认执行不可恢复的清除 (--off 必带)') a = ap.parse_args() if a.on: print('[X] --on 已移除: 按用户令"清除产物不留备份", 没有备份可挪回。\n' ' 运行期恢复: 放原始件后 python scripts/rebuild_all.py (缺生成端的族由研发补, docs §13.6)\n' ' 要重新算出数据面产物: python scripts/rebuild_all.py (SCADA 侧加 --scada)') sys.exit(3) sys.exit(do_off(assume_yes=a.yes) if a.off else do_status())