"""outdoor_frame_mode — 레인2 샷별 구도 판정 (E2E6 피드백 ③, 2026-07-16). structure_plate(레인2) 샷마다 LLM 이 frame_mode 를 판정한다: - structure_dominant: 인공 구조물이 프레임의 주 피사체(정면·클로즈· 디테일·개구부 중심) — 마커 맵→스케치가 구조물 표현을 왜곡하므로 스케치를 생략하고 스틸은 seed 직참조. - layout_dominant: 배치·이동·방향 관계가 주(부감·개활) — 현행 마커 맵→스케치 유지. 계약(Codex 설계 합의): 독립 샷별 CP(lane plan 확장 금지), evidence literal 인용 실재 검증+대상 샷 완전성 parity fail-closed, mixed 임의 라우팅 금지. 판별 어휘 하드코딩 0 — 전부 LLM 계약. """ from __future__ import annotations import logging from typing import Any, Callable, Dict, List, Optional, Sequence from app.modules.prompt_loader import load_prompt from app.modules.pipeline.outdoor_lane_plan import _check_evidence from app.modules.pipeline.shot_ref_classify import tag_of logger = logging.getLogger(__name__) _MODULE = "outdoor_frame_mode" FRAME_MODES = ("structure_dominant", "layout_dominant") _CONF = ("high", "medium", "low") MAX_RETRY = 2 PROMPT_VERSION_MAP = { "1": "1.202607161700", } def resolve_prompt_version(selector: str) -> str: try: return PROMPT_VERSION_MAP[selector] except KeyError: raise ValueError( f"unknown outdoor_frame_mode prompt version selector " f"{selector!r} (known: {sorted(PROMPT_VERSION_MAP)})" ) def build_frame_mode_schema() -> Dict[str, Any]: return { "type": "object", "additionalProperties": False, "properties": { "items": { "type": "array", "items": { "type": "object", "additionalProperties": False, "properties": { "shot": {"type": "string"}, "frame_mode": { "type": "string", "enum": list(FRAME_MODES)}, "confidence": { "type": "string", "enum": list(_CONF)}, "evidence": { "type": "object", "additionalProperties": False, "properties": { "scene_index": {"type": "integer"}, "quote_ko": {"type": "string"}, }, "required": ["scene_index", "quote_ko"], }, "rationale_ko": {"type": "string"}, }, "required": ["shot", "frame_mode", "confidence", "evidence", "rationale_ko"], }, } }, "required": ["items"], } def validate_frame_modes( items: Sequence[Dict[str, Any]], target_tags: Sequence[str], scene_texts: Dict[int, str], ) -> List[str]: """완전성 parity + 증거 무결성 — 위반 리스트 (fail-closed 재시도용). - 대상 샷 전부 entry 필수(누락=위반), 대상 밖/중복 entry=위반. - evidence literal 인용 실재(whitespace 정규화만 — 의미 판단 없음) + **해당 샷의 씬에서만 인용**(Codex f7c81336 BLOCKING-2: 다른 씬 인용이 통과하던 결함 — 프롬프트 '해당 씬' 계약과 동기). - 저신뢰(low)=위반 아님(재시도 소진 후 conservative 처리는 caller 정책이 아니라 **위반으로 fail-closed** — mixed 임의 라우팅 금지, Codex 설계 합의 2). """ from app.modules.pipeline.shot_ref_classify import parse_tag violations: List[str] = [] seen: set = set() target = set(target_tags) for it in items or []: tag = it.get("shot") if tag in seen: violations.append(f"샷 {tag} entry 중복") seen.add(tag) if tag not in target: violations.append(f"샷 {tag} 은 판정 대상 밖") continue if it.get("frame_mode") not in FRAME_MODES: violations.append(f"샷 {tag} frame_mode 값 위반") if it.get("confidence") == "low": violations.append( f"샷 {tag} confidence=low — 근거를 보강해 high/medium 으로" " 재판정할 것") ev = it.get("evidence") violations.extend(_check_evidence(f"샷 {tag}", ev, scene_texts)) if isinstance(ev, dict): try: shot_si = parse_tag(str(tag))[0] except Exception: # noqa: BLE001 — 대상 tag 는 항상 정형 shot_si = None if shot_si is not None and ev.get("scene_index") != shot_si: violations.append( f"샷 {tag} evidence.scene_index={ev.get('scene_index')}" f" — 해당 샷의 씬({shot_si})에서만 인용할 것") missing = sorted(target - seen) if missing: violations.append(f"판정 누락 샷: {missing}") return violations def validate_frame_cp_entries( frame_shots: Dict[str, Dict[str, Any]], target_tags: Sequence[str], scene_texts: Dict[int, str], ) -> List[str]: """persisted frame_mode CP 소비 직전 재검증 (Codex f7c81336 HIGH-3). entry 존재만 믿지 않는다 — status=ok 강제 + 저작 시점과 동일한 검증(validate_frame_modes)을 현재 authoritative 대상 집합으로 재실행 (stale/조작/enum 밖 값이 layout 경로로 조용히 하강하는 창 제거). """ violations: List[str] = [] items: List[Dict[str, Any]] = [] for tag in sorted(frame_shots or {}): e = frame_shots[tag] if not isinstance(e, dict): violations.append(f"샷 {tag} entry 형식 위반") continue if e.get("status") != "ok": violations.append(f"샷 {tag} status={e.get('status')!r} — ok 아님") items.append({"shot": tag, **{ k: e.get(k) for k in ("frame_mode", "confidence", "evidence", "rationale_ko") }}) violations.extend( validate_frame_modes(items, target_tags, scene_texts)) return violations def run_outdoor_frame_mode( *, shots: List[Dict[str, Any]], scene_texts: Dict[int, str], scene_headings: Dict[int, 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, ) -> Dict[str, Any]: """판정 실행 → {"shots": {tag: entry}} (검증 통과분만 반환 — 위반 시 MAX_RETRY 재시도, 소진=AppError fail-closed). shots=[{scene_index, shot_index, description, camera_direction}] — structure_plate 바인딩 샷 전체(caller 가 authoritative 재구성). 씬 원문 전문은 절대 자르지 않는다. """ from app.core.errors import AppError if call_structured_fn is None: from app.modules.llm.llm_client import call_structured call_structured_fn = call_structured if not shots: return {"shots": {}} resolved = resolve_prompt_version(prompt_version) pc = {**(project_config or {}), f"{_MODULE}_judge": {"model": "gpt"}} ordered = sorted( shots, key=lambda s: (int(s["scene_index"]), int(s["shot_index"]))) tags = [tag_of(int(s["scene_index"]), int(s["shot_index"])) for s in ordered] scene_ids = sorted({int(s["scene_index"]) for s in ordered}) scenes_txt = "\n\n".join( f"[씬 {si}] {scene_headings.get(si, '')}\n{scene_texts.get(si, '')}" for si in scene_ids ) shots_txt = "\n".join( f"{t}: {s.get('description') or ''}" + (f"\n [연출] {s['camera_direction']}" if (s.get("camera_direction") or "").strip() else "") for t, s in zip(tags, ordered) ) system = load_prompt(_MODULE, "judge_system", version=resolved) schema = build_frame_mode_schema() user = f"씬 원문 전문:\n\n{scenes_txt}\n\n샷 목록(시간순):\n{shots_txt}" last_violations: List[str] = [] for attempt in range(1, MAX_RETRY + 2): res = call_structured_fn( f"{_MODULE}_judge", system, user, schema, project_config=pc, schema_name=f"{_MODULE}_judge", opik_metadata=opik_metadata, ) items = res.get("items") or [] last_violations = validate_frame_modes(items, tags, scene_texts) if not last_violations: return { "shots": { it["shot"]: { "status": "ok", "frame_mode": it["frame_mode"], "confidence": it["confidence"], "evidence": it["evidence"], "rationale_ko": it.get("rationale_ko") or "", } for it in items } } logger.warning( "outdoor_frame_mode attempt=%d 위반 %d건: %s", attempt, len(last_violations), last_violations[:6], ) user = ( f"{user}\n\n직전 응답의 계약 위반 — 전부 교정해 다시:\n- " + "\n- ".join(last_violations[:12]) ) raise AppError( code=f"step.contract_violation.{_MODULE}", message=( "frame_mode 판정 재시도 소진 — fail-closed: " + "; ".join(last_violations[:8]) ), status_code=422, )