"""Shot 연관 분석 — 코드 기반 엔티티 오버랩 스코어링. 각 selected shot에 대해 같은 배경(location)의 이전 shot 중 교집합(공통 엔티티) 최대 + 여집합(현재에 없는 엔티티) 최소인 것을 선택. LLM 호출 없이 shot_director VE + scene_director location 데이터로 계산. """ import json import logging from pathlib import Path from typing import Any, Dict, List from app.core.name_matcher import build_name_index, lookup_name from app.core.step_runner import StepRunner logger = logging.getLogger(__name__) class ShotDependencyStep(StepRunner): """Shot 연관 분석 — 배경 기준 + 엔티티 오버랩 스코어링.""" def _load_prev_checkpoint(self, step_id: str): 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 _execute(self, mode="resume") -> Dict[str, Any]: # 데이터 로드 shot_cp = self._load_prev_checkpoint("shot_validator") if not shot_cp or not shot_cp.get("data", {}).get("scenes"): from app.core.errors import AppError raise AppError(code="step.no_shots", message="shot_extract 결과 없음", status_code=400) from app.core.steps.shot_validator_step import assert_no_failed_scenes assert_no_failed_scenes(shot_cp, self.project_config, consumer_step="shot_dependency") sel_cp = self._load_prev_checkpoint("shot_selection") selected_map: Dict[int, set] = {} if sel_cp and sel_cp.get("data", {}).get("scenes"): for s in sel_cp["data"]["scenes"]: selected_map[s["scene_index"]] = set(s.get("selected_shot_indices", [])) # scene_director — 씬별 primary_location director_cp = self._load_prev_checkpoint("scene_director") scene_location: Dict[int, str] = {} if director_cp and director_cp.get("data", {}).get("scenes"): for ds in director_cp["data"]["scenes"]: scene_location[ds["scene_index"]] = ds.get("primary_location", "") # shot_director — shot별 VE shot_director_cp = self._load_prev_checkpoint("shot_director") shot_ve: Dict[tuple, set] = {} # (scene_index, shot_index) → set of entity short_ids if shot_director_cp and shot_director_cp.get("data", {}).get("scenes"): for sc in shot_director_cp["data"]["scenes"]: si = sc["scene_index"] for sh in sc.get("shots", []): key = (si, sh["shot_index"]) shot_ve[key] = set(sh.get("visible_entity_ids", [])) # fallback: shot_director 없으면 scene_director VE 사용 scene_ve: Dict[int, set] = {} if director_cp and director_cp.get("data", {}).get("scenes"): for ds in director_cp["data"]["scenes"]: scene_ve[ds["scene_index"]] = set(ds.get("present_entity_ids", [])) # shot_extract characters → short_id 매핑 (T2I보다 먼저 확정됨) # entity_character_list 또는 entity_t2i에서 이름→short_id 매핑 # name_matcher: 원본 + 괄호·공백 정규화 둘 다 키로 (드리프트 대응) _ent_items: List[Dict[str, Any]] = [] t2i_cp = self._load_prev_checkpoint("entity_t2i") if t2i_cp and t2i_cp.get("data"): for etype in ["characters", "locations", "props"]: for e in t2i_cp["data"].get(etype, []): if e.get("name") and e.get("short_id"): _ent_items.append(e) name_to_sid: Dict[str, str] = build_name_index( _ent_items, key_fn=lambda e: e.get("name", ""), value_fn=lambda e: e["short_id"], ) # shot별 인물 short_id — shot_extract의 characters 필드 기반 shot_characters: Dict[tuple, set] = {} # (scene_index, shot_index) → set of C## IDs for sc in shot_cp["data"]["scenes"]: si = sc["scene_index"] for sh in sc.get("shots", []): key = (si, sh.get("shot_index", 0)) char_sids = set() for name in sh.get("characters", []): sid = lookup_name(name_to_sid, name) if sid and sid.startswith("C"): char_sids.add(sid) shot_characters[key] = char_sids # selected shots 순서대로 수집 all_selected: List[Dict] = [] for sc in shot_cp["data"]["scenes"]: si = sc["scene_index"] sel_indices = selected_map.get(si) for sh in sc.get("shots", []): shot_idx = sh.get("shot_index", 0) if sel_indices is None or shot_idx in sel_indices: key = (si, shot_idx) # shot_director VE + shot_extract characters 결합 ve = shot_ve.get(key, scene_ve.get(si, set())) # shot_extract의 인물 목록으로 보정 (VE보다 정확) shot_chars = shot_characters.get(key, set()) if shot_chars: # VE에서 C##만 shot_extract 기준으로 교체, L##/P##은 유지 non_chars = {e for e in ve if not e.startswith("C")} ve = shot_chars | non_chars loc = scene_location.get(si, "") all_selected.append({ "scene_index": si, "shot_index": shot_idx, "location": loc, "entities": ve, }) if not all_selected: return {"completed_count": 0, "applicable_count": 0, "failed_count": 0, "data": {"dependencies": []}} # 연관 계산: 같은 배경 이전 shot 중 교집합↑ 여집합↓ dependencies = [] for i, current in enumerate(all_selected): cur_key = (current["scene_index"], current["shot_index"]) cur_loc = current["location"] cur_entities = current["entities"] best_ref = None best_score = -999 if cur_loc: # 같은 배경의 이전 shot만 후보 for j in range(i): prev = all_selected[j] if prev["location"] != cur_loc: continue prev_entities = prev["entities"] intersection = cur_entities & prev_entities complement = prev_entities - cur_entities # 이전에 있지만 현재에 없는 # 캐릭터(C##) 여집합은 가중 감점 — 이미지에서 인물이 가장 눈에 띄므로 char_complement = {e for e in complement if e.startswith("C")} non_char_complement = complement - char_complement score = len(intersection) - len(char_complement) * 3 - len(non_char_complement) if score >= best_score: best_score = score best_ref = prev dep = { "scene_index": current["scene_index"], "shot_index": current["shot_index"], "location_refs": [], "character_refs": [], } if best_ref: dep["location_refs"] = [{ "scene_index": best_ref["scene_index"], "shot_index": best_ref["shot_index"], "score": best_score, "shared_entities": sorted(current["entities"] & best_ref["entities"]), "extra_entities": sorted(best_ref["entities"] - current["entities"]), }] dependencies.append(dep) linked = sum(1 for d in dependencies if d["location_refs"]) logger.info("shot_dependency: %d/%d shots linked (code-based scoring)", linked, len(all_selected)) return { "completed_count": len(dependencies), "applicable_count": len(all_selected), "failed_count": 0, "data": {"dependencies": dependencies}, }