"""Shot 선택 단계 — 씬별 중요 샷 N개 선정. 2개 이상 shot이 있는 씬에서만 선정 (1개 shot은 자동 선택). 씬별 병렬 처리. v4 (2026-04-19): 2중 캡 (절대 + 비율 50%) 적용, reason 필수. """ import logging import os from concurrent.futures import ThreadPoolExecutor, as_completed from typing import Any, Dict, List from app.core.step_runner import StepRunner from app.modules.llm.llm_client import call_structured from app.modules.prompt_loader import load_prompt, load_schema logger = logging.getLogger(__name__) SHOT_SELECTION_MAX = int(os.environ.get("SHOT_SELECTION_MAX", "5")) MAX_WORKERS = 4 def _half_cap(total: int) -> int: """씬 총 샷 수의 50% 비율 캡 (최소 1).""" return max(1, total // 2) def _effective_max(total: int, absolute_max: int) -> int: """절대 상한과 50% 비율 상한 중 작은 값. 최소 1.""" return max(1, min(absolute_max, _half_cap(total))) class ShotSelectionStep(StepRunner): """중요 샷 선택 — 씬별 2중 캡 (절대 + 비율).""" def _config_hash(self) -> str: """E2E13 Codex HIGH-3(재리뷰 HIGH-2): 실제 로드되는 effective prompt(system/user/selection_schema — DB active row 우선 포함)의 내용 해시를 접는다 — 팩 내 내용 변경·DB winner 변경도 CP 무효화.""" import hashlib import json as _json from app.core.step_runner import compute_config_hash from app.modules.prompt_loader import load_prompt, load_schema h = hashlib.sha256() for text in ( load_prompt("shot_selection", "system", db=self.db), load_prompt("shot_selection", "user", db=self.db), ): h.update(text.encode("utf-8")) h.update(b"\x00") h.update(_json.dumps( load_schema("shot_selection", "selection_schema", db=self.db), 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 _load_prev_checkpoint(self, step_id: str): import json as _json from pathlib import Path as _Path from app.core.config import settings cp = ( _Path(settings.projects_dir) / self.project_id / "checkpoints" / "episodes" / self.episode_id / step_id / "manifest.json" ) if cp.exists(): return _json.loads(cp.read_text(encoding="utf-8")) return None def _execute(self, mode="resume") -> Dict[str, Any]: shot_cp = self._load_prev_checkpoint("shot_validator") if not shot_cp or not shot_cp.get("data", {}).get("scenes"): from app.core.errors import AppError raise AppError(code="step.no_shots", message="shot_extract 결과 없음", status_code=400) from app.core.steps.shot_validator_step import assert_no_failed_scenes assert_no_failed_scenes(shot_cp, self.project_config, consumer_step="shot_selection") scenes = shot_cp["data"]["scenes"] system = load_prompt("shot_selection", "system", db=self.db) user_template = load_prompt("shot_selection", "user", db=self.db) schema = load_schema("shot_selection", "selection_schema", db=self.db) max_n = SHOT_SELECTION_MAX # v4: SHOT_SELECTION_ENABLED=false → 모든 shot 자동 선택 (reason 플래그) from app.core.config import settings as _settings if not _settings.shot_selection_enabled: all_auto = [] for s in scenes: shots = s.get("shots", []) sel_shots = [ {"shot_index": sh["shot_index"], "reason": "SHOT_SELECTION_ENABLED=false"} for sh in shots ] all_auto.append({ "scene_index": s["scene_index"], "selected_shot_indices": [sh["shot_index"] for sh in shots], "selected_shots": sel_shots, "total_shots": len(shots), "selected_count": len(shots), }) self._apply_episode_max_shots(all_auto, _settings.episode_max_shots) total_selected = sum(r["selected_count"] for r in all_auto) total_shots = sum(r["total_shots"] for r in all_auto) logger.info("shot_selection DISABLED: %d/%d shots selected", total_selected, total_shots) return { "completed_count": len(all_auto), "applicable_count": len(scenes), "failed_count": 0, "data": {"scenes": all_auto, "total_selected": total_selected, "total_shots": total_shots, "max_per_scene": max_n}, "config_hash": self._config_hash(), } # 1 shot: 자동 선택. 2+ shot: 모두 LLM 평가 (작은 씬도 포함 — v4) scenes_to_select = [] auto_selected = [] for s in scenes: shots = s.get("shots", []) if len(shots) <= 1: sel_shots = ( [{"shot_index": shots[0]["shot_index"], "reason": "씬 내 유일 샷"}] if shots else [] ) auto_selected.append({ "scene_index": s["scene_index"], "selected_shot_indices": [shots[0]["shot_index"]] if shots else [], "selected_shots": sel_shots, "total_shots": len(shots), "selected_count": len(shots), }) else: scenes_to_select.append(s) logger.info( "shot_selection: %d scenes auto-selected (1-shot), %d scenes to evaluate " "(abs_max=%d, ratio_cap=50%%)", len(auto_selected), len(scenes_to_select), max_n, ) llm_selected = [] if scenes_to_select: with ThreadPoolExecutor(max_workers=MAX_WORKERS) as pool: futures = { pool.submit( self._select_for_scene, s, idx + 1, system, user_template, schema, max_n ): idx for idx, s in enumerate(scenes_to_select) } for fut in as_completed(futures): result = fut.result() if result: llm_selected.append(result) all_results = auto_selected + llm_selected all_results.sort(key=lambda x: x["scene_index"]) # v4: 에피소드 최대 shot 수 제한 — 비례 분배 from app.core.config import settings self._apply_episode_max_shots(all_results, settings.episode_max_shots) total_selected = sum(r["selected_count"] for r in all_results) total_shots = sum(r["total_shots"] for r in all_results) logger.info("shot_selection: %d/%d shots selected across %d scenes", total_selected, total_shots, len(all_results)) return { "completed_count": len(all_results), "applicable_count": len(scenes), "failed_count": 0, "data": { "scenes": all_results, "total_selected": total_selected, "total_shots": total_shots, "max_per_scene": max_n, }, # E2E13 Codex HIGH-3: step-local hash persist — resume 비교 동일 method "config_hash": self._config_hash(), } @staticmethod def _apply_episode_max_shots(results: List[Dict[str, Any]], ep_max: int) -> None: """에피소드 총 shot 수를 ep_max 이하로 비례 삭감. 0이면 비활성.""" if ep_max <= 0: return total_before = sum(r["selected_count"] for r in results) if total_before <= ep_max: return # selected_shots를 single source로 하고 selected_shot_indices는 파생 — 두 리스트 순서 불일치 방지 def _sync_from_shots(r: Dict[str, Any]) -> None: r["selected_shot_indices"] = [s["shot_index"] for s in r["selected_shots"]] r["selected_count"] = len(r["selected_shots"]) # 1단계: 비례 할당 (최소 1개 보장) for r in results: quota = max(1, round(ep_max * r["selected_count"] / total_before)) if r["selected_count"] > quota: r["selected_shots"] = r["selected_shots"][:quota] _sync_from_shots(r) # 2단계: 아직 초과하면 많은 씬에서 1개씩 삭감 while sum(r["selected_count"] for r in results) > ep_max: max_r = max( (r for r in results if r["selected_count"] > 0), key=lambda r: r["selected_count"], default=None, ) if not max_r: break max_r["selected_shots"].pop() _sync_from_shots(max_r) final_total = sum(r["selected_count"] for r in results) if final_total > ep_max: logger.warning( "episode_max_shots=%d could not be met → final %d", ep_max, final_total, ) else: logger.info("episode_max_shots=%d applied: %d → %d", ep_max, total_before, final_total) def _select_for_scene(self, scene_data, call_idx, system, user_template, schema, absolute_max): scene_idx = scene_data["scene_index"] shots = scene_data.get("shots", []) total = len(shots) half_cap = _half_cap(total) effective_max = _effective_max(total, absolute_max) shots_block = "\n".join( f"Shot {sh['shot_index']} (beat:{sh.get('based_on_beat', 0)}): {sh['description']}" for sh in shots ) user_prompt = user_template.format( scene_index=scene_idx, scene_heading=scene_data.get("scene_heading", ""), max_n=absolute_max, total_shots=total, half_cap=half_cap, shots_block=shots_block, ) try: result = call_structured( step="shot_selection", system_prompt=system, user_prompt=user_prompt, response_schema=schema, project_config=self.project_config, schema_name=f"shot_selection_{call_idx}", opik_metadata=self.build_opik_metadata(), ) # v4 우선: selected_shots. v3 legacy fallback: selected_shot_indices raw = result.get("selected_shots") if not raw: legacy = result.get("selected_shot_indices") or [] raw = [{"shot_index": i, "reason": ""} for i in legacy] valid_indices = {sh["shot_index"] for sh in shots} seen = set() deduped: List[Dict[str, Any]] = [] for item in raw: if not isinstance(item, dict): continue idx = item.get("shot_index") if idx in valid_indices and idx not in seen: reason = item.get("reason") or "" deduped.append({"shot_index": idx, "reason": reason}) seen.add(idx) # 2중 캡: effective_max 초과분 절단 selected = deduped[:effective_max] if not selected: logger.warning( "shot_selection scene %d: LLM empty → fallback to first shot", scene_idx, ) selected = [{ "shot_index": shots[0]["shot_index"], "reason": "fallback: LLM 결과 없음", }] logger.info( "shot_selection scene %d: %d/%d selected (eff_max=%d)", scene_idx, len(selected), total, effective_max, ) return { "scene_index": scene_idx, "selected_shot_indices": [s["shot_index"] for s in selected], "selected_shots": selected, "total_shots": total, "selected_count": len(selected), } except Exception as exc: logger.warning( "shot_selection scene %d FAILED: %s → fallback to first %d", scene_idx, exc, effective_max, ) first = shots[:effective_max] fallback_sel = [ {"shot_index": s["shot_index"], "reason": f"fallback: {exc}"} for s in first ] return { "scene_index": scene_idx, "selected_shot_indices": [s["shot_index"] for s in first], "selected_shots": fallback_sel, "total_shots": total, "selected_count": len(first), }