"""SceneAnalysisContext DTO. Phase 3.6 — detail_steps._execute가 로드하는 20+개 체크포인트 dict를 단일 dataclass로 집약. 목적: - `_execute`와 `_analyze_one` 사이 **명시적 계약**. - 테스트에서 fixture로 SceneAnalysisContext만 조립하면 `_analyze_one` 호출 가능. - 향후 리팩토링(Phase 3b) 시 closure → ctx.field 전환 base. 본 Phase 3.6 범위: - dataclass 정의 + `SceneContextLoader.load_all()` 호출 경로 확립. - `_analyze_one`의 closure 소비는 유지 (점진 전환 — Phase 3b/후속). """ from __future__ import annotations from dataclasses import dataclass, field from typing import Any, Dict, List, Optional, Tuple @dataclass class SceneAnalysisContext: """scene_detail step이 생산/소비하는 전체 context. 각 필드는 `_load_prev_checkpoint` 또는 DB에서 로드된 원본 구조를 유지. 체크포인트가 없으면 해당 필드는 빈 기본값. """ # ── scene_save ───────────────────────────────────────── segments: List[Dict[str, Any]] = field(default_factory=list) # ── scene_director ────────────────────────────────────── director_scenes: List[Dict[str, Any]] = field(default_factory=list) # scene_index → present_entity_ids (V only, NO LLM dependency) scene_visible: Dict[int, List[str]] = field(default_factory=dict) # ── dependency (shot_dependency 우선, fallback scene_dependency) ─ dependencies: List[Dict[str, Any]] = field(default_factory=list) # ── outlook_phase3 (또는 legacy outlook_extraction) ───── outlook_data: Dict[str, Any] = field(default_factory=dict) # ── shot_type DB (촬영 기법 리스트, 시스템 프롬프트 주입용) ─ shot_types_block: str = "" shot_type_rows: List[Any] = field(default_factory=list) # sqlalchemy Row # ── shot_staging ──────────────────────────────────────── # "{scene_index}_{shot_index}" → staging dict staging_map: Dict[str, Dict[str, Any]] = field(default_factory=dict) # ── scene_consistency ─────────────────────────────────── # scene_index → [fixed_element dict, ...] fixed_elements_by_scene: Dict[int, List[Dict[str, Any]]] = field(default_factory=dict) # ── location_consistency ──────────────────────────────── # L## → fixed_visual_description (씬 간 동일한 외형 문장, T2I용 영어) location_visuals_by_id: Dict[str, str] = field(default_factory=dict) # ── shot_cinematography (legacy) ──────────────────────── scene_shots_map: Dict[int, List[Dict[str, Any]]] = field(default_factory=dict) # ── entity_t2i ────────────────────────────────────────── entities: Dict[str, Any] = field(default_factory=dict) # ── shot_director ─────────────────────────────────────── # (scene_index, shot_index) → visible_entity_ids shot_director_ve: Dict[Tuple[int, int], List[str]] = field(default_factory=dict) # (scene_index, shot_index) → variant_resolved {base→actual} shot_director_vr: Dict[Tuple[int, int], Dict[str, str]] = field(default_factory=dict) # ── shot_selection ────────────────────────────────────── # scene_index → set of selected shot_indices selected_map: Dict[int, set] = field(default_factory=dict) # ── shot_extract (selected shots 필터링됨) ────────────── # scene_index → [shot dict, ...] shot_scenes_map: Dict[int, List[Dict[str, Any]]] = field(default_factory=dict) # ── beat_extract ──────────────────────────────────────── # scene_index → {beat_index: beat dict} beats_by_scene: Dict[int, Dict[int, Dict[str, Any]]] = field(default_factory=dict) # ── scene_summary ─────────────────────────────────────── # scene_index → summary text summaries: Dict[int, str] = field(default_factory=dict) # ── visual_world_rules ────────────────────────────────── # 원본 data (system prompt enrichment에서 사용) world_rules: Optional[Dict[str, Any]] = None # ── planning_doc ──────────────────────────────────────── # get_planning_context() 결과 (inject_if_available 용) planning_context: Optional[Any] = None # ── Phase 2 — chain_bg_render의 shot_guides ───────────── # (scene_index, shot_index) → guide str (영어 자연어 한 단락) chain_bg_guide_by_shot: Dict[Tuple[int, int], str] = field(default_factory=dict) # ── Phase 9.1 — chain_bg.camera_recommendations ───────── # (scene_index, shot_index) → camera meta dict # ({camera_position, camera_height, lens_hint, framing_notes}). # background_render cp의 data.groups[bg_id].camera_recommendations와 # shot_ids[]를 join하여 shot 단위로 펼침. scene_detail이 user_prompt에 # prepend하여 카메라 일관성 룰 적용에 사용. chain_bg_camera_meta_by_shot: Dict[Tuple[int, int], Dict[str, str]] = field(default_factory=dict) # ── G3.2 — background_prompt.objects_owned_by_background consumer ── # (scene_index, shot_index) → owned object names (English canonical, sorted). # background_prompt cp의 backgrounds[bid].objects_owned_by_background를 # spec.applies_to_shots(우선) 또는 shot_guides[].shot_id(fallback)로 join. # scene_detail이 post-parse owned judge / sentinel 검증에 소비. chain_bg_owned_by_shot: Dict[Tuple[int, int], List[str]] = field(default_factory=dict) # ── G4.1 Wave 4 R4 B3 — background_prompt bg_id (bid) 매핑 ── # (scene_index, shot_index) → bg_id (e.g. "cb_main_room"). # background_prompt cp 의 backgrounds dict key (bid) 를 owned 와 동일한 # spec.applies_to_shots / shot_guides[].shot_id 로 join. card 의 # background_binding.bg_id source — 부재 시 builder 가 not_applicable # mode 로 처리. multi-bg 가 같은 shot 에 적용되면 첫 번째 ok bid (deterministic # alpha-sorted) 만 반환 — 다중 bg shot 은 G4.x 후속 lift 대상. chain_bg_id_by_shot: Dict[Tuple[int, int], str] = field(default_factory=dict) # ── Phase 1b — shot_essence_extraction consumer ───────── # (scene_index, shot_index) → essence list. status='failed' shot은 빈 list essence_by_shot: Dict[Tuple[int, int], List[str]] = field(default_factory=dict) # ── G4.6 Phase 4 (Codex iter 1 B1) — main thread prebuild ─── # ThreadPool worker 안에서 SQLAlchemy session race 회피용. # _execute() 시작 시점 (ThreadPool 진입 전) project_id + entity_type='character' # scoped EntityCanon row 한 번 query → map. _build_entity_traits_block / # validate_visible_entities_contract 가 db 대신 본 map 사용. # short_id (e.g., "C08") → entity_canon.name (e.g., generic placeholder) name_by_short_id: Dict[str, str] = field(default_factory=dict) # short_id (e.g., "C08") → list[str] (entity_canon.stable_traits parsed) traits_by_short_id: Dict[str, List[str]] = field(default_factory=dict) # ── Task 14 Wave 2 — main thread prebuild (prop EntityCanon term map) ─ # `reconcile_owned_prop_namespace_overlap` 의 Gate C (term-match) 입력. # ThreadPool worker 진입 전 main thread 에서 project_id + entity_type='prop' # + status='active' scoped EntityCanon + EntityAlias row 한 번 query → # short_id (e.g., "P06") → frozenset of normalized prop term strings. # term source 는 entity_canon.{name, description, t2i_prompt} + 모든 alias. # 정규화 = strip + lowercase + collapse internal whitespace (helper 와 producer # 동일 함수 `_normalize_prop_term`). 빈 / whitespace-only 입력 제외. prop_term_map: Dict[str, frozenset] = field(default_factory=dict) # ── episode_reference_policy (Phase 2) ────────────────── episode_reference_policy: Optional[Dict[str, Any]] = None # ── meta ──────────────────────────────────────────────── project_id: str = "" episode_id: str = ""