#!/usr/bin/env python3 """QUARANTINED: dev-only — atomic claim 우회. production dispatch 에서 호출 금지. 본 스크립트는 `runner._execute` 를 직접 호출 (line 68/122) — Block C 의 atomic claim (`StepRunner._try_claim_running`) + finalize (`_execute_and_finalize`) 를 모두 우회한다. 동시 실행 시 race 가능 + step_run row 가 'running' 으로 leak 될 수 있다. 운영 dispatch 경로 (api/v1/steps, run_steps_batch) 에서 호출 금지. 개발/디버그 용도 한정 — 단일 씬/shot scene_detail 만 fan-out filter 후 직접 실행해 결과를 확인하는 목적. AC-C3 grep 의 allowlist (디렉토리 quarantine `backend/scripts/_quarantined/`) 로 관리. plan v2.1.3 §3744 옵션 B. Usage (dev only): .venv/bin/python scripts/_quarantined/test_scene_detail.py [--save] .venv/bin/python scripts/_quarantined/test_scene_detail.py --shot [--save] --shot N: 특정 shot만 재실행 (v4 shot 모드) --save: 결과를 체크포인트에 병합 저장 (없으면 stdout 출력만) """ import argparse import json import logging import os import sys # backend/ 기준 경로 설정 (한 단계 더 상위) sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))) logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s") logger = logging.getLogger("test_scene_detail") def main(): parser = argparse.ArgumentParser(description="단일 씬 scene_detail 테스트") parser.add_argument("project_id", help="프로젝트 ID") parser.add_argument("episode_id", help="에피소드 ID") parser.add_argument("scene_index", type=int, help="테스트할 씬 번호") parser.add_argument("--shot", type=int, default=None, help="v4: 특정 shot만 재실행") parser.add_argument("--save", action="store_true", help="체크포인트에 병합 저장") args = parser.parse_args() from app.core.database import SessionLocal from app.core.steps.detail_steps import SceneDetailStep db = SessionLocal() try: # project_config 로드 from sqlalchemy import text row = db.execute(text( "SELECT llm_config_json FROM project_settings WHERE project_id = :pid" ), {"pid": args.project_id}).fetchone() project_config = json.loads(row[0]) if row and row[0] else {} runner = SceneDetailStep( step_id="scene_detail", project_id=args.project_id, episode_id=args.episode_id, db=db, project_config=project_config, ) # 기존 체크포인트 로드 existing_cp = runner.load_checkpoint() existing_scenes = {} if existing_cp and existing_cp.get("data", {}).get("scenes"): for s in existing_cp["data"]["scenes"]: existing_scenes[s.get("scene_index", 0)] = s # _execute 내부에서 사용할 segments 필터링을 위해 monkey-patch original_execute = runner._execute def _filtered_execute(mode="resume"): """원본 _execute를 호출하되 segments를 필터링.""" # 직접 실행 로직 재구현 (단일 씬만) from pathlib import Path from app.core.config import settings cp_base = ( Path(settings.projects_dir) / args.project_id / "checkpoints" / "episodes" / args.episode_id ) save_cp_path = cp_base / "scene_save" / "manifest.json" save_cp = json.loads(save_cp_path.read_text()) if save_cp_path.exists() else {} all_segments = save_cp.get("data", {}).get("segments", []) # 타겟 씬만 필터 target_segs = [s for s in all_segments if s.get("scene_index") == args.scene_index] if not target_segs: logger.error("씬 %d을 찾을 수 없습니다 (전체 %d씬)", args.scene_index, len(all_segments)) return {"completed_count": 0, "applicable_count": 0, "failed_count": 0, "data": {"scenes": []}} logger.info("씬 %d 단독 실행 시작", args.scene_index) # segments를 임시로 교체해서 원본 실행 import types # segments 패치: _execute 안에서 segments를 로드하므로, save checkpoint를 임시 교체 original_load = runner._load_prev_checkpoint def _patched_load(step_id): cp = original_load(step_id) if step_id == "scene_save" and cp: cp = json.loads(json.dumps(cp)) # deep copy cp["data"]["segments"] = target_segs # v4: --shot 지정 시 shot_validator에서 해당 shot만 포함 # (shot_validator로 다운스트림 로더가 전환됨. shot_extract 체크포인트는 # 있어도 scene_context_loader가 읽지 않음) if args.shot is not None and step_id == "shot_validator" and cp: cp = json.loads(json.dumps(cp)) for sc in cp.get("data", {}).get("scenes", []): if sc.get("scene_index") == args.scene_index: sc["shots"] = [sh for sh in sc.get("shots", []) if sh.get("shot_index") == args.shot] if args.shot is not None and step_id == "shot_selection" and cp: cp = json.loads(json.dumps(cp)) for sc in cp.get("data", {}).get("scenes", []): if sc.get("scene_index") == args.scene_index: sc["selected_shot_indices"] = [args.shot] sc["selected_count"] = 1 return cp runner._load_prev_checkpoint = _patched_load try: result = original_execute(mode) finally: runner._load_prev_checkpoint = original_load return result result = _filtered_execute() new_scenes = result.get("data", {}).get("scenes", []) if not new_scenes: logger.error("씬 %d 결과 없음", args.scene_index) return # 결과 출력 for s in new_scenes: si = s["scene_index"] print(f"\n=== 씬 {si} === visible: {s.get('visible_entities', [])}") for vi, v in enumerate(s.get("t2i_variations", []), 1): print(f" var_{vi}: {v.get('t2i_prompt', '')}") print(f" outfits: {v.get('outfit_assignments', [])}") # 체크포인트 병합 저장 if args.save: # v4: (scene_index, _shot_index) 키로 병합 new_keys = {(s.get("scene_index"), s.get("_shot_index")) for s in new_scenes} merged = [s for s in existing_cp.get("data", {}).get("scenes", []) if (s.get("scene_index"), s.get("_shot_index")) not in new_keys] merged.extend(new_scenes) merged.sort(key=lambda r: (r.get("scene_index", 0), r.get("_shot_index") or 0)) save_data = existing_cp or {} save_data.update({ "status": "completed", "completed_count": len(merged), "applicable_count": len(merged), "failed_count": 0, "data": {"scenes": merged}, }) runner.save_checkpoint(save_data) logger.info("체크포인트 저장 완료 (%d씬 병합)", len(merged)) else: logger.info("--save 없음, 출력만 완료") finally: db.close() if __name__ == "__main__": main()