"""엔티티/씬 이미지 생성 순서 결정 — 시각적 의존성 기반. Visual dependency graph를 구축하여 엔티티 참조 이미지를 올바른 순서로 생성하고, 독립적인 엔티티는 병렬로 처리한다. 씬 이미지도 동일한 토폴로지 정렬을 사용하여 같은 배경을 공유하는 씬은 순차 처리, 독립 씬은 병렬 처리한다. """ # Version: 1.1.0 — add scene dependency graph for concurrent scene generation # updated_at: 2026-03-16 import json from typing import Any, Dict, List, Set # Only these (entity_type, entity_type) + relation_family combos create # visual dependencies that require one entity's image as reference for another. VISUAL_REFERENCE_RULES: Dict[tuple, Set[str]] = { # Character ↔ Character: family resemblance / same-person variants ("character", "character"): {"identity", "transformation"}, # Character ↔ Prop: wearing/carrying ("character", "prop"): {"possession"}, ("prop", "character"): {"possession"}, # Location ↔ Location: same place at different times / variants ("location", "location"): {"transformation", "identity"}, } def build_visual_dependency_graph( entities: list, relations: list, participants: list, ) -> Dict[str, Set[str]]: """Build dependency graph based on visual reference rules only. Args: entities: list of dicts with at least {"id", "entity_type"}. relations: list of dicts with {"id", "relation_family"}. participants: list of dicts with {"relation_id", "canon_id"}. Returns: {entity_id: set of entity_ids it depends on (must be generated first)} """ entity_types: Dict[str, str] = {e["id"]: e["entity_type"] for e in entities} deps: Dict[str, Set[str]] = {e["id"]: set() for e in entities} # Build relation → participants mapping rel_parts: Dict[str, List] = {} for p in participants: rel_parts.setdefault(p["relation_id"], []).append(p) for rel in relations: parts = rel_parts.get(rel["id"], []) if len(parts) < 2: continue family = rel["relation_family"] for i, p1 in enumerate(parts): for p2 in parts[i + 1 :]: id1, id2 = p1["canon_id"], p2["canon_id"] if id1 not in entity_types or id2 not in entity_types: continue type1, type2 = entity_types[id1], entity_types[id2] type_pair = (type1, type2) type_pair_rev = (type2, type1) allowed = VISUAL_REFERENCE_RULES.get( type_pair ) or VISUAL_REFERENCE_RULES.get(type_pair_rev) if allowed and family in allowed: # The first participant listed is treated as the # "primary" — generate it first so the second can # use its image as a reference. deps[id2].add(id1) return deps def topological_sort_entities( entities: list, deps: Dict[str, Set[str]], ) -> List[List[str]]: """Topological sort into batches (each batch can run in parallel). Args: entities: list of dicts with at least {"id"}. deps: dependency graph from build_visual_dependency_graph(). Returns: List of batches. Each batch is a list of entity IDs whose dependencies have all been completed in earlier batches. """ remaining: Set[str] = {e["id"] for e in entities} completed: Set[str] = set() batches: List[List[str]] = [] while remaining: # Find entities whose dependencies are all completed ready: List[str] = [] for eid in sorted(remaining): # sorted for deterministic order entity_deps = deps.get(eid, set()) if entity_deps.issubset(completed): ready.append(eid) if not ready: # Circular dependency — break the cycle by adding all remaining batches.append(sorted(remaining)) break batches.append(ready) completed.update(ready) remaining -= set(ready) return batches # ── Scene dependency graph ─────────────────────────────────────────── def build_scene_dependency_graph( stills: List[Dict[str, Any]], entity_lookup: Dict[str, Dict[str, Any]], ) -> Dict[int, Set[int]]: """Build scene dependency graph based on shared background locations. Scenes sharing the same location must be generated sequentially (for visual continuity from previous scene reference). Scenes at independent locations can be generated in parallel. Args: stills: list of still dicts (ordered by still_index). Each must have "visible_entities_json". entity_lookup: entity_id -> entity dict (with "entity_type"). Returns: {scene_index: set of scene_indices it depends on} """ deps: Dict[int, Set[int]] = {si: set() for si in range(len(stills))} location_to_latest_scene: Dict[str, int] = {} for si, still_data in enumerate(stills): # Parse visible entities to find location IDs try: visible_ids = json.loads(still_data.get("visible_entities_json", "[]")) except (json.JSONDecodeError, TypeError): visible_ids = [] location_ids: List[str] = [] for v in visible_ids: if isinstance(v, dict): eid = v.get("id") or v.get("entity_id", "") elif isinstance(v, str): eid = v else: continue entity = entity_lookup.get(eid) if entity and entity.get("entity_type") == "location": location_ids.append(eid) # Each location creates a dependency chain for loc_id in location_ids: if loc_id in location_to_latest_scene: deps[si].add(location_to_latest_scene[loc_id]) location_to_latest_scene[loc_id] = si # dependent_scene_id도 의존성에 추가 (씬 분석에서 결정된 시각적 연관) dep_scene_id = still_data.get("dependent_scene_id") if dep_scene_id: # dependent_scene_id는 still UUID → scene_index로 변환 for sj, other in enumerate(stills): if other.get("id") == dep_scene_id: deps[si].add(sj) break return deps def topological_sort_scenes( scene_count: int, deps: Dict[int, Set[int]], ) -> List[List[int]]: """Topological sort scene indices into batches for concurrent generation. Args: scene_count: total number of scenes. deps: dependency graph from build_scene_dependency_graph(). Returns: List of batches. Each batch is a list of scene indices whose dependencies have all been completed in earlier batches. """ remaining: Set[int] = set(range(scene_count)) completed: Set[int] = set() batches: List[List[int]] = [] while remaining: ready: List[int] = [] for si in sorted(remaining): scene_deps = deps.get(si, set()) if scene_deps.issubset(completed): ready.append(si) if not ready: # Circular dependency — break the cycle batches.append(sorted(remaining)) break batches.append(ready) completed.update(ready) remaining -= set(ready) return batches