"""scene_consistency StepRunner — 씬 내 교차 샷 시각적 일관성 고정 요소 추출.""" import logging import os from concurrent.futures import ThreadPoolExecutor, as_completed from typing import Any, Dict, List, Optional, Set, Tuple from app.core.errors import AppError from app.core.step_runner import StepRunner from app.core.steps._evidence_helpers import ( _normalize_scene_consistency_result, assert_fresh_llm_evidence, ) from app.core.steps.detail_steps import _DetailStepMixin from app.modules.llm.llm_client import call_structured from app.modules.llm.safety import sanitize_for_safety as _sanitize_for_safety # legacy alias from app.modules.prompt_loader import load_prompt, load_schema logger = logging.getLogger(__name__) # 글로벌 3-tier fallback layer (`app.modules.llm.llm_client.call_structured`)가 # sanitize + GPT fallback을 자동 처리한다. step에서는 단일 호출로 충분. # `_sanitize_for_safety` alias는 외부 import(detail_steps 등) 호환을 위해 보존한다. # Group 2 #1 (visual_pipeline_contracts_plan) — scene 처리 결과의 명시적 status. # LLM 출력에는 없는 코드 부여 필드. consumer (detail_steps._load_fixed_elements, # shot_dependency_t2i_step) 는 ``fixed_elements`` 만 읽으므로 status 추가는 안 깨짐. # verify_completion 이 status 기반으로 partial/clean 분류 → silent fallback 차단. STATUS_OK = "ok" # LLM 정상 결과 STATUS_SKIPPED_SINGLE_SHOT = "skipped_single_shot" # 1 샷만 선택 → 교차 일관성 N/A STATUS_SKIPPED_NO_SELECTION = "skipped_no_selection" # selected_shot_indices=[] (의도적 deselect) STATUS_BLOCKED_NO_SELECTION = "blocked_no_selection" # shot_selection cp 자체 누락 — partial STATUS_FAILED_ALL_TIERS = "failed_all_tiers" # gemini→sanitize→gpt 모두 fail — partial STATUS_VALIDATOR_VIOLATIONS = "validator_violations" # G2.2 framing × element conflict (partial) STATUS_CONTRACT_VIOLATION = "contract_violation" # G3.1 evidence/inference 4-field 위반 (partial) # G3.1 (Codex BLOCKING 1): cp invalidation 보장 — manifest schema_version 와 일치 필수. # Area #4 (2026-05-18): 2→3 (element_scope enum required — fixed_elements item 에 # "full" | "close" SOT). 옛 v6 cp 는 schema_version 2 → mismatch 로 cp invalidation # 강제 v7 rerun. v6 의 code-side close/full inference helper trio 는 모두 폐기. SCENE_CONSISTENCY_SCHEMA_VERSION = 3 # Group 2 #2 (visual_pipeline_contracts_plan) — framing × fixed_element deterministic validator. # Area #4 (2026-05-18) — element_scope LLM emit SOT 로 전환: # - 옛 v6 의 code-side close/full inference (regex + keyword + classifier 3 helper) # 모두 폐기 (Gate 1 — semantic regex ban). # - producer (scene_consistency v7) 가 `element_scope: "full" | "close"` enum 을 # schema required 로 emit. code 는 enum 직접 비교만 (Gate 3 — structured SOT). # - `_detect_framing_conflicts` 는 element_scope 직접 read 후 비교. def _normalize_applies_to_shots(raw: Any) -> Set[int]: """G2.2 helper: applies_to_shots → int set (LLM 변형 출력 강건 처리). iter#1 review fix (Codex MINOR): scalar / non-iterable / str 입력 안전 default 빈 set. str 은 iterable 이지만 char-by-char iter 가 의미 없음 → 명시적 거부. """ out: Set[int] = set() if raw is None: return out if isinstance(raw, (str, bytes)) or not hasattr(raw, "__iter__"): # malformed cp / LLM scalar 출력 — 빈 set return (caller 가 0 overlap 판정). return out for v in raw: try: out.add(int(v)) except (TypeError, ValueError): continue return out def _detect_framing_conflicts(scene_result: Dict[str, Any]) -> List[Dict[str, Any]]: """G2.2 (Area #4 v7): element_scope enum 직접 비교 — close/full overlap 위반 검출. 같은 character_name 의 character_state element 쌍 중 ``applies_to_shots`` 가 겹치고 ``element_scope`` enum 값이 다른 위반 검출. v6 의 code-side regex/keyword inference 는 폐기됐고, producer (v7 prompt + schema) 가 emit 한 enum 신뢰 (Gate 3). fail-fast (Gate 4): - ``element_scope`` 누락 → ``KeyError`` (schema violation, call_structured retry 유도) - enum 외 값 → ``AppError(step.contract_violation.scene_consistency.element_scope)`` Returns: violation list. 각 항목은 충돌이 발견된 (shot_index, character_name, element_id 쌍, framings 쌍) 정보. 발견 0 = clean. Metadata key 의 ``"framings"`` 는 legacy 호환 유지 — 값만 element_scope enum (``"full"`` | ``"close"``). """ fixed = scene_result.get("fixed_elements", []) or [] if not fixed: return [] # 1) character_state 만 + character_name 별 그룹핑. by_char: Dict[str, List[Dict[str, Any]]] = {} for fe in fixed: if not isinstance(fe, dict): continue if fe.get("element_type") != "character_state": continue name = (fe.get("character_name") or "").strip() if not name: continue by_char.setdefault(name, []).append(fe) # 2) 같은 character 의 element 쌍 중 applies_to_shots 겹침 + framing 다름 검출. violations: List[Dict[str, Any]] = [] for name, elements in by_char.items(): if len(elements) < 2: continue for i in range(len(elements)): for j in range(i + 1, len(elements)): e1, e2 = elements[i], elements[j] shots1 = _normalize_applies_to_shots(e1.get("applies_to_shots", [])) shots2 = _normalize_applies_to_shots(e2.get("applies_to_shots", [])) overlap = shots1 & shots2 if not overlap: continue # Area #4 v7 (Codex iter 1 C2 fix): element_scope enum 직접 read. # required by v7 schema, KeyError = schema violation (Gate 4 fail-fast). s1 = e1["element_scope"] s2 = e2["element_scope"] if s1 not in ("full", "close") or s2 not in ("full", "close"): raise AppError( code="step.contract_violation.scene_consistency.element_scope", message=( f"scene_consistency: invalid element_scope pair " f"{s1!r}/{s2!r} for {name!r}" ), ) # 서로 다른 element_scope 만 conflict (close vs full = 시각 중복 위험). if s1 == s2: continue for shot_idx in sorted(overlap): violations.append({ "shot_index": shot_idx, "character_name": name, "elements": [ e1.get("element_id", ""), e2.get("element_id", ""), ], # legacy metadata key "framings" 보존 (Codex iter 1 I2 — rename 0). # 값만 v7 element_scope enum ("full" | "close") 에서 옴. "framings": [s1, s2], }) return violations def is_scene_result_consumer_safe(scene_result: Dict[str, Any]) -> bool: """G2.1+G2.2 consumer-side defense single-source helper (iter#4 review fix). consumer (scene_context_loader, shot_dependency_t2i_step) 가 ``fixed_elements`` 를 LLM prompt 에 주입하기 전에 호출. ``False`` 면 skip + warning. 2-tier 검사: 1) status 직접 차단 — failed_all_tiers / blocked_no_selection / validator_violations. 2) status='ok' 또는 status=None 옛 cp — ``_detect_framing_conflicts`` 직접 호출 (downstream-only force-run 시 dispatcher gate 통과 path 보호). consumer 마다 같은 가드 로직 복제 → drift 위험. 본 helper 가 single source. """ status = (scene_result.get("status") or "") if status in ( STATUS_FAILED_ALL_TIERS, STATUS_BLOCKED_NO_SELECTION, STATUS_VALIDATOR_VIOLATIONS, STATUS_CONTRACT_VIOLATION, ): return False if _detect_framing_conflicts(scene_result): return False return True class SceneConsistencyStep(_DetailStepMixin, StepRunner): """씬 내 2+ 샷에 걸친 고정 시각 요소 추출 (scene_detail 직전 실행).""" def _execute(self, mode="resume") -> Dict[str, Any]: # ── 의존 체크포인트 로드 ── scene_save_cp = self._load_prev_checkpoint("scene_save") segments = scene_save_cp.get("data", {}).get("segments", []) if scene_save_cp else [] shot_extract_cp = self._load_prev_checkpoint("shot_validator") shot_selection_cp = self._load_prev_checkpoint("shot_selection") if not shot_extract_cp or not shot_selection_cp: raise AppError( code="step.no_input", message="shot_extract 또는 shot_selection 결과 없음", status_code=400, ) if not shot_extract_cp.get("data", {}).get("scenes"): raise AppError( code="step.no_input", message="shot_extract data.scenes가 비어있음", status_code=400, ) if not shot_selection_cp.get("data", {}).get("scenes"): raise AppError( code="step.no_input", message="shot_selection data.scenes가 비어있음", status_code=400, ) beat_extract_cp = self._load_prev_checkpoint("beat_extract") staging_cp = self._load_prev_checkpoint("shot_staging") shot_director_cp = self._load_prev_checkpoint("shot_director") entity_merge_cp = self._load_prev_checkpoint("entity_merge") # ── 선택된 샷 그룹핑 ── selected_map: Dict[int, set] = {} if shot_selection_cp and shot_selection_cp.get("data", {}).get("scenes"): for sc in shot_selection_cp["data"]["scenes"]: selected_map[sc["scene_index"]] = set(sc.get("selected_shot_indices", [])) shots_by_scene: Dict[int, List[dict]] = {} # G2.1: shot_selection 누락 씬은 "단일 샷" silent fallback 으로 흘러가지 # 않도록 별도 set 으로 추적 — 처리 단계에서 BLOCKED_NO_SELECTION status 부여. blocked_no_selection_scenes: Set[int] = set() if shot_extract_cp and shot_extract_cp.get("data", {}).get("scenes"): for sc in shot_extract_cp["data"]["scenes"]: si = sc["scene_index"] sel = selected_map.get(si) if sel is None: # shot_selection 체크포인트에서 해당 씬이 누락된 경우 — 명시적 blocked # 마킹. "전체 포함" 기본값은 위험 (선택을 명시적으로 빈 set 으로 만든 # 경우와 구분 불가). 옛 silent skip 패턴 (단일 샷처럼 처리) 차단. logger.warning( "scene_consistency: scene %d missing from shot_selection — blocked", si, ) blocked_no_selection_scenes.add(si) shots_by_scene[si] = [] else: shots_by_scene[si] = [ sh for sh in sc.get("shots", []) if sh.get("shot_index") in sel ] # beat 데이터 beats_by_scene: Dict[int, Dict[int, dict]] = {} if beat_extract_cp and beat_extract_cp.get("data", {}).get("scenes"): for sc in beat_extract_cp["data"]["scenes"]: si = sc.get("scene_index") beats_by_scene[si] = { b["beat_index"]: b for b in sc.get("beats", []) if b.get("beat_index") } # staging 데이터 staging_map: Dict[str, dict] = {} if staging_cp and staging_cp.get("data", {}).get("shots"): for st in staging_cp["data"]["shots"]: key = f"{st.get('scene_index')}_{st.get('shot_index')}" staging_map[key] = st # shot_director VE 데이터 shot_director_ve: Dict[tuple, list] = {} 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", []): shot_director_ve[(si, sh["shot_index"])] = sh.get("visible_entity_ids", []) # 엔티티 이름 매핑 (short_id → name) entity_names: Dict[str, str] = {} if entity_merge_cp and entity_merge_cp.get("data"): for etype in ["characters", "locations", "props"]: for e in entity_merge_cp["data"].get(etype, []): sid = e.get("short_id", "") if sid: entity_names[sid] = e.get("name", "") # 씬 텍스트 매핑 seg_by_scene: Dict[int, dict] = {} for seg in segments: seg_by_scene[seg.get("scene_index", 0)] = seg # ── 프롬프트 로드 ── system_prompt = load_prompt("scene_consistency", "system") schema = load_schema("scene_consistency", "schema") # visual_world_rules → system prompt에 지역/인종 맥락 주입 rules_cp = self._load_prev_checkpoint("visual_world_rules") if rules_cp and rules_cp.get("data"): rd = rules_cp["data"] world_lines = [] if rd.get("era"): world_lines.append(f"시대: {rd['era']}") if rd.get("region"): world_lines.append(f"지역/국가: {rd['region']}") if world_lines: system_prompt += ( "\n\n## 세계관 맥락 (인물 묘사 시 인종/국적 참고)" "\n" + "\n".join(world_lines) ) # ── resume: 기존 결과에서 성공한 씬 보존 ── # G2.1 (Codex BLOCKING + Claude BLOCKING fix): status field 우선. # 옛 prefix-only filter 는 (1) ``analysis_summary`` text drift (정확 prefix 위반) # 시 실패 결과 슬립 + (2) blocked 결과 가 prior cp 에 있으면 그대로 보존 → # silent fallback 회귀. status 가 있으면 그것을 신뢰, 없으면 옛 cp prefix fallback. # G2.2: validator_violations 도 retry 대상 (LLM 재호출이 위반 해소 가능). _retry_statuses = ( STATUS_FAILED_ALL_TIERS, STATUS_BLOCKED_NO_SELECTION, STATUS_VALIDATOR_VIOLATIONS, STATUS_CONTRACT_VIOLATION, ) existing_cp = self._load_prev_checkpoint("scene_consistency") existing_ok: Dict[int, dict] = {} # scene_index → result (정상/skipped 만) if mode == "resume" and existing_cp and existing_cp.get("data", {}).get("scenes"): for sc in existing_cp["data"]["scenes"]: si = sc.get("scene_index") sc_status = sc.get("status") if sc_status in _retry_statuses: # 신 status — 명시적 partial → 재시도 (선택이 채워졌을 수도 있음). continue if sc_status is None: # backward-compat: 옛 cp 는 status 없음. analysis_summary prefix 로 # "분석 실패" / "분석 차단" 둘 다 retry 대상으로 분류. summary = sc.get("analysis_summary", "") or "" if summary.startswith("분석 실패") or summary.startswith("분석 차단"): continue existing_ok[si] = sc if existing_ok: logger.info("scene_consistency resume: %d scenes already done, skipping", len(existing_ok)) # ── 씬별 처리 (병렬) ── scene_results: List[dict] = [] processed = 0 skipped = 0 failed = 0 blocked = 0 # 1) 빠른 경로: blocked · 단일 샷 · 기존 성공 씬 재사용 — 직렬 처리 후 분리 tasks: List[Tuple[int, List[dict]]] = [] for si, shots in sorted(shots_by_scene.items()): # G2.1 (Codex IMPORTANT fix): blocked / single_shot 분기는 existing_ok # 무시 — 현재 input 진실 (blocked / single shot) 이 prior ok 결과 보다 # 우선. 옛 결과 보존하면 stale ok 가 silent 통과 → fail-fast 위반. if si in blocked_no_selection_scenes: scene_results.append({ "scene_index": si, "analysis_summary": "분석 차단 — shot_selection 누락", "fixed_elements": [], "status": STATUS_BLOCKED_NO_SELECTION, }) blocked += 1 continue # G2.1 (Codex MINOR fix): selected_shot_indices=[] 명시적 deselect 와 # selected_shot_indices=[1] 단일 샷은 의미가 다름. detail_steps L317 # ("selected_shot_indices=[] → 전부 deselect → 이 씬 스킵") 와 같은 맥락. # blocked (cp 부재) 와도 구분 — 사용자/LLM 이 명시적으로 0 선택. if len(shots) == 0: scene_results.append({ "scene_index": si, "analysis_summary": "스킵 — selected_shot_indices=[] (명시적 deselect)", "fixed_elements": [], "status": STATUS_SKIPPED_NO_SELECTION, }) skipped += 1 continue if len(shots) < 2: scene_results.append({ "scene_index": si, "analysis_summary": "단일 샷 — 교차 일관성 분석 불필요", "fixed_elements": [], "status": STATUS_SKIPPED_SINGLE_SHOT, }) skipped += 1 continue if si in existing_ok: # G3.1 (Claude BLOCKING 1): existing_ok reuse 직전 normalize. # 옛 v5 cp 에 confidence 만 부분적으로 있는 케이스에서 contract # validator (L467) 가 false positive 로 retry 격상하는 회귀 차단. # normalize 가 4 필드 default + confidence='legacy' 마킹 → contract # validator 의 두 번째 skip 가드 (legacy) 통과. _normalize_scene_consistency_result( existing_ok[si], where="scene_consistency.existing_ok_reuse", ) scene_results.append(existing_ok[si]) processed += 1 continue tasks.append((si, shots)) # 2) 병렬 LLM 처리: 씬 간 독립 → ThreadPoolExecutor if tasks: _max_workers = int(os.environ.get("SCENE_CONSISTENCY_WORKERS", "5")) _max_workers = max(1, min(_max_workers, len(tasks))) logger.info( "scene_consistency: %d scenes need LLM (workers=%d, already_done=%d, single_shot_skipped=%d)", len(tasks), _max_workers, processed, skipped, ) with ThreadPoolExecutor(max_workers=_max_workers) as pool: futures = { pool.submit( self._process_one_scene, si, shots, seg_by_scene.get(si, {}).get("text", ""), system_prompt, schema, beats_by_scene, shot_director_ve, entity_names, staging_map, ): si for si, shots in tasks } for f in as_completed(futures): result, is_failed = f.result() scene_results.append(result) if is_failed: failed += 1 else: processed += 1 scene_results.sort(key=lambda r: r.get("scene_index", 0)) # G2.2: framing × fixed_element deterministic post-validator 일괄 적용. # status==ok 또는 backward-compat (status 없는 옛 cp) 의 정상 결과만 검증 — failed/ # blocked/skipped 는 fixed_elements 가 비어있어 의미 없음. 위반 발견 시 result 의 # status 를 STATUS_VALIDATOR_VIOLATIONS 로 격상 + violations 메타데이터 추가. # fixed_elements 자체는 보존 (LLM 결과 silent strip 안 함 — 정직한 노출). validator_violation_count = 0 for r in scene_results: r_status = r.get("status") # status 부재 (옛 cp) + 정상 케이스 ("분석 실패" / "분석 차단" prefix 아님) 도 검증. if r_status not in (None, STATUS_OK): continue if r_status is None: summary = (r.get("analysis_summary", "") or "") if summary.startswith("분석 실패") or summary.startswith("분석 차단"): continue violations = _detect_framing_conflicts(r) if violations: r["validator_violations"] = violations r["status"] = STATUS_VALIDATOR_VIOLATIONS validator_violation_count += 1 logger.warning( "scene_consistency S%s: %d framing conflict(s) detected", r.get("scene_index"), len(violations), ) # ★위반을 **알려 주고** 한 번 다시 묻는다 (2026-09-18 컨트리로드). # 같은 입력으로 다시 물으면 같은 답이다 — S42 가 resume 두 번 모두 같은 충돌 # (같은 인물의 full·close 요소가 같은 샷에 겹침)로 서서 주행이 멈췄다. # 못 고치면 위반 결과를 그대로 둔다(partial — 조용히 지우지 않는다). # ★이번에 새로 물은 씬만 — 재사용한 옛 결과는 모델을 부르지 않는다(다음 resume 이 다시 묻는다). asked_now = {t_si for t_si, _ in tasks} for idx, r in enumerate(scene_results): if r.get("status") != STATUS_VALIDATOR_VIOLATIONS or r["scene_index"] not in asked_now: continue si = r["scene_index"] retried, retry_failed = self._process_one_scene( si, shots_by_scene.get(si, []), seg_by_scene.get(si, {}).get("text", ""), system_prompt, schema, beats_by_scene, shot_director_ve, entity_names, staging_map, corrections=r["validator_violations"], ) if retry_failed or _detect_framing_conflicts(retried): r["status"] = STATUS_VALIDATOR_VIOLATIONS logger.warning("scene_consistency S%s: 충돌을 알려 주고 다시 물어도 남았다", si) continue scene_results[idx] = retried validator_violation_count -= 1 logger.info("scene_consistency S%s: 충돌을 알려 주고 다시 물어 해소", si) # G3.1: post-parse contract validator (evidence/inference 4-field). # LLM strict schema 가 4 필드 존재만 강제 — contract consistency 추가 검증 # (confidence='legacy' 출력 / source_facts=[] 인데 confidence!=low 등) 은 # assert_fresh_llm_evidence 가 AppError 로 신호. status 격상 + retry 대상. # G2.2 validator 와 동일 패턴 — fixed_elements 보존, status 만 격상. contract_violation_count = 0 for r in scene_results: r_status = r.get("status") # 이미 다른 partial status 면 skip (validator/blocked/failed). if r_status not in (None, STATUS_OK): continue if r_status is None: summary = (r.get("analysis_summary", "") or "") if summary.startswith("분석 실패") or summary.startswith("분석 차단"): continue contract_errors: List[Dict[str, Any]] = [] for element in r.get("fixed_elements", []) or []: # 옛 cp / 옛 fixture (4 필드 전부 부재) — skip. # _execute 후처리는 새 LLM 결과 strict 검증이 목적. 옛 cp lazy # backfill 은 verify_completion 의 normalize 가 처리 (4 필드 default # + confidence='legacy' 마킹). existing_ok reuse 된 옛 v5 cp 도 여기. if not any(f in element for f in ( "source_facts", "visual_inferences", "creative_decisions", "confidence" )): continue # normalize 가 마킹한 옛 cp adapter — strict assert 우회 (legacy 는 # LLM 출력 금지 sentinel). assert_fresh 호출하면 violation 으로 # 잡혀 옛 cp 가 partial 로 격상되는 회귀. if element.get("confidence") == "legacy": continue try: assert_fresh_llm_evidence(element, "scene_consistency") except AppError as exc: if exc.code != "step.contract_violation": raise contract_errors.append({ "element_id": element.get("element_id"), "error": exc.message, }) if contract_errors: r["contract_violations"] = contract_errors r["status"] = STATUS_CONTRACT_VIOLATION contract_violation_count += 1 logger.warning( "scene_consistency S%s: %d contract violation(s) detected", r.get("scene_index"), len(contract_errors), ) total = processed + skipped + failed + blocked # G2.1: blocked 는 정직하게 fail 카운트에 합산 (silent skip 차단). UI/리포트 가 # completed_count 만 보면 blocked 가 invisible 했던 회귀 패턴 차단. # G2.2 (Claude MINOR fix): validator_violation_count 도 completed_count 에서 차감 # — LLM 정상 결과 인데 framing 충돌로 격상된 scene 을 completed 로 두면 신호 # 불일치. dispatcher cascade 는 별도 (allow_partial_downstream=False + verify # partial). 사용자/UI 가 보는 completed 는 진짜 clean 결과만. return { "completed_count": (processed + skipped) - validator_violation_count - contract_violation_count, "applicable_count": total, "failed_count": failed + blocked, # G3.1 (Codex BLOCKING 1): manifest schema_version 와 일치. cp 에 명시 # 보존되어야 step_runner P0-3 가 옛 cp(schema=1)를 mismatch 로 reject. "schema_version": SCENE_CONSISTENCY_SCHEMA_VERSION, "blocked_count": blocked, "validator_violation_count": validator_violation_count, "contract_violation_count": contract_violation_count, "data": {"scenes": scene_results}, } def verify_completion(self): """Group 1 #3 + Group 2 #1 (visual_pipeline_contracts_plan): 산출물 검증. scene 별 ``status`` 필드 (G2.1) 우선: - ``failed_all_tiers`` → partial (3-tier 모두 실패) - ``blocked_no_selection`` → partial (shot_selection 누락 silent skip 차단) - ``validator_violations`` → partial (G2.2 framing × fixed_element 충돌) - ``skipped_single_shot`` / ``skipped_no_selection`` / ``ok`` → clean backward-compat: status 없는 옛 cp 는 ``analysis_summary`` prefix 로 fallback — "분석 실패" → failed, "분석 차단" → blocked. (G1.3 startswith 정렬 유지.) source 우선순위: step_runner 가 보존한 ``_last_execute_result`` (Codex review Critical: exit verify 가 save_checkpoint 전 호출 → fresh run 에서 cp 부재) → cp ``_load_prev_checkpoint`` fallback. 둘 다 없으면 missing. fail-fast 정책 — 빈 결과 (``fixed_elements: []``) silent 통과 차단. """ from app.core.integrity_report import CompletionReport # Codex review Critical: exit verify 시점 fresh run 에서 cp 가 아직 없음. # step_runner.run() 에서 _last_execute_result 로 보존된 결과 우선 사용. result = getattr(self, "_last_execute_result", None) if result is None: cp = self._load_prev_checkpoint("scene_consistency") if not cp: return CompletionReport( is_complete=False, missing=["scene_consistency cp + result 모두 부재"], severity="missing", metadata={ "total_scenes": 0, "failed_scenes": [], "blocked_scenes": [], }, ) result = cp scene_results = result.get("data", {}).get("scenes", []) or [] # G3.1: 옛 cp (4 필드 누락) lazy backfill — fixed_elements 각 element 에 # source_facts/visual_inferences/creative_decisions/confidence default 주입, # confidence='legacy' 마킹. 새 LLM 출력은 _execute 가 strict schema + assert # 로 보장하므로 normalize 가 도달해도 noop. for r in scene_results: _normalize_scene_consistency_result( r, where="scene_consistency.verify_completion" ) failed_scenes: List[int] = [] blocked_scenes: List[int] = [] violation_scenes: List[int] = [] contract_scenes: List[int] = [] for r in scene_results: si = r.get("scene_index") status = r.get("status") if status == STATUS_FAILED_ALL_TIERS: failed_scenes.append(si) continue if status == STATUS_BLOCKED_NO_SELECTION: blocked_scenes.append(si) continue if status == STATUS_VALIDATOR_VIOLATIONS: # G2.2: framing × fixed_element 충돌 — partial. violation_scenes.append(si) continue if status == STATUS_CONTRACT_VIOLATION: # G3.1: evidence/inference 4-field contract 위반 — partial. contract_scenes.append(si) continue if status is None: # backward-compat: 옛 cp (status 필드 없음) — analysis_summary prefix. summary = (r.get("analysis_summary", "") or "") if summary.startswith("분석 실패"): failed_scenes.append(si) continue if summary.startswith("분석 차단"): blocked_scenes.append(si) continue # G2.2 (Codex BLOCKING fix): status="ok" 또는 status=None+clean-summary 인 # scene 의 fixed_elements 도 verify_completion 에서 직접 검증. step_runner.run # 의 entry verify path 가 완료된 cp 에 대해 _execute 호출 안 하므로 (skip # 분기), validator 가 _execute 안에만 있으면 옛 cp 가 silent clean 으로 # 통과하는 회귀 발생. 두 곳에서 동일 검증 — performance 영향 0 (deterministic # in-memory check), 안전성 ↑. if status in (None, STATUS_OK): if _detect_framing_conflicts(r): violation_scenes.append(si) total = len(scene_results) issues_count = ( len(failed_scenes) + len(blocked_scenes) + len(violation_scenes) + len(contract_scenes) ) if issues_count: missing_lines: List[str] = [] if failed_scenes: missing_lines.append( f"{len(failed_scenes)}/{total} scene(s) all-tier failed: " f"{failed_scenes[:5]}" ) if blocked_scenes: missing_lines.append( f"{len(blocked_scenes)}/{total} scene(s) blocked " f"(shot_selection 누락): {blocked_scenes[:5]}" ) if violation_scenes: missing_lines.append( f"{len(violation_scenes)}/{total} scene(s) framing conflict " f"(G2.2 validator): {violation_scenes[:5]}" ) if contract_scenes: missing_lines.append( f"{len(contract_scenes)}/{total} scene(s) evidence contract " f"violation (G3.1): {contract_scenes[:5]}" ) return CompletionReport( is_complete=False, missing=missing_lines, severity="partial", metadata={ "total_scenes": total, "failed_scenes": failed_scenes, "blocked_scenes": blocked_scenes, "violation_scenes": violation_scenes, "contract_scenes": contract_scenes, }, ) return CompletionReport( is_complete=True, missing=[], severity="clean", metadata={ "total_scenes": total, "failed_scenes": [], "blocked_scenes": [], "violation_scenes": [], "contract_scenes": [], }, ) def _process_one_scene( self, si: int, shots: List[dict], scene_text: str, system_prompt: str, schema: dict, beats_by_scene: Dict[int, Dict[int, dict]], shot_director_ve: Dict[tuple, list], entity_names: Dict[str, str], staging_map: Dict[str, dict], corrections: Optional[List[Dict[str, Any]]] = None, ) -> Tuple[dict, bool]: """한 씬 처리 (Gemini → sanitize fallback → GPT fallback). Returns: (result_dict, is_failed). result_dict는 scene_results에 바로 append 가능한 형태 (scene_index 포함). is_failed는 3-tier 모두 실패한 경우만 True. ``corrections`` — 직전 답의 G2.2 위반(`_detect_framing_conflicts` 결과). 있으면 그 자리만 짚어 다시 묻는다. 규칙은 system 의 「scope 중복 금지」 그대로다. """ # user prompt 구성 user_prompt = "" if corrections: user_prompt += "[수정 요청] 직전 답이 scope 중복 금지를 어겼다:\n" + "".join( f"- {v.get('character_name')}: " + " · ".join(f"{e}({f})" for e, f in zip(v.get("elements", []), v.get("framings", []))) + f" 가 Shot {v.get('shot_index')} 에 함께 적용됨\n" for v in corrections ) + "\n" user_prompt += f"씬 {si}번 분석\n\n" user_prompt += f"씬 텍스트:\n{scene_text}\n\n" user_prompt += f"[선택된 샷 목록 — {len(shots)}개]\n" for sh in shots: sh_idx = sh.get("shot_index", 0) user_prompt += f"\n--- Shot {sh_idx} ---\n" user_prompt += f"설명: {sh.get('description', '')}\n" beat_idx = sh.get("based_on_beat") if beat_idx: beat_data = beats_by_scene.get(si, {}).get(beat_idx) if beat_data: user_prompt += ( f"Beat #{beat_idx} [{beat_data.get('change_type', '')}]: " f"{beat_data.get('before_state', '')} → {beat_data.get('after_state', '')}\n" ) ve_list = shot_director_ve.get((si, sh_idx), []) if ve_list: ve_names = [] for vid in ve_list: name = entity_names.get(vid, "") ve_names.append(f"{vid}({name})" if name else vid) user_prompt += f"등장 요소: {', '.join(ve_names)}\n" staging_key = f"{si}_{sh_idx}" staging = staging_map.get(staging_key) if staging: cam = staging.get("camera_direction", "") light = staging.get("lighting_mood", "") if cam: user_prompt += f"카메라: {cam}\n" if light: user_prompt += f"조명: {light}\n" char_angles = staging.get("character_angles", []) if char_angles: for a in char_angles: _cn = a.get("character", "") _angle = a.get("angle", "") _pose = a.get("body_pose", "") # Area #2 W6: v13 3 field labeling — gaze_direction_kind + gaze_target_id + subject_state # raw enum/ID label 그대로 노출 (code semantic translation 금지, Gate 1). _kind = a["gaze_direction_kind"] # required by v13 schema, KeyError = schema violation (Gate 4 fail-fast) _target_id = a.get("gaze_target_id") or "" # conditional nullable/omitted OK (schema oneOf) _state = a["subject_state"] # required by v13 schema, KeyError = schema violation (Gate 4 fail-fast) _gaze_str = f", gaze={_kind}" + (f"→{_target_id}" if _target_id else "") _state_str = f", state={_state}" user_prompt += f" 인물 배치: {_cn} — {_angle}, {_pose}{_gaze_str}{_state_str}\n" user_prompt += ( "\n[지시]\n" "위 샷들을 분석하여, 2개 이상의 샷에 걸쳐 동일해야 할 시각적 요소를 추출하세요.\n" "특히 다음에 주의:\n" "- subject_state가 immobilized인 인물의 자세와 위치\n" "- key_bg_elements/state가 명시한 환경 상태\n" "- shot 간 고정되어야 하는 registered prop/background element 위치\n" "- 샷 사이에 상태가 바뀌는 요소는 제외\n" "- description에 엔티티 ID(C##, L##, P##)를 절대 사용하지 마세요. 보통명사로만 묘사하세요.\n" ) # 글로벌 3-tier fallback (call_structured 내부) — gemini → sanitize → gpt 자동. try: result = call_structured( step="scene_consistency", system_prompt=system_prompt, user_prompt=user_prompt, response_schema=schema, project_config=self.project_config, schema_name=f"scene_consistency_{si}" + ("_retry" if corrections else ""), opik_metadata=self.build_opik_metadata( extra_tags=["retry"] if corrections else None, extra_metadata={"scene_index": si}), ) result["scene_index"] = si # G2.1: ok status 명시 (LLM 출력엔 없음, 코드가 부여). result["status"] = STATUS_OK logger.info("scene_consistency S%d: %d fixed elements", si, len(result.get("fixed_elements", []))) return result, False except Exception as exc: logger.error("scene_consistency S%d all tiers failed: %s", si, exc) return { "scene_index": si, "analysis_summary": "분석 실패", "fixed_elements": [], "status": STATUS_FAILED_ALL_TIERS, }, True