"""scene_camera_flow StepRunner — 씬 단위 카메라 연속 이동 경로 설계. shot_staging 이전에 실행되어, 선택된 샷들을 씬 관통 카메라 플로우 위에 배정한다. shot_staging은 이 플로우에서 개별 샷 카메라를 파생시킨다. """ import logging from concurrent.futures import ThreadPoolExecutor, as_completed from typing import Any, Dict, List from app.core.errors import AppError from app.core.step_runner import StepRunner from app.core.steps.detail_steps import _DetailStepMixin from app.modules.llm.llm_client import call_structured from app.modules.prompt_loader import load_prompt, load_schema logger = logging.getLogger(__name__) MAX_WORKERS = 4 class SceneCameraFlowStep(_DetailStepMixin, StepRunner): """씬 단위 카메라 연속 흐름 설계 — 선택된 샷을 플로우 위에 배정.""" def _config_hash(self) -> str: """실제 로드되는 effective prompt(system/user/schema)의 내용 해시를 접는다 — 같은 팩 내 내용 변경도 CP 무효화 (shot_staging 관례 동형). 2026-08-12 카메라 문법 팩 v3 발행에서 드러난 구멍: 기본 config_hash 는 프로젝트 설정만 봐서, 팩을 올려도 완료 스텝이 clean SKIP 되고 하류 staging 이 **구 flow 를 계승**해 팩 효과가 무력화된다(정면·정적 수렴의 발원지가 flow 라 이 스텝의 재실행이 수정의 성패 지점이다).""" import hashlib import json as _json from app.core.step_runner import compute_config_hash h = hashlib.sha256() h.update(load_prompt("scene_camera_flow", "system").encode("utf-8")) h.update(b"\x00") h.update(load_prompt("scene_camera_flow", "user").encode("utf-8")) h.update(b"\x00") h.update(_json.dumps( load_schema("scene_camera_flow", "schema"), sort_keys=True, ).encode("utf-8")) h.update(b"\x00") h.update(compute_config_hash(self.project_config).encode("utf-8")) return h.hexdigest()[:16] def _execute(self, mode="resume") -> Dict[str, Any]: # 필수 체크포인트 검증 (파일 존재 + 최소 data shape) scene_save_cp = self._load_prev_checkpoint("scene_save") if not scene_save_cp or not scene_save_cp.get("data", {}).get("segments"): raise AppError( code="step.no_input", message="scene_save 체크포인트 누락 또는 segments 없음", status_code=400, ) segments = scene_save_cp["data"]["segments"] seg_by_scene: Dict[int, dict] = {seg.get("scene_index", 0): seg for seg in segments} shot_extract_cp = self._load_prev_checkpoint("shot_validator") shot_selection_cp = self._load_prev_checkpoint("shot_selection") if not shot_extract_cp or not shot_selection_cp: raise AppError( code="step.no_input", message="shot_extract 또는 shot_selection 결과 없음", status_code=400, ) if not shot_extract_cp.get("data", {}).get("scenes"): raise AppError( code="step.no_input", message="shot_extract data.scenes가 비어있음", status_code=400, ) from app.core.steps.shot_validator_step import assert_no_failed_scenes assert_no_failed_scenes(shot_extract_cp, self.project_config, consumer_step="scene_camera_flow") scene_director_cp = self._load_prev_checkpoint("scene_director") if not scene_director_cp or not scene_director_cp.get("data", {}).get("scenes"): raise AppError( code="step.no_input", message="scene_director 체크포인트 누락 또는 scenes 없음", status_code=400, ) entity_merge_cp = self._load_prev_checkpoint("entity_merge") if not entity_merge_cp or not entity_merge_cp.get("data"): raise AppError( code="step.no_input", message="entity_merge 체크포인트 누락", status_code=400, ) # 선택된 shot_index 집합 (씬별) selected_map: Dict[int, set] = {} for sc in (shot_selection_cp.get("data", {}).get("scenes") or []): selected_map[sc["scene_index"]] = set(sc.get("selected_shot_indices", [])) # 모든 샷 (선택 + 비선택) 씬별 그룹핑 all_shots_by_scene: Dict[int, List[dict]] = {} for sc in shot_extract_cp["data"]["scenes"]: si = sc["scene_index"] all_shots_by_scene[si] = sc.get("shots", []) # 엔티티 이름 매핑 (ID는 프롬프트 노출 금지지만 인물 이름 참고용) entity_names_by_scene: Dict[int, List[str]] = {} if scene_director_cp and scene_director_cp.get("data", {}).get("scenes"): id_to_name: Dict[str, str] = {} if entity_merge_cp and entity_merge_cp.get("data"): for etype in ["characters", "locations", "props"]: for e in entity_merge_cp["data"].get(etype, []): sid = e.get("short_id", "") if sid: id_to_name[sid] = e.get("name", "") for sc in scene_director_cp["data"]["scenes"]: si = sc.get("scene_index") visible = sc.get("visible_entity_ids", []) or [] names = [id_to_name.get(v, "") for v in visible if id_to_name.get(v)] entity_names_by_scene[si] = names # 프롬프트 / 스키마 로드 system_prompt = load_prompt("scene_camera_flow", "system") user_template = load_prompt("scene_camera_flow", "user") schema = load_schema("scene_camera_flow", "schema") # resume: 기존 성공 결과 보존 existing_cp = self._load_prev_checkpoint("scene_camera_flow") existing_ok: Dict[int, dict] = {} if mode == "resume" and existing_cp and existing_cp.get("data", {}).get("scenes"): for sc in existing_cp["data"]["scenes"]: si = sc.get("scene_index") if sc.get("flow_stages"): existing_ok[si] = sc if existing_ok: logger.info("scene_camera_flow resume: %d scenes already done", len(existing_ok)) # 씬별 작업 리스트 tasks: List[dict] = [] scene_results: List[dict] = [] processed = 0 skipped = 0 failed = 0 for si, all_shots in sorted(all_shots_by_scene.items()): sel = selected_map.get(si, set()) selected_shots = [sh for sh in all_shots if sh.get("shot_index") in sel] unselected_shots = [sh for sh in all_shots if sh.get("shot_index") not in sel] if not selected_shots: scene_results.append({ "scene_index": si, "flow_summary": "선택된 샷 없음 — 플로우 없음", "flow_stages": [], "shot_assignments": [], }) skipped += 1 continue if si in existing_ok: scene_results.append(existing_ok[si]) processed += 1 continue tasks.append({ "scene_index": si, "selected_shots": selected_shots, "unselected_shots": unselected_shots, "seg": seg_by_scene.get(si, {}), "entities": entity_names_by_scene.get(si, []), }) if tasks: with ThreadPoolExecutor(max_workers=MAX_WORKERS) as pool: futures = { pool.submit( self._process_scene, task, system_prompt, user_template, schema ): task["scene_index"] for task in tasks } for fut in as_completed(futures): result = fut.result() if result.get("flow_stages"): processed += 1 else: failed += 1 scene_results.append(result) scene_results.sort(key=lambda r: r.get("scene_index", 0)) total = processed + skipped + failed return { "completed_count": processed + skipped, "applicable_count": total, "failed_count": failed, "data": {"scenes": scene_results}, # 팩 내용 해시 저장 — resume 대조가 같은 계산을 쓰게 한다 # (shot_staging 관례 동형, 위 _config_hash docstring 참조) "config_hash": self._config_hash(), } def _process_scene(self, task: dict, system_prompt: str, user_template: str, schema: dict) -> dict: si = task["scene_index"] selected_shots = task["selected_shots"] unselected_shots = task["unselected_shots"] seg = task["seg"] entity_names = task["entities"] scene_heading = seg.get("heading", "") scene_text = seg.get("text", "") sel_lines = [] for sh in selected_shots: sel_lines.append( f" Shot {sh.get('shot_index', '?')}: {sh.get('description', '')}" ) selected_block = "\n".join(sel_lines) if sel_lines else " (없음)" unsel_lines = [] for sh in unselected_shots: unsel_lines.append( f" Shot {sh.get('shot_index', '?')}: {sh.get('description', '')}" ) unselected_block = "\n".join(unsel_lines) if unsel_lines else " (없음)" entities_block = ", ".join(entity_names) if entity_names else "(엔티티 정보 없음)" user_prompt = user_template.format( scene_index=si, scene_heading=scene_heading, scene_text=scene_text, selected_shots_block=selected_block, unselected_shots_block=unselected_block, entities_block=entities_block, ) # 스키마에 shot_index enum 주입 — 선택된 샷만 허용 import copy call_schema = copy.deepcopy(schema) selected_indices = [sh["shot_index"] for sh in selected_shots] if selected_indices: call_schema["properties"]["shot_assignments"]["items"]["properties"]["shot_index"] = { "type": "integer", "enum": selected_indices, } try: result = call_structured( step="scene_camera_flow", system_prompt=system_prompt, user_prompt=user_prompt, response_schema=call_schema, project_config=self.project_config, schema_name=f"scene_camera_flow_{si}", opik_metadata=self.build_opik_metadata(extra_metadata={"scene_index": si}), ) result["scene_index"] = si logger.info( "scene_camera_flow S%d: %d stages, %d shot assignments", si, len(result.get("flow_stages", [])), len(result.get("shot_assignments", [])), ) return result except Exception as exc: logger.error("scene_camera_flow S%d failed: %s", si, exc) return { "scene_index": si, "flow_summary": f"분석 실패: {exc}", "flow_stages": [], "shot_assignments": [], }