"""shot_staging — 샷별 촬영 연출 + 배경 중요 요소 분석. 촬영감독(DP) 역할로 각 샷의 구성요소를 분석하고 창의적 카메라 연출 + 조명 + 배경 핵심 요소를 결정. Patch C / Area A — content_surface / reflective_surface element 의 orientation 이 비면 batch retry (max_attempts=3) 후 소진 시 ShotStagingOrientationError raise. validator 는 call_structured try/ except **밖** 에서 실행 — broad except 가 swallow 하지 않도록. """ import logging from typing import Any, Dict, List, Optional from app.core.errors import AppError, ShotStagingOrientationError from app.core.frame_spatial_contract import ( _format_retry_hint as _format_fsc_retry_hint, find_frame_spatial_contract_violations, ) from app.core.gaze_direction import validate_pairing from app.modules.llm.llm_client import call_structured from app.modules.prompt_loader import load_prompt, load_schema logger = logging.getLogger(__name__) BATCH_SIZE = 10 MAX_ATTEMPTS = 3 ORIENTATION_REQUIRED_CLASSES = ("content_surface", "reflective_surface") def _find_orientation_violations(shots: List[Dict[str, Any]]) -> List[Dict[str, Any]]: """directionality_class 가 orientation 필수 class 인데 orientation 이 빈 entry.""" out: List[Dict[str, Any]] = [] for shot in shots or []: si = shot.get("scene_index") shi = shot.get("shot_index") for el in shot.get("key_bg_elements", []) or []: cls = el.get("directionality_class") or "" orient = (el.get("orientation") or "").strip() if cls in ORIENTATION_REQUIRED_CLASSES and not orient: out.append({ "scene_index": si, "shot_index": shi, "element": el.get("element", ""), "directionality_class": cls, "orientation_raw": el.get("orientation") or "", }) return out def _validate_gaze_pairings(shots: List[Dict[str, Any]]) -> None: """character_angles[] 의 gaze_direction_kind ↔ gaze_target_id pairing 강제 검증. v15 schema 는 OpenAI strict structured-output 호환을 위해 oneOf 조건부를 제거했다. kind↔target 정합(looks_at_* → registered short_id, 그 외 → null)은 이 코드측 fail-fast 검증으로 옮겨졌다. 첫 위반에서 validate_pairing 이 AppError(step.contract_violation.gaze_target_id.*) 를 raise — batch loop 위로 propagate. semantic regex / fallback 없음 (closed-world enum + short_id shape). """ for shot in shots or []: si = shot.get("scene_index") shi = shot.get("shot_index") for cangle in shot.get("character_angles", []) or []: validate_pairing( cangle.get("gaze_direction_kind"), cangle.get("gaze_target_id"), where=f"shot_staging scene={si} shot={shi} character={cangle.get('character')!r}", ) def _format_retry_hint(violations: List[Dict[str, Any]]) -> str: lines = [ "", "", "[재시도 — 직전 응답에서 다음 element 의 directionality_class 가", " 'content_surface' 또는 'reflective_surface' 인데 orientation 이", " 비어 있었습니다. 이 두 class 는 어느 면이 보이는지 / 무엇이", " 반사되는지 NL 로 반드시 작성하세요:]", ] for v in violations: lines.append( f" - S{v['scene_index']} Shot{v['shot_index']}: " f"element='{v['element']}' class='{v['directionality_class']}' " f"orientation 누락" ) return "\n".join(lines) def run_shot_staging( shot_extract_data: Dict, shot_selection_data: Dict, scene_save_data: Dict, entity_merge_data: Dict, vwr_data: Dict, camera_flow_data: Optional[Dict] = None, opik_metadata: Optional[Dict] = None, creator_corrections_block: str = "", ) -> Dict[str, Any]: """shot_staging 실행 — 선택된 샷별 창의적 연출 분석. Patch C / Area A — batch 마다 LLM 응답의 content_surface / reflective_surface orientation 누락을 validator 로 검사. 위반 시 retry (max_attempts=3). 소진 시 ShotStagingOrientationError raise (HTTP 422, step fail). """ t2i_context = vwr_data.get("t2i_context", "") system = load_prompt("shot_staging", "system") schema = load_schema("shot_staging", "schema") system = system.replace("{t2i_context}", t2i_context) # 제작자 정정 채널 (wave3) — 빈 문자열이면 기존과 byte-identical system += creator_corrections_block or "" # camera_flow 인덱싱 flow_by_shot: Dict[tuple, Dict] = {} flow_stages_by_scene: Dict[int, Dict[int, Dict]] = {} flow_summary_by_scene: Dict[int, str] = {} if camera_flow_data: for sc in (camera_flow_data.get("scenes") or []): si = sc.get("scene_index") flow_summary_by_scene[si] = sc.get("flow_summary", "") stages_by_idx = { s["stage_index"]: s for s in sc.get("flow_stages", []) if s.get("stage_index") is not None } flow_stages_by_scene[si] = stages_by_idx for a in sc.get("shot_assignments", []): shi = a.get("shot_index") stage_idx = a.get("stage_index") if shi is None or stage_idx is None: continue stage = stages_by_idx.get(stage_idx) if stage: flow_by_shot[(si, shi)] = {"stage": stage, "assignment": a} # 선택된 shot 인덱스 맵 selected_map = {} for s in shot_selection_data.get("scenes", []): si = s.get("scene_index") selected = s.get("selected_shot_indices", []) if selected: selected_map[si] = set(selected) # 씬 원본 텍스트 맵 scene_text_map = {} for seg in scene_save_data.get("segments", []): si = seg.get("scene_index") scene_text_map[si] = seg.get("text", "") # 인물 이름 목록 char_names = [c.get("name", "") for c in entity_merge_data.get("characters", [])] # 선택된 샷 수집 items = [] for s in shot_extract_data.get("scenes", []): si = s.get("scene_index") selected = selected_map.get(si, set()) for sh in s.get("shots", []): shi = sh.get("shot_index") if shi in selected: items.append({ "scene_index": si, "shot_index": shi, "description": sh.get("description", ""), "characters": sh.get("characters", []), "beat_title": sh.get("based_on_beat_title", ""), "scene_text": scene_text_map.get(si, ""), }) if not items: logger.info("shot_staging: no selected shots found") return {"shots": [], "total": 0} logger.info("shot_staging: %d selected shots to analyze", len(items)) all_results = [] failed_batches = 0 total_batches = (len(items) - 1) // BATCH_SIZE + 1 for batch_start in range(0, len(items), BATCH_SIZE): batch = items[batch_start:batch_start + BATCH_SIZE] batch_num = batch_start // BATCH_SIZE + 1 user_lines = [] for bi in batch: chars_str = ", ".join(bi["characters"]) if bi["characters"] else "(인물 없음)" flow_lines: List[str] = [] flow_entry = flow_by_shot.get((bi["scene_index"], bi["shot_index"])) if flow_entry: stage = flow_entry["stage"] assignment = flow_entry["assignment"] summary = flow_summary_by_scene.get(bi["scene_index"], "") if summary: flow_lines.append(f" 씬 플로우 요약: {summary}") flow_lines.append( f" 플로우 단계 {stage.get('stage_index')}[{stage.get('stage_label', '')}] " f"position={assignment.get('flow_position', '')}" ) if stage.get("camera_position"): flow_lines.append(f" camera_position: {stage['camera_position']}") if stage.get("camera_motion"): flow_lines.append(f" camera_motion: {stage['camera_motion']}") if stage.get("visual_focus"): flow_lines.append(f" visual_focus: {stage['visual_focus']}") if stage.get("transition_to_next"): flow_lines.append(f" transition_to_next: {stage['transition_to_next']}") dev = assignment.get("deviation_note", "") if dev: flow_lines.append(f" 이 샷의 미세 조정: {dev}") flow_block = ("\n" + "\n".join(flow_lines)) if flow_lines else "" user_lines.append( f"[씬{bi['scene_index']} Shot{bi['shot_index']}]\n" f" Beat: {bi['beat_title']}\n" f" 인물: {chars_str}\n" f" 묘사: {bi['description']}" f"{flow_block}\n" f" 씬 원문: {bi['scene_text']}" ) base_user_prompt = ( f"등록된 인물 목록: {', '.join(char_names)}\n\n" f"아래 샷들의 촬영 연출을 설계하세요:\n\n" + "\n\n".join(user_lines) ) # Patch C / Area A — per-attempt validator + retry. validator raise 는 try 외부. # Area Frame Spatial Contract Task 3 — orientation + fsc 둘 다 검사, # retry hint 양쪽 모두 append, 소진 시 raise 우선순위 (spec §8.1). violations: List[Dict[str, Any]] = [] fsc_v: List[Dict[str, Any]] = [] batch_failed_at_call = False for attempt in range(1, MAX_ATTEMPTS + 1): user_prompt = base_user_prompt if attempt == 1 else ( base_user_prompt + _format_retry_hint(violations) + _format_fsc_retry_hint(fsc_v) ) try: result = call_structured( step="shot_staging", system_prompt=system, user_prompt=user_prompt, response_schema=schema, opik_metadata=opik_metadata, ) except Exception as e: failed_batches += 1 logger.warning( "shot_staging batch %d/%d attempt %d failed: %s", batch_num, total_batches, attempt, e, ) batch_failed_at_call = True break # ↓ try 외부 — validator raise 가 batch loop 위로 propagate. batch_shots = result.get("shots", []) # v15: schema oneOf 제거 후 gaze kind↔target pairing 은 코드측 fail-fast. _validate_gaze_pairings(batch_shots) orientation_violations = _find_orientation_violations(batch_shots) fsc_violations = find_frame_spatial_contract_violations(batch_shots) if not orientation_violations and not fsc_violations: all_results.extend(batch_shots) logger.info( "shot_staging batch %d/%d attempt %d: %d shots ok", batch_num, total_batches, attempt, len(batch_shots), ) break if attempt < MAX_ATTEMPTS: logger.info( "shot_staging batch %d/%d attempt %d: orient=%d fsc=%d, retry", batch_num, total_batches, attempt, len(orientation_violations), len(fsc_violations), ) violations = orientation_violations # for orientation retry hint fsc_v = fsc_violations # for fsc retry hint continue # 소진 — raise 우선순위 (spec §8.1): if fsc_violations and not orientation_violations: raise AppError( code="shot_staging.frame_spatial_contract_invalid", message=( f"shot_staging batch {batch_num}/{total_batches}: " f"fsc post-validation failed after {MAX_ATTEMPTS} attempts" ), details={"fsc_violations": fsc_violations}, ) if fsc_violations and orientation_violations: raise AppError( code="shot_staging.frame_spatial_contract_invalid", message=( f"shot_staging batch {batch_num}/{total_batches}: " f"both orientation + fsc violations after {MAX_ATTEMPTS} attempts" ), details={ "orientation_violations": orientation_violations, "fsc_violations": fsc_violations, }, ) # orientation only — legacy typed error 보존. raise ShotStagingOrientationError( batch_num=batch_num, total_batches=total_batches, attempts=MAX_ATTEMPTS, violations=orientation_violations, ) if batch_failed_at_call: continue if failed_batches: logger.warning("shot_staging: %d/%d batches failed", failed_batches, total_batches) return { "shots": all_results, "total": len(all_results), "failed_batches": failed_batches, }