"""outdoor_shot_grounding — 샷 grounding (W22 직행 체인 ③, s30 stage_ground 계약). 야외 선택 샷마다 캐논 맵+마커 범례+샷 원문+**씬 원문 전문**을 주고 존/앵커 마커/카메라 배치를 접지한다. s30 대비 변경(사용자 확정): - concept 선행 콜 없음 (1콜 축약 — 씬 원문이 구상 컨텍스트를 대신) - moment_context_en 신설: 씬 원문에만 있는 순간의 세계 사실 보강 (S13_3 '차'→경찰차 소실 사례 대응, evidence-bound·연출 금지) 스키마는 스펙에서 동적 생성 — map_zone/anchor_markers 가 해당 그룹의 존 라벨/마커 코드 enum 으로 잠긴다. 서술 필드는 ID-free (코드는 anchor_markers 에만) — 결정론 검증+재시도. 설계: docs/w22-outdoor-canon-direct-compose-design-20260709/design.md """ import base64 import logging import re from typing import Any, Callable, Dict, List, Optional, Sequence from app.core.errors import AppError from app.modules.prompt_loader import load_prompt logger = logging.getLogger(__name__) _MODULE = "outdoor_shot_grounding" PROMPT_VERSION_MAP = { "1": "1.202607100125", } MAX_ATTEMPTS = 3 # 서술 필드 — 마커 코드 금지 (ID-free, 코드는 anchor_markers 에만) _PROSE_FIELDS = ( "camera_position_en", "look_direction_en", "in_frame_en", "moment_context_en", ) def resolve_prompt_version(selector: str) -> str: try: return PROMPT_VERSION_MAP[selector] except KeyError: raise ValueError( f"unknown outdoor_shot_grounding prompt version selector " f"{selector!r} (known: {sorted(PROMPT_VERSION_MAP)})" ) def spec_codes(spec: Dict[str, Any]) -> List[str]: return [ it.get("code", "") for it in spec.get("items", []) or [] if it.get("code") ] def build_ground_schema(spec: Dict[str, Any]) -> Dict[str, Any]: """그룹 스펙에서 동적 enum 스키마 생성 (s30 계약 + moment_context_en).""" zones = list(spec.get("zone_labels_en", []) or []) codes = spec_codes(spec) if not zones or not codes: raise AppError( code="step.outdoor_shot_grounding.bad_spec", message="스펙에 존 라벨 또는 마커 코드가 없어 스키마 생성 불가", status_code=422, ) return { "type": "object", "additionalProperties": False, "required": [ "map_zone", "anchor_markers", "camera_position_en", "look_direction_en", "in_frame_en", "rationale_ko", "moment_context_en", ], "properties": { "map_zone": {"type": "string", "enum": zones}, "anchor_markers": { "type": "array", "minItems": 1, "items": {"type": "string", "enum": codes}, }, "camera_position_en": {"type": "string", "minLength": 8}, "look_direction_en": {"type": "string", "minLength": 5}, "in_frame_en": {"type": "string", "minLength": 8}, "rationale_ko": {"type": "string", "minLength": 5}, "moment_context_en": {"type": "string"}, }, } def build_legend_block(spec: Dict[str, Any]) -> str: """VLM 내부용 코드 포함 범례 (s30 계약 — 하류 t2i 프롬프트와 무관).""" return "\n".join( f"- ({it.get('code', '')}) {it.get('name_en', '')} — " f"{it.get('placement_en', '')}" for it in spec.get("items", []) or [] ) def build_shot_block(shot: Dict[str, Any]) -> str: si = shot.get("scene_index") shi = shot.get("shot_index") lines = [f"S{si}_Shot{shi}: {shot.get('description', '')}"] chars = shot.get("characters") or [] if chars: lines.append(f"등장 인물: {', '.join(str(c) for c in chars)}") cam = shot.get("camera_direction") or "" if cam: lines.append(f"연출 카메라 메모 (참고): {cam}") elements = [ el.get("element", "") for el in shot.get("key_bg_elements") or [] if el.get("element") ] if elements: lines.append(f"연출 배경 요소 (참고): {', '.join(elements)}") return "\n".join(lines) def validate_ground( result: Dict[str, Any], codes: Sequence[str] ) -> List[str]: """서술 필드 ID-free 결정론 검증 — 위반 리스트 반환 (enum 은 스키마가 잠금).""" violations: List[str] = [] for field in _PROSE_FIELDS: text = result.get(field) or "" leaked = [ c for c in codes if c and re.search(rf"\b{re.escape(c)}\b", text) ] if leaked: violations.append( f"{field} 에 마커 코드 {leaked} 포함 — 서술명(name_en)으로 " "지칭할 것 (코드는 anchor_markers 에만)" ) return violations def _png_part(png_bytes: bytes) -> Dict[str, Any]: b64 = base64.b64encode(png_bytes).decode() return {"type": "image_url", "image_url": {"url": f"data:image/png;base64,{b64}"}} def build_user_parts( template: str, *, spec: Dict[str, Any], shot: Dict[str, Any], scene_text: str, map_png: bytes, ) -> List[Dict[str, Any]]: """멀티모달 user 조립 — s30 파트 순서(맵→범례→샷→씬 전문).""" filled = template for key, val in { "legend_block": build_legend_block(spec), "zones_block": "\n".join( f"- {z}" for z in spec.get("zone_labels_en", []) or [] ), "shot_block": build_shot_block(shot), # 씬 원문 전문 — 절대 자르지 않는다 (CLAUDE.md 절대 규칙) "scene_text_block": scene_text, }.items(): filled = filled.replace("{" + key + "}", val) return [ {"type": "text", "text": "SITE PLAN of the filming property:"}, _png_part(map_png), {"type": "text", "text": filled}, ] def _format_retry_hint(violations: List[str]) -> str: lines = [ "", "", "[재시도 — 직전 응답이 아래 계약을 위반했습니다. 전부 고쳐서", " 전체 결과를 다시 출력하세요:]", ] lines.extend(f" - {v}" for v in violations) return "\n".join(lines) def run_outdoor_shot_grounding_shot( *, spec: Dict[str, Any], shot: Dict[str, Any], scene_text: str, map_png: bytes, creator_corrections_block: str = "", prompt_version: str = "1", call_structured_fn: Optional[Callable[..., Dict[str, Any]]] = None, project_config: Optional[Dict[str, Any]] = None, opik_metadata: Optional[Dict[str, Any]] = None, max_attempts: int = MAX_ATTEMPTS, ) -> Dict[str, Any]: """샷 1개 grounding — ID-free 위반 시 위반 힌트 재시도. 반환 {"ground": <검증 통과 결과>, "attempts": n}. 소진 시 AppError(step.contract_violation.outdoor_shot_grounding). """ if call_structured_fn is None: from app.modules.llm.llm_client import call_structured call_structured_fn = call_structured resolved = resolve_prompt_version(prompt_version) system = load_prompt(_MODULE, "system", version=resolved) system += creator_corrections_block or "" template = load_prompt(_MODULE, "user_template", version=resolved) schema = build_ground_schema(spec) codes = spec_codes(spec) base_parts = build_user_parts( template, spec=spec, shot=shot, scene_text=scene_text, map_png=map_png ) attempts = 0 parts = base_parts last_violations: List[str] = [] while attempts < max_attempts: attempts += 1 result = call_structured_fn( "outdoor_shot_grounding", system, parts, schema, project_config=project_config, schema_name="outdoor_shot_grounding", opik_metadata=opik_metadata, ) violations = validate_ground(result or {}, codes) if not violations: return {"ground": result, "attempts": attempts} last_violations = violations logger.warning( "outdoor_shot_grounding S%s_Shot%s attempt=%d violations=%d", shot.get("scene_index"), shot.get("shot_index"), attempts, len(violations), ) parts = base_parts + [ {"type": "text", "text": _format_retry_hint(violations)} ] raise AppError( code="step.contract_violation.outdoor_shot_grounding", message=( f"S{shot.get('scene_index')}_Shot{shot.get('shot_index')} " f"grounding 계약 위반 (attempts={max_attempts}): " + "; ".join(last_violations[:8]) ), status_code=422, )