"""감독 Phase StepRunner -- scene_director, scene_cinematography, scene_dependency.""" import json import logging from pathlib import Path from typing import Any, Dict, List, Optional from app.core.name_matcher import build_name_index, lookup_name from app.core.step_runner import StepRunner from app.modules.llm.llm_client import call_structured logger = logging.getLogger(__name__) class _DirectorStepMixin: """감독 단계 공통 -- fulltext 로드, 이전 단계 결과 로드.""" def _opik_meta(self, extra_tags: list = None) -> Dict: if hasattr(self, "build_opik_metadata"): return self.build_opik_metadata(extra_tags) return {} def _load_prev_checkpoint(self, step_id: str) -> Optional[Dict]: 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 _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_cleaned_text(self) -> str: """text_cleanup 결과 사용, 없으면 원본 fulltext.""" cp = self._load_prev_checkpoint("text_cleanup") if cp and cp.get("data", {}).get("cleaned_text"): return cp["data"]["cleaned_text"] return self._load_fulltext() class SceneDirectorStep(_DirectorStepMixin, StepRunner): """Step 12: 씬 감독 (V/A/H 3분류).""" def _execute(self, mode="resume") -> Dict[str, Any]: # Load segments (scene_save > scene_segmentation) — text 필드 포함 save_cp = self._load_prev_checkpoint("scene_save") seg_cp = self._load_prev_checkpoint("scene_segmentation") segments = ( (save_cp or {}).get("data", {}).get("segments") or (seg_cp or {}).get("data", {}).get("segments") or [] ) # Load entities + inject short_ids from DB (mid-pipeline sync 이후) t2i_cp = self._load_prev_checkpoint("entity_t2i") entities = t2i_cp.get("data", {}) if t2i_cp else {} try: from app.modules.short_id import build_short_id_info sid_info = build_short_id_info(self.db, self.project_id, self.episode_id) if sid_info: # name_matcher: 원본 + 괄호·공백 정규화 둘 다 키로 (드리프트 대응) _sid_items = list(sid_info.items()) # [(short_id, entity_dict), ...] name_to_short = build_name_index( _sid_items, key_fn=lambda it: it[1].get("name", ""), value_fn=lambda it: it[0], ) injected, missing = 0, [] for etype in ["characters", "locations", "props"]: for e in entities.get(etype, []): if not e.get("short_id"): sid = lookup_name(name_to_short, e.get("name", "")) or "" if sid: e["short_id"] = sid injected += 1 else: missing.append(e.get("name", "?")) logger.info("Injected %d short_ids from DB", injected) if missing: logger.warning("short_id 미매칭 %d개: %s", len(missing), missing[:5]) else: logger.warning("DB에 short_id 없음 — scene_director가 이름으로 fallback") except Exception as exc: logger.warning("short_id injection failed (non-fatal): %s", exc) # 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".join(f"- {n}" for n in notes) else: # fallback: rules에서 possession 관련만 추출 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".join(rules_lines) # v4: selected shot 컨텍스트 추가 (있으면, 토큰 절약을 위해 selected만) sel_cp = self._load_prev_checkpoint("shot_selection") shot_cp = self._load_prev_checkpoint("shot_validator") if sel_cp and shot_cp and shot_cp.get("data", {}).get("scenes"): selected_map = {} for s in sel_cp.get("data", {}).get("scenes", []): selected_map[s["scene_index"]] = set(s.get("selected_shot_indices", [])) shot_context_lines = [] for s in shot_cp["data"]["scenes"]: si = s["scene_index"] sel_indices = selected_map.get(si, set()) for sh in s.get("shots", []): if not sel_indices or sh["shot_index"] in sel_indices: chars = ", ".join(sh.get("characters", [])) shot_context_lines.append( f"S{si} Shot{sh['shot_index']}: {sh['description']} [{chars}]" ) if shot_context_lines: visual_rules += "\n\n[Shot 분석 참고]\n" + "\n".join(shot_context_lines) # 기획서 컨텍스트 보강 (줄거리 + 톤) from app.core.planning_doc_context import get_planning_context pctx = get_planning_context(self.project_id, self.episode_id, self.db) visual_rules += pctx.inject_if_available("story_arc", "## 기획서: 전체 줄거리") visual_rules += pctx.inject_if_available("tone_mood", "## 기획서: 톤/분위기") from app.modules.pipeline.scene_director_v2 import direct_scenes result = direct_scenes( segments, entities=entities, visual_rules=visual_rules, project_config=self.project_config, opik_metadata=self.build_opik_metadata(), ) scenes = result.get("scenes", []) # entity 필터링은 entity_filter 단계에서 처리 — director는 판정만 수행 # failed_count 는 파생값으로 둔다 — 하드코딩 0 이던 시절, LLM 이 입력 # 113개에 출력 116개를 돌려줘도 completed 로 넘어갔다(2026-08-04 실측). # direct_scenes 의 파리티 게이트가 먼저 raise 하므로 정상 경로에서는 # 항상 0 이지만, 게이트를 우회하는 경로가 생겨도 여기서 드러난다. return { "completed_count": len(scenes), "applicable_count": len(segments), "failed_count": max(0, len(segments) - len(scenes)), "data": result, } # -- Step 13: scene_cinematography -- class SceneCinematographyStep(_DirectorStepMixin, StepRunner): """촬영 감독 -- 전체 씬 일괄로 촬영 기법 N가지 선택.""" def _execute(self, mode="resume") -> Dict[str, Any]: seg_prev = self._load_prev_checkpoint("scene_save") if not seg_prev: seg_prev = self._load_prev_checkpoint("scene_segmentation") if not seg_prev or not seg_prev.get("data"): from app.core.errors import AppError raise AppError( code="step.no_input", message="세그먼트 결과 없음", status_code=400, ) segments = seg_prev["data"]["segments"] from app.modules.prompt_loader import load_prompt, load_schema from app.core.config import settings from sqlalchemy import text as sql_text shot_count = settings.scene_variation_count system_prompt = load_prompt( "scene_cinematography", "system", db=self.db, shot_count=shot_count, ) # analyze는 shot_types_block, scenes_block 등이 있으므로 raw로 로드 후 별도 format analyze_template = load_prompt( "scene_cinematography", "analyze", db=self.db, ) schema = load_schema("scene_cinematography", "analyze_schema", db=self.db) # DB에서 활성 샷 타입 로드 shot_rows = self.db.execute( sql_text( "SELECT name, category, description FROM shot_type " "WHERE is_active = true ORDER BY sort_order" ) ).fetchall() shot_types_block = "\n".join( f"- {r[0]} [{r[1]}]: {r[2]}" for r in shot_rows ) # 전체 씬 (전문) scenes_block_items = [] for seg in segments: scene_text = seg.get("text", "") scenes_block_items.append( f"씬 {seg['scene_index']}: {seg['heading']}\n{scene_text}" ) scenes_block = "\n\n".join(scenes_block_items) user_prompt = analyze_template.format( shot_types_block=shot_types_block, scenes_block=scenes_block, shot_count=shot_count, ) logger.info( "Scene cinematography: %d scenes, %d shot types, %d variations", len(segments), len(shot_rows), shot_count, ) result = call_structured( step="scene_cinematography", system_prompt=system_prompt, user_prompt=user_prompt, response_schema=schema, project_config=self.project_config, schema_name="scene_cinematography", opik_metadata=self._opik_meta(), ) scenes = result.get("scenes", []) # shot_count 강제 (LLM이 초과/미달 시) for sc in scenes: shots = sc.get("shots", []) if len(shots) > shot_count: sc["shots"] = shots[:shot_count] elif len(shots) < shot_count: logger.warning("Scene %d: %d shots returned, expected %d", sc.get("scene_index"), len(shots), shot_count) return { "completed_count": len(scenes), "applicable_count": len(segments), "failed_count": len(segments) - len(scenes), "data": {"scenes": scenes, "shot_count": shot_count}, } # -- Step 14: scene_dependency v2 -- class SceneDependencyStep(_DirectorStepMixin, StepRunner): """씬 연관 분석 v2 -- 배경 중심 + 인물 중심 분리.""" def _execute(self, mode="resume") -> Dict[str, Any]: seg_prev = self._load_prev_checkpoint("scene_save") if not seg_prev: seg_prev = self._load_prev_checkpoint("scene_segmentation") if not seg_prev or not seg_prev.get("data"): from app.core.errors import AppError raise AppError( code="step.no_input", message="세그먼트 결과 없음", status_code=400, ) segments = seg_prev["data"]["segments"] # scene_director 결과 로드 director_prev = self._load_prev_checkpoint("scene_director") director_result = director_prev.get("data", {}) if director_prev else {} # entity 결과 로드 entity_prev = self._load_prev_checkpoint("entity_t2i") if not entity_prev: entity_prev = self._load_prev_checkpoint("entity_detail_batch") entities = entity_prev.get("data", {}) if entity_prev else {} from app.modules.pipeline.scene_dependency_v2 import extract_dependencies result = extract_dependencies( segments=segments, director_result=director_result, entities=entities, project_config=self.project_config, opik_metadata=self._opik_meta(), ) deps = result.get("dependencies", []) return { "completed_count": len(deps), "applicable_count": len(segments), "failed_count": 0, "data": {"dependencies": deps}, } DIRECTOR_STEP_CLASSES = { "scene_director": SceneDirectorStep, "scene_cinematography": SceneCinematographyStep, "scene_dependency": SceneDependencyStep, }