diff --git a/app/harness/build_split.py b/app/harness/build_split.py index cd70b73..197d4e1 100644 --- a/app/harness/build_split.py +++ b/app/harness/build_split.py @@ -12,6 +12,7 @@ test,再以视频组为原子切出诊断 / 验证池,原子冻结 pools.jso from __future__ import annotations import hashlib +import json import sqlite3 from collections import Counter, defaultdict from dataclasses import asdict, dataclass, field @@ -104,6 +105,38 @@ class SplitBuildResult: return getattr(self, key) +def _guard_frozen_products(out_path: Path, manifest_path: Path, *, force: bool) -> None: + """冻结前的覆盖保护:产物已存在时按 force 决定报错或备份。 + + 参数: + out_path: 目标 pools.json 路径。 + manifest_path: 目标 split_manifest.json 路径。 + force: False 时已存在即 FileExistsError;True 时把旧产物重命名为 + .bak.<旧 pools_sha256 前 8 位或 'prev'> 再放行。 + + 异常: + FileExistsError: force=False 且产物已存在(防静默覆盖冻结锚点)。 + """ + if not out_path.exists() and not manifest_path.exists(): + return + if not force: + raise FileExistsError( + f"已存在冻结产物 {out_path}(或其 manifest)。重跑切分会覆盖训练依赖的" + "冻结锚点——确认要替换请加 --force(旧产物将备份为 .bak.*)。" + ) + # 备份后缀取旧 manifest 的 pools_sha256 前 8 位,无则用 'prev' + suffix = "prev" + if manifest_path.exists(): + try: + old = json.loads(manifest_path.read_text(encoding="utf-8")) + suffix = str(old.get("pools_sha256", "prev"))[:8] or "prev" + except (json.JSONDecodeError, OSError): + suffix = "prev" + for p in (out_path, manifest_path): + if p.exists(): + p.rename(p.with_name(f"{p.name}.bak.{suffix}")) + + def build_split( *, db_path: Path, @@ -115,6 +148,7 @@ def build_split( out_path: Path, manifest_path: Path, generated_at: str, + force: bool = False, ) -> SplitBuildResult: """顶层编排结果驱动视频级切分,冻结 pools.json + manifest 并跑防御断言。 @@ -137,6 +171,9 @@ def build_split( out_path: 冻结 pools.json 目标路径(原子写)。 manifest_path: 溯源 manifest 目标路径(原子写)。 generated_at: 生成时间戳(ISO 字符串),由调用方传入以保证可复现。 + force: 覆盖保护开关。False(默认)时若 out_path/manifest_path 已存在即 + FileExistsError(防静默覆盖训练依赖的冻结锚点);True 时先把旧产物备份为 + .bak.* 再放行覆盖。 返回: SplitBuildResult,含 pools / manifest / assignment,支持字典式访问。 @@ -199,6 +236,7 @@ def build_split( val_wrong_min=config.val_wrong_min, wrong_tier_by_video=dict(wrong_tier_by_video), ) + _guard_frozen_products(out_path, manifest_path, force=force) save_pools(pools, out_path) # Phase 4: 溯源 manifest(pools_sha256 锚定冻结内容)。 diff --git a/app/harness/video_split_cli.py b/app/harness/video_split_cli.py index 1317b8b..90a7d4f 100644 --- a/app/harness/video_split_cli.py +++ b/app/harness/video_split_cli.py @@ -519,6 +519,7 @@ async def run_pipeline( questions_dir: Path, out_dir: Path, generated_at: str, + force: bool = False, ) -> SplitBuildResult: """内联三阶段:Phase 0 INFRA T0 补录 → Phase 1 诊断 → Phase 2 冻结切分 → McNemar 护栏。 @@ -534,6 +535,7 @@ async def run_pipeline( questions_dir: benchmark 题库目录(Phase 2 加载题库切池)。 out_dir: 冻结产物目录(pools.json + split_manifest.json)。 generated_at: 生成时间戳(ISO 字符串,由调用方传入;见模块 C-2 复现锚点约定)。 + force: 覆盖已存在冻结产物开关,透传给 build_split(False 时已存在即报错)。 返回: SplitBuildResult(冻结三池 + manifest + assignment)。 @@ -577,6 +579,7 @@ async def run_pipeline( out_path=out_dir / "pools.json", manifest_path=out_dir / "split_manifest.json", generated_at=generated_at, + force=force, ) # McNemar 功效护栏(build_split 契约外的 capstone 层校验)。 @@ -650,6 +653,7 @@ def _execute_real(config: VideoSplitConfig, fingerprint: str, args: argparse.Nam questions_dir=questions_dir, out_dir=out_dir, generated_at=generated_at, + force=args.force, ) ) finally: @@ -773,6 +777,11 @@ def build_arg_parser() -> argparse.ArgumentParser: "对 manifest 做字节级复现比对。" ), ) + parser.add_argument( + "--force", + action="store_true", + help="覆盖已存在的冻结 pools.json/manifest(旧产物备份为 .bak.*)", + ) return parser diff --git a/tests/unit/test_video_split_cli.py b/tests/unit/test_video_split_cli.py index c15b196..60cb702 100644 --- a/tests/unit/test_video_split_cli.py +++ b/tests/unit/test_video_split_cli.py @@ -303,3 +303,29 @@ def test_git_short_sha_nonempty(): sha = cli.git_short_sha() assert sha assert len(sha) >= 4 + + +def test_build_split_refuses_overwrite_without_force(tmp_path): + """已存在指纹不同的 pools.json 时,force=False 必须报错不覆盖。""" + from app.harness.build_split import _guard_frozen_products + + out_path = tmp_path / "pools.json" + out_path.write_text('{"split_mode":"global"}', encoding="utf-8") + manifest_path = tmp_path / "split_manifest.json" + + with pytest.raises(FileExistsError, match="已存在冻结产物"): + _guard_frozen_products(out_path, manifest_path, force=False) + + +def test_build_split_force_backs_up_old(tmp_path): + """force=True 时旧产物被备份为 .bak.* 再允许覆盖。""" + from app.harness.build_split import _guard_frozen_products + + out_path = tmp_path / "pools.json" + out_path.write_text('{"old":1}', encoding="utf-8") + manifest_path = tmp_path / "split_manifest.json" + manifest_path.write_text('{"pools_sha256":"deadbeef00000000"}', encoding="utf-8") + + _guard_frozen_products(out_path, manifest_path, force=True) + baks = list(tmp_path.glob("pools.json.bak.*")) + assert len(baks) == 1, f"未备份旧产物: {list(tmp_path.iterdir())}"