"""요약 Phase StepRunner — episode_summary, visual_world_rules, scene_summary, project_summary.""" import json import logging from typing import Any, Dict, List, Optional from app.core.step_runner import StepRunner logger = logging.getLogger(__name__) class _SummaryStepMixin: 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 결과가 있으면 사용, 없으면 원본.""" 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() def _load_prev_checkpoint(self, step_id: str) -> Optional[Dict]: from pathlib import 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 class EpisodeSummaryStep(_SummaryStepMixin, StepRunner): """Step 3: 에피소드 요약 (500자).""" def _execute(self, mode="resume") -> Dict[str, Any]: fulltext = self._load_cleaned_text() from app.core.config import settings from app.modules.pipeline.episode_summarizer import summarize_episode result = summarize_episode( fulltext=fulltext, max_length=settings.max_episode_summary_length, project_config=self.project_config, opik_metadata=self.build_opik_metadata(), ) # episode.summary 에 저장 from app.models.project import Episode ep = self.db.query(Episode).filter(Episode.id == self.episode_id).first() if ep: ep.summary = result.get("summary", "") self.db.commit() return { "completed_count": 1, "applicable_count": 1, "failed_count": 0, "data": result, } class VisualWorldRulesStep(_SummaryStepMixin, StepRunner): """Step 4: 시각적 세계관 규칙 추출. 입력: - text_cleanup 체크포인트 (정리된 텍스트) - episode_summary 체크포인트 (에피소드 요약) 출력: - rules: 시각적 규칙 목록 - era: 시대 배경 - region: 지역/국가 배경 """ def _execute(self, mode: str = "resume") -> Dict[str, Any]: # 1. 정리된 텍스트 로드 fulltext = self._load_cleaned_text() # 2. 에피소드 요약 로드 episode_summary = "" ep_cp = self._load_prev_checkpoint("episode_summary") if ep_cp and ep_cp.get("data", {}).get("summary"): episode_summary = ep_cp["data"]["summary"] # 3. 기획서 컨텍스트 보강 from app.core.planning_doc_context import get_planning_context pctx = get_planning_context(self.project_id, self.episode_id, self.db) planning_supplement = ( pctx.inject_if_available("world_setting", "## 기획서: 세계관") + pctx.inject_if_available("tone_mood", "## 기획서: 톤/분위기") + pctx.inject_if_available("visual_concepts", "## 기획서: 비주얼 컨셉") ) # 4. LLM 호출 from app.modules.pipeline.visual_world_rules import extract_visual_rules result = extract_visual_rules( fulltext=fulltext, episode_summary=episode_summary + planning_supplement, project_config=self.project_config, opik_metadata=self.build_opik_metadata(extra_tags=["visual_world_rules"]), ) # ★★선언된 좌표는 **그 글자 그대로** — 모델이 원고를 다시 읽어 바꾸지 못한다 # (Codex BLOCK 2026-09-02: 「1960년대」선언이 「20세기 전반, 일제강점기」로 변질). from app.modules.pipeline.grounding_coordinates import apply_declared_coordinates result = apply_declared_coordinates(result, self.project_config) rules_count = len(result.get("rules", [])) era = result.get("era", "") region = result.get("region", "") logger.info( "VisualWorldRulesStep: %d rules, era=%s, region=%s, planning_doc=%s", rules_count, era, region, pctx.has_planning_doc, ) return { "completed_count": 1, "applicable_count": 1, "failed_count": 0, "data": result, # ★좌표 계약 판을 지문에 접지 **않는다** (Codex NON-BLOCK 을 실측으로 거부, 2026-09-02 밤): # 선언이 생기면 project_config 해시가 이미 움직여 옛 CP 는 어긋난다. 판까지 접으면 # 선언 없는 모든 프로젝트의 기존 CP 가 재개마다 drift→BLOCK 이다(로그 실측). } class SceneSummaryStep(_SummaryStepMixin, StepRunner): """Step 7: 씬별 요약 (병렬 ThreadPool). 입력: - scene_save 체크포인트 (segments) - text_cleanup 체크포인트 (정리된 텍스트) - episode_summary 체크포인트 (에피소드 요약) - visual_world_rules 체크포인트 (시각적 규칙) 출력: - scene_summaries: [{scene_index, scene_summary}, ...] """ def _execute(self, mode: str = "resume") -> Dict[str, Any]: # 1. 씬 세그먼트 로드 (scene_save 체크포인트) scene_cp = self._load_prev_checkpoint("scene_save") if not scene_cp or not scene_cp.get("data", {}).get("segments"): from app.core.errors import AppError raise AppError( code="step.no_input", message="씬 저장 결과 없음 (scene_save 체크포인트 필요)", status_code=400, ) segments: List[Dict] = scene_cp["data"]["segments"] # 2. 에피소드 요약 로드 episode_summary = "" ep_cp = self._load_prev_checkpoint("episode_summary") if ep_cp and ep_cp.get("data", {}).get("summary"): episode_summary = ep_cp["data"]["summary"] # 3. 시각적 세계관 규칙 로드 visual_rules = "" vw_cp = self._load_prev_checkpoint("visual_world_rules") if vw_cp and vw_cp.get("data"): rules_list = vw_cp["data"].get("rules", []) if rules_list: visual_rules = "\n".join( f"- [{r.get('rule_type', '')}] {r.get('description', '')}" for r in rules_list ) # 4. 병렬 요약 호출 from app.modules.pipeline.scene_summarizer import summarize_scenes scene_summaries = summarize_scenes( segments=segments, episode_summary=episode_summary, visual_rules=visual_rules, project_config=self.project_config, opik_metadata=self.build_opik_metadata(extra_tags=["scene_summary"]), ) completed = sum(1 for s in scene_summaries if s.get("scene_summary")) total = len(segments) failed = total - completed logger.info( "SceneSummaryStep: %d/%d scenes summarized (%d failed)", completed, total, failed, ) return { "completed_count": completed, "applicable_count": total, "failed_count": failed, "data": {"summaries": scene_summaries}, }