"""Beat 추출 + Shot 추출 단계. v4: 원본 씬 → beat(상태 변화) → shot(스틸컷) 추출. """ import logging import time from concurrent.futures import ThreadPoolExecutor, as_completed from typing import Any, Dict, List, Optional 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__) BUNDLE_TARGET = 3000 REF_MAX = 2000 MAX_WORKERS = 4 PREV_SHOTS_MAX = 0 # 0 = 무제한 (이전 shot 전부 전달) class _BeatShotMixin: """Beat/Shot 공통 — segments, 인물 리스트 로드.""" def _load_prev_checkpoint(self, step_id: str) -> Optional[Dict]: import json as _json from app.core.config import settings from pathlib import Path as _Path 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 _load_fulltext(self) -> str: from app.models.project import Episode from sqlalchemy.orm import undefer ep = ( self.db.query(Episode) .options(undefer(Episode.fulltext)) .filter(Episode.id == self.episode_id) .first() ) if not ep or not ep.fulltext: from app.core.errors import AppError raise AppError(code="step.no_fulltext", message="시나리오 텍스트가 없습니다.", status_code=400) return ep.fulltext def _load_scene_texts(self) -> List[Dict]: """scene_save 체크포인트에서 segments → 씬 텍스트 리스트.""" save_cp = self._load_prev_checkpoint("scene_save") if not save_cp or not save_cp.get("data", {}).get("segments"): from app.core.errors import AppError raise AppError(code="step.no_scenes", message="scene_save 결과 없음", status_code=400) segments = save_cp["data"]["segments"] # 구 체크포인트 fallback: text 필드 없으면 fulltext 슬라이스 fulltext = "" if any(not seg.get("text") for seg in segments): fulltext = self._load_fulltext() logger.warning("beat_shot: %d segments missing text, using fulltext fallback", sum(1 for s in segments if not s.get("text"))) scene_texts = [] for seg in segments: text = seg.get("text") or fulltext[seg.get("start_char", 0):seg.get("end_char", 0)] scene_texts.append({ "idx": seg["scene_index"], "heading": seg.get("heading", ""), "text": text, "length": seg.get("length", len(text)), }) return scene_texts def _load_character_list(self) -> List[str]: """entity_character_list 체크포인트에서 인물 이름 목록 로드.""" cp = self._load_prev_checkpoint("entity_character_list") if not cp or not cp.get("data", {}).get("characters"): return [] return [c["name"] for c in cp["data"]["characters"]] # ───────────────────────────────────────────────────────── # Beat Extract # ───────────────────────────────────────────────────────── class BeatExtractStep(_BeatShotMixin, StepRunner): """Beat 추출 — 원본 씬에서 인물 상태 변화 최소 단위 추출.""" def _execute(self, mode="resume") -> Dict[str, Any]: scene_texts = self._load_scene_texts() # visual_world_rules → director_notes (물리적 존재 판단 기준) rules_cp = self._load_prev_checkpoint("visual_world_rules") visual_rules = "" if rules_cp and rules_cp.get("data"): rules_data = rules_cp["data"] notes = rules_data.get("director_notes", []) if notes: visual_rules = "\n[물리적 존재 판단 기준 — 인물 언급 시 카메라에 보이는 바디 기준으로 기술]\n" + "\n".join(f"- {n}" for n in notes) else: rules_lines = [] for r in rules_data.get("rules", []): if r.get("rule_type") in ("possession", "projection", "ghost"): rules_lines.append(f"[{r.get('rule_type', '')}] {r.get('visual_guideline', '')}") if rules_lines: visual_rules = "\n[물리적 존재 판단 기준]\n" + "\n".join(f"- {l}" for l in rules_lines) if not visual_rules: logger.info("beat_extract: visual_world_rules에 director_notes 없음 — 물리적 존재 기준 미적용") # 인물 리스트 (entity_character_list 체크포인트) character_names = self._load_character_list() if character_names: char_block = "\n[인물 목록 — 인물 언급 시 이 목록의 이름을 사용]\n" + "\n".join(f"- {n}" for n in character_names) + "\n" if visual_rules: visual_rules = visual_rules + "\n" + char_block else: visual_rules = char_block logger.info("beat_extract: %d characters injected", len(character_names)) system = load_prompt("beat_extract", "system", db=self.db) user_template = load_prompt("beat_extract", "user", db=self.db) schema = load_schema("beat_extract", "beat_schema", db=self.db) bundles = self._build_bundles(scene_texts) logger.info("beat_extract: %d scenes → %d bundles", len(scene_texts), len(bundles)) results_by_idx = {} with ThreadPoolExecutor(max_workers=MAX_WORKERS) as pool: futures = { pool.submit( self._call_bundle, b, i + 1, system, user_template, schema, visual_rules ): i + 1 for i, b in enumerate(bundles) } for fut in as_completed(futures): idx, scenes, err = fut.result() results_by_idx[idx] = scenes all_scenes = [] for i in sorted(results_by_idx): all_scenes.extend(results_by_idx[i]) total_beats = sum(len(s.get("beats", [])) for s in all_scenes) completed = min(len(all_scenes), len(scene_texts)) logger.info("beat_extract: %d scenes, %d beats", completed, total_beats) return { "completed_count": completed, "applicable_count": len(scene_texts), "failed_count": max(0, len(scene_texts) - completed), "data": {"scenes": all_scenes, "total_beats": total_beats}, } def _build_bundles(self, scene_texts: List[Dict]) -> List[Dict]: bundles = [] i = 0 while i < len(scene_texts): bundle = [] bundle_len = 0 while i < len(scene_texts) and bundle_len + scene_texts[i]["length"] <= BUNDLE_TARGET: bundle.append(scene_texts[i]) bundle_len += scene_texts[i]["length"] i += 1 if not bundle and i < len(scene_texts): bundle.append(scene_texts[i]) bundle_len = scene_texts[i]["length"] i += 1 # 앞쪽 참조 (REF_MAX) — scene_index 기반 조회 ref_parts = [] ref_len = 0 scene_by_idx = {s["idx"]: s for s in scene_texts} j = bundle[0]["idx"] - 1 while j >= 0: ref_scene = scene_by_idx.get(j) if not ref_scene or ref_len + ref_scene["length"] > REF_MAX: break ref_parts.insert(0, ref_scene["text"]) ref_len += ref_scene["length"] j -= 1 bundles.append({ "scenes": bundle, "bundle_len": bundle_len, "ref_text": "".join(ref_parts) if ref_parts else "", }) return bundles def _call_bundle(self, bundle_info, call_idx, system, user_template, schema, visual_rules=""): import copy bundle = bundle_info["scenes"] ref_text = bundle_info["ref_text"] expected_indices = [s["idx"] for s in bundle] ref_section = "" if ref_text: ref_section = f"[앞쪽 씬 — 참조만, 이 씬들의 beat는 추출하지 마세요]\n{ref_text}\n\n" bundle_text = "\n".join( f"--- Scene {s['idx']}: {s['heading']} ---\n{s['text']}" for s in bundle ) if visual_rules: bundle_text = visual_rules + "\n\n" + bundle_text user_prompt = user_template.format( ref_section=ref_section, bundle_text=bundle_text, ) # 스키마에 scene_index enum 주입 call_schema = copy.deepcopy(schema) call_schema["properties"]["scenes"]["items"]["properties"]["scene_index"] = { "type": "integer", "enum": expected_indices, } scene_range = f"{bundle[0]['idx']}~{bundle[-1]['idx']}" max_retry = 5 for attempt in range(max_retry + 1): try: result = call_structured( step="beat_extract", system_prompt=system, user_prompt=user_prompt, response_schema=call_schema, project_config=self.project_config, schema_name=f"beat_extract_{call_idx}", opik_metadata=self.build_opik_metadata(), ) scenes = result.get("scenes", []) # 검증: 반환된 scene_index가 expected와 일치하는지 returned_indices = [s.get("scene_index") for s in scenes] if set(returned_indices) != set(expected_indices): missing = set(expected_indices) - set(returned_indices) extra = set(returned_indices) - set(expected_indices) # fallback: 갯수 일치하면 순서 매핑 if len(scenes) == len(expected_indices) and not extra: logger.warning( "beat_extract call %d (scenes %s): index mismatch, remapping by order. missing=%s", call_idx, scene_range, missing, ) for i, s in enumerate(scenes): s["scene_index"] = expected_indices[i] elif attempt < max_retry: logger.warning( "beat_extract call %d (scenes %s): count mismatch (got %d, expected %d) — retry %d/%d", call_idx, scene_range, len(scenes), len(expected_indices), attempt + 1, max_retry, ) time.sleep(2) continue else: # 최종: 있는 것만 사용 + 빠진 씬은 빈 beats로 채움 logger.warning( "beat_extract call %d (scenes %s): filling missing scenes %s with empty beats", call_idx, scene_range, missing, ) returned_set = set(returned_indices) for mi in expected_indices: if mi not in returned_set: scenes.append({"scene_index": mi, "scene_heading": "", "beats": []}) total_beats = sum(len(s.get("beats", [])) for s in scenes) logger.info("beat_extract call %d (scenes %s): %d scenes, %d beats", call_idx, scene_range, len(scenes), total_beats) return call_idx, scenes, None except Exception as exc: if attempt < max_retry: logger.warning("beat_extract call %d (scenes %s) retry %d/%d: %s", call_idx, scene_range, attempt + 1, max_retry, exc) time.sleep(2) continue logger.warning("beat_extract call %d (scenes %s) FAILED after %d retries: %s", call_idx, scene_range, max_retry, exc) return call_idx, [], str(exc) # ───────────────────────────────────────────────────────── # Shot Extract # ───────────────────────────────────────────────────────── class ShotExtractStep(_BeatShotMixin, StepRunner): """Shot 추출 — beat 기반으로 씬별 스틸컷 추출.""" def _config_hash(self) -> str: """E2E13 Codex HIGH-3(재리뷰 HIGH-2): 실제 로드되는 effective prompt(system/user/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_extract", "system", db=self.db), load_prompt("shot_extract", "user", db=self.db), ): h.update(text.encode("utf-8")) h.update(b"\x00") h.update(_json.dumps( load_schema("shot_extract", "shot_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 _execute(self, mode="resume") -> Dict[str, Any]: scene_texts = self._load_scene_texts() # beat 결과 로드 beat_cp = self._load_prev_checkpoint("beat_extract") if not beat_cp or not beat_cp.get("data", {}).get("scenes"): from app.core.errors import AppError raise AppError(code="step.no_beats", message="beat_extract 결과 없음", status_code=400) beats_data = beat_cp["data"]["scenes"] # visual_world_rules → director_notes (물리적 존재 판단 기준) rules_cp = self._load_prev_checkpoint("visual_world_rules") visual_rules = "" if rules_cp and rules_cp.get("data"): rules_data = rules_cp["data"] notes = rules_data.get("director_notes", []) if notes: visual_rules = "\n[물리적 존재 판단 기준 — description에는 카메라에 보이는 인물/바디만 묘사]\n" + "\n".join(f"- {n}" for n in notes) else: # fallback: possession/projection 관련 rules에서 추출 rules_lines = [] for r in rules_data.get("rules", []): if r.get("rule_type") in ("possession", "projection", "ghost"): rules_lines.append(f"[{r.get('rule_type', '')}] {r.get('visual_guideline', '')}") if rules_lines: visual_rules = "\n[물리적 존재 판단 기준]\n" + "\n".join(f"- {l}" for l in rules_lines) if not visual_rules: logger.info("shot_extract: visual_world_rules에 director_notes 없음 — 물리적 존재 기준 미적용") # visual_world_rules → t2i_context (시각 컨텍스트 요약) t2i_context = "" if rules_cp and rules_cp.get("data"): _ctx = rules_cp["data"].get("t2i_context", "") if _ctx: t2i_context = f"\n[시각적 배경 — 시나리오를 이미지화하는 작업입니다. description 묘사 시 아래 맥락을 반영하세요]\n{_ctx}\n" logger.info("shot_extract: t2i_context injected (%d chars)", len(_ctx)) if t2i_context: visual_rules = t2i_context + visual_rules # 인물 리스트 (entity_character_list 체크포인트) character_names = self._load_character_list() if character_names: logger.info("shot_extract: %d characters injected as enum", len(character_names)) system = load_prompt("shot_extract", "system", db=self.db) user_template = load_prompt("shot_extract", "user", db=self.db) schema = load_schema("shot_extract", "shot_schema", db=self.db) # 제작자 정정 채널 (wave3) — 정정 없으면 빈 문자열 = byte-identical from app.core.creator_corrections import project_corrections_block system += project_corrections_block(self.project_id) # scene_index 기반 beat 매칭 beats_by_scene_idx = {} for bd in beats_data: idx = bd.get("scene_index") if idx is not None: beats_by_scene_idx[idx] = bd.get("beats", []) bundles = self._build_bundles(scene_texts, beats_by_scene_idx) logger.info("shot_extract: %d scenes → %d bundles (sequential)", len(scene_texts), len(bundles)) # 순차 실행 — 이전 번들의 shot 결과를 누적하여 다음 번들에 전달 all_scenes = [] prev_shots_ctx = "" for i, bundle in enumerate(bundles): call_idx = i + 1 idx, scenes, err = self._call_bundle( bundle, call_idx, system, user_template, schema, visual_rules, prev_shots_ctx, character_names ) if err: logger.warning("shot_extract: bundle %d failed — %s", call_idx, err) all_scenes.extend(scenes) # 누적 컨텍스트 — 전체 처리된 shot 결과에서 최근 부분 유지 prev_shots_ctx = self._build_prev_shots_context(all_scenes) # 빈 shots 씬 경고 empty_shot_scenes = [s["scene_index"] for s in all_scenes if not s.get("shots")] if empty_shot_scenes: logger.warning("shot_extract: %d scenes with 0 shots: %s", len(empty_shot_scenes), empty_shot_scenes) total_shots = sum(len(s.get("shots", [])) for s in all_scenes) completed = min(len(all_scenes), len(scene_texts)) logger.info("shot_extract: %d scenes, %d shots", completed, total_shots) return { "completed_count": completed, "applicable_count": len(scene_texts), "failed_count": max(0, len(scene_texts) - completed), "data": {"scenes": all_scenes, "total_shots": total_shots}, # E2E13 Codex HIGH-3: step-local hash persist — resume 비교와 # 동일 method (프롬프트 팩 교체=CP 무효화) "config_hash": self._config_hash(), } def _build_bundles(self, scene_texts: List[Dict], beats_by_scene_idx: Dict[int, List]) -> List[Dict]: bundles = [] i = 0 while i < len(scene_texts): bundle = [] bundle_len = 0 while i < len(scene_texts) and bundle_len + scene_texts[i]["length"] <= BUNDLE_TARGET: st = scene_texts[i] bundle.append({**st, "beats": beats_by_scene_idx.get(st["idx"], [])}) bundle_len += st["length"] i += 1 if not bundle and i < len(scene_texts): st = scene_texts[i] bundle.append({**st, "beats": beats_by_scene_idx.get(st["idx"], [])}) bundle_len = st["length"] i += 1 bundles.append({"scenes": bundle, "bundle_len": bundle_len}) return bundles def _build_prev_shots_context(self, scenes: List[Dict]) -> str: """처리된 shot 결과 → 다음 번들 컨텍스트 (전체 전달, truncation 없음).""" lines = [] for s in scenes: si = s.get("scene_index", "?") for sh in s.get("shots", []): desc = sh.get("description", "") lines.append(f" 씬{si} Shot{sh.get('shot_index', '?')}: {desc}") if not lines: return "" if PREV_SHOTS_MAX > 0: # 글자 수 제한 시 최근 shot 우선 kept = [] total = 0 for line in reversed(lines): needed = len(line) + (1 if kept else 0) if total + needed > PREV_SHOTS_MAX: break kept.insert(0, line) total += needed if len(kept) < len(lines): return " ...\n" + "\n".join(kept) return "\n".join(kept) return "\n".join(lines) def _format_beats(self, beats): if not beats: return " (beat 없음 — 상황 묘사 기반으로 Shot 추출)" lines = [] for b in beats: lines.append(f" Beat {b['beat_index']}: [{b['change_type']}] " f"{b['before_state']} → {b['after_state']}") return "\n".join(lines) def _call_bundle(self, bundle_info, call_idx, system, user_template, schema, visual_rules="", prev_shots_ctx="", character_names=None): import copy bundle = bundle_info["scenes"] expected_indices = [s["idx"] for s in bundle] # 이전 번들 shot 결과 → ref_section ref_section = "" if prev_shots_ctx: ref_section = ( "[이전 씬의 Shot 결과 — 참조만, 인물 호칭·존재 일관성 유지]\n" + prev_shots_ctx + "\n\n" ) # 인물 리스트 섹션 character_list_section = "" if character_names: character_list_section = ( "[인물 목록 — characters에는 반드시 이 목록의 이름만 사용]\n" + "\n".join(f"- {n}" for n in character_names) + "\n\n" ) scenes_parts = [] for s in bundle: beat_text = self._format_beats(s["beats"]) scenes_parts.append( f"--- Scene {s['idx']}: {s['heading']} ---\n" f"[Beats]\n{beat_text}\n\n" f"[원문]\n{s['text']}" ) scenes_section = "\n\n".join(scenes_parts) if visual_rules: scenes_section = visual_rules + "\n\n" + scenes_section # 구 프롬프트 호환: {character_list_section} 없으면 무시 if "{character_list_section}" in user_template: user_prompt = user_template.format( ref_section=ref_section, scenes_section=scenes_section, character_list_section=character_list_section, ) else: user_prompt = user_template.format( ref_section=ref_section, scenes_section=scenes_section, ) if character_list_section: user_prompt = character_list_section + user_prompt # 스키마에 scene_index enum 주입 call_schema = copy.deepcopy(schema) call_schema["properties"]["scenes"]["items"]["properties"]["scene_index"] = { "type": "integer", "enum": expected_indices, } # 스키마에 characters enum 주입 (인물 목록에서만 선택 강제) if character_names: shot_props = call_schema["properties"]["scenes"]["items"]["properties"].get("shots", {}).get("items", {}).get("properties", {}) if "characters" in shot_props: shot_props["characters"]["items"] = { "type": "string", "enum": character_names, } scene_range = f"{bundle[0]['idx']}~{bundle[-1]['idx']}" max_retry = 5 for attempt in range(max_retry + 1): try: result = call_structured( step="shot_extract", system_prompt=system, user_prompt=user_prompt, response_schema=call_schema, project_config=self.project_config, schema_name=f"shot_extract_{call_idx}", opik_metadata=self.build_opik_metadata(), ) scenes = result.get("scenes", []) # 검증: scene_index 일치 확인 returned_indices = [s.get("scene_index") for s in scenes] if set(returned_indices) != set(expected_indices): missing = set(expected_indices) - set(returned_indices) if len(scenes) == len(expected_indices): logger.warning( "shot_extract call %d (scenes %s): index mismatch, remapping by order. missing=%s", call_idx, scene_range, missing, ) for i, s in enumerate(scenes): s["scene_index"] = expected_indices[i] elif attempt < max_retry: logger.warning( "shot_extract call %d (scenes %s): count mismatch (got %d, expected %d) — retry %d/%d", call_idx, scene_range, len(scenes), len(expected_indices), attempt + 1, max_retry, ) time.sleep(2) continue else: returned_set = set(returned_indices) for mi in expected_indices: if mi not in returned_set: scenes.append({"scene_index": mi, "scene_heading": "", "shots": []}) total_shots = sum(len(s.get("shots", [])) for s in scenes) logger.info("shot_extract call %d (scenes %s): %d scenes, %d shots", call_idx, scene_range, len(scenes), total_shots) return call_idx, scenes, None except Exception as exc: if attempt < max_retry: logger.warning("shot_extract call %d (scenes %s) retry %d/%d: %s", call_idx, scene_range, attempt + 1, max_retry, exc) time.sleep(2) continue logger.warning("shot_extract call %d (scenes %s) FAILED after %d retries: %s", call_idx, scene_range, max_retry, exc) return call_idx, [], str(exc)