"""background_chain_planning — location별 배경 이미지 chain 트리 설계. set_design 교체용 신규 step (PR #3, 2026-04-27). - on_demand: 자동 run-all에서 실행 X. 사용자 명시 트리거 시만. - 입력: shot_validator + shot_selection + shot_staging + scene_director + entity_merge + entity_detail + visual_world_rules - 출력: data.locations[loc_id] = {location_id, rationale_summary, nodes, execution_order, unassigned_shots} 흐름: Phase 0: 선택된 shot을 primary_location 기준으로 그루핑 Phase 1: location 단위로 LLM(gpt-5.5) 호출 + chain 트리 설계 Phase 2: semantic invariant validator """ from __future__ import annotations import json import logging import re from concurrent.futures import ThreadPoolExecutor, as_completed from dataclasses import dataclass, field from typing import Any, Dict, List, Optional from app.modules.llm.llm_client import call_structured from app.modules.prompt_loader import load_prompt, load_schema logger = logging.getLogger(__name__) # 한글/한자/kana 검출 — universal noun 검증용. # - Hangul Compat Jamo (U+3131-U+318E) + Hangul Syllables (U+AC00-U+D7A3) # - CJK Unified Ideographs (U+4E00-U+9FFF) + Extension A (U+3400-U+4DBF) # - CJK Compatibility Ideographs (U+F900-U+FAFF) # - Hiragana/Katakana 영역 (U+3040-U+30FF) _NON_ASCII_TEXT_RE = re.compile( r"[ㄱ-ㆎ가-힣" r"一-鿿㐀-䶿豈-﫿" r"぀-ヿ]" ) @dataclass class ShotInfo: scene_index: int shot_index: int shot_id: str description: str camera_direction: str = "" lighting_mood: str = "" character_angles: List[Dict[str, Any]] = field(default_factory=list) beat_title: str = "" beat_change: str = "" visible_entity_ids: List[str] = field(default_factory=list) @dataclass class LocationGroup: location_id: str location_name: str location_description: str visual_traits: List[str] shots: List[ShotInfo] = field(default_factory=list) # ────────────────────────────────────────────── # Phase 0: location 그루핑 # ────────────────────────────────────────────── def _phase0_group_by_location( shot_extract_data: Dict[str, Any], shot_selection_data: Dict[str, Any], shot_staging_data: Dict[str, Any], director_data: Dict[str, Any], entity_merge_data: Dict[str, Any], entity_detail_data: Optional[Dict[str, Any]] = None, ) -> Dict[str, LocationGroup]: """선택된 샷을 primary_location 기준으로 location별 그루핑. director.scenes[].primary_location → 시작 short_id (e.g. "L04"). legacy 한글 이름 fallback도 지원 (set_design Phase 0와 동등). """ # scene_director → primary_location 매핑 scene_location: Dict[int, str] = {} for ds in director_data.get("scenes", []): si = ds.get("scene_index") loc = ds.get("primary_location", "") if si is not None and loc: scene_location[si] = loc # entity 메타: short_id → (name, description, visual_traits) locs_meta: Dict[str, Dict[str, Any]] = {} name_to_sid: Dict[str, str] = {} for loc in entity_merge_data.get("locations", []): sid = loc.get("short_id", "") if not sid: continue locs_meta[sid] = { "name": loc.get("name", sid), "description": loc.get("description", "") or "", "visual_traits": loc.get("visual_traits", []) or [], } if loc.get("name"): name_to_sid[loc["name"]] = sid # entity_detail에서 description/visual_traits 보강 (있으면 우선) if entity_detail_data: for loc in entity_detail_data.get("locations", []): sid = loc.get("short_id", "") if not sid or sid not in locs_meta: continue if loc.get("description"): locs_meta[sid]["description"] = loc["description"] if loc.get("visual_traits"): locs_meta[sid]["visual_traits"] = loc["visual_traits"] # 선택된 shot index 셋 selected_map: Dict[int, Optional[set]] = {} for s in shot_selection_data.get("scenes", []): si = s.get("scene_index") selected = s.get("selected_shot_indices") if selected is not None: selected_map[si] = set(selected) # staging 매핑 staging_map: Dict[str, Dict[str, Any]] = {} for st in shot_staging_data.get("shots", []): key = f"{st.get('scene_index')}_{st.get('shot_index')}" staging_map[key] = st groups: Dict[str, LocationGroup] = {} for sc in shot_extract_data.get("scenes", []): si = sc.get("scene_index") sel = selected_map.get(si) for sh in sc.get("shots", []): shi = sh.get("shot_index", 0) if sel is not None and shi not in sel: continue loc_raw = scene_location.get(si, "") if not loc_raw: logger.debug("Phase 0: scene %d no primary_location — skip shot %d", si, shi) continue # primary_location → short_id 정규화 if loc_raw.startswith("L") and loc_raw[1:].isdigit(): loc_id = loc_raw else: # C4/P043: name substring 양방향 매칭 fallback 폐기 (Semantic Regex Ban). # exact name match·L-id 모두 실패 = primary_location contract 위반 → # silent loc_id=loc_raw fallback 금지, fail-loud (No Silent Fallback). loc_id = name_to_sid.get(loc_raw, "") if not loc_id: raise ValueError( f"Phase 0: primary_location {loc_raw!r} did not resolve to " f"a known location short_id or exact entity name " f"(substring fallback removed — C4/P043)" ) staging = staging_map.get(f"{si}_{shi}", {}) shot_id = f"S{si:02d}_Shot{shi}" # visible_entities — shot 자체 + scene_director에서 가져옴 (둘 중 하나에 있으면 OK) visible: List[str] = [] for v in sh.get("visible_entities", []) or []: if isinstance(v, dict): eid = v.get("id") or v.get("entity_id") or v.get("short_id") if eid: visible.append(str(eid)) elif isinstance(v, str): visible.append(v) beat_change = "" for bf in ("change_type", "after_state"): val = sh.get(bf) if val: beat_change = (beat_change + " " + str(val)).strip() if loc_id not in groups: meta = locs_meta.get(loc_id, {}) groups[loc_id] = LocationGroup( location_id=loc_id, location_name=meta.get("name", loc_id), location_description=meta.get("description", ""), visual_traits=meta.get("visual_traits", []), ) groups[loc_id].shots.append(ShotInfo( scene_index=si, shot_index=shi, shot_id=shot_id, description=sh.get("description", "") or "", camera_direction=staging.get("camera_direction", ""), lighting_mood=staging.get("lighting_mood", ""), character_angles=staging.get("character_angles", []) or [], beat_title=sh.get("based_on_beat_title", ""), beat_change=beat_change, visible_entity_ids=visible, )) for group in groups.values(): group.shots.sort(key=lambda s: (s.scene_index, s.shot_index)) logger.info("Phase 0: %d locations grouped (%d total shots)", len(groups), sum(len(g.shots) for g in groups.values())) return groups # ────────────────────────────────────────────── # Phase 1: LLM 호출 # ────────────────────────────────────────────── def _build_shots_block(shots: List[ShotInfo]) -> str: parts: List[str] = [] for s in shots: ca_summary = "" if s.character_angles: angle_strs = [] for ca in s.character_angles: if not isinstance(ca, dict): continue ang = ca.get("angle") or ca.get("camera_relative", "") pose = ca.get("pose", "") if ang or pose: angle_strs.append(f"{ang} {pose}".strip()) ca_summary = "; ".join(a for a in angle_strs if a) parts.append( f"\n--- {s.shot_id} ---\n" f"description (Korean — context only, ignore proper names): {s.description}\n" f"beat: {s.beat_title or '(none)'} ({s.beat_change or 'no state change'})\n" f"camera_direction: {s.camera_direction or '(none)'}\n" f"lighting_mood: {s.lighting_mood or '(none)'}\n" f"character_angles: {ca_summary or '(none)'}\n" f"visible_entities: {json.dumps(s.visible_entity_ids, ensure_ascii=False)}" ) return "\n".join(parts) def _build_world_rules_excerpt(world_rules_data: Optional[Dict[str, Any]]) -> str: """visual_world_rules 체크포인트에서 location/visual 관련 발췌.""" if not world_rules_data: return "(no world rules available)" rules = world_rules_data.get("rules") or world_rules_data.get("data", {}).get("rules") or [] if not rules: return "(no rules listed)" lines: List[str] = [] for r in rules: if isinstance(r, dict): label = r.get("label") or r.get("category") or "" content = r.get("content") or r.get("description") or "" if content: lines.append(f"- [{label}] {content}" if label else f"- {content}") elif isinstance(r, str): lines.append(f"- {r}") return "\n".join(lines) if lines else "(rules empty)" def build_planning_user_prompt( location_id: str, base_prompt: str, floor_plan_prompts: Dict[str, str], ) -> str: """chain_bg_planning user_prompt 빌드. floor plan prompt가 있으면 prepend. Phase 4: floor_plan_prompts.get(location_id)이 비어있지 않으면 [FLOOR PLAN] + [CHAIN BG PLANNING TASK] 블록으로 prepend. 빈 dict / 누락 / 빈 string 시 base_prompt 그대로 (mode=off / chain_only 회귀 보장). """ fp = (floor_plan_prompts.get(location_id) or "").strip() if not fp: return base_prompt return ( f"[FLOOR PLAN — spatial layout authority for location {location_id}]\n" f"{fp}\n\n" f"[CHAIN BG PLANNING TASK]\n" f"{base_prompt}" ) def _build_parent_chain_bg_block(parent_ctx: str) -> str: """parent_ctx가 있으면 [PARENT CHAIN BG ...] 블록 + 빈 줄로 prepend. Phase 5 (planner-driven): chain_bg_order대로 순차 처리할 때 부모 group의 rationale + node descriptions 요약을 받아 visual continuity anchor로 LLM에 전달한다. parent_ctx가 빈 문자열/None이면 빈 문자열을 반환 — template의 {parent_chain_bg_block} placeholder가 그냥 사라진다. 포맷은 system prompt의 PARENT CHAIN BG CONTEXT 섹션이 인식하는 형태와 일치. """ text = (parent_ctx or "").strip() if not text: return "" return ( "[PARENT CHAIN BG — visual continuity anchor]\n" f"{text}\n\n" ) def _plan_one_location( group: LocationGroup, world_rules_excerpt: str, opik_metadata: Optional[Dict[str, Any]] = None, floor_plan_prompts: Optional[Dict[str, str]] = None, parent_chain_bg_ctx: str = "", ) -> Dict[str, Any]: """단일 location/group의 chain 트리를 LLM으로 설계. Phase 4 legacy path: parent_chain_bg_ctx="" (template placeholder가 빈 문자열로 채워져 prompt 텍스트에 영향 없음). floor_plan_prompts만 prepend된다. Phase 5 planner-driven path: parent_chain_bg_ctx가 있으면 user_template의 {parent_chain_bg_block} 자리에 [PARENT CHAIN BG ...] 블록이 삽입돼 LLM이 visual continuity를 유지하도록 안내한다. """ system = load_prompt("background_chain_planning", "system") template = load_prompt("background_chain_planning", "user_template") schema = load_schema("background_chain_planning", "schema") # str.format()은 user content의 `{`/`}`를 placeholder로 오인하므로 escape. # 시나리오 description, 한국어 location name, world rules 등에 중괄호가 # 들어가면 KeyError/ValueError 발생 — set_design 시절 잠재 버그를 본 PR에서 차단. def _esc(s: str) -> str: return (s or "").replace("{", "{{").replace("}", "}}") # parent_chain_bg_block은 우리가 합성한 텍스트 — 시나리오 raw text가 아니라 # 자체적으로 brace를 포함하지 않지만, 만약을 대비해 escape 적용. parent_block = _build_parent_chain_bg_block(parent_chain_bg_ctx) user = template.format( parent_chain_bg_block=_esc(parent_block), location_id=group.location_id, location_name=_esc(group.location_name), location_description=_esc(group.location_description) or "(no description)", visual_traits_json=_esc(json.dumps(group.visual_traits, ensure_ascii=False)), shot_count=len(group.shots), shots_block=_esc(_build_shots_block(group.shots)), world_rules_excerpt=_esc(world_rules_excerpt), ) # Phase 4: floor plan prompt prepend (있을 때만) user = build_planning_user_prompt( location_id=group.location_id, base_prompt=user, floor_plan_prompts=floor_plan_prompts or {}, ) logger.info("Phase 1: planning %s (%d shots)", group.location_id, len(group.shots)) plan = call_structured( step="background_chain_planning", system_prompt=system, user_prompt=user, response_schema=schema, opik_metadata=opik_metadata, ) return plan # ────────────────────────────────────────────── # Phase 2: semantic validator # ────────────────────────────────────────────── def validate_plan(plan: Dict[str, Any], group: LocationGroup) -> List[str]: """Plan의 의미적 invariant 검증. JSON schema가 못 잡는 것들. 검사: 1. location_id 일치 2. 입력 shot이 정확히 1번 등장 (누락/중복 0) 3. parent_id가 nodes 안에 존재 (또는 빈 문자열) 4. execution_order: 모든 노드 1번 + parent-before-child 5. ≥1 root anchor 6. 모든 영문 필드에 한글/한자/kana 0건 (universal noun 검증) skip_chain=true plan은 nodes/execution_order/parent/root 검증을 우회한다. rationale_summary의 universal-noun 검증만 적용 (LLM이 한글로 빠지지 않게). """ errors: List[str] = [] if plan.get("location_id") != group.location_id: errors.append( f"location_id mismatch: expected {group.location_id!r}, got {plan.get('location_id')!r}" ) # skip_chain=true → outdoor/open-air. 노드 검증 우회 + ASCII/full-empty 검증. if plan.get("skip_chain") is True: summary = plan.get("rationale_summary") or "" if _NON_ASCII_TEXT_RE.search(summary): errors.append("rationale_summary contains non-ASCII text (universal-noun rule violated)") reason = plan.get("skip_reason") or "" if _NON_ASCII_TEXT_RE.search(reason): errors.append("skip_reason contains non-ASCII text") # SKIP 응답은 nodes / execution_order / unassigned_shots 모두 빈 배열이어야 한다. # 그래야 downstream(render/coordinator)이 chain 미생성으로 안전히 분기한다. for empty_key in ("nodes", "execution_order", "unassigned_shots"): val = plan.get(empty_key) if val: # None / [] 모두 falsy 통과, 비어있지 않은 list만 잡음 errors.append( f"skip_chain=true but {empty_key} is non-empty ({len(val)} items) — must be empty array" ) return errors nodes = plan.get("nodes", []) # 1) node id 중복 검사 — dict로 collapse되기 전에. duplicate id가 있으면 # parent_id 매칭 등이 silently 잘못된 노드를 가리킬 수 있다. id_counts: Dict[str, int] = {} for n in nodes: nid = n.get("id", "") id_counts[nid] = id_counts.get(nid, 0) + 1 for nid, count in id_counts.items(): if count > 1: errors.append(f"duplicate node id: {nid!r} appears {count} times") nodes_by_id: Dict[str, Dict[str, Any]] = {n["id"]: n for n in nodes} # 2) parent_id 검증 — 빈 문자열만 허용되는 missing sentinel. # "null"/"None"/"NULL" 같은 literal sentinel은 프롬프트 스펙 위반. _SENTINEL_NULLS = {"null", "none", "nil"} def _is_sentinel_null(pid: str) -> bool: return pid.strip().lower() in _SENTINEL_NULLS input_shot_ids = {s.shot_id for s in group.shots} seen_in_nodes: Dict[str, str] = {} for n in nodes: for sid in (n.get("shot_ids") or []): if sid in seen_in_nodes: errors.append( f"shot {sid} duplicated: nodes={seen_in_nodes[sid]} and {n['id']}" ) seen_in_nodes[sid] = n["id"] missing = input_shot_ids - set(seen_in_nodes.keys()) if missing: errors.append(f"shots missing from any node.shot_ids: {sorted(missing)}") extra = set(seen_in_nodes.keys()) - input_shot_ids if extra: errors.append(f"shots in node.shot_ids but not in input shots: {sorted(extra)}") for n in nodes: pid_raw = n.get("parent_id", "") if pid_raw and _is_sentinel_null(pid_raw): errors.append( f"node {n.get('id', '?')} parent_id={pid_raw!r} is a literal null sentinel — " f"use empty string '' for missing parent" ) continue pid = pid_raw.strip() if pid and pid not in nodes_by_id: errors.append(f"node {n['id']} parent_id={pid!r} not found in nodes") execution_order = plan.get("execution_order", []) seen_in_order: set = set() for nid in execution_order: n = nodes_by_id.get(nid) if not n: errors.append(f"execution_order entry {nid!r} not in nodes") continue if nid in seen_in_order: errors.append(f"execution_order entry {nid!r} duplicated") pid_raw = n.get("parent_id", "") # sentinel null은 위에서 별도로 errored — 여기서는 정상 부모로 간주 X pid = pid_raw.strip() if not _is_sentinel_null(pid_raw) else "" if pid and pid not in seen_in_order: errors.append(f"execution_order: {nid} comes before its parent {pid}") seen_in_order.add(nid) missing_in_order = set(nodes_by_id.keys()) - set(execution_order) if missing_in_order: errors.append(f"execution_order missing nodes: {sorted(missing_in_order)}") roots = [ n for n in nodes if n.get("kind") == "anchor_root" or (not n.get("parent_id", "").strip() and not _is_sentinel_null(n.get("parent_id", ""))) ] if not roots: errors.append("no anchor_root node — at least one root with empty parent_id required") text_fields = ("id", "label", "description", "rationale") for n in nodes: for field_name in text_fields: val = n.get(field_name) or "" if _NON_ASCII_TEXT_RE.search(val): errors.append( f"node {n.get('id', '?')}.{field_name} contains non-ASCII text " f"(Korean/Hanja/kana detected — universal-noun rule violated)" ) for sva in n.get("shared_visual_anchors_with_parent") or []: if isinstance(sva, str) and _NON_ASCII_TEXT_RE.search(sva): errors.append( f"node {n.get('id', '?')}.shared_visual_anchors_with_parent contains " f"non-ASCII text (Korean/Hanja/kana detected)" ) summary = plan.get("rationale_summary") or "" if _NON_ASCII_TEXT_RE.search(summary): errors.append("rationale_summary contains non-ASCII text (universal-noun rule violated)") return errors # ────────────────────────────────────────────── # 진입점 # ────────────────────────────────────────────── # 저빈도 location skip 임계값 — selected shot 수 ≤ 이 값이면 chain 생성 안 함 (prev_shot_ref fallback). _LOW_FREQUENCY_SHOT_THRESHOLD = 3 def run_background_chain_planning( shot_extract_data: Dict[str, Any], shot_selection_data: Dict[str, Any], shot_staging_data: Dict[str, Any], director_data: Dict[str, Any], entity_merge_data: Dict[str, Any], entity_detail_data: Optional[Dict[str, Any]] = None, world_rules_data: Optional[Dict[str, Any]] = None, opik_metadata: Optional[Dict[str, Any]] = None, max_workers: int = 3, floor_plan_prompts: Optional[Dict[str, str]] = None, planner_groups: Optional[Dict[str, Any]] = None, ) -> Dict[str, Any]: """전체 파이프라인 실행. 두 경로: - planner_groups가 None이면 LEGACY (Phase 4): location별 _phase0 그루핑 → ThreadPool 병렬 (max_workers=3). mode=chain_only 회귀 보장. - planner_groups가 dict면 PLANNER-DRIVEN (Phase 5): chain_bg_order대로 SEQUENTIAL 처리, 부모 group의 chain bg 컨텍스트를 자식에게 inject. Skip 조건 (chain bg 생성 안 함, scene_image_pipeline에서 prev_shot_ref로 fallback): 1. selected shot 수 ≤ _LOW_FREQUENCY_SHOT_THRESHOLD — 코드 단에서 즉시 skip (LLM 호출 X) 2. LLM이 outdoor/open-air로 판정 → skip_chain=true 응답 Phase 4: floor_plan_prompts 제공 시 _plan_one_location의 user_prompt에 도면 prompt가 prepend된다. mode=off / chain_only 시 빈 dict. Phase 5: planner_groups = {"groups": {gid: spec}, "order": [gid, ...]}. 각 group = 별도 LLM call (HARD 제약). 부모 group의 chain_bg 결과를 합성해 `[PARENT CHAIN BG]` 블록으로 자식 group prompt에 inject. """ floor_plan_prompts = floor_plan_prompts or {} # Phase 5 planner-driven 분기 — planner_groups가 None이 아니면 신규 path. if planner_groups is not None: return _run_planner_driven( planner_groups=planner_groups, shot_extract_data=shot_extract_data, shot_selection_data=shot_selection_data, shot_staging_data=shot_staging_data, director_data=director_data, entity_merge_data=entity_merge_data, entity_detail_data=entity_detail_data, world_rules_data=world_rules_data, opik_metadata=opik_metadata, floor_plan_prompts=floor_plan_prompts, ) # Phase 4 LEGACY path — 아래 로직은 변경 없음. groups = _phase0_group_by_location( shot_extract_data, shot_selection_data, shot_staging_data, director_data, entity_merge_data, entity_detail_data, ) if not groups: logger.info("background_chain_planning: no locations — nothing to do") return {"locations": {}, "_failed_count": 0} world_rules_excerpt = _build_world_rules_excerpt(world_rules_data) locations_out: Dict[str, Dict[str, Any]] = {} failed = 0 # 저빈도 location은 LLM 호출 전 즉시 skip 처리. groups에서 제거. low_freq_loc_ids = [ lid for lid, g in groups.items() if len(g.shots) <= _LOW_FREQUENCY_SHOT_THRESHOLD ] for lid in low_freq_loc_ids: g = groups.pop(lid) # 저빈도 skip은 LLM 호출 없이 즉시 반환 — floor_plan prepend 발생 X. locations_out[lid] = { "location_id": lid, "location_name": g.location_name, "shot_count": len(g.shots), "skip_chain": True, "skip_reason": f"low_frequency (<= {_LOW_FREQUENCY_SHOT_THRESHOLD} selected shots)", "rationale_summary": ( f"Skipped chain: only {len(g.shots)} selected shot(s); " f"renderer will use previous-related-shot reference." ), "nodes": [], "execution_order": [], "unassigned_shots": [], "floor_plan_used": False, "status": "skipped", } if low_freq_loc_ids: logger.info( "background_chain_planning: %d low-frequency locations skipped (no LLM call): %s", len(low_freq_loc_ids), low_freq_loc_ids, ) def _process_one(loc_id: str, group: LocationGroup) -> tuple: # 도면 prompt가 prepend되었는지 마커 (debug용 — Phase 4 traceability). # mode=floor_plan_anchored일 때 어떤 location이 실제로 도면을 안았는지 # 체크포인트에서 사후 식별 가능. floor_plan_used = bool( (floor_plan_prompts or {}).get(loc_id, "").strip() ) try: plan = _plan_one_location( group, world_rules_excerpt, opik_metadata, floor_plan_prompts=floor_plan_prompts, ) errors = validate_plan(plan, group) if errors: logger.error( "background_chain_planning: %s validation FAILED — %d issues:\n %s", loc_id, len(errors), "\n ".join(errors), ) return loc_id, { "location_id": loc_id, "location_name": group.location_name, "shot_count": len(group.shots), "validation_errors": errors, "plan": plan, "floor_plan_used": floor_plan_used, "status": "failed", }, True # LLM이 outdoor로 판정 → skip 처리. if plan.get("skip_chain") is True: logger.info( "background_chain_planning: %s — LLM marked as outdoor/open-air, skipping chain", loc_id, ) return loc_id, { "location_id": loc_id, "location_name": group.location_name, "shot_count": len(group.shots), "skip_chain": True, "skip_reason": plan.get("skip_reason", "outdoor open-air"), "rationale_summary": plan.get("rationale_summary", ""), "nodes": [], "execution_order": [], "unassigned_shots": [], "floor_plan_used": floor_plan_used, "status": "skipped", }, False return loc_id, { "location_id": loc_id, "location_name": group.location_name, "shot_count": len(group.shots), "skip_chain": False, "skip_reason": "", "rationale_summary": plan.get("rationale_summary", ""), "nodes": plan.get("nodes", []), "execution_order": plan.get("execution_order", []), "unassigned_shots": plan.get("unassigned_shots", []), "floor_plan_used": floor_plan_used, "status": "ok", }, False except Exception as exc: logger.error("background_chain_planning: %s exception: %s", loc_id, exc, exc_info=True) return loc_id, { "location_id": loc_id, "location_name": group.location_name, "shot_count": len(group.shots), "error": str(exc), "floor_plan_used": floor_plan_used, "status": "exception", }, True if groups: workers = min(max_workers, len(groups)) or 1 with ThreadPoolExecutor(max_workers=workers) as pool: futures = {pool.submit(_process_one, lid, g): lid for lid, g in groups.items()} for fut in as_completed(futures): loc_id, result, is_failed = fut.result() locations_out[loc_id] = result if is_failed: failed += 1 skipped_total = sum(1 for v in locations_out.values() if v.get("status") == "skipped") logger.info( "background_chain_planning: done — %d locations (%d skipped, %d failed)", len(locations_out), skipped_total, failed, ) return {"locations": locations_out, "_failed_count": failed} # ────────────────────────────────────────────── # Phase 5 — planner-driven path (group-based sequential) # ────────────────────────────────────────────── def _build_group_shot_index( shot_extract_data: Dict[str, Any], shot_staging_data: Dict[str, Any], ) -> Dict[str, Dict[str, Any]]: """shot_id (S{scene}_Shot{idx}) → 풀 shot 메타 + staging 인덱스. Phase 5 planner-driven에서 group.shot_ids로 ShotInfo를 즉시 lookup하기 위함. legacy `_phase0_group_by_location`이 scene_director.primary_location 기준 그루핑을 했다면, planner는 group spec에 shot_ids를 직접 가지고 있어 director 의존성을 우회한다. """ staging_map: Dict[str, Dict[str, Any]] = {} for st in shot_staging_data.get("shots", []) or []: si = st.get("scene_index") shi = st.get("shot_index") if si is None or shi is None: continue staging_map[f"S{si}_Shot{shi}"] = st shot_meta: Dict[str, Dict[str, Any]] = {} for sc in shot_extract_data.get("scenes", []) or []: si = sc.get("scene_index") if si is None: continue for sh in sc.get("shots", []) or []: shi = sh.get("shot_index", 0) sid = f"S{si}_Shot{shi}" shot_meta[sid] = { "scene_index": si, "shot_index": shi, "shot_data": sh, "staging": staging_map.get(sid, {}), } return shot_meta def _location_meta_index( entity_merge_data: Dict[str, Any], entity_detail_data: Optional[Dict[str, Any]] = None, ) -> Dict[str, Dict[str, Any]]: """short_id → {name, description, visual_traits} (entity_detail 우선 보강). legacy `_phase0_group_by_location`의 locs_meta 빌드 로직과 동일 — Phase 5 path도 동일한 location 메타로 LLM context를 채운다. """ locs_meta: Dict[str, Dict[str, Any]] = {} for loc in entity_merge_data.get("locations", []) or []: sid = loc.get("short_id", "") if not sid: continue locs_meta[sid] = { "name": loc.get("name", sid), "description": loc.get("description", "") or "", "visual_traits": loc.get("visual_traits", []) or [], } if entity_detail_data: for loc in entity_detail_data.get("locations", []) or []: sid = loc.get("short_id", "") if not sid or sid not in locs_meta: continue if loc.get("description"): locs_meta[sid]["description"] = loc["description"] if loc.get("visual_traits"): locs_meta[sid]["visual_traits"] = loc["visual_traits"] return locs_meta def _build_shot_info_for_group( shot_id: str, shot_meta_idx: Dict[str, Dict[str, Any]], ) -> Optional[ShotInfo]: """group.shot_ids 한 entry → ShotInfo. 매칭 실패 시 None.""" meta = shot_meta_idx.get(shot_id) if not meta: return None sh = meta["shot_data"] staging = meta["staging"] visible: List[str] = [] for v in sh.get("visible_entities", []) or []: if isinstance(v, dict): eid = v.get("id") or v.get("entity_id") or v.get("short_id") if eid: visible.append(str(eid)) elif isinstance(v, str): visible.append(v) beat_change = "" for bf in ("change_type", "after_state"): val = sh.get(bf) if val: beat_change = (beat_change + " " + str(val)).strip() return ShotInfo( scene_index=meta["scene_index"], shot_index=meta["shot_index"], shot_id=shot_id, description=sh.get("description", "") or "", camera_direction=staging.get("camera_direction", ""), lighting_mood=staging.get("lighting_mood", ""), character_angles=staging.get("character_angles", []) or [], beat_title=sh.get("based_on_beat_title", ""), beat_change=beat_change, visible_entity_ids=visible, ) def _build_group_location_group( group_spec: Dict[str, Any], shot_meta_idx: Dict[str, Dict[str, Any]], locs_meta: Dict[str, Dict[str, Any]], ) -> LocationGroup: """planner.chain_bg_groups[i] → LocationGroup (Phase 5). location_id는 group_spec.location_id를 그대로 사용 (planner가 schema enum으로 검증 완료). shot_meta_idx에서 ShotInfo를 lookup해 group.shots로 묶는다. 매칭 실패한 shot_id는 logger warning + 스킵. """ loc_id = group_spec.get("location_id", "") or "" meta = locs_meta.get(loc_id, {}) shots: List[ShotInfo] = [] missing: List[str] = [] for sid in group_spec.get("shot_ids", []) or []: info = _build_shot_info_for_group(sid, shot_meta_idx) if info is None: missing.append(sid) continue shots.append(info) if missing: logger.warning( "Phase 5 planner: group %s missing shot meta for %s — skipped", group_spec.get("id", "?"), missing, ) shots.sort(key=lambda s: (s.scene_index, s.shot_index)) return LocationGroup( location_id=loc_id, location_name=meta.get("name", loc_id), location_description=meta.get("description", ""), visual_traits=meta.get("visual_traits", []), shots=shots, ) def _summarize_parent_for_continuity(parent_result: Dict[str, Any]) -> str: """부모 group의 chain bg 결과 → 자식 group prompt의 PARENT CHAIN BG block 본문. rationale_summary + 각 node의 label/description을 ASCII English로 합성. skip_chain=true / failed parent는 빈 문자열 — 자식이 자유 plan하도록. """ if not parent_result: return "" if parent_result.get("status") not in ("ok",): return "" if parent_result.get("skip_chain") is True: return "" lines: List[str] = [] summary = (parent_result.get("rationale_summary") or "").strip() if summary: lines.append(f"Parent group rationale: {summary}") nodes = parent_result.get("nodes") or [] if nodes: lines.append("Parent nodes:") for n in nodes: label = (n.get("label") or n.get("id") or "").strip() desc = (n.get("description") or "").strip() if label and desc: lines.append(f"- {label}: {desc}") elif label: lines.append(f"- {label}") elif desc: lines.append(f"- {desc}") return "\n".join(lines) def _run_planner_driven( *, planner_groups: Dict[str, Any], shot_extract_data: Dict[str, Any], shot_selection_data: Dict[str, Any], shot_staging_data: Dict[str, Any], director_data: Dict[str, Any], entity_merge_data: Dict[str, Any], entity_detail_data: Optional[Dict[str, Any]] = None, world_rules_data: Optional[Dict[str, Any]] = None, opik_metadata: Optional[Dict[str, Any]] = None, floor_plan_prompts: Optional[Dict[str, str]] = None, ) -> Dict[str, Any]: """Phase 5 planner-driven path. chain_bg_order대로 SEQUENTIAL 처리 — 각 group은 별도 LLM call (HARD 제약). 부모 group의 chain bg 결과를 합성해 자식 group prompt에 [PARENT CHAIN BG] 블록으로 inject한다. Returns: { "groups": {group_id: result, ...}, # Phase 5 신규 shape "locations": {loc_id: result, ...}, # Phase 4 chain_bg_render reader compat alias "_failed_count": int } compat alias `locations`: Phase 4 chain_bg_render는 `data.locations[loc_id]`을 읽는다 (background_chain_render.py:583-585). T7에서 `data.groups`로 pivot 예정이지만, 그 전까지는 동일 location의 마지막 group result로 집계해 reader가 깨지지 않게 한다. 동일 location에 여러 group이 있을 때 마지막 ok result를 우선하고, 모든 group이 fail/skipped면 그 중 마지막 entry를 그대로 사용한다. """ floor_plan_prompts = floor_plan_prompts or {} groups_spec: Dict[str, Dict[str, Any]] = planner_groups.get("groups") or {} order: List[str] = list(planner_groups.get("order") or []) if not groups_spec or not order: logger.info( "background_chain_planning (planner-driven): empty planner — " "groups=%d, order=%d", len(groups_spec), len(order), ) return {"groups": {}, "locations": {}, "_failed_count": 0} # 인덱스 빌드 (full meta 일관성 확보) shot_meta_idx = _build_group_shot_index(shot_extract_data, shot_staging_data) locs_meta = _location_meta_index(entity_merge_data, entity_detail_data) world_rules_excerpt = _build_world_rules_excerpt(world_rules_data) parent_prompts: Dict[str, str] = {} # group_id → PARENT CHAIN BG body results_by_group: Dict[str, Dict[str, Any]] = {} locations_compat: Dict[str, Dict[str, Any]] = {} failed = 0 for group_id in order: group_spec = groups_spec.get(group_id) if not group_spec: logger.warning( "Phase 5 planner: group_id %s in chain_bg_order not found in groups — skip", group_id, ) continue loc_id = group_spec.get("location_id", "") or "" location_group = _build_group_location_group( group_spec, shot_meta_idx, locs_meta, ) floor_plan_used = bool( (floor_plan_prompts or {}).get(loc_id, "").strip() ) # 부모 group의 chain bg context — 없으면 빈 문자열 parent_id = (group_spec.get("parent_id") or "").strip() parent_ctx = parent_prompts.get(parent_id, "") if parent_id else "" if not location_group.shots: logger.warning( "Phase 5 planner: group %s has no resolvable shots — skipping LLM call", group_id, ) result = { "group_id": group_id, "location_id": loc_id, "location_name": location_group.location_name, "shot_count": 0, "error": "no resolvable shots in group", "floor_plan_used": floor_plan_used, "parent_id": parent_id, "parent_context_used": False, "status": "exception", } results_by_group[group_id] = result failed += 1 # compat alias도 채워둔다 (loc_id가 있을 때만) if loc_id: locations_compat[loc_id] = _to_legacy_location_shape(result) continue try: plan = _plan_one_location( location_group, world_rules_excerpt, opik_metadata, floor_plan_prompts=floor_plan_prompts, parent_chain_bg_ctx=parent_ctx, ) errors = validate_plan(plan, location_group) if errors: logger.error( "Phase 5 planner: group %s validation FAILED — %d issues:\n %s", group_id, len(errors), "\n ".join(errors), ) result = { "group_id": group_id, "location_id": loc_id, "location_name": location_group.location_name, "shot_count": len(location_group.shots), "validation_errors": errors, "plan": plan, "floor_plan_used": floor_plan_used, "parent_id": parent_id, "parent_context_used": bool(parent_ctx), "status": "failed", } results_by_group[group_id] = result failed += 1 elif plan.get("skip_chain") is True: logger.info( "Phase 5 planner: group %s — LLM skip_chain (outdoor)", group_id, ) result = { "group_id": group_id, "location_id": loc_id, "location_name": location_group.location_name, "shot_count": len(location_group.shots), "skip_chain": True, "skip_reason": plan.get("skip_reason", "outdoor open-air"), "rationale_summary": plan.get("rationale_summary", ""), "nodes": [], "execution_order": [], "unassigned_shots": [], "floor_plan_used": floor_plan_used, "parent_id": parent_id, "parent_context_used": bool(parent_ctx), "status": "skipped", } results_by_group[group_id] = result else: result = { "group_id": group_id, "location_id": loc_id, "location_name": location_group.location_name, "shot_count": len(location_group.shots), "skip_chain": False, "skip_reason": "", "rationale_summary": plan.get("rationale_summary", ""), "nodes": plan.get("nodes", []), "execution_order": plan.get("execution_order", []), "unassigned_shots": plan.get("unassigned_shots", []), "floor_plan_used": floor_plan_used, "parent_id": parent_id, "parent_context_used": bool(parent_ctx), "status": "ok", } results_by_group[group_id] = result # 후속 자식 group을 위해 부모 컨텍스트 저장 parent_prompts[group_id] = _summarize_parent_for_continuity(result) except Exception as exc: logger.error( "Phase 5 planner: group %s exception: %s", group_id, exc, exc_info=True, ) result = { "group_id": group_id, "location_id": loc_id, "location_name": location_group.location_name, "shot_count": len(location_group.shots), "error": str(exc), "floor_plan_used": floor_plan_used, "parent_id": parent_id, "parent_context_used": bool(parent_ctx), "status": "exception", } results_by_group[group_id] = result failed += 1 # compat alias 갱신 — 동일 loc_id에 여러 group이 있을 때 ok > others 우선. if loc_id: existing = locations_compat.get(loc_id) new_shape = _to_legacy_location_shape(result) if existing is None: locations_compat[loc_id] = new_shape else: # ok가 우선, 그 다음은 마지막 등장 — 부분 fail이어도 ok 결과 유지. if existing.get("status") != "ok" or new_shape.get("status") == "ok": locations_compat[loc_id] = new_shape skipped_total = sum(1 for v in results_by_group.values() if v.get("status") == "skipped") logger.info( "background_chain_planning (planner-driven): done — " "%d groups (%d skipped, %d failed) across %d locations", len(results_by_group), skipped_total, failed, len(locations_compat), ) return { "groups": results_by_group, "locations": locations_compat, "_failed_count": failed, } def _to_legacy_location_shape(group_result: Dict[str, Any]) -> Dict[str, Any]: """group result → Phase 4 chain_bg_render가 읽는 location result shape. 필드 매핑은 legacy `_process_one`의 ok/skipped/failed/exception 분기와 동일. group_id / parent_id 등 Phase 5 전용 필드는 보존돼 downstream이 읽을 수 있다. """ base = { "location_id": group_result.get("location_id", ""), "location_name": group_result.get("location_name", ""), "shot_count": group_result.get("shot_count", 0), "floor_plan_used": group_result.get("floor_plan_used", False), "status": group_result.get("status", "exception"), # Phase 5 traceability 필드 (chain_bg_render는 이 키들을 무시함) "group_id": group_result.get("group_id", ""), "parent_id": group_result.get("parent_id", ""), "parent_context_used": group_result.get("parent_context_used", False), } status = base["status"] if status == "ok": base.update({ "skip_chain": False, "skip_reason": "", "rationale_summary": group_result.get("rationale_summary", ""), "nodes": group_result.get("nodes", []) or [], "execution_order": group_result.get("execution_order", []) or [], "unassigned_shots": group_result.get("unassigned_shots", []) or [], }) elif status == "skipped": base.update({ "skip_chain": True, "skip_reason": group_result.get("skip_reason", ""), "rationale_summary": group_result.get("rationale_summary", ""), "nodes": [], "execution_order": [], "unassigned_shots": [], }) elif status == "failed": base.update({ "validation_errors": group_result.get("validation_errors", []), "plan": group_result.get("plan"), }) else: # exception base.update({ "error": group_result.get("error", ""), }) return base