"""location_consistency StepRunner — 씬 간 location 외형 고정 문장 추출. scene 이미지 생성은 현재 location 참조 이미지를 주입하지 않으므로, scene_detail이 t2i_prompt에 쓰는 `[L##: 설명]` 블록이 유일한 location 시각 정보다. 이 step은 각 L##의 "씬 간 동일해야 할 외형 문장"을 미리 확정하여 scene_detail이 그대로 삽입하도록 강제한다. 환경 상태(날씨/시간/조명)는 제외. """ import logging from concurrent.futures import ThreadPoolExecutor, as_completed from typing import Any, Dict, List from app.core.errors import AppError 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__) MAX_WORKERS = 4 class LocationConsistencyStep(StepRunner): """Location별 씬 간 고정 외형 문장 추출 (scene_detail 직전).""" def _load_prev_checkpoint(self, step_id: str): import json as _json from pathlib import Path as _Path from app.core.config import settings as _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]: # ── 의존 체크포인트 로드 ── entity_merge_cp = self._load_prev_checkpoint("entity_merge") if not entity_merge_cp or not entity_merge_cp.get("data"): raise AppError( code="step.no_input", message="entity_merge 결과 없음", status_code=400, ) entity_detail_cp = self._load_prev_checkpoint("entity_detail") entity_details = {} if entity_detail_cp and entity_detail_cp.get("data"): entity_details = entity_detail_cp["data"].get("entity_details", {}) or {} scene_save_cp = self._load_prev_checkpoint("scene_save") segments: List[dict] = [] if scene_save_cp and scene_save_cp.get("data"): segments = scene_save_cp["data"].get("segments", []) or [] # ── Location 목록 구성 (entity_merge가 source of truth) ── merge_locations = entity_merge_cp["data"].get("locations", []) or [] if not merge_locations: logger.info("location_consistency: no locations in entity_merge — nothing to do") return { "completed_count": 0, "applicable_count": 0, "failed_count": 0, "data": {"locations": []}, } # ── 등장 씬 매핑 ── # 1순위: scene_director의 present_entity_ids로 ID 기반 매칭 (오염 無). # 2순위 (fallback): 이름 substring — "강" vs "강가" 오염 방지 위해 최소 2자 + 단어 경계. scenes_by_location: Dict[str, List[int]] = {} director_cp = self._load_prev_checkpoint("scene_director") id_based: Dict[int, List[str]] = {} if director_cp and director_cp.get("data", {}).get("scenes"): for ds in director_cp["data"]["scenes"]: si = ds.get("scene_index") if si is not None: id_based[si] = ds.get("present_entity_ids", []) or [] for loc in merge_locations: sid = loc.get("short_id", "") name = loc.get("name", "") if not sid: continue scenes_by_location[sid] = [] if id_based: # ID 기반 매핑 (권장 경로) for si, present in id_based.items(): if sid in present: scenes_by_location[sid].append(si) elif name and len(name) >= 2: # Fallback: 이름 단어 경계 매칭 (heading 우선, heading 없으면 text) for seg in segments: si = seg.get("scene_index") if si is None: continue heading = seg.get("heading", "") or "" # heading에 location 이름이 명확히 등장하면 매칭 # (heading은 "L##. 항구 - 낮" 같은 정형 구조) if name in heading: scenes_by_location[sid].append(si) scenes_by_location[sid].sort() # ── 프롬프트 로드 ── system_prompt = load_prompt("location_consistency", "system") schema = load_schema("location_consistency", "location_schema") # ── resume: 기존 성공 결과 보존 ── existing_cp = self._load_prev_checkpoint("location_consistency") existing_ok: Dict[str, dict] = {} if mode == "resume" and existing_cp and existing_cp.get("data", {}).get("locations"): for loc in existing_cp["data"]["locations"]: lid = loc.get("location_id") if not lid: continue desc = loc.get("fixed_visual_description", "") or "" # 실패는 빈 description 또는 analysis_summary="실패"로 표현 if desc and not loc.get("analysis_summary", "").startswith("실패"): existing_ok[lid] = loc if existing_ok: logger.info( "location_consistency resume: %d locations already done, skipping", len(existing_ok), ) # ── location별 병렬 LLM 호출 ── location_results: List[dict] = [] processed = 0 failed = 0 to_process: List[dict] = [] for loc in merge_locations: sid = loc.get("short_id", "") if not sid: continue if sid in existing_ok: location_results.append(existing_ok[sid]) continue to_process.append(loc) if to_process: with ThreadPoolExecutor(max_workers=MAX_WORKERS) as pool: futures = { pool.submit( self._process_location, loc, entity_details, scenes_by_location, segments, system_prompt, schema, ): loc for loc in to_process } for fut in as_completed(futures): orig_loc = futures[fut] try: result = fut.result() except Exception as fut_exc: # _process_location 내부에서 잡지 못한 예외 방어 (e.g. 환경 오류) logger.error( "location_consistency %s future raised: %s", orig_loc.get("short_id", "?"), fut_exc, ) result = { "location_id": orig_loc.get("short_id", ""), "name": orig_loc.get("name", ""), "fixed_visual_description": "", "analysis_summary": f"실패 — future 예외: {fut_exc}", } if not result: continue location_results.append(result) # analysis_summary가 "실패" prefix면 항상 failed 집계 (Codex Critical 1). # fallback의 entity_detail description은 한국어·조명/분위기 포함이라 # scene_detail에 주입하면 외형 고정 목적이 깨진다. summary = result.get("analysis_summary", "") or "" if summary.startswith("실패"): failed += 1 elif result.get("fixed_visual_description"): processed += 1 else: failed += 1 # 정렬 (L01, L02, ...) location_results.sort(key=lambda r: r.get("location_id", "")) total = processed + failed + len(existing_ok) logger.info( "location_consistency: %d processed, %d failed, %d cached (total=%d)", processed, failed, len(existing_ok), total, ) return { "completed_count": processed + len(existing_ok), "applicable_count": total, "failed_count": failed, "data": {"locations": location_results}, } def _process_location( self, loc: dict, entity_details: Dict[str, dict], scenes_by_location: Dict[str, List[int]], segments: List[dict], system_prompt: str, schema: dict, ) -> dict: sid = loc.get("short_id", "") name = loc.get("name", "") # entity_detail 조회. key는 주로 "name:location"이지만, LLM이 entity_type을 # "location" 외 값으로 반환한 경우(예: "배경"/"장소") 또는 key가 name만인 경우를 # 위해 3단계 lookup: "name:location" → "name" → name만 포함 key 중 첫 location-like. detail_key = f"{name}:location" detail = entity_details.get(detail_key) if not detail: detail = entity_details.get(name) if not detail and name: # 마지막 fallback: name으로 시작하고 ":"이 들어간 key 중 location 힌트 가진 것 for k, v in entity_details.items(): if k.startswith(f"{name}:") and isinstance(v, dict): detail = v break detail = detail or {} base_description = detail.get("description", "") if isinstance(detail, dict) else "" visual_traits = detail.get("visual_traits", []) if isinstance(detail, dict) else [] # 등장 씬 텍스트 (최대 3개만 — 맥락 이해 충분) appear_scene_indices = scenes_by_location.get(sid, []) seg_map = {seg.get("scene_index"): seg for seg in segments} # 씬 텍스트는 절대 자르지 않는다 (CLAUDE.md 절대 규칙). # 샘플 수만 제한해서 토큰을 관리한다. scene_samples = [] for si in appear_scene_indices[:3]: seg = seg_map.get(si, {}) heading = seg.get("heading", "") or "" text = seg.get("text", "") or "" scene_samples.append((si, heading, text)) # User prompt 구성 user_prompt = f"[Location {sid}]\n이름: {name}\n\n" if base_description: user_prompt += f"기존 설명 (entity_detail):\n{base_description}\n\n" if visual_traits: user_prompt += f"시각 특성 (entity_detail):\n- " + "\n- ".join(visual_traits) + "\n\n" if scene_samples: user_prompt += f"등장 씬 텍스트 (맥락 참고용, {len(scene_samples)}개 샘플):\n" for si, heading, text in scene_samples: user_prompt += f"\n--- 씬 {si} {('[' + heading + ']') if heading else ''} ---\n" user_prompt += text user_prompt += "\n" user_prompt += "\n" user_prompt += ( "[지시]\n" "위 정보를 바탕으로 이 location의 **씬 간 동일하게 유지되어야 할 외형만** 추출하여 " "fixed_visual_description을 영어로 작성하세요.\n" "- 포함: 크기·형태·재질·색상·구조적 디테일·고정 소품\n" "- 제외: 날씨·시간·조명·대기 상태·인물·차량·감정·분위기·순간적 상태\n" "- 3~5 문장, 40~80 단어\n" "- 엔티티 ID(C##/L##/P##) 사용 금지, 고유명사 사용 금지\n" "- T2I 프롬프트에 직접 삽입 가능한 구체적 묘사\n" ) # 글로벌 3-tier fallback (call_structured 내부) — gemini → sanitize → gpt 자동. # P2-3: validate_response로 빈 locations 배열을 semantic empty로 분류하여 # Tier 2/3 진행 보장. 이전엔 valid JSON `{"locations": []}` 가 Tier 1 통과로 # 간주되어 글로벌 fallback이 trigger 안 됐다. def _validate_locations(resp: Dict[str, Any]) -> bool: """semantic 검증: locations 배열에 최소 1개 요소 + fixed_visual_description 존재.""" locs = resp.get("locations") if isinstance(resp, dict) else None if not locs or not isinstance(locs, list): return False first = locs[0] if not isinstance(first, dict): return False desc = first.get("fixed_visual_description", "") or "" return bool(desc.strip()) try: result = call_structured( step="location_consistency", system_prompt=system_prompt, user_prompt=user_prompt, response_schema=schema, project_config=self.project_config, schema_name=f"location_consistency_{sid}", opik_metadata=self.build_opik_metadata(extra_metadata={"location_id": sid}), validate_response=_validate_locations, ) # schema는 locations 배열을 기대 — 실제 반환에서 첫 항목 추출 locations = result.get("locations", []) if locations: first = locations[0] # location_id·name은 호출 입력으로 강제 (LLM이 잘못 쓰는 경우 방지) first["location_id"] = sid first["name"] = name desc = first.get("fixed_visual_description", "") or "" logger.info( "location_consistency %s (%s): %d words", sid, name, len(desc.split()), ) return first raise ValueError("locations 배열이 비어있음") except Exception as exc: logger.error("location_consistency %s all tiers failed: %s", sid, exc) return { "location_id": sid, "name": name, "fixed_visual_description": base_description or "", "analysis_summary": "실패 — 기존 entity_detail description 사용", }