guanlan-src/scripts/products_state.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
r"""产物开关 —— 把 outputs\<场>\ 下的产物整体挪走 / 挪回 (人工检查页面空状态用)。

## 为什么做成开关而不是删除

产物是"上一次算出来的结果", 人工检查"没数据时页面长什么样"必须把它们清掉; 但清掉之后一定要能**一条命令还原**,
否则就得重跑整条摄入链(而且像油样那 102 行华创合并报告根本重算不出来)。所以: **移动**(同盘瞬间完成) + 清单留痕。

## 保留不动的

- `data\raw\` 现场数据 (检查的是"产物没了会怎样", 不是"数据没了会怎样")
- `outputs\<场>\windscada\_pre_rebuild_20260911\` 随包基线备份 (它是备份, 不是产物; `rebuild_from_raw.py --verify` 要用)
- `release\` 门户与三维/交付件 (那是外壳, 不是数据产物)
- `logs\` `run\` 运行期文件

用法:
    python scripts/products_state.py --off      # 挪走全部产物, 打印清单
    python scripts/products_state.py --status   # 看当前是开还是关
    python scripts/products_state.py --on       # 全部挪回
"""
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

OFF = ROOT / '_products_off'                                       # 挪走后的存放处 (仓库外无需, 放安装目录下便于还原)
MANIFEST = OFF / 'manifest.json'
KEEP = ('_pre_rebuild_20260911',)                                  # 随包基线备份, 不算产物


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[tuple[pathlib.Path, pathlib.Path]]:
    """→ [(原位置, 挪走后的位置)]"""
    items = []
    for fam in farms():
        for d in sorted(fam.iterdir()):
            if d.name in KEEP:
                continue
            items.append((d, OFF / fam.name / d.name))
    return items


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 archive_old() -> bool:
    """把**上一代**暂存区改名存档, 给这一次挪走腾出干净的落点。

    为什么必须做: 挪走的落点是 `_products_off/<场>/<目录>`; 若该目录已存在(shutil.move 的语义是
    "目的地是已存在目录 → 把源挪进去"), 会嵌套成 `_products_off/rudong/windscada/windscada/…`,
    随后 `--on` 就会把新旧两代混在一起还原 —— 2026-09-12 实逮这个隐患。
    存档时**保留** `_baseline_kept/`(随包基线, `rebuild_from_raw.py --verify` 要用), 只动场站目录与清单。
    """
    arch = ROOT / f'_products_off_prev_{time.strftime("%Y%m%d_%H%M%S")}'
    moved = []
    for fam in farms():
        src = OFF / fam.name
        if src.exists():
            arch.mkdir(parents=True, exist_ok=True)
            shutil.move(str(src), str(arch / fam.name))
            moved.append(fam.name)
    if MANIFEST.exists():
        arch.mkdir(parents=True, exist_ok=True)
        shutil.move(str(MANIFEST), str(arch / 'manifest.json'))
    if not moved and not arch.exists():
        return False
    print(f'  上一代暂存区已存档 → {P.rel(arch)}/  (场站: {", ".join(moved) or "无"}; 基线 _baseline_kept 留在原地)')
    return True


def do_off(archive=False) -> int:
    items = plan()
    if not items:
        print('产物已经全部挪走了 (用 --status 看清单)')
        return 0
    OFF.mkdir(parents=True, exist_ok=True)
    conflicts = [dst for _, dst in items if dst.exists()]
    if conflicts:
        if not archive:
            print(f'[X] 暂存区里已经有同名产物 ({len(conflicts)} 个, 例: {P.rel(conflicts[0])}) —— 直接挪会嵌套成'
                  f' <目录>/<目录>/, 之后 --on 会把两代混在一起。\n'
                  f'    二选一: ① 加 --archive-old 把上一代存档后重跑 (推荐);\n'
                  f'           ② 手工把整个 {P.rel(OFF)} 改名后再跑 (例: Move-Item _products_off _products_off_旧)。')
            return 1
        archive_old()
    man = []
    # 随包基线备份藏在 windscada 里面: 先摘出来, 它不是产物 (rebuild_from_raw.py --verify 要用)
    baseline = P.store() / KEEP[0]
    baseline_kept = False
    kept_at = OFF / '_baseline_kept' / KEEP[0]
    if baseline.exists():
        if not kept_at.exists():
            kept_at.parent.mkdir(parents=True, exist_ok=True)
            n, mb = count(baseline)
            shutil.move(str(baseline), str(kept_at))
            print(f'  保留 {P.rel(baseline):34s} {n:5d} 件 {mb:8.1f} MB  (随包基线, 不算产物)')
    kept_at = OFF / '_baseline_kept' / KEEP[0]
    baseline_kept = kept_at.exists()          # ★必须在 move **之后**判 (2026-09-11 曾写成之前 → 恒 False, --on 就不还原基线)
    for src, dst in items:
        if not src.exists():
            continue
        n, mb = count(src)
        dst.parent.mkdir(parents=True, exist_ok=True)
        shutil.move(str(src), str(dst))
        man.append(dict(rel=P.rel(src), to=P.rel(dst), files=n, mb=round(mb, 1)))
        print(f'  挪走 {P.rel(src):34s} {n:5d} 件 {mb:8.1f} MB')
    MANIFEST.write_text(json.dumps(dict(at=time.strftime('%Y-%m-%d %H:%M:%S'),
                                        baseline_kept=baseline_kept, items=man),
                                   ensure_ascii=False, indent=1), encoding='utf-8')
    print(f'\n共 {len(man)} 项已挪到 {P.rel(OFF)} (清单: {P.rel(MANIFEST)})')
    print(f'还原: python scripts/products_state.py --on')
    return 0


def do_on() -> int:
    if not MANIFEST.exists():
        print('没有清单 (没挪过, 或清单已删)')
        return 1
    man = json.loads(MANIFEST.read_text(encoding='utf-8'))
    n = 0
    for it in man['items']:
        src, dst = P.resolve(it['to']), P.resolve(it['rel'])
        if not src.exists():
            print(f'  [!] 备份里没有 {it["to"]}, 跳过')
            continue
        dst.parent.mkdir(parents=True, exist_ok=True)
        if dst.exists():
            shutil.rmtree(dst) if dst.is_dir() else dst.unlink()
        shutil.move(str(src), str(dst))
        n += 1
        print(f'  还原 {it["rel"]:34s} {it["files"]:5d} 件')
    if man.get('baseline_kept') or (OFF / '_baseline_kept' / KEEP[0]).exists():
        b = OFF / '_baseline_kept' / KEEP[0]        # 不看清单标记, 直接看备份在不在 (更稳)
        if b.exists():
            target = P.store() / KEEP[0]
            target.parent.mkdir(parents=True, exist_ok=True)
            if target.exists():
                shutil.rmtree(target)
            shutil.move(str(b), str(target))
            print(f'  还原 {P.rel(target):34s} (随包基线)')
    if OFF.exists() and not any(OFF.rglob('*')):
        shutil.rmtree(OFF, ignore_errors=True)
    print(f'\n共还原 {n} 项')
    return 0


def do_status() -> int:
    items = plan()
    if not items:
        print('当前: 产物**已全部挪走** (页面应呈空状态)')
        if MANIFEST.exists():
            man = json.loads(MANIFEST.read_text(encoding='utf-8'))
            print(f'  挪走时间 {man["at"]}, 共 {len(man["items"])} 项')
            for it in man['items']:
                print(f'    {it["rel"]:34s} {it["files"]:5d} 件 {it["mb"]:8.1f} MB')
        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')
    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('--on', action='store_true', help='全部挪回')
    g.add_argument('--status', action='store_true', help='看当前状态')
    ap.add_argument('--archive-old', action='store_true',
                    help='--off 前把上一代暂存区存档到 _products_off_prev_<时间戳>/ (暂存区已有同名产物时必须加)')
    a = ap.parse_args()
    sys.exit(do_off(archive=a.archive_old) if a.off else do_on() if a.on else do_status())