"""씬 visible_entities 검증기 — 분석 후 재검증 단계. 각 씬의 visible_entities가 실제로 물리적으로 보이는 대상인지 LLM으로 재확인하고, 비물리적 엔티티를 제거한다. """ import json import logging import time from concurrent.futures import ThreadPoolExecutor, as_completed from typing import Any, Dict, List, Optional from app.modules.llm.llm_client import call_structured, _resolve_model logger = logging.getLogger(__name__) # 검증 단계 전용 — 메인 분석과 다른 모델로 교차 검증 VERIFY_STEP = "scene_verify" VERIFY_SCHEMA = { "type": "object", "properties": { "scene_index": {"type": "integer"}, "verified_entities": { "type": "array", "items": { "type": "object", "properties": { "entity_name": {"type": "string"}, "entity_type": {"type": "string"}, "physically_visible": {"type": "boolean", "description": "카메라에 물리적으로 보이는가"}, "reason": {"type": "string", "description": "판단 이유 (한 줄)"}, }, "required": ["entity_name", "entity_type", "physically_visible", "reason"], "additionalProperties": False, }, }, }, "required": ["scene_index", "verified_entities"], "additionalProperties": False, } def _get_cross_model_config(project_config: Optional[Dict]) -> Dict: """메인 분석 모델과 다른 provider의 모델로 교차 검증. scene_detail이 GPT면 → 검증은 Gemini Pro scene_detail이 Gemini면 → 검증은 GPT """ main_model = _resolve_model("scene_detail", project_config) if main_model.startswith("gpt"): cross = {"scene_verify": {"model": "gemini-pro"}} else: cross = {"scene_verify": {"model": "gpt"}} # project_config 위에 교차 모델 오버라이드 merged = dict(project_config or {}) merged.update(cross) logger.info("Cross-model verification: main=%s → verify=%s", main_model, cross["scene_verify"]["model"]) return merged def validate_visible_entities( scenes: List[Dict[str, Any]], fulltext: str, segments: List[Dict[str, Any]], project_config: Optional[Dict] = None, max_workers: int = 5, ) -> List[Dict[str, Any]]: """씬별 visible_entities를 재검증하고 비물리적 엔티티를 제거. Args: scenes: 씬 분석 결과 리스트 (scene_index, visible_entities, ...) fulltext: 시나리오 전문 segments: 세그먼트 리스트 (start_char, end_char) project_config: LLM 모델 설정 Returns: 수정된 scenes 리스트 (visible_entities에서 비물리적 엔티티 제거됨) """ seg_map = {s["scene_index"]: s for s in segments} scenes_to_verify = [] for scene in scenes: vis = scene.get("visible_entities", []) chars = [v for v in vis if v.get("entity_type") == "character"] scene_type = scene.get("scene_type", "normal") # 캐릭터 2명 이상 또는 voiceover 씬에 캐릭터가 있으면 검증 if len(chars) >= 2 or (scene_type == "voiceover" and len(chars) >= 1): scenes_to_verify.append(scene) if not scenes_to_verify: logger.info("Visible entity validation: no scenes need verification") return scenes logger.info("Visible entity validation: %d scenes to verify", len(scenes_to_verify)) cross_config = _get_cross_model_config(project_config) def _verify_one(scene: Dict) -> Dict: scene = dict(scene) # shallow copy — 스레드 안전 si = scene.get("scene_index", 0) seg = seg_map.get(si) if not seg: return scene scene_text = seg.get("text") or fulltext[seg.get("start_char", 0):seg.get("end_char", 0)] vis = scene.get("visible_entities", []) entity_list = "\n".join( f"- {v.get('short_id', '')} {v.get('entity_name', '?')} ({v.get('entity_type', '?')})" if v.get('short_id') else f"- {v.get('entity_name', '?')} ({v.get('entity_type', '?')})" for v in vis ) # 앞쪽 연관 씬 정보 prev_idx = scene.get("dependent_scene_index", -1) prev_context = "" if prev_idx > 0: prev_seg = seg_map.get(prev_idx) if prev_seg: prev_text = prev_seg.get("text") or fulltext[prev_seg.get("start_char", 0):prev_seg.get("end_char", 0)] prev_context = f"\n\n[앞쪽 연관 씬 #{prev_idx} 일부]\n{prev_text}" scene_type = scene.get("scene_type", "normal") type_note = "" if scene_type in ("flashback", "dream", "montage"): type_note = f"\n참고: 이 씬은 '{scene_type}' 유형입니다. {scene_type} 내에서 물리적으로 보이는 인물은 visible로 판단하세요." prompt = f"""씬 #{si}의 visible_entities가 실제로 물리적으로 보이는지 검증하세요. 씬 유형: {scene_type}{type_note} [씬 텍스트] {scene_text} {prev_context} [현재 visible_entities] {entity_list} 각 엔티티에 대해 판단하세요: - physically_visible = true: 이 장소에 몸이 있고 카메라로 찍을 수 있음 - physically_visible = false: 목소리만 들림, 원격 조종자, 상상/기억 속 인물, 다른 장소에 있음 주의: - 대사에 이름이 나와도, 실제로 그 인물의 몸이 이 장소에 없으면 false - "()" 형태 표현 → visible_body 의 몸만 현장에 있고, remote_identity 는 원격에서 조종 중이므로 remote_identity 는 false - 전화/무전/방송으로만 등장하는 인물은 false""" try: result = call_structured( step=VERIFY_STEP, # 크로스 모델로 검증 system_prompt="영화 촬영 감독. 각 씬에서 카메라에 실제로 보이는 인물/물체만 식별한다.", user_prompt=prompt, response_schema=VERIFY_SCHEMA, project_config=cross_config, schema_name="entity_verify", ) verified = result.get("verified_entities", []) # 검증 결과를 원본 엔티티 기준으로 매칭 (entity_name 또는 short_id) verified_map = {v["entity_name"]: v for v in verified} removed = [] kept = [] for orig in vis: orig_name = orig.get("entity_name", "") orig_sid = orig.get("short_id", "") v = verified_map.get(orig_name) or verified_map.get(orig_sid) if v and not v.get("physically_visible", True): removed.append(v) else: kept_item = {"entity_name": orig_name, "entity_type": orig.get("entity_type", "")} if orig_sid: kept_item["short_id"] = orig_sid kept.append(kept_item) if removed: removed_log = [f"{v['entity_name']}({v.get('reason', '')})" for v in removed] logger.info("Scene %d: removed non-physical entities: %s", si, removed_log) scene["visible_entities"] = kept _clean_t2i_markers(scene, removed) return scene except Exception as exc: logger.warning("Scene %d verification failed, keeping original: %s", si, exc) return scene # 병렬 검증 results_map = {} with ThreadPoolExecutor(max_workers=max_workers) as executor: futures = {} for scene in scenes_to_verify: if futures: time.sleep(1) futures[executor.submit(_verify_one, scene)] = scene["scene_index"] for future in as_completed(futures): si = futures[future] try: result = future.result() results_map[si] = result except Exception as exc: logger.error("Verification worker failed for scene %d: %s", si, exc) # 결과 병합 verified_count = 0 for i, scene in enumerate(scenes): si = scene.get("scene_index", 0) if si in results_map: scenes[i] = results_map[si] verified_count += 1 logger.info("Visible entity validation complete: %d scenes verified", verified_count) return scenes def _clean_t2i_markers(scene: Dict, removed_entities: List[Dict]) -> None: """제거된 엔티티의 마커를 T2I 프롬프트에서 정리 — short_id(C01O02) + 레거시([[]]).""" import re removed_names = {e["entity_name"] for e in removed_entities if e.get("entity_name")} removed_sids = {e.get("short_id", "") for e in removed_entities if e.get("short_id")} def _clean_prompt(prompt: str) -> str: if not prompt: return prompt # short_id 패턴 제거: C01O02 for sid in removed_sids: if sid.startswith("C"): prompt = re.sub(rf'{re.escape(sid)}O\d{{2,3}}', '', prompt) else: prompt = re.sub(rf'(?