"""아웃룩 Phase StepRunner — outlook_phase1/2/3 (분할), outlook_dedup.""" import json import logging from typing import Any, Dict, List, Optional from app.core.step_runner import StepRunner logger = logging.getLogger(__name__) class _OutlookStepMixin: """아웃룩 단계 공통 — fulltext 로드, 이전 단계 결과 로드.""" 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_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 def _load_characters_and_scene_map(self): """scene_director 기반 캐릭터 목록 + 씬별 캐릭터 맵 로드.""" director_prev = self._load_prev_checkpoint("scene_director") director_scenes = [] characters = [] if director_prev and director_prev.get("data"): director_scenes = director_prev["data"].get("scenes", []) from app.modules.short_id import build_short_id_info sid_info = build_short_id_info(self.db, self.project_id, self.episode_id) for sid, info in sorted(sid_info.items()): if info["type"] == "character": characters.append({"short_id": sid, "name": info["name"]}) if not characters: entity_prev = self._load_prev_checkpoint("entity_t2i") if entity_prev and entity_prev.get("data"): for c in entity_prev["data"].get("characters", []): characters.append({"short_id": c.get("short_id", c["name"]), "name": c["name"]}) scene_char_map: Dict[int, list] = {} for ds in director_scenes: si = ds.get("scene_index") chars = [sid for sid in ds.get("present_entity_ids", []) if sid.startswith("C")] scene_char_map[si] = chars return characters, scene_char_map, director_scenes def _load_segments(self): seg_prev = self._load_prev_checkpoint("scene_save") if not seg_prev or not seg_prev.get("data", {}).get("segments"): seg_prev = self._load_prev_checkpoint("scene_segmentation") if not seg_prev or not seg_prev.get("data", {}).get("segments"): from app.core.errors import AppError raise AppError(code="step.no_input", message="씬 세그먼테이션 결과 없음", status_code=400) return seg_prev["data"]["segments"] def _load_visual_rules(self) -> str: vwr_cp = self._load_prev_checkpoint("visual_world_rules") vwr_data = vwr_cp.get("data", {}) if vwr_cp else {} visual_rules = "" if vwr_data.get("era") or vwr_data.get("region"): visual_rules = f"시대: {vwr_data.get('era', '')}\n지역: {vwr_data.get('region', '')}" for r in vwr_data.get("rules", []): if r.get("rule_type") == "costume": visual_rules += f"\n의상: {r.get('description', '')}" t2i_ctx = vwr_data.get("t2i_context", "") if t2i_ctx: visual_rules += f"\n\n[T2I 시각 컨텍스트]\n{t2i_ctx}" return visual_rules # ── Phase 1: 아웃룩 목록 추출 + orphan retry ── class OutlookPhase1Step(_OutlookStepMixin, StepRunner): """아웃룩 Phase1 — 캐릭터별 아웃룩 목록 추출 + orphan retry.""" def _execute(self, mode="resume") -> Dict[str, Any]: segments = self._load_segments() characters, scene_char_map, _ = self._load_characters_and_scene_map() visual_rules = self._load_visual_rules() opik_meta = self.build_opik_metadata() from app.modules.pipeline.outlook_extractor_v2 import extract_outlooks_phase1 logger.info("Outlook phase1: %d scenes, %d characters", len(segments), len(characters)) phase1_result = extract_outlooks_phase1( segments=segments, characters=characters, scene_character_map=scene_char_map, visual_rules=visual_rules, project_config=self.project_config, opik_metadata=opik_meta, ) phase1_outlooks = phase1_result.get("outlooks", []) # short_id 부여 + character_id 정규화 _char_name_to_sid = {c.get("name", ""): c.get("short_id", "") for c in characters} _char_sid_set = {c.get("short_id", "") for c in characters} def _normalize_char_id(cid): if cid in _char_sid_set: return cid return _char_name_to_sid.get(cid, cid) for i, ol in enumerate(phase1_outlooks): ol["short_id"] = f"O{i + 1:02d}" if ol.get("character_id"): ol["character_id"] = _normalize_char_id(ol["character_id"]) logger.info("Outlook phase1 complete: %d outlooks (O01~O%02d)", len(phase1_outlooks), len(phase1_outlooks)) # Orphan retry (최대 2회) active_char_sids = {cid for chars in scene_char_map.values() for cid in chars} for p1_retry in range(1, 3): chars_with_outlook = {ol.get("character_id", "") for ol in phase1_outlooks} orphan_chars = [c for c in characters if c.get("short_id", "") in active_char_sids and c.get("short_id", "") not in chars_with_outlook] if not orphan_chars: break logger.warning( "Outlook phase1 retry %d/2: %d orphan characters — %s", p1_retry, len(orphan_chars), [c.get("short_id", "") + "(" + c.get("name", "") + ")" for c in orphan_chars], ) orphan_char_map = {si: [cid for cid in chars if cid in {c["short_id"] for c in orphan_chars}] for si, chars in scene_char_map.items()} orphan_char_map = {si: chars for si, chars in orphan_char_map.items() if chars} retry1_result = extract_outlooks_phase1( segments=[seg for seg in segments if seg.get("scene_index") in orphan_char_map], characters=orphan_chars, scene_character_map=orphan_char_map, visual_rules=visual_rules, project_config=self.project_config, opik_metadata=opik_meta, ) new_outlooks = retry1_result.get("outlooks", []) if not new_outlooks: logger.warning("Outlook phase1 retry %d: LLM returned 0 outlooks, stopping", p1_retry) break next_num = len(phase1_outlooks) + 1 for ol in new_outlooks: ol["short_id"] = f"O{next_num:02d}" if ol.get("character_id"): ol["character_id"] = _normalize_char_id(ol["character_id"]) next_num += 1 phase1_outlooks.extend(new_outlooks) logger.info("Outlook phase1 retry %d: +%d outlooks (total %d)", p1_retry, len(new_outlooks), len(phase1_outlooks)) # ── O00 (Null Outlook) 자동 할당 ── # 소스 1) LLM phase1 결과의 non_humanoid_characters # 소스 2) entity_relation visual_similarity=false 변형 캐릭터 null_outlook_chars = set() # 1) LLM이 판별한 비인간형 for cid in phase1_result.get("non_humanoid_characters", []): if cid in active_char_sids: null_outlook_chars.add(cid) # 2) entity_relation visual_similarity=false rel_cp = self._load_prev_checkpoint("entity_relation") if rel_cp and rel_cp.get("data", {}).get("relations"): for rel in rel_cp["data"]["relations"]: if not rel.get("visual_similarity", True): var_sid = rel.get("variant_short_id", "") if var_sid and var_sid in active_char_sids: null_outlook_chars.add(var_sid) # orphan 캐릭터 경고 (O00 할당 안 함 — LLM 미스 가능) chars_with_outlook = {ol.get("character_id", "") for ol in phase1_outlooks} orphan_chars = [ c.get("short_id", "") for c in characters if c.get("short_id", "") in active_char_sids and c.get("short_id", "") not in chars_with_outlook and c.get("short_id", "") not in null_outlook_chars ] if orphan_chars: logger.warning( "Outlook phase1: %d orphan characters without outlook (not O00): %s", len(orphan_chars), orphan_chars, ) if null_outlook_chars: # LLM이 비인간형 캐릭터에 아웃룩을 할당했으면 제거 removed = [ol.get("short_id") for ol in phase1_outlooks if ol.get("character_id") in null_outlook_chars] phase1_outlooks = [ol for ol in phase1_outlooks if ol.get("character_id") not in null_outlook_chars] if removed: logger.info("O00: removed LLM-assigned outlooks for non-human chars: %s", removed) # O00 추가 phase1_outlooks.append({ "short_id": "O00", "name": "Null Outlook", "description": "Non-human or non-outfit entity. Uses entity reference image as-is.", "character_id": None, "is_null": True, }) logger.info( "O00 (Null Outlook) assigned to %d characters: %s", len(null_outlook_chars), sorted(null_outlook_chars), ) return { "completed_count": len(phase1_outlooks), "applicable_count": len(characters), "failed_count": 0, "data": { "outlooks": phase1_outlooks, "null_outlook_chars": sorted(null_outlook_chars), }, } # ── Phase 2: 씬별 매핑 + retry ── class OutlookPhase2Step(_OutlookStepMixin, StepRunner): """아웃룩 Phase2 — 씬별 캐릭터→아웃룩 매핑 + 누락 retry.""" def _execute(self, mode="resume") -> Dict[str, Any]: segments = self._load_segments() characters, scene_char_map, _ = self._load_characters_and_scene_map() opik_meta = self.build_opik_metadata() # Phase1 결과 로드 p1_cp = self._load_prev_checkpoint("outlook_phase1") if not p1_cp or not p1_cp.get("data", {}).get("outlooks"): from app.core.errors import AppError raise AppError(code="step.no_input", message="outlook_phase1 결과 없음", status_code=400) outlooks = p1_cp["data"]["outlooks"] null_outlook_chars = set(p1_cp["data"].get("null_outlook_chars", [])) # O00을 LLM에 보내지 않음 (LLM이 O00을 정상 아웃룩으로 오인 방지) regular_outlooks = [ol for ol in outlooks if ol.get("short_id") != "O00" and not ol.get("is_null")] # null_outlook_chars 캐릭터도 LLM 입력에서 제외 regular_chars = [c for c in characters if c.get("short_id", "") not in null_outlook_chars] regular_scene_map = { si: [cid for cid in chars if cid not in null_outlook_chars] for si, chars in scene_char_map.items() } regular_scene_map = {si: chars for si, chars in regular_scene_map.items() if chars} from app.modules.pipeline.outlook_extractor_v2 import extract_outlooks_phase2 logger.info("Outlook phase2: mapping %d outlooks to %d scenes (O00: %d chars excluded)", len(regular_outlooks), len(segments), len(null_outlook_chars)) phase2_result = extract_outlooks_phase2( segments=segments, outlooks=regular_outlooks, characters=regular_chars, scene_character_map=regular_scene_map, project_config=self.project_config, opik_metadata=opik_meta, ) assignments = phase2_result.get("scene_assignments", []) logger.info("Outlook phase2 complete: %d scene assignments", len(assignments)) # 누락 retry (최대 3회) MAX_RETRY = 3 for retry_round in range(1, MAX_RETRY + 1): scenes_with_chars = {si for si, chars in scene_char_map.items() if chars} assigned_scenes = {sa["scene_index"] for sa in assignments if sa.get("assignments")} missing_scenes = scenes_with_chars - assigned_scenes for sa in assignments: si = sa["scene_index"] expected = set(scene_char_map.get(si, [])) assigned = {a.get("character_id", "") for a in sa.get("assignments", [])} if expected - assigned: missing_scenes.add(si) if not missing_scenes: break logger.warning("Outlook phase2 retry %d/%d: %d씬 누락", retry_round, MAX_RETRY, len(missing_scenes)) missing_segments = [seg for seg in segments if seg.get("scene_index") in missing_scenes] if not missing_segments: break missing_char_map = {si: chars for si, chars in scene_char_map.items() if si in missing_scenes} missing_char_ids = {cid for chars in missing_char_map.values() for cid in chars} missing_characters = [c for c in characters if c.get("short_id", c.get("name")) in missing_char_ids] retry_result = extract_outlooks_phase2( segments=missing_segments, outlooks=outlooks, characters=missing_characters, scene_character_map=missing_char_map, project_config=self.project_config, opik_metadata=opik_meta, ) retry_assignments = retry_result.get("scene_assignments", []) for sa in retry_assignments: if not sa.get("assignments"): continue si_match = sa["scene_index"] existing_sa = next((s for s in assignments if s["scene_index"] == si_match), None) if not existing_sa or not existing_sa.get("assignments"): assignments = [s for s in assignments if s["scene_index"] != si_match] assignments.append(sa) else: existing_chars = {a.get("character_id", "") for a in existing_sa["assignments"]} for a in sa["assignments"]: if a.get("character_id", "") not in existing_chars: existing_sa["assignments"].append(a) filled = sum(1 for sa in retry_assignments if sa.get("assignments")) logger.info("Outlook phase2 retry %d: %d/%d 복구", retry_round, filled, len(missing_scenes)) if filled == 0: logger.warning("Outlook phase2 retry %d: 복구 0건", retry_round) # ── O00 자동 주입: null_outlook_chars가 등장하는 모든 씬에 O00 할당 ── if null_outlook_chars: assignments_map: Dict[int, Dict] = {sa["scene_index"]: sa for sa in assignments} injected_count = 0 for si, chars in scene_char_map.items(): null_in_scene = [cid for cid in chars if cid in null_outlook_chars] if not null_in_scene: continue sa = assignments_map.get(si) if not sa: sa = {"scene_index": si, "assignments": []} assignments.append(sa) assignments_map[si] = sa # LLM이 비인간형 캐릭터에 정상 아웃룩을 할당했을 수 있으므로 강제 교체 for cid in null_in_scene: sa["assignments"] = [ a for a in sa["assignments"] if a.get("character_id") != cid ] sa["assignments"].append({"character_id": cid, "outlook_id": "O00"}) injected_count += 1 if injected_count: logger.info("O00 injected into %d scene-character slots", injected_count) return { "completed_count": len(assignments), "applicable_count": len(segments), "failed_count": 0, "data": { "outlooks": outlooks, "scene_assignments": assignments, "null_outlook_chars": sorted(null_outlook_chars), }, } # ── Phase 3: 병합 판별 + 코드 정리 ── class OutlookPhase3Step(_OutlookStepMixin, StepRunner): """아웃룩 Phase3 — LLM 병합 판별 + 코드에서 정리.""" def _execute(self, mode="resume") -> Dict[str, Any]: # Phase2 결과 로드 p2_cp = self._load_prev_checkpoint("outlook_phase2") if not p2_cp or not p2_cp.get("data"): from app.core.errors import AppError raise AppError(code="step.no_input", message="outlook_phase2 결과 없음", status_code=400) outlooks = p2_cp["data"]["outlooks"] assignments = p2_cp["data"]["scene_assignments"] null_outlook_chars = set(p2_cp["data"].get("null_outlook_chars", [])) # O00을 LLM 입력에서 제외 (보호) o00_outlook = None regular_outlooks = [] for ol in outlooks: if ol.get("short_id") == "O00" or ol.get("is_null"): o00_outlook = ol else: regular_outlooks.append(ol) _, scene_char_map, director_scenes = self._load_characters_and_scene_map() character_routes: Dict[str, list] = {} for ds in director_scenes: si = ds.get("scene_index") for sid in ds.get("present_entity_ids", []): if sid.startswith("C"): character_routes.setdefault(sid, []).append(si) summary_cp = self._load_prev_checkpoint("scene_summary") scene_summaries: Dict[int, str] = {} if summary_cp: for ss in summary_cp.get("data", {}).get("summaries", []): scene_summaries[ss.get("scene_index", 0)] = ss.get("scene_summary", "") from app.modules.pipeline.outlook_extractor_v2 import extract_outlooks_phase3 # O00 제외한 일반 아웃룩만 LLM에 전달 logger.info("Outlook phase3: cleaning %d outlooks, %d assignments (O00 protected)", len(regular_outlooks), len(assignments)) phase3_result = extract_outlooks_phase3( outlooks=regular_outlooks, scene_assignments=assignments, scene_summaries=scene_summaries, character_routes=character_routes, project_config=self.project_config, opik_metadata=self.build_opik_metadata(), ) cleaned_outlooks = phase3_result.get("cleaned_outlooks", []) cleaned_assignments = phase3_result.get("cleaned_assignments", []) removed = phase3_result.get("removed", []) # merge_map 적용 (retry 결과에도) merge_map = {r["outlook_id"]: r["merge_into"] for r in removed if r.get("merge_into")} if merge_map: for sa in cleaned_assignments: for a in sa.get("assignments", []): oid = a.get("outlook_id", "") if oid in merge_map: a["outlook_id"] = merge_map[oid] logger.info("Phase3 merge applied: %s", merge_map) # O00을 결과에 복원 if o00_outlook: cleaned_outlooks.append(o00_outlook) logger.info("O00 (Null Outlook) preserved in phase3 result") logger.info("Outlook phase3 complete: %d outlooks (removed %d)", len(cleaned_outlooks), len(removed)) return { "completed_count": len(cleaned_outlooks), "applicable_count": 1, "failed_count": 0, "data": { "outlooks": cleaned_outlooks, "scene_assignments": cleaned_assignments, "removed": removed, "null_outlook_chars": sorted(null_outlook_chars), }, } # ── Legacy wrapper (하위 호환) ── class OutlookExtractionStep(_OutlookStepMixin, StepRunner): """Legacy: outlook_extraction → phase1+2+3 순차 실행.""" def _execute(self, mode="resume") -> Dict[str, Any]: # Phase1 p1 = OutlookPhase1Step.__new__(OutlookPhase1Step) p1.__dict__.update(self.__dict__) p1_result = p1._execute(mode) self.save_checkpoint({"status": "running", "data": p1_result["data"]}) # Phase2 # phase1 체크포인트를 outlook_phase1로 저장 (phase2가 읽을 수 있도록) from pathlib import Path from app.core.config import settings p1_cp_dir = Path(settings.projects_dir) / self.project_id / "checkpoints" / "episodes" / self.episode_id / "outlook_phase1" p1_cp_dir.mkdir(parents=True, exist_ok=True) (p1_cp_dir / "manifest.json").write_text( json.dumps({"status": "completed", "data": p1_result["data"]}, ensure_ascii=False, indent=2), encoding="utf-8", ) p2 = OutlookPhase2Step.__new__(OutlookPhase2Step) p2.__dict__.update(self.__dict__) p2_result = p2._execute(mode) # phase2 체크포인트 저장 p2_cp_dir = Path(settings.projects_dir) / self.project_id / "checkpoints" / "episodes" / self.episode_id / "outlook_phase2" p2_cp_dir.mkdir(parents=True, exist_ok=True) (p2_cp_dir / "manifest.json").write_text( json.dumps({"status": "completed", "data": p2_result["data"]}, ensure_ascii=False, indent=2), encoding="utf-8", ) p3 = OutlookPhase3Step.__new__(OutlookPhase3Step) p3.__dict__.update(self.__dict__) p3_result = p3._execute(mode) return p3_result # ── Outlook Dedup ── class OutlookDedupStep(_OutlookStepMixin, StepRunner): """아웃룩 중복 판별 — 기존 outlook_dedup 로직 재사용.""" def _execute(self, mode="resume") -> Dict[str, Any]: from app.modules.pipeline.outlook_dedup import ( find_duplicate_outlooks, apply_outlook_merge, ) # outlook_phase3 또는 outlook_extraction 체크포인트 로드 ol_cp = self._load_prev_checkpoint("outlook_phase3") if not ol_cp or ol_cp.get("status") != "completed": ol_cp = self._load_prev_checkpoint("outlook_extraction") if not ol_cp or ol_cp.get("status") != "completed": return { "completed_count": 0, "applicable_count": 0, "failed_count": 0, "data": {"message": "outlook phase3/extraction not completed"}, } outlooks = ol_cp.get("data", {}).get("outlooks", []) # O00 (Null Outlook)은 dedup 대상에서 제외 regular_outlooks = [ol for ol in outlooks if ol.get("short_id") != "O00" and not ol.get("is_null")] if len(regular_outlooks) < 2: return { "completed_count": 0, "applicable_count": 0, "failed_count": 0, "data": {"message": "too few outlooks to dedup"}, } duplicates = find_duplicate_outlooks(regular_outlooks, project_config=self.project_config) merge_count = len(duplicates) if isinstance(duplicates, list) else 0 if merge_count > 0: apply_outlook_merge(self.db, self.project_id, duplicates) logger.info("Outlook dedup: merged %d groups", merge_count) else: logger.info("Outlook dedup: no duplicates found") return { "completed_count": 1, "applicable_count": 1, "failed_count": 0, "data": {"merge_groups": merge_count, "duplicates": duplicates}, }