"""ShotContiLightStep — 샷별 경량 콘티 생성 (order 21.92). s40/s41 레시피 이식: 인물 샷(비 bgonly·비 prev)마다 배경 플레이트 느슨 참조 LIGHT_FRAME 콘티를 gpt-image-2 로 생성해 `//images//conti/conti_{tag}.png` 에 저장한다. data.contis[tag] = {image_path, plate_path, prompt, skipped_reason, error} 플레이트는 background_render/background_chain_render 체크포인트 soft-load (manifest 의존 아님 — mode off 시 게이트에 막히지 않도록, outdoor_place_canon 스타일 FP soft-load 선례). 플레이트 부재 샷=no_plate 격리(스틸은 진행). moderation 거절 시 PromptSanitizer 1회 재시도. """ import hashlib import json import logging from pathlib import Path from typing import Any, Dict, List, Optional from app.core.step_runner import StepRunner from app.modules.pipeline.shot_continuity_author import ( carried_clause_for, ) # ★모듈 수준으로 올린다 — 세 메서드가 지역 import 를 반복하고 있었고, # lane 갈래는 **부르면서 import 를 안 했다**(NameError 가 호출 시점까지 # 안 드러나는 부류). from app.modules.pipeline.shot_ref_classify import tag_of logger = logging.getLogger(__name__) # v2 (2026-07-16 R1/R2): plate_authority 키(plate_select ON) + lane_conti # structure_plate 정책 entry 재편(ab_select_*) — 구 CP 소비 금지. # v3 (2026-07-17 seed-bg): 배경 권위 typed 해석(bg_source/bg_path) — # 플레이트 producer 구조적 부재 그룹의 seed-bg 승격 (Codex 합의). SCHEMA_VERSION = 4 PROMPT_VERSION = "1" # Stage D (2026-07-15): 야외 lane 분기 팩 — outdoor_lane_pipe_enabled ON 시만 # v2 (2026-07-24 Codex 합의): per-marker landmark anchor 저작 # v3 (2026-07-25 사용자 지적①): 시야 쐐기(view_left/right)·피사체 # facing 저작 — v2 는 origin→look_target 선 하나뿐이라 "카메라가 어느 # 방향으로 무엇을 찍는지"와 "엔티티가 어디를 향하는지"가 데이터에 # 아예 없었다. # v4 (2026-07-26): 카메라 대면 관계 선언(camera_facing_relation/ # _evidence) — SHOT TEXT 가 '뒷모습'이라 말해도 v3 는 facing 좌표만 # 저작돼 어긋난 방향을 잡을 수단이 없었다(실측 S15sh5). 선언을 좌표와 # 결정론 대조하고 인용은 SHOT BLOCK 원문과 대조한다. LANE_GEOMETRY_PACK_VERSION = "4" # v4 (2026-07-16 R4·R5): 마커 굽기 전면 제거 — 클린 맵+ID-free 텍스트 # geometry (코드로 이미지 그리기 절대 금지 준수) # v5 (2026-07-24 사용자 강질책): 마커 맵 단계 복원 — i2i 로. 클린 맵→ # 이미지 모델이 CAM·시선·엔티티 마커 작화(marker_annotate)→VLM 검증 # (marker_check)→스케치 참조=마커 맵. 코드 드로잉은 계속 금지. # v6 (Codex 합의 ②): 배치 권위=anchor/검증 통과 마커 맵 — 좌표는 # coarse estimate (캔ary 3회 실측: i2i 수치 분율 정밀도 한계). # v7 (Codex 재리뷰 ①②): geometry-text 분기 편입(BLOCKING — v6 은 # legacy 분기로 하강)+staging_geometry_head 권위 정합. v6=캔ary 소비 # 불변. # v8 (2026-07-25): lane 자연 연기 사각 봉합 — naturalism_clause 이식 # (일반 콘티와 동일 문구, default-only). # v9 (2026-07-25 사용자 지적①): 마커 작화 계약 개정 — CAM 시야 쐐기 # (좌/우 프레임 경계+틴트)·피사체 facing 화살표·슬롯 라벨에 짧은 설명. # marker_check 도 쐐기/화살표 검증, sketch_head 는 쐐기=프레임 포함 # 범위로 정합. geometry v3 과 짝(_VIEW_* 팩 집합). v8=커밋 소비분 불변. # v10 (v9 캔ary 실측): i2i 가 쐐기를 캔버스에 맞추려 CAM 을 지정 위치 # 밖으로 끌어내려 marker_check 소진 — 마커 이동 금지 조항 추가. # v11: marker_check 판정을 측량이 아닌 작화 가독성 기준으로 완화 # v12: 화면 기준 방향 서술이 작화 기준임을 명시 (맵 분율 오독 차단) # v13: marker_check 에서 틴트 등 스타일 축 제외 (배치·방향만 판정) # v14 (2026-07-26 사용자 확정 — 3변형 격리 실측): 인물 방향 반전 원인이 # 렌더 스타일임을 확정 — 콘티=마네킹(mannequin_frame), 콘티 참조=마커 맵 # 1장(plate_look 제거 → plate_look_on=False), 관할 명문화(수평 배치= # staging 우선). 배경·인물 외형은 뒤 단계에서 i2i 로 얹는다. LANE_SKETCH_PACK_VERSION = "16" #: lane 기하 판정의 **기본 별칭**. 2026-08-27 (#92): `gpt`(Sol) → gemini. #: #: ★조립과 지문이 **같은 자리**를 봐야 한다. 앞 판은 조립만 고쳤는데 #: 지문에는 `lane_marker_judge`(= `lane_marker_map_check` 의 모델)만 #: 접혀 있어서, 그것이 **이미 gemini 물리명**이라 지문이 안 움직였다. #: 완주 CP 가 **Sol 산출을 그대로 재사용하면서 Gemini 판처럼 기록**된다 #: (Codex BLOCK). 「우연히 덮이는 경우가 더 고약하다」의 그 자리다. LANE_GEOMETRY_JUDGE_ALIAS = "gemini-pro" def lane_geometry_judge_identity( project_config: Optional[Dict[str, Any]] = None, ) -> Dict[str, str]: """이 주행에서 lane 기하 판정이 **실제로 쓰는** 별칭과 물리 모델. ★`project_config` 가 덮을 수 있으므로 **effective 값**을 돌려준다 — 기본값만 접으면 프로젝트가 다른 모델로 바꿨을 때 지문이 안 움직인다. """ cfg = dict(project_config or {}) entry = cfg.get("outdoor_marker_geometry") alias = LANE_GEOMETRY_JUDGE_ALIAS if isinstance(entry, dict) and isinstance(entry.get("model"), str): alias = entry["model"] from app.modules.pipeline.era_research import resolve_model_physical return {"alias": alias, "physical": resolve_model_physical(alias)} LANE_LEAKAGE_MAX_ATTEMPTS = 3 # 3 (2026-07-25): 마커 계약이 쐐기·화살표·라벨로 늘어 1차 시도 실패 # 여지가 커졌다 — 검증 재시도 1회 추가 (i2i 1콜 비용, 소진=fail-closed). LANE_MARKER_ANNOTATE_MAX_ATTEMPTS = 3 # 마네킹 콘티 ↔ 배경/엔티티 마네킹 교체 계약의 짝 버전. 어느 한쪽만 # 바뀌면 CP 를 무효화해야 하므로 config hash 에 접힌다. MANNEQUIN_CHAIN_CONTRACT_VERSION = "1" def _mannequin_sketch_packs(): """마네킹 저작 sketch 팩 집합 (지연 import — 모듈 로드 순환 방지).""" from app.modules.pipeline.outdoor_marker_map import ( _MANNEQUIN_SKETCH_PACKS, ) return _MANNEQUIN_SKETCH_PACKS def _resolve_openai_client(): """background_render_step._resolve_openai_client 와 동일 패턴.""" from app.core.openai_keys import openai_client from app.core.config import settings return openai_client( timeout=float(settings.llm_timeout_image_gen), ) class ShotContiLightStep(StepRunner): """경량 콘티 — 플레이트 느슨 참조 단일 프레임 (레시피 이식).""" def _config_hash(self) -> str: import hashlib import json as _json from app.core.config import settings from app.modules.pipeline.shot_conti_light import ( CONTI_IMAGE_MODEL, resolve_prompt_version, ) from app.core.step_runner import compute_config_hash from app.modules.pipeline.shot_continuity_author import ( CARRIED_PERSON_PROJECTION_VERSION as _carried_person_proj, ) payload = { "still_recipe_mode": settings.still_recipe_mode, "schema_version": SCHEMA_VERSION, "prompt_version": resolve_prompt_version( self._prompt_selector() ), "image_model": CONTI_IMAGE_MODEL, # HIGH-5: project_config step override(모델 라우팅) 변경 감지 "project_config_hash": compute_config_hash(self.project_config), # ★CARRIED 인물 문장을 정본 표기에 결속하는 계약 (#104, # 2026-08-29). 이 스텝은 `carried_clause_for` 를 두 곳에서 # 부른다(`shot_conti_light_step:1354`·`shot_conti_light:763`) # — 프롬프트 바이트가 바뀌므로 여기도 같이 접는다. 한쪽만 # 접으면 다른 소비 경로가 clean-skip 해 안 닿는다. "carried_person_projection": _carried_person_proj, } # BGFIRST2 이식 ② (2026-07-20): 원근 가이드 콘티는 프롬프트·참조를 # 실질 변경(팩 v2 는 위 prompt_version 이 이미 반영) — flag 와 # CAMERA/FRAME 절 스템 팩도 ON 시만 스탬프 (OFF byte-identical). if bool(getattr(settings, "still_bgfirst_enabled", False)): from app.modules.pipeline.still_recipe import ( CAMERA_FRAME_PROMPT_VERSION as _cam_pack_sel, resolve_prompt_version as _recipe_pack, ) payload["still_bgfirst_enabled"] = True payload["conti_camera_frame_pack"] = _recipe_pack(_cam_pack_sel) # fix④ full (2026-07-21): no_plate 콘티 저작이 산출 집합을 실질 # 변경 (팩 v3 은 위 prompt_version 이 반영) — ON 시만 스탬프. if self._no_plate_conti_on(): payload["still_bgfirst_full_enabled"] = True # 맵 기반 분기는 출력(플레이트 대체)을 실질 변경 — ON 일 때만 스탬프 # (OFF byte-identical, opt-in stamping 관례). 2차 리뷰 H4: 맵 팩· # nb2 모델도 출력 실질 입력. if bool(getattr(settings, "outdoor_map_conti_enabled", False)): from app.modules.pipeline.outdoor_map_conti import ( resolve_prompt_version as map_pack_version, ) from app.modules.pipeline.outdoor_shot_grounding import ( resolve_prompt_version as grounding_pack_version, ) payload["outdoor_map_conti_enabled"] = True payload["outdoor_map_conti_pack"] = map_pack_version( PROMPT_VERSION ) # 3차 리뷰 H2: 맵 lane 의 실제 grounding 계약도 접는다 payload["outdoor_shot_grounding_pack"] = grounding_pack_version( "1" ) payload["map_nb2_model"] = settings.gemini_image_model # Stage D: lane 분기도 출력 실질 입력 — ON 일 때만 스탬프 (opt-in # stamping 관례, OFF byte-identical). Codex 배선 조건 ①. if bool(getattr(settings, "outdoor_lane_pipe_enabled", False)): from app.core.steps.outdoor_lane_plan_step import ( EVIDENCE_VALIDATION_VERSION as _lane_evidence_version, ) from app.modules.pipeline.outdoor_marker_map import ( GEOMETRY_TEXT_VERSION as _geometry_text_version, resolve_prompt_version as geometry_pack_version, resolve_sketch_pack_version, ) payload["outdoor_lane_pipe_enabled"] = True payload["lane_geometry_pack"] = geometry_pack_version( LANE_GEOMETRY_PACK_VERSION ) from app.modules.pipeline.outdoor_marker_map import ( MARKER_ANNOTATE_CONTRACT_VERSION as _marker_contract, MARKER_MAP_SIZE as _marker_size, _ANNOTATE_PACKS as _annotate_packs, ) # v5 (Codex HIGH-3): 마커 작화·검증 계약도 lane 출력 실질 # 입력 — annotate 팩에서만 스탬프 (v4 이하 byte-identical) if LANE_SKETCH_PACK_VERSION in _annotate_packs: from app.modules.pipeline.outdoor_marker_map import ( MARKER_GEOMETRY_CONTRACT_VERSION as _geo_contract, ) payload["lane_marker_contract"] = _marker_contract payload["lane_marker_geometry_contract"] = _geo_contract payload["lane_marker_max_attempts"] = ( LANE_MARKER_ANNOTATE_MAX_ATTEMPTS) payload["lane_marker_judge"] = { "alias": "gemini-pro", "physical": settings.gemini_text_model, } # ★#92 (2026-08-27): 위 칸은 `lane_marker_map_check` 의 # 모델이지 **기하 판정이 아니다.** 기하 판정을 Sol 에서 # gemini 로 바꿔도 이 칸은 안 움직여, 완주 CP 가 Sol # 산출을 재사용하면서 Gemini 판처럼 기록됐다. payload["lane_geometry_judge"] = ( lane_geometry_judge_identity(self.project_config)) payload["lane_marker_map_size"] = _marker_size payload["lane_sketch_pack"] = resolve_sketch_pack_version( LANE_SKETCH_PACK_VERSION ) # 재리뷰 HIGH-1: 직렬화 계약(v4 스케치 실질 입력)도 스탬프 — # completed CP 는 _execute 를 안 타 sidecar 지문만으로는 # 무효화되지 않는다. payload["lane_geometry_text_version"] = _geometry_text_version payload["lane_evidence_validation_version"] = ( _lane_evidence_version ) # 2026-07-26: 마네킹 팩은 하류 교체 계약과 짝이다 — lane-prev # 플래그까지 접지 않으면 v14 CP 를 만든 뒤 그 플래그만 내려 # CP 재사용+legacy 경로 유출이 가능하다(구 hash 는 bgfirst/ # full 만 접었다). lane pipe OFF 는 lane 콘티 자체가 없어 # 유출 표면이 없으므로 opt-in stamping 관례를 지킨다 # (OFF byte-identical — 무관 프로젝트 전량 재렌더 금지). if LANE_SKETCH_PACK_VERSION in _mannequin_sketch_packs(): payload["lane_mannequin_chain"] = { "contract": MANNEQUIN_CHAIN_CONTRACT_VERSION, "bgfirst": bool(getattr( settings, "still_bgfirst_enabled", False)), "bgfirst_full": bool(getattr( settings, "still_bgfirst_full_enabled", False)), "lane_prev": bool(getattr( settings, "still_lane_prev_bgfirst_enabled", False)), } # R1 (Codex 설계 리뷰 BLOCKING-1): 플레이트 권위 선행 판정도 출력 # (콘티가 그려지는 플레이트) 실질 입력 — ON 시만 스탬프. if bool(getattr(settings, "still_plate_select_enabled", False)): from app.modules.pipeline.plate_select import ( PLATE_AUTHORITY_VERSION, resolve_prompt_version as _plate_select_pack, ) payload["still_plate_select_enabled"] = True payload["plate_select_pack"] = _plate_select_pack("1") payload["plate_authority_version"] = PLATE_AUTHORITY_VERSION payload["plate_select_judge_model"] = settings.gemini_text_model # 2026-07-16 사용자 확정(복잡 구조물=A/B): outdoor_frame_mode 소비 # 제거 — 스탬프도 함께 제거 (stale flag 가 hash 에 영향 주지 않게, # Codex R 열린쟁점④). return hashlib.sha256( _json.dumps(payload, sort_keys=True).encode("utf-8") ).hexdigest()[:16] def _prompt_selector(self) -> str: """콘티 팩 selector — full=v5(키샷 레이아웃 가치, E2E13 fix①⑦ — v4 소품 방향+자연 연기+no_plate 헤더 승계), BGFIRST2 ON=v2(원근 가이드), OFF=v1.""" from app.core.config import settings if self._no_plate_conti_on(): return "5" return ( "2" if bool(getattr(settings, "still_bgfirst_enabled", False)) else PROMPT_VERSION ) @staticmethod def _no_plate_conti_on() -> bool: """fix④ full 게이트 — bgfirst 자체가 ON 일 때만 유효 (OFF 인데 full 만 ON 이면 콘티는 v1 그대로, 서비스가 fail-closed).""" from app.core.config import settings return bool( getattr(settings, "still_bgfirst_full_enabled", False) ) and bool(getattr(settings, "still_bgfirst_enabled", False)) @staticmethod def _mannequin_chain_ready() -> bool: """마네킹 콘티를 만들어도 되는 조건 — 하류 교체 계약이 살아 있나. 마네킹 산출은 배경 i2i·엔티티 단계의 교체 계약과 짝으로만 유효하다. 세 플래그 중 하나라도 꺼지면 서비스가 lane_chain=False 로 계산해 마네킹을 legacy 조립 참조로 넣고, 그 경로엔 교체 계약이 없어 마네킹이 최종 스틸까지 유출된다(2026-07-26 설계 실측). """ from app.core.config import settings return all( bool(getattr(settings, name, False)) for name in ( "still_bgfirst_enabled", "still_bgfirst_full_enabled", "still_lane_prev_bgfirst_enabled", ) ) def _visible_char_sids_by_tag(self) -> Dict[str, set]: """샷별 **확정 배정 인물의 short_id** (감사 1-B, 2026-08-27). ★한 실행에서 lane 갈래와 일반 콘티 갈래가 **각자 부른다.** 그때마다 프로젝트 전체 `EntityCanon` 과 에피소드 전체 `SceneStill` 을 훑으므로 인스턴스에 담아 둔다 (큰 판은 923 + 980 행이다). CARRIED 를 이 집합으로 거른다 — 그 샷에 없는 인물이 "they must be one of: …" 와 함께 나가던 것을 막는다. ★`_char_refs_by_tag` 를 재사용하지 않는다. 그쪽은 ㉠ 이름과 참조 경로만 돌려주고 short_id 를 버리며 ㉡ 팩 selector 가 바뀐 경우 에만 불린다. **visibility 는 언제나 필요하다.** ★참조 이미지 유무와 **묶지 않는다.** 참조가 없어도 그 사람은 화면에 있다 — 참조로 거르면 연속성이 조용히 사라진다. ★계산은 `shot_ref_classify.load_character_sids_by_tag` 하나다 (2026-09-19) — 스틸 조립이 「캐릭터가 보이면 사람 샷」 규칙에 같은 집합을 쓴다. 둘이 따로 세면 콘티 대상과 스틸이 갈린다. """ cached = getattr(self, "_vis_sids_cache", None) if cached is not None: return cached from app.modules.pipeline.shot_ref_classify import ( load_character_sids_by_tag, ) out = load_character_sids_by_tag( self.db, self.project_id, self.episode_id) self._vis_sids_cache = out return out def _char_refs_by_tag(self) -> Dict[str, Any]: """샷별 캐릭터 비례 참조 — VE(scene_still SOT) → character 엔티티 primary reference ImageAsset 경로. (BGFIRST2 v2 콘티 전용 — outlook v14 보편 계약으로 primary=기본 모습.) 결손=해당 샷 참조 없음(fail-safe — 콘티 자체를 막지 않는다).""" from app.core.file_paths import resolve_image_path from app.models.project import EntityCanon, ImageAsset, SceneStill from app.modules.pipeline.shot_ref_classify import tag_of char_name_by_id: Dict[str, str] = { e.id: (e.name or e.id) for e in self.db.query(EntityCanon) .filter( EntityCanon.project_id == self.project_id, EntityCanon.entity_type == "character", ) .all() } ref_path_cache: Dict[str, Optional[Path]] = {} def _primary_ref(eid: str) -> Optional[Path]: if eid not in ref_path_cache: row = ( self.db.query(ImageAsset) .filter( ImageAsset.project_id == self.project_id, ImageAsset.entity_id == eid, ImageAsset.asset_type == "reference", ImageAsset.is_primary == 1, ) .first() ) p = resolve_image_path(row.file_path) if row else None ref_path_cache[eid] = ( p if p is not None and p.exists() else None ) return ref_path_cache[eid] out: Dict[str, Any] = {} rows = ( self.db.query(SceneStill) .filter( SceneStill.project_id == self.project_id, SceneStill.episode_id == self.episode_id, ) .all() ) for row in rows: si, shi = row.scene_index, row.shot_index if si is None or shi is None: continue try: ve = json.loads(row.visible_entities_json or "[]") except json.JSONDecodeError: continue refs = [] for v in ve: eid = ( (v.get("id") or v.get("entity_id")) if isinstance(v, dict) else (v if isinstance(v, str) else None) ) if not eid or eid not in char_name_by_id: continue p = _primary_ref(eid) if p is not None: refs.append((char_name_by_id[eid], p)) if refs: out[tag_of(int(si), int(shi))] = refs return out def _load_prev_checkpoint(self, step_id: str) -> Optional[Dict[str, Any]]: from app.core.config import settings cp = ( Path(settings.projects_dir) / self.project_id / "checkpoints" / "episodes" / self.episode_id / step_id / "manifest.json" ) if cp.exists(): try: return json.loads(cp.read_text(encoding="utf-8")) except Exception as exc: # noqa: BLE001 logger.warning( "shot_conti_light: %s 로드 실패: %s", step_id, exc ) return None def _make_gen_fn(self): """gpt-image edit + moderation 시 sanitizer 1회 재시도.""" from app.modules.llm.gpt_image_primitive import call_gpt_image_bytes from app.core.image_call_budget import reserve_current_call from app.modules.pipeline.shot_conti_light import ( CONTI_IMAGE_MODEL, CONTI_SIZE, ) from app.modules.prompt_sanitizer import PromptSanitizer client = _resolve_openai_client() sanitizer = PromptSanitizer(project_config=self.project_config) call_kwargs = { "model": CONTI_IMAGE_MODEL, "size": CONTI_SIZE, "quality": "high", "n": 1, } def gen_fn( tag: str, prompt: str, labeled_refs, out_path: Path ) -> Path: # labeled_refs = [(라벨, Path)] — v1=[("plate", 플레이트)] # 단일(기존 GPT 입력과 동일), v2=+캐릭터 비례 참조 ref_paths = [p for _lab, p in labeled_refs if p is not None] current = prompt for attempt in (1, 2): try: # ★이미지 문 — run-wide cap 을 **모든** gpt-image 자리가 지난다 reserve_current_call(source="shot_conti_light.conti") png = call_gpt_image_bytes( client, mode="edit" if ref_paths else "generate", prompt=current, ref_paths=ref_paths or None, call_kwargs=call_kwargs, capture_role="conti_light", capture_metadata={ "operation_type": "shot_conti_light", "shot_tag": tag, }, ) from app.modules.pipeline.multiroll_gemini import ( atomic_write_bytes, ) return atomic_write_bytes(out_path, png) except Exception as exc: # noqa: BLE001 msg = str(exc).lower() moderated = ( "moderation" in msg or "safety" in msg or "content_policy" in msg or "rejected" in msg ) if attempt == 1 and moderated: sr = sanitizer.sanitize(current, str(exc), [], attempt) sanitized = sr.get("sanitized_prompt") or "" if sanitized: logger.warning( "shot_conti_light %s: moderation — sanitize " "재시도", tag, ) current = sanitized continue raise raise RuntimeError("unreachable") return gen_fn def _register_intermediate_assets( self, contis: Dict[str, Any], map_plates: Dict[str, Any], lane_contis: Optional[Dict[str, Any]] = None, ) -> None: """콘티/맵 플레이트/lane 스케치를 intermediate ImageAsset 으로 영속 (Codex 1차 리뷰 BLOCKING-2) — 스틸 lineage(input_image_ids)가 실을 asset UUID 제공. file_path 기준 idempotent(재실행 시 기존 row 재사용), 삭제 없음. 결과 asset_id 는 각 entry 에 병기되어 체크포인트로 내려간다.""" import uuid as _uuid from datetime import datetime, timezone from app.core.file_paths import to_relative_image_path # 단독 프로세스(헤드리스 러너)에서 ImageAsset FK(project_registry) # 매퍼 configure 가 실패하지 않도록 catalog 모델 명시 로드 from app.models import catalog as _catalog # noqa: F401 from app.models.project import ImageAsset, SceneStill from app.modules.pipeline.shot_conti_light import CONTI_IMAGE_MODEL from app.modules.pipeline.shot_ref_classify import parse_tag from app.services.image_capture.annotate import ( annotate_generated_asset, ) still_rows = ( self.db.query(SceneStill) .filter( SceneStill.project_id == self.project_id, SceneStill.episode_id == self.episode_id, ) .all() ) still_by_key = { (s.scene_index, s.shot_index): s.id for s in still_rows } from app.core.config import settings as _s from app.modules.pipeline.shot_conti_light import ( register_intermediate_assets, ) from app.modules.pipeline.shot_ref_classify import parse_tag as _pt def find_asset_by_rel(rel: str): return ( self.db.query(ImageAsset) .filter( ImageAsset.project_id == self.project_id, ImageAsset.file_path == rel, ) .first() ) def new_asset(*, rel, tag, asset_type, model, prompt): si, shi = _pt(tag) asset = ImageAsset( id=str(_uuid.uuid4()), project_id=self.project_id, asset_type=asset_type, entity_id=None, still_id=still_by_key.get((si, shi)), episode_id=self.episode_id, file_path=rel, prompt_used=prompt, generation_model=model, status="generated", created_at=datetime.now(timezone.utc).isoformat(), ) self.db.add(asset) return asset def annotate_fn(asset, *, role, input_ids, meta): annotate_generated_asset( asset, pipeline_role=role, stage="conti", input_image_ids=input_ids, pipeline_metadata=meta, ) try: register_intermediate_assets( contis=contis, map_plates=map_plates, lane_contis=lane_contis, rel_fn=to_relative_image_path, find_asset_by_rel=find_asset_by_rel, new_asset=new_asset, annotate_fn=annotate_fn, conti_model=CONTI_IMAGE_MODEL, plate_model=_s.gemini_image_model, ) self.db.commit() except Exception: # 2차 리뷰 M6: 등록은 lineage 필수 계약 — fail-open 금지. # rollback 후 상위로 propagate (스텝 failed). self.db.rollback() raise def _run_map_conti( self, *, shots: list, classify_shots: Dict[str, Any], classify_scenes: Dict[str, Any], scene_texts: Dict[int, str], scene_primary: Dict[int, str], selected_keys, out_dir: Path, force: bool = False, ) -> Dict[str, Any]: """야외 복잡 구도 → 맵 기반 플레이트 (outdoor_map_conti_enabled). 캐논 자산(outdoor_place_spec/canon)·grounding 은 W22 스택 재사용. 반환 = run_outdoor_map_conti 결과 {"map_plates": {tag: {...}}}. """ from app.modules.pipeline.outdoor_direct_common import ( build_shot_loc_map, filter_group_shots, ) from app.modules.pipeline.outdoor_map_conti import ( judge_complex_shots, run_outdoor_map_conti, ) from app.modules.pipeline.outdoor_shot_grounding import ( run_outdoor_shot_grounding_shot, ) from app.modules.pipeline.shot_conti_light import conti_targets from app.modules.pipeline.shot_ref_classify import parse_tag, tag_of spec_cp = self._load_prev_checkpoint("outdoor_place_spec") spec_groups = (spec_cp or {}).get("data", {}).get("groups", {}) or {} canon_cp = self._load_prev_checkpoint("outdoor_place_canon") canon_groups = (canon_cp or {}).get("data", {}).get("groups", {}) or {} if not spec_groups: logger.info("shot_conti_light: 야외 스펙 그룹 없음 — 맵 분기 no-op") return {"map_plates": {}} shot_loc_by_key = build_shot_loc_map( self._load_prev_checkpoint("shot_validator") ) group_assets: Dict[str, Any] = {} group_by_tag: Dict[str, str] = {} for gid, entry in spec_groups.items(): if not isinstance(entry, dict) or not entry.get("spec"): continue canon = canon_groups.get(gid) or {} master = canon.get("master_png_path") or "" map_png = canon.get("map_png_path") or "" if ( canon.get("status") == "ok" and master and map_png and Path(master).exists() and Path(map_png).exists() ): group_assets[gid] = { "spec": entry["spec"], "master_png": Path(master), "map_png": Path(map_png), } gshots = filter_group_shots( shots, scene_indices=entry.get("scene_indices") or [], loc_ids=set(entry.get("outdoor_loc_ids") or []), scene_primary=scene_primary, shot_loc_by_key=shot_loc_by_key, selected_keys=selected_keys, ) for sh in gshots: group_by_tag[ tag_of(int(sh["scene_index"]), int(sh["shot_index"])) ] = gid desc_by_tag = { tag_of(int(s["scene_index"]), int(s["shot_index"])): ( s.get("description") or "" ) for s in shots } outdoor_targets = [ t for t in conti_targets(classify_shots) if t in group_by_tag ] if not outdoor_targets: return {"map_plates": {}} complex_by_tag = judge_complex_shots( shot_desc_by_tag={t: desc_by_tag.get(t, "") for t in outdoor_targets}, prompt_version=PROMPT_VERSION, project_config=self.project_config, opik_metadata=self.build_opik_metadata(), ) # 3차 리뷰 H2: 제작자 정정 SOT 를 맵 콘티 lane 에서도 전달 # (정식 outdoor_shot_grounding_step 과 동일 계약) from app.core.creator_corrections import project_corrections_block corrections_block = project_corrections_block(self.project_id) def ground_fn(tag, spec, shot_desc, scene_text, map_png_path: Path): si, shi = parse_tag(tag) out = run_outdoor_shot_grounding_shot( spec=spec, shot={ "scene_index": si, "shot_index": shi, "description": shot_desc, }, scene_text=scene_text, map_png=map_png_path.read_bytes(), creator_corrections_block=corrections_block, project_config=self.project_config, opik_metadata=self.build_opik_metadata(), ) return out["ground"] from app.modules.llm.gemini_image_client import GeminiImageClient gemini = GeminiImageClient() def bg_gen_fn(tag, prompt, labeled_refs, out_path: Path) -> Path: gemini.set_context( project_id=self.project_id, episode_id=self.episode_id, operation_type="outdoor_map_conti_plate", ) png, _ms = gemini.generate_image( prompt, labeled_references=[ (label, Path(p).read_bytes()) for label, p in labeled_refs ], aspect_ratio="16:9", ) from app.modules.pipeline.multiroll_gemini import ( atomic_write_bytes, ) return atomic_write_bytes(out_path, png) from app.core.config import settings as _settings from app.modules.pipeline.outdoor_map_conti import ( resolve_prompt_version as map_pack_version, ) from app.modules.pipeline.outdoor_shot_grounding import ( resolve_prompt_version as grounding_pack_version, ) from app.core.step_runner import compute_config_hash return run_outdoor_map_conti( conti_target_tags=outdoor_targets, shot_desc_by_tag=desc_by_tag, group_by_tag=group_by_tag, group_assets=group_assets, complex_by_tag=complex_by_tag, time_of_day_by_scene=classify_scenes, scene_texts=scene_texts, ground_fn=ground_fn, bg_gen_fn=bg_gen_fn, out_dir=out_dir, prompt_version=PROMPT_VERSION, extra_fingerprint={ "nb2_model": _settings.gemini_image_model, "map_pack": map_pack_version(PROMPT_VERSION), # 3차 리뷰 H2: grounding 계약 실질 입력 — 팩·정정 SOT· # 모델 라우팅(project_config) "grounding_pack": grounding_pack_version("1"), "corrections_block": corrections_block or "", "project_config_hash": compute_config_hash( self.project_config ), }, force=force, ) def _make_lane_sketch_fn(self): """레인1 스토리보드 스케치 — 참조=클린 canon 맵, gpt edit + sanitize 재시도 (R4: 마커 굽힌 control 소멸). 2026-08-25 — `model` 인자를 받는다. 마커 맵 검사가 소진되면 호출부가 **다른 모델로** 다시 시킨다(같은 모델 되풀이는 값이 없다는 실측). 슬러그 꼴(`provider/model`)이면 OpenRouter 로 나가고, 그 밖은 종전대로 OpenAI edit 이다 — 두 경로는 API 모양이 다르다. """ from app.core.config import settings as _cfg from app.modules.llm.gpt_image_primitive import call_gpt_image_bytes from app.core.image_call_budget import reserve_current_call from app.modules.pipeline.marker_map_engine import draw_marker_map_via_model from app.modules.pipeline.shot_conti_light import CONTI_SIZE from app.modules.prompt_sanitizer import PromptSanitizer client = _resolve_openai_client() sanitizer = PromptSanitizer(project_config=self.project_config) # 기본 엔진은 **설정**이 정한다 — 종전에는 소스 상수였다. _default_model = str( getattr(_cfg, "marker_map_image_model", "") or "gpt-image-2.5-sunburst") call_kwargs = { "model": _default_model, "size": CONTI_SIZE, "quality": "high", "n": 1, } def sketch_fn( tag: str, prompt: str, ref_paths: list, role: str = "lane_storyboard_sketch", size: str = None, out_meta: dict = None, model: str = "", ) -> bytes: _model = model or _default_model png = draw_marker_map_via_model( model=_model, prompt=prompt, ref_paths=list(ref_paths), role=role, capture_context={ "project_id": self.project_id, "episode_id": self.episode_id, "operation_type": "shot_conti_light_lane", "multiroll_tag": tag, }, # OpenAI 갈래만 이 자리의 사정(client·sanitizer·size)을 # 안다 — 엔진으로 끌어올리지 않고 함수로 넘긴다. openai_render=lambda m: _sketch_via_openai( tag, prompt, ref_paths, role, size, out_meta, m), ) if out_meta is not None: out_meta.setdefault("effective_prompt", prompt) out_meta["model"] = _model return png def _sketch_via_openai( tag: str, prompt: str, ref_paths: list, role: str, size, out_meta, model: str, ) -> bytes: # size: v5 마커 작화=canon 맵과 동일 1:1 캔버스 (Codex # BLOCKING-1 — 기본 None=스케치 16:9 불변). out_meta: # provenance 용 effective prompt 회수(sanitize 반영, HIGH-2). current = prompt for attempt in (1, 2): try: reserve_current_call(source="shot_conti_light.marker") png = call_gpt_image_bytes( client, mode="edit", prompt=current, ref_paths=[str(p) for p in ref_paths], call_kwargs=dict( call_kwargs, model=model, **({"size": size} if size else {})), capture_role=role, capture_metadata={ "operation_type": "shot_conti_light_lane", "shot_tag": tag, }, ) if out_meta is not None: out_meta["effective_prompt"] = current out_meta["model"] = model return png except Exception as exc: # noqa: BLE001 msg = str(exc).lower() moderated = ( "moderation" in msg or "safety" in msg or "content_policy" in msg or "rejected" in msg ) if attempt == 1 and moderated: sr = sanitizer.sanitize(current, str(exc), [], attempt) sanitized = sr.get("sanitized_prompt") or "" if sanitized: current = sanitized continue raise raise RuntimeError("unreachable") return sketch_fn def _run_lane_conti( self, *, scene_texts: Dict[int, str], continuity: Dict[str, Any], location_by_scene: Dict[int, str], scene_headings: Dict[int, str], selected_keys, out_dir: Path, plate_map: Dict[str, Path], assign_by_key: Dict[str, str], force: bool = False, ) -> Dict[str, Any]: """야외 lane 샷 처리 (Stage D → 2026-07-16 사용자 확정 재설계). - 레인1(map_marker), 팩 v5 (2026-07-24 사용자 지시 — 마커 맵 단계 복원, i2i): base map(클린 canon map)→geometry(LLM)→ **i2i 마커 맵**(이미지 모델이 1:1 캔버스로 CAM·시선·슬롯 마커 작화, 코드 드로잉 금지 준수)→마커 VLM 검증(원본 대조 2이미지, 소진=fail-closed)→스토리보드 스케치(gpt edit, 참조=마커 맵+geometry ID-free 텍스트 절)→leakage VLM gate. - 레인2(structure_plate): 맵·마커·마커 스케치 **제거** — A/B 정책 entry(ab_select_pending)만 기록하고 일반 경량 콘티 경로에 병합. finalize(finalize_structure_ab_entries)가 ready/bypass/failed 확정. persisted plan 소비 직전 재검증(fail-closed), camera_direction 결손=AppError(Codex 배선 조건 ③). 팩이 마네킹 저작(v14+)이면 진입 시 하류 교체 계약 3플래그를 요구한다(마네킹 유출 1층). 반환 {"lane_contis": {tag: {...}}}. 샷 단위 실패 격리(leakage 소진 =failed). 사이드카 지문(lane_records.json)으로 재개. """ from app.core.errors import AppError from app.modules.llm.llm_client import call_structured from app.modules.pipeline.multiroll_gemini import ( atomic_write_bytes, png_part, ) from app.modules.pipeline.multiroll_select import ( compute_input_fingerprint, ) from app.modules.pipeline.outdoor_lane_plan import ( build_group_shot_reconstructor, lane_bindings_by_tag, revalidate_persisted_plan, validate_group_parity, ) from app.modules.pipeline.outdoor_marker_map import ( GEOMETRY_TEXT_VERSION, MARKER_ANNOTATE_CONTRACT_VERSION, MARKER_GEOMETRY_CONTRACT_VERSION, MARKER_MAP_SIZE, _ANCHOR_SKETCH_PACKS as _anchor_sketch_packs, _PLATE_LOOK_PACKS as _plate_look_packs, _VIEW_SKETCH_PACKS as _view_sketch_packs, _coord_first, build_leakage_schema, build_marker_annotate_prompt, build_marker_check_schema, build_marker_check_user_lines, build_marker_sketch_prompt, lane_conti_fingerprint_payload, load_marker_check_sys, marker_check_user_header, resolve_prompt_version, resolve_sketch_pack_version, run_marker_geometry_shot, ) from app.modules.pipeline.shot_conti_light import ( archive_stale_output, load_sidecar_records, save_sidecar_records, ) from app.modules.pipeline.shot_ref_classify import parse_tag from app.modules.prompt_loader import load_prompt from app.modules.pipeline.shot_conti_light import ( CONTI_IMAGE_MODEL as _conti_model, ) lane_cp = self._load_prev_checkpoint("outdoor_lane_plan") if lane_cp is None: # Codex Stage D BLOCKING-2: lane pipe ON 인데 plan CP 부재 = # fail-closed — 야외 샷이 일반 콘티 경로로 조용히 하강 금지 raise AppError( code="step.missing_input.shot_conti_light_lane", message=( "outdoor_lane_plan 체크포인트 부재 — lane pipe ON " "상태에서는 lane plan 선행 필수" ), status_code=422, ) lane_data = lane_cp.get("data", {}) or {} bindings = lane_bindings_by_tag(lane_data) spec_cp = self._load_prev_checkpoint("outdoor_place_spec") spec_groups = (spec_cp or {}).get("data", {}).get("groups", {}) or {} # Codex 배선 조건 ②(BLOCKING-2 강화): persisted plan 재검증을 # plan 자기 바인딩이 아니라 **현재 authoritative 선택 샷 집합** # (Stage A 와 동일 재구성)으로 수행 — stale CP 의 샷 누락이 # 일반 콘티로 하강하는 것을 커버리지 위반으로 차단. reconstruct = build_group_shot_reconstructor( staging_cp=self._load_prev_checkpoint("shot_staging"), validator_cp=self._load_prev_checkpoint("shot_validator"), selection_cp=self._load_prev_checkpoint("shot_selection"), director_cp=self._load_prev_checkpoint("scene_director"), ) # 그룹 단위 parity (Codex 재리뷰 BLOCKING-1): 현재 샷이 있는 그룹의 # lane entry 누락/failed = fail-closed — 일반 콘티 하강 차단 parity = validate_group_parity( lane_data=lane_data, spec_groups=spec_groups, reconstruct=reconstruct, ) if parity: raise AppError( code="step.missing_input.shot_conti_light_lane", message=( "lane plan 그룹 parity 위반 — outdoor_lane_plan " "재실행 필요: " + "; ".join(parity[:8]) ), status_code=422, ) for gid in sorted(lane_data.get("groups", {}) or {}): entry = lane_data["groups"][gid] if not isinstance(entry, dict) or entry.get("status") != "ok": continue violations = revalidate_persisted_plan( entry.get("plan") or {}, scene_texts, group_shots=reconstruct(spec_groups.get(gid) or {}), ) if violations: raise AppError( code="step.contract_violation.shot_conti_light_lane", message=( f"persisted lane plan 재검증 실패 (group={gid}): " + "; ".join(violations[:8]) ), status_code=422, ) if not bindings: logger.info("shot_conti_light: lane 바인딩 없음 — lane 분기 no-op") return {"lane_contis": {}} canon_cp = self._load_prev_checkpoint("outdoor_place_canon") canon_groups = (canon_cp or {}).get("data", {}).get( "groups", {}) or {} seed_cp = self._load_prev_checkpoint("outdoor_structure_seed") seed_groups = (seed_cp or {}).get("data", {}).get("groups", {}) or {} # 2026-07-16 사용자 확정(복잡 구조물=맵 제외+A/B): outdoor_frame_mode # 소비 제거 — structure_plate 샷은 스케치 자체가 없어 frame_mode # (structure_dominant=스케치 생략) 판정이 무의미해짐. # 샷 dict: 서술 SOT=shot_validator + staging camera_direction 병합 # (Stage A 실측 교훈 — staging 엔 description 없음) validator_cp = self._load_prev_checkpoint("shot_validator") vshot_by_key: Dict[str, Dict[str, Any]] = {} for sc in (validator_cp or {}).get("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") if shi is None: continue v = dict(sh) v["scene_index"] = int(si) v["shot_index"] = int(shi) vshot_by_key[f"S{int(si)}sh{int(shi)}"] = v staging_cp = self._load_prev_checkpoint("shot_staging") cam_by_tag: Dict[str, str] = {} for sh in (staging_cp or {}).get("data", {}).get("shots", []) or []: si, shi = sh.get("scene_index"), sh.get("shot_index") if si is not None and shi is not None: cam_by_tag[f"S{int(si)}sh{int(shi)}"] = ( sh.get("camera_direction") or "") from app.modules.pipeline.outdoor_direct_common import is_selected from app.modules.pipeline.shot_conti_light import ( build_pose_clauses, ) # 마네킹 유출 1층(producer 거부): 마네킹 팩으로 콘티를 굽기 전에 # 하류 교체 계약이 살아 있는지 확인한다. 계약이 없으면 마네킹이 # legacy 조립 참조로 흘러 최종 스틸까지 남는다 — 굽지 않는 쪽이 # 안전하다(fail-closed). # # 대상은 **실제로 마네킹 스케치를 굽는 샷**뿐이다: bindings 는 # lane "none"/"structure_plate" 까지 포함하는데 none 은 아래에서 # continue(일반 파이프), structure_plate 는 정책 entry 만 만들고 # continue 라 둘 다 스케치가 없다 — 유출 표면이 0 인 프로젝트를 # 기본 플래그에서 hard-fail 시키면 안 된다(선택 샷 기준도 동일). _mannequin_shots = [ t for t, b in bindings.items() if b.get("lane") == "map_marker" and is_selected(parse_tag(t), selected_keys) ] if ( LANE_SKETCH_PACK_VERSION in _mannequin_sketch_packs() and _mannequin_shots and not self._mannequin_chain_ready() ): raise AppError( code="step.config.lane_mannequin_chain_off", message=( f"lane sketch 팩 v{LANE_SKETCH_PACK_VERSION}(마네킹)" "인데 still_bgfirst_enabled/" "still_bgfirst_full_enabled/" "still_lane_prev_bgfirst_enabled 가 모두 ON 이 아님 " "— 마네킹 콘티는 배경/엔티티 교체 계약과 짝으로만 " "유효하다 (마네킹 유출 방지, fail-closed)" ), status_code=422, ) # 2026-07-26: 소비자(still_recipe_service)가 팩 계약을 exact # revalidate 할 수 있도록 resolved 팩명을 CP entry 에 영속한다 — # 이전에는 lane_records sidecar 지문 extra 에만 있어서 구 팩 # 콘티를 신 계약으로 조용히 소비할 수 있었다. 생성/reuse 양 # 경로에 같은 스탬프를 쓴다(reuse 는 지문에 sketch_pack 이 접혀 # 있어 항상 동일 팩). _pack_stamp: Dict[str, Any] = { "lane_sketch_pack": resolve_sketch_pack_version( LANE_SKETCH_PACK_VERSION), "lane_geometry_pack": resolve_prompt_version( LANE_GEOMETRY_PACK_VERSION), "mannequin_chain_contract": ( MANNEQUIN_CHAIN_CONTRACT_VERSION if LANE_SKETCH_PACK_VERSION in _mannequin_sketch_packs() else "" ), } leakage_sys = load_prompt( "marker_map_sketch", "leakage_judge", version=resolve_sketch_pack_version(LANE_SKETCH_PACK_VERSION), ) # v5: i2i 마커 맵 검증 계약 (존재·근사 위치·플랜 보존) marker_check_sys = load_marker_check_sys(LANE_SKETCH_PACK_VERSION) sketch_fn = self._make_lane_sketch_fn() # v9(케이스1 단계 C·D): 스케치 2번째 참조=장소 실사(canon master) plate_look_on = LANE_SKETCH_PACK_VERSION in _plate_look_packs # ★#92 (2026-08-27): Sol → gemini-pro. 사용자 지시 # 「gpt sol vlm 은 성능이 안좋아」. Opik 실측에서 이 자리가 # **아직 Sol 로 도는 두 곳 중 하나**였다(30일 12건). # # ★바로 아래 두 자리(`marker_sketch_leakage`· # `lane_marker_map_check`)는 이미 `gemini-pro` 다 — 같은 도면· # 같은 마커를 보는데 판정 모델만 갈려 있었다. # # ★**이중화(두 모델)는 여기서 안 한다.** 도면 읽기는 실측 8장으로 # 「exact 를 요구하면 안 된다」를 알아낸 뒤에 계약을 정했는데, # 이 자리는 그런 실측이 없다. 합의 규칙을 실측 없이 넣는 것은 # 바로 그 함정이다. geometry_config = { "outdoor_marker_geometry": {"model": LANE_GEOMETRY_JUDGE_ALIAS}} geometry_config.update(self.project_config or {}) # ★지문이 보는 것과 **같은 자리**에서 실효 신원을 읽는다. _geo_judge = lane_geometry_judge_identity(self.project_config) leakage_config = {"marker_sketch_leakage": {"model": "gemini-pro"}} leakage_config.update(self.project_config or {}) marker_check_config = { "lane_marker_map_check": {"model": "gemini-pro"}} marker_check_config.update(self.project_config or {}) # HIGH-3: 판정 alias 의 물리 모델 — 지문·hash 스탬프용 from app.core.config import settings as _settings _marker_judge_model = _settings.gemini_text_model records_path = out_dir / "lane_records.json" sidecar = load_sidecar_records(records_path) out_dir.mkdir(parents=True, exist_ok=True) # Codex 합의 BLOCK1 (2026-08-12): 운영자 무콘티 예외 태그 — 유효 # 성공 캐시가 없으면 유료 체인에 들어가지 않고 단락한다 (아래 # per-shot 루프에서 소비). from app.core.config import settings as _exempt_settings from app.services.still_recipe_service import ( parse_no_conti_exempt_tags as _parse_exempt, ) _lane_exempt_tags = _parse_exempt( getattr(_exempt_settings, "bgfirst_no_conti_exempt_tags", "")) lane_contis: Dict[str, Any] = {} # 감사 1-B — 샷별 확정 배정 인물. **루프 밖에서 한 번** 읽는다. _vis_sids = self._visible_char_sids_by_tag() for tag in sorted(bindings, key=lambda t: parse_tag(t)): b = bindings[tag] si, shi = parse_tag(tag) if not is_selected((si, shi), selected_keys): continue gid = b["group_id"] lane = b["lane"] # 재설계 C (2026-07-19): lane "none"=일반 파이프 — lane entry 를 # 만들지 않아야 일반 경량 콘티가 담당한다(집계·finalize·일반 # 경로 제외는 전부 map_marker/structure_plate positive 매칭). # entry 를 만들면 아래 레인1 경로가 canon 맵을 요구해 fail-closed. if lane == "none": continue entry: Dict[str, Any] = { "lane": lane, "group_id": gid, "segment_id": b["segment_id"], } lane_contis[tag] = entry # ── 복잡 구조물(structure_plate) = A/B 정책 entry (2026-07-16 # 사용자 확정 — 맵·마커·마커 스케치 제거). 이미지 생성 없음: # 일반 경량 콘티(플레이트 느슨 참조)는 일반 경로가 생성하고, # 콘티/플레이트/seed 완비 검사는 일반 콘티 후 finalize # (finalize_structure_ab_entries)가 수행 — Codex R2 policy/ # asset entry 분리 계약. seed 상태는 여기서 그대로 기록 # (결손도 finalize 의 fail-closed 판정 입력). if lane == "structure_plate": seed_entry = seed_groups.get(gid) or {} # seed-bg 조건 2 (Codex): 배경 권위 typed 해석 — # valid plate > 구조적 부재+valid seed > failure. from app.modules.pipeline.shot_conti_light import ( resolve_structure_bg_source, ) bg_source, bg_path, bg_reason = resolve_structure_bg_source( shot_key=f"{si}_{shi}", plate_map=plate_map, assign_by_key=assign_by_key, seed_path=seed_entry.get("seed_png_path"), seed_status=seed_entry.get("status") or "missing", ) entry.update( status="ab_select_pending", seed_path=seed_entry.get("seed_png_path") or None, seed_asset_id=seed_entry.get("seed_asset_id"), seed_status=seed_entry.get("status") or "missing", bg_source=bg_source, bg_path=bg_path, bg_reason=bg_reason, ) continue spec = (spec_groups.get(gid) or {}).get("spec") if not spec: entry.update(status="failed", error="outdoor_place_spec 부재") continue # Codex 3차 리뷰 BLOCK-2 반영 (2026-08-12): 예외 태그 샷은 # 여기서 **자기 완결로 즉시 무료 exempt** 한다 — base/master/ # camera fail-closed raise 보다 먼저이고, 어떤 파일·입력 # 상태(디렉터리·권한·TOCTOU)에서도 raise 에 도달할 수 없다. # 캐시 재사용은 의도적으로 포기한다: 재사용 판정을 여기서 # 복제하면 본 경로와의 드리프트 표면이 새로 생기고(측정 # 도구≠프로덕션 재발 축), exempt 는 "이 샷은 콘티 없이 # 가라"는 운영자 선언이라 캐시 사용 가치도 낮다. 콘티를 # 다시 원하면 force(명시 재시도)가 정상 경로다. # non-exempt 의 fail-closed 는 그대로다. if not force and tag in _lane_exempt_tags: entry.update( status="failed", error=("operator_exempt short-circuit — 운영자 무콘티 " "선언, 유료 시도 생략"), ) logger.warning( "shot_conti_light %s: 운영자 무콘티 예외 — 유료 시도 " "없이 계수 제외 대상으로 표시", tag) continue # base map 결정(레인1) — 결손 = fail-closed (레인 배정 샷의 # 상류 자산 부재는 조용한 degrade 금지) canon = canon_groups.get(gid) or {} base_map = canon.get("map_png_path") or "" base_asset_id = canon.get("map_asset_id") if ( canon.get("status") != "ok" or not base_map or not Path(base_map).exists() ): raise AppError( code="step.missing_input.shot_conti_light_lane", message=( f"{tag}(lane={lane}) base 자산 결손 — " f"outdoor_place_canon 산출 필요 (group={gid})" ), status_code=422, ) # 2026-07-25 사용자 스펙 케이스1 단계 C·D: 콘티는 맵(A)만이 # 아니라 **장소 배경**도 참조한다 — lane 샷만 배경 권위가 # 없어 같은 장소 일반 샷과 다른 장소처럼 그려지던 실측 결함 # (S15sh1 vs S15sh5). canon master=맵과 짝인 장소 실사라 # 결손은 map 과 동일하게 fail-closed. master_png = canon.get("master_png_path") or "" master_asset_id = canon.get("master_asset_id") if plate_look_on: # 파일 실재+비어있지 않음+asset UUID 까지 fail-closed — # lineage(장소 LOOK edge)가 UUID 없이 성립하지 않는다 _mp = Path(master_png) if master_png else None if ( _mp is None or not _mp.is_file() or _mp.stat().st_size == 0 or not master_asset_id ): raise AppError( code="step.missing_input.shot_conti_light_lane", message=( f"{tag}(lane={lane}) canon master 결손/무효 " f"({master_png!r}, asset={master_asset_id!r}) " f"— lane 콘티의 장소 외형 권위 필요 " f"(group={gid})" ), status_code=422, ) # Codex 배선 조건 ③: camera_direction fail-closed camera_direction = (cam_by_tag.get(tag) or "").strip() if not camera_direction: raise AppError( code="step.missing_input.shot_conti_light_lane", message=( f"{tag} shot_staging camera_direction 결손 — " "lane 스케치는 카메라 연출 메모 필수" ), status_code=422, ) vshot = dict(vshot_by_key.get(tag) or {}) vshot.update({"scene_index": si, "shot_index": shi, "camera_direction": camera_direction}) scene_text = scene_texts.get(si, "") map_png = Path(base_map).read_bytes() # 스케치 결정 입력 선계산 — 지문에 전부 포함 (Codex Stage D # HIGH-5: pose/carried/place/vshot 전체 필드가 바뀌어도 # 구 스케치가 reused 되던 결함) # ★★**키가 틀려 lane 은 연속성이 통째로 안 나가고 있었다** # (2026-08-27 Codex BLOCK, main 부터 있던 결함). # # 생산자는 `tag_of` = `S{si}sh{shi}` 로 적는데 이 갈래만 # `S{si}_Shot{shi}` 로 찾았다. 체크포인트 실측: 키 258개 중 # `_Shot` 모양 **0개** — `pose_clauses`·`pose_fix`·`carried` # 셋이 전부 빈 값으로 나갔다. # # ★고치면 **동작이 바뀐다** — lane 샷에 자세 정본과 이어받는 # 상태가 처음으로 실린다. 그것이 원래 의도다. _tag = tag_of(int(si), int(shi)) pose_clauses = build_pose_clauses( continuity.get("pose_canon") or [], _tag) fix = (continuity.get("pose_fix") or {}).get(_tag) or {} # 감사 1-B (2026-08-27) — 그 샷에 배정된 인물만. # ★lane 은 일반 콘티와 달리 무인 샷도 닿으므로 이 필터가 특히 # 필요하다(Codex 판정). carried = carried_clause_for( continuity.get("carried"), _tag, fix, visible_short_ids=_vis_sids.get(_tag) or set(), bg_only=not (vshot.get("person_visible", True))) place_text = (location_by_scene.get(si) or scene_headings.get(si) or "") # 입력 지문 — geometry(LLM) 없이 재사용 판정 (B3 관례). # payload=결정 입력 전체 직렬화 (HIGH-5, 순수 함수 잠금) fingerprint = compute_input_fingerprint( prompt=lane_conti_fingerprint_payload( spec=spec, vshot=vshot, scene_text=scene_text, lane=lane, segment_label_en=b["segment_label_en"], place_text=place_text, pose_clauses=pose_clauses, carried_en=carried, ), labeled_refs=( [("map", Path(base_map))] + ([("master", Path(master_png))] if plate_look_on else []) ), roll_count=1, critique_enabled=False, extra={ # Codex ④: selector 가 아닌 resolved 팩명 "geometry_pack": resolve_prompt_version( LANE_GEOMETRY_PACK_VERSION), "marker_geometry_contract": ( MARKER_GEOMETRY_CONTRACT_VERSION), # HIGH-3: selector 가 아닌 resolved 팩명 — 동일 # selector 재발행(타임스탬프 drift)도 실질 입력 "sketch_pack": resolve_sketch_pack_version( LANE_SKETCH_PACK_VERSION), # R5: 직렬화 형식 변경도 스케치 실질 입력 "geometry_text_version": GEOMETRY_TEXT_VERSION, "conti_model": _conti_model, # v5 (HIGH-3): 마커 작화·검증 계약 — 스키마/2이미지 # 비교/재시도 정책·판정 물리 모델·캔버스 크기가 # 바뀌면 마커 맵·스케치 재생성 "marker_contract": MARKER_ANNOTATE_CONTRACT_VERSION, "marker_max_attempts": LANE_MARKER_ANNOTATE_MAX_ATTEMPTS, "marker_judge_model": _marker_judge_model, # ★기하 판정은 **별도 칸**이다 (위 칸은 검증 모델). "geometry_judge": _geo_judge, "marker_map_size": MARKER_MAP_SIZE, }, ) sketch_path = out_dir / f"lane_sketch_{tag}.png" marker_map_path = out_dir / f"lane_marker_map_{tag}.png" prev_rec = sidecar.get(tag) or {} if ( not force and sketch_path.exists() and marker_map_path.exists() and prev_rec.get("input_fingerprint") == fingerprint and (prev_rec.get("leakage") or {}).get( "has_marker_leakage") is False and (prev_rec.get("marker_check") or {}).get( "markers_ok") is True # HIGH-3 후속: 마커 맵 파일 무결성 — sidecar 기록 SHA 와 # 현재 파일 대조 (마커 맵은 지문 입력이 아니므로 별도 # 감사값으로 잠근다) and prev_rec.get("marker_map_sha256") and hashlib.sha256( marker_map_path.read_bytes()).hexdigest() == prev_rec.get("marker_map_sha256") ): entry.update( status="ok", image_path=str(sketch_path), base_map_path=base_map, base_map_asset_id=base_asset_id, marker_map_path=str(marker_map_path), marker_check=prev_rec.get("marker_check"), marker_prompt=prev_rec.get("marker_prompt"), marker_attempts=prev_rec.get("marker_attempts"), # 재사용 판도 「무엇을 골라 썼나」를 잃지 않는다 — # 여기서 떨어뜨리면 재개 한 번에 기록이 사라진다. marker_best_effort=prev_rec.get( "marker_best_effort", False), marker_pick_index=prev_rec.get("marker_pick_index"), marker_pick_count=prev_rec.get("marker_pick_count"), marker_pick_why_ko=prev_rec.get("marker_pick_why_ko"), marker_model=prev_rec.get("marker_model"), geometry=prev_rec.get("geometry"), leakage=prev_rec.get("leakage"), prompt=prev_rec.get("prompt"), reused=True, **_pack_stamp, ) if plate_look_on: entry.update( master_png_path=master_png, master_asset_id=master_asset_id, ) continue # stale 산출 아카이브 (지문 불일치/force) archive_stale_output(sketch_path) archive_stale_output(marker_map_path) try: g_out = run_marker_geometry_shot( spec=spec, shot=vshot, scene_text=scene_text, map_png=map_png, segment_label_en=( b["segment_label_en"] or "fixed structure and its surroundings"), prompt_version=LANE_GEOMETRY_PACK_VERSION, project_config=geometry_config, opik_metadata=self.build_opik_metadata(), ) geometry = g_out["geometry"] # v5 (2026-07-24 사용자 지시): 마커 맵 단계 = i2i — # 이미지 모델이 클린 canon 맵 위에 CAM·시선·엔티티 # 마커를 그리고(코드 드로잉 금지 준수), VLM 이 존재· # 근사 위치·플랜 보존을 검증한다. 스케치 참조=마커 맵. annotate_prompt = build_marker_annotate_prompt( geometry=geometry, prompt_version=LANE_SKETCH_PACK_VERSION, ) # ★2026-08-25 방향 수리 — 이 머리말을 스텝이 손으로 들고 # 있었고, 그 문장이 검사기에게 "분율은 coarse estimate" # 라고 말했다. 그래서 검사기는 랜드마크 산문(방위 낱말)을 # 권위로 읽었고, 좌표가 중앙에 둔 피사체를 「남동에 있어야 # 한다」며 반려했다. 도면에 나침반이 없으니 그 말은 그림에서 # 확인할 수 없다. 머리말을 팩 버전과 함께 움직이게 옮겼다. check_user = ( marker_check_user_header(LANE_SKETCH_PACK_VERSION) + "\n" + "\n".join(build_marker_check_user_lines( geometry, require_anchors=( LANE_SKETCH_PACK_VERSION in _anchor_sketch_packs), require_view=( LANE_SKETCH_PACK_VERSION in _view_sketch_packs), coord_first=_coord_first( LANE_SKETCH_PACK_VERSION))) ) marker_check: Dict[str, Any] = {} marker_meta: Dict[str, Any] = {} marker_png = b"" marker_ok = False # ★그리는 모델 (2026-08-25 사용자 지시 "그냥 nb2 만" — # "공간이나 방향은 nb2 가 훨씬 잘하는 것 확인했어"). # 설정이 준 목록을 그대로 돈다. 지금 목록은 nb2 한 칸이라 # 종전과 **같은 횟수·같은 순서**로 돌고, 후보를 다시 넣는 # 날에는 이 자리가 그대로 후퇴 사다리가 된다. # ★중첩을 늘리지 않고 (모델, 시도) 목록 하나로 편다 — # 루프를 겹치면 안쪽 break 가 바깥을 못 끊어 다음 모델까지 # 그리고 만다(돈이 새는 모양). from app.modules.pipeline.marker_map_engine import ( marker_map_models, ) from app.modules.pipeline.outdoor_marker_map import ( pick_best_marker_map, ) _models = marker_map_models() _plan = [(_m, _a) for _m in _models for _a in range( 1, LANE_MARKER_ANNOTATE_MAX_ATTEMPTS + 1)] current_annotate = annotate_prompt _prev_model = "" # ★사용자 지시 (2026-08-25): "그냥 nb2 만, nb2 가 3번 # 실패하면 그 셋 중에 하나 하라고 VLM 으로." # 그래서 시도마다 **산출을 버리지 않고 쥐고 있다가**, 통과가 # 없으면 그 셋을 VLM 에게 보여 주고 하나 고르게 한다. # 종전에는 마지막 시도만 남아 있어 고를 수가 없었고, 그래서 # 실패로 떨어뜨리는 수밖에 없었다(스텝이 partial 로 멈췄다). _cands: List[Dict[str, Any]] = [] for _model, m_attempt in _plan: if _model != _prev_model: # 모델이 바뀌면 문안을 처음 것으로 되돌린다 — 앞 # 모델에 준 재지시("이전 시도가 …로 실패했다")가 # 새 모델의 첫 시도를 오염시키지 않게. current_annotate = annotate_prompt if _prev_model: logger.warning( "shot_conti_light %s: 마커 맵 검사 소진 — " "모델을 %s 에서 %s 로 바꿔 다시 그린다", tag, _prev_model, _model) _prev_model = _model # BLOCKING-1: 캔버스=canon 맵과 동일 1:1 — 16:9 로 # edit 하면 '동일 맵+마커만' 계약과 좌표계가 깨진다 marker_png = sketch_fn( tag, current_annotate, [Path(base_map)], role="lane_marker_map", size=MARKER_MAP_SIZE, out_meta=marker_meta, model=_model) atomic_write_bytes(marker_map_path, marker_png) # BLOCKING-1: 보존 판정=원본 대조 — 클린 맵+마커 맵 # 2이미지를 명시 라벨 순서로 전달 marker_check = call_structured( "lane_marker_map_check", marker_check_sys, [{"type": "text", "text": check_user}, {"type": "text", "text": "ORIGINAL CLEAN SITE PLAN:"}, png_part(map_png), {"type": "text", "text": "ANNOTATED SITE PLAN (to check):"}, png_part(marker_png)], build_marker_check_schema(), project_config=marker_check_config, schema_name="lane_marker_map_check", opik_metadata=self.build_opik_metadata(), ) entry["marker_attempts"] = m_attempt if marker_check.get("markers_ok"): marker_ok = True break # 통과는 못 했지만 **버리지 않는다** — 다 못 하면 # 이 셋을 VLM 에게 보여 주고 하나 고르게 한다. _cands.append({"png": marker_png, "check": marker_check, "model": _model, "attempt": m_attempt}) current_annotate = ( annotate_prompt + "\n\n[REGENERATE — the previous attempt failed " + "the marker check: " + str(marker_check.get("details_ko", "")) + ". Draw exactly the specified markers and " + "change nothing else on the plan.]" ) if not marker_ok: # ★계속 그리지 않는다 — **그 셋 중 하나를 VLM 이 고른다** # (2026-08-25 사용자 지시). 종전에는 여기서 실패로 # 떨어뜨렸고, 그 한 샷 때문에 스텝이 partial 로 멈춰 # 뒤가 통째로 막혔다. if not _cands: entry.update( status="failed", error=("marker map 을 한 장도 못 그렸다 — 모델 " f"{', '.join(_models)}"), marker_models_tried=list(_models), ) continue _idx = pick_best_marker_map( base_png=map_png, candidates=_cands, geometry=geometry, prompt_version=LANE_SKETCH_PACK_VERSION, opik_metadata=self.build_opik_metadata(), project_config=marker_check_config, ) _pick = _cands[_idx] marker_png = _pick["png"] marker_check = _pick["check"] atomic_write_bytes(marker_map_path, marker_png) logger.warning( "shot_conti_light %s: 통과한 마커 맵이 없다 — %d장 중 " "%d번째를 VLM 이 골랐다 (%s). 시도한 모델 %s", tag, len(_cands), _idx + 1, _pick["model"], ", ".join(_models)) entry.update( marker_best_effort=True, marker_pick_index=_idx + 1, marker_pick_count=len(_cands), marker_pick_why_ko=_pick.get("pick_why_ko", ""), marker_model=_pick["model"], marker_check=marker_check, marker_models_tried=list(_models), ) sketch_prompt = build_marker_sketch_prompt( shot_desc=vshot.get("description") or "", place_text=place_text, geometry=geometry, pose_clauses=pose_clauses, carried_en=carried, camera_direction_en=camera_direction, seed_look_attached=False, plate_attached=plate_look_on, prompt_version=LANE_SKETCH_PACK_VERSION, ) # v5: 스케치 참조=i2i 마커 맵 (검증 통과본) # v9: +장소 실사(canon master) — 배치·카메라=마커 맵, # 장소 외형=사진 (plate_look 조항이 관할 분리) ref_paths = [marker_map_path] + ( [Path(master_png)] if plate_look_on else []) current = sketch_prompt verdict: Dict[str, Any] = {} ok = False for attempt in range(1, LANE_LEAKAGE_MAX_ATTEMPTS + 1): png = sketch_fn(tag, current, ref_paths) atomic_write_bytes(sketch_path, png) verdict = call_structured( "marker_sketch_leakage", leakage_sys, [{"type": "text", "text": ("Shot CAMERA DIRECTION (the sketch is " "allowed to follow this): " + camera_direction)}, {"type": "text", "text": "The storyboard sketch to check:"}, png_part(png)], build_leakage_schema(), project_config=leakage_config, schema_name="marker_sketch_leakage", opik_metadata=self.build_opik_metadata(), ) entry["sketch_attempts"] = attempt if not verdict.get("has_marker_leakage"): ok = True break current = ( sketch_prompt + "\n\n[REGENERATE — the previous attempt leaked " + "staging markers: " + str(verdict.get("details_ko", "")) + ". Remove every marker symbol.]" ) if not ok: entry.update( status="failed", error=( "marker leakage " f"{LANE_LEAKAGE_MAX_ATTEMPTS}회 소진: " + str(verdict.get("details_ko", ""))), leakage=verdict, ) continue except AppError: raise except Exception as exc: # noqa: BLE001 logger.exception( "shot_conti_light lane: %s 실패", tag) entry.update(status="failed", error=str(exc)) continue _marker_prompt = ( marker_meta.get("effective_prompt") or annotate_prompt) _marker_sha = hashlib.sha256(marker_png).hexdigest() entry.update( status="ok", image_path=str(sketch_path), base_map_path=base_map, base_map_asset_id=base_asset_id, marker_map_path=str(marker_map_path), marker_check=marker_check, marker_prompt=_marker_prompt, geometry=geometry, geometry_attempts=g_out["attempts"], leakage=verdict, prompt=sketch_prompt, **_pack_stamp, ) if plate_look_on: # 케이스1 단계 C 소비 기록 — 콘티가 실제로 참조한 장소 # 실사 (lineage/감사; 소비자는 이 경로로 배경 권위 추적) entry.update( master_png_path=master_png, master_asset_id=master_asset_id, ) sidecar[tag] = { "input_fingerprint": fingerprint, "geometry": geometry, "marker_check": marker_check, # HIGH-2: 마커 provenance — reuse 후에도 감사 필드 보존 "marker_prompt": _marker_prompt, "marker_attempts": entry.get("marker_attempts"), # ★검사를 못 통과하고 **VLM 이 골라 쓴 판**이면 그 사실을 # 기록에 남긴다. 종전에는 `entry` 에만 담겨 스텝 반환값과 # 함께 사라졌고, 파일에는 「검사 실패」만 남아 무엇을 쓰기로 # 했는지 알 길이 없었다 — 무엇을 골랐는지 모르면 다음 판과 # 대조할 수도 없다. (실측 2026-08-26: Opik 에는 고르기 호출이 # 남았는데 sidecar 에는 한 칸도 없었다.) "marker_best_effort": entry.get("marker_best_effort", False), "marker_pick_index": entry.get("marker_pick_index"), "marker_pick_count": entry.get("marker_pick_count"), "marker_pick_why_ko": entry.get("marker_pick_why_ko"), "marker_model": entry.get("marker_model"), # 마커 맵 무결성 감사값 (지문 미접힘의 보완) "marker_map_sha256": _marker_sha, "leakage": verdict, "prompt": sketch_prompt, } save_sidecar_records(records_path, sidecar) return {"lane_contis": lane_contis} def _execute(self, mode: str = "resume") -> Dict[str, Any]: from app.core.config import settings from app.modules.pipeline.outdoor_direct_common import ( build_selected_keys, is_selected, ) from app.modules.pipeline.shot_conti_light import ( resolve_shot_plate_map, run_shot_conti_light, ) classify_cp = self._load_prev_checkpoint("shot_ref_classify") classify_shots = ( (classify_cp or {}).get("data", {}).get("shots", {}) or {} ) # B-3 (2026-07-19): 배경 공유 계획=classify prev 의 상위 권위. # 스틸 소비(still_recipe_service)와 동일 override 를 이 스텝의 # classify SOT 에 1회 적용 — 콘티 생성 대상(conti_targets)과 # finalize_structure_ab_entries 가 같은 effective prev 를 본다 # (E2E9 실측: 계획 prev 지휘 complex 샷이 ready 로 확정돼 스틸 # "ready 인데 prev" fail-closed — S7sh5/S28sh9). 계획 부재/스텝 # OFF=빈 dict 로 기존 판정 불변. 앵커 결손=ValueError 전파 # (fail-closed — B-2 validator 가 앵커 실재 보증). _sp_cp = self._load_prev_checkpoint("background_share_plan") _sp_plans = ( ((_sp_cp or {}).get("data", {}) or {}).get("plan", {}) or {} ).get("shot_plans") or {} if _sp_plans: from app.modules.pipeline.still_recipe import ( apply_share_plan_prev, ) classify_shots = { _t: ( {**(_c or {}), "prev": apply_share_plan_prev( (_c or {}).get("prev"), _sp_plans[_t])} if _sp_plans.get(_t) is not None else dict(_c or {}) ) for _t, _c in classify_shots.items() } # ★캐릭터(로봇 포함)가 보이는 샷은 무조건 사람 샷 (2026-09-19 사용자 # 지시). 스틸 조립과 **같은 함수**다 — 콘티 대상과 스틸의 bg_only 가 # 갈리면 콘티 없는 인물 샷이 생긴다. from app.modules.pipeline.shot_ref_classify import ( force_person_visible_for_characters, ) classify_shots = force_person_visible_for_characters( classify_shots, self._visible_char_sids_by_tag()) continuity = ( (self._load_prev_checkpoint("shot_continuity") or {}) .get("data", {}) ) validator_cp = self._load_prev_checkpoint("shot_validator") scenes = (validator_cp or {}).get("data", {}).get("scenes", []) or [] selected_keys = build_selected_keys( self._load_prev_checkpoint("shot_selection") ) shots: list = [] scene_headings: Dict[int, str] = {} for sc in scenes: si = sc.get("scene_index") if si is None: continue si = int(si) if sc.get("scene_heading"): scene_headings[si] = sc["scene_heading"] for sh in sc.get("shots", []) or []: shi = sh.get("shot_index") if shi is None: continue if not is_selected((si, int(shi)), selected_keys): continue shots.append( { "scene_index": si, "shot_index": int(shi), "description": sh.get("description") or "", } ) if not shots or not classify_shots: logger.info( "shot_conti_light: 샷/분류 데이터 없음 — no-op " "(shots=%d, classify=%d)", len(shots), len(classify_shots), ) return { "applicable_count": 0, "completed_count": 0, "failed_count": 0, "schema_version": SCHEMA_VERSION, "config_hash": self._config_hash(), "data": {"contis": {}}, } director_cp = self._load_prev_checkpoint("scene_director") scene_primary: Dict[int, str] = {} for sc in (director_cp or {}).get("data", {}).get("scenes", []) or []: si = sc.get("scene_index") primary = sc.get("primary_location", "") or "" if si is not None and primary: scene_primary[int(si)] = primary # 장소 서술 — shot_ref_classify 와 동일 파생 (EntityCanon 이름, # 없으면 헤딩) from app.modules.pipeline.shot_ref_classify import ( derive_location_by_scene, ) location_by_scene = derive_location_by_scene( self.db, self.project_id, scene_primary, scene_headings ) plate_map = resolve_shot_plate_map( settings.projects_dir, self.project_id, self.episode_id ) out_dir = ( Path(settings.projects_dir) / self.project_id / "images" / self.episode_id / "conti" ) lane_pipe_on = bool( getattr(settings, "outdoor_lane_pipe_enabled", False) ) and bool(getattr(settings, "outdoor_lane_plan_enabled", False)) map_conti_on = bool( getattr(settings, "outdoor_map_conti_enabled", False)) scene_texts: Dict[int, str] = {} if lane_pipe_on or map_conti_on: scene_save_cp = self._load_prev_checkpoint("scene_save") for seg in ( (scene_save_cp or {}).get("data", {}).get("segments", []) or [] ): si = seg.get("scene_index") if isinstance(si, int): # 씬 원문 전문 — 절대 자르지 않는다 scene_texts[si] = seg.get("text") or "" # ── 야외 lane 분기 = 마커 스케치 콘티 (Stage D, opt-in) ───── lane_contis: Dict[str, Any] = {} if lane_pipe_on: # B2 재리뷰: 배정 기록은 status **불문** 로더 — producer # 실패 그룹의 기록 소실이 '구조적 무생산' 오인(seed fail-open) # 되는 것을 차단 from app.modules.pipeline.plate_select import ( load_bg_assignment_any_status as _load_assign_any, ) _assign_by_key_lane = _load_assign_any( settings.projects_dir, self.project_id, self.episode_id) lane_result = self._run_lane_conti( scene_texts=scene_texts, continuity=continuity, location_by_scene=location_by_scene, scene_headings=scene_headings, selected_keys=selected_keys, out_dir=out_dir, plate_map=plate_map, assign_by_key=_assign_by_key_lane, force=(mode == "force"), ) lane_contis = lane_result.get("lane_contis") or {} # seed-bg 승격 (Codex 조건 1): 플레이트 producer 구조적 부재 # 그룹의 A/B 샷은 콘티=seed 느슨 참조 — local effective map 에 # seed 를 주입 (bg_source/bg_path 가 lane entry 에 명시 기록돼 # plate 위장이 아님; finalize 가 conti.plate_path 와 대조). from app.modules.pipeline.shot_ref_classify import ( parse_tag as _pt_seedbg, ) for _t, _e in lane_contis.items(): if ( isinstance(_e, dict) and _e.get("bg_source") == "seed" and _e.get("bg_path") ): _si2, _shi2 = _pt_seedbg(_t) plate_map[f"{_si2}_{_shi2}"] = Path(_e["bg_path"]) # 레인1(map_marker) 샷만 일반 콘티/맵 콘티 대상에서 제외 — # 콘티=마커 스케치. structure_plate(복잡 구조물) 샷은 2026-07-16 # 사용자 확정으로 **일반 경량 콘티 대상에 포함** (정책 entry 는 # 아래 finalize 가 확정 — Codex R2 policy/asset 분리). if lane_contis: from app.modules.pipeline.shot_ref_classify import tag_of _map_lane_tags = { t for t, e in lane_contis.items() if isinstance(e, dict) and e.get("lane") == "map_marker" } shots = [ s for s in shots if tag_of(int(s["scene_index"]), int(s["shot_index"])) not in _map_lane_tags ] # ── 야외 복잡 구도 = 맵 기반 플레이트 분기 (opt-in, dormant) ── map_plates: Dict[str, Any] = {} if map_conti_on: classify_scenes = ( (classify_cp or {}).get("data", {}).get("scenes", {}) or {} ) map_result = self._run_map_conti( shots=shots, classify_shots=classify_shots, classify_scenes=classify_scenes, scene_texts=scene_texts, scene_primary=scene_primary, selected_keys=selected_keys, out_dir=out_dir, force=(mode == "force"), ) map_plates = map_result.get("map_plates") or {} # 맵 플레이트가 성립한 샷은 일반 플레이트를 대체 — 이후 콘티·스틸 # 은 일반 샷과 완전 동일 파이프 (사용자 확정: 병합) from app.modules.pipeline.shot_ref_classify import parse_tag for tag, entry in map_plates.items(): p = entry.get("plate_path") if p: si, shi = parse_tag(tag) plate_map[f"{si}_{shi}"] = Path(p) # ── R1 (Codex 설계 리뷰 BLOCKING-1): 플레이트 권위 선행 고정 ── # plate_select 판정을 콘티 생성 **전** 1회 수행·영속 — 콘티와 # 스틸(A/B 양 브랜치)이 정확히 같은 플레이트를 소비한다. 스틸 # 시점 재판정은 bg_only(콘티 없음) 샷만 잔존. plate_select_on = bool( getattr(settings, "still_plate_select_enabled", False)) plate_authority: Dict[str, Any] = {} if plate_select_on: from app.modules.llm.llm_client import call_structured from app.modules.pipeline.plate_select import ( load_bg_groups_and_assignment, plate_candidates_by_location, run_plate_authority, ) from app.modules.pipeline.shot_conti_light import ( conti_targets, load_sidecar_records, save_sidecar_records, ) from app.modules.pipeline.shot_ref_classify import ( parse_tag as _parse_tag, tag_of as _tag_of, ) classify_scenes_all = ( (classify_cp or {}).get("data", {}).get("scenes", {}) or {} ) shot_desc_by_tag = { _tag_of(int(s["scene_index"]), int(s["shot_index"])): s.get("description") or "" for s in shots } place_by_scene = { int(s["scene_index"]): ( (classify_scenes_all.get(str(int(s["scene_index"]))) or {}).get("place_en") or location_by_scene.get(int(s["scene_index"]), "") ) for s in shots } _bg_groups, assign_by_key = load_bg_groups_and_assignment( settings.projects_dir, self.project_id, self.episode_id) eligible = [ t for t in conti_targets(classify_shots) if t in shot_desc_by_tag and t not in map_plates ] out_dir.mkdir(parents=True, exist_ok=True) authority_records_path = out_dir / "plate_authority.json" auth_sidecar = load_sidecar_records(authority_records_path) plate_authority = run_plate_authority( tags=eligible, plate_map=plate_map, assign_by_key=assign_by_key, cands_by_loc=plate_candidates_by_location(_bg_groups), shot_desc_by_tag=shot_desc_by_tag, place_text_by_scene=place_by_scene, sidecar=auth_sidecar, call_structured_fn=call_structured, judge_model=settings.gemini_text_model, project_config=self.project_config, opik_metadata=self.build_opik_metadata(), force=(mode == "force"), ) save_sidecar_records(authority_records_path, auth_sidecar) for _tag, _entry in plate_authority.items(): _p = _entry.get("plate_path") if _p: _si, _shi = _parse_tag(_tag) plate_map[f"{_si}_{_shi}"] = Path(_p) from app.modules.pipeline.shot_conti_light import ( CONTI_IMAGE_MODEL as _conti_model, resolve_prompt_version as _conti_pack, ) # ── BGFIRST2 이식 ② (2026-07-20, opt-in): 원근 가이드 콘티 v2 — # 샷별 서브공간 place + staging CAMERA/FRAME 절 + 캐릭터 비례 참조. # OFF = 세 입력 전부 None → run 은 v1 경로 byte-identical. _selector = self._prompt_selector() _classify_scenes = None _staging_by_key = None _char_refs = None _extra_fp = { "conti_model": _conti_model, "conti_pack": _conti_pack(_selector), } if _selector != PROMPT_VERSION: _classify_scenes = ( (classify_cp or {}).get("data", {}).get("scenes", {}) or {} ) _staging_cp = self._load_prev_checkpoint("shot_staging") _staging_by_key = { f"{sh['scene_index']}_{sh['shot_index']}": sh for sh in ( ((_staging_cp or {}).get("data", {}) or {}) .get("shots", []) or [] ) if sh.get("scene_index") is not None and sh.get("shot_index") is not None } _char_refs = self._char_refs_by_tag() # CAMERA/FRAME 절 스템 팩도 콘티 산출 실질 입력 (sidecar 지문) from app.modules.pipeline.still_recipe import ( CAMERA_FRAME_PROMPT_VERSION as _cam_sel, resolve_prompt_version as _recipe_pack, ) _extra_fp["conti_camera_frame_pack"] = _recipe_pack(_cam_sel) result = run_shot_conti_light( shots=shots, classify_shots=classify_shots, continuity=continuity, location_by_scene=location_by_scene, plate_map=plate_map, out_dir=out_dir, gen_fn=self._make_gen_fn(), prompt_version=_selector, extra_fingerprint=_extra_fp, force=(mode == "force"), classify_scenes=_classify_scenes, staging_by_key=_staging_by_key, char_refs_by_tag=_char_refs, no_plate_conti=self._no_plate_conti_on(), # 감사 1-B — 그 샷에 배정된 인물만 CARRIED 에 실린다. # ★`_char_refs` 와 **묶지 않는다**: 그쪽은 팩 selector 가 # 바뀐 경우에만 채워지고 참조가 없는 인물을 빠뜨린다. visible_char_sids_by_tag=self._visible_char_sids_by_tag(), ) # 콘티/맵 플레이트/lane 스케치 intermediate ImageAsset 영속 + # asset_id 병기 — 등록은 lineage 필수 계약(2차 리뷰 M6): 실패 시 # 스텝 failed 로 propagate (조용한 completed 금지). self._register_intermediate_assets( result["contis"], map_plates, lane_contis) # ── 복잡 구조물 A/B 정책 finalize (Codex R2) — 일반 콘티 산출과 # 대조해 ab_select_ready/bypass/failed 확정. 반환=lane 집계 가산분 # (unique shot 계약 — 일반 콘티에 잡힌 샷은 이중 가산 금지). structure_delta = {"applicable": 0, "completed": 0, "failed": 0} if lane_contis: from app.modules.pipeline.shot_conti_light import ( finalize_structure_ab_entries, ) # B-3: classify_shots 는 스텝 진입 시점에 share plan prev # override 적용 완료(위) — finalize 도 동일 effective prev. structure_delta = finalize_structure_ab_entries( lane_contis=lane_contis, contis=result["contis"], classify_shots=classify_shots, ) # lane 집계 = 레인1(map_marker) 스케치 샷 + 구조물 보정 delta. # 구조물 ready/bypass 샷은 일반 콘티 집계가 담당(unique shot — # 배치 리뷰 BLOCKING-1: 콘티 성공 후 정책 실패는 completed -1 / # failed +1 보정으로 샷당 최종 status 1회 집계). _map_entries = [ e for e in lane_contis.values() if isinstance(e, dict) and e.get("lane") == "map_marker" ] # 운영자 선언 무콘티 예외(2026-07-30 장치)의 스텝 계수 연장 # (2026-08-11): 마커 검사 소진 등으로 끝내 실패한 lane 샷을 # 운영자가 선언하면 실패로 세지 않는다 — 소비측(still_recipe)의 # 같은 예외가 그 샷을 기존 무콘티 경로로 보낸다. 선언이 없으면 # 빈 집합 no-op 으로 기존 계수와 byte-identical. from app.services.still_recipe_service import ( parse_no_conti_exempt_tags, ) _exempt_tags = parse_no_conti_exempt_tags( getattr(settings, "bgfirst_no_conti_exempt_tags", "")) for _t, _e in lane_contis.items(): if ( _t in _exempt_tags and isinstance(_e, dict) and _e.get("lane") == "map_marker" and _e.get("status") == "failed" ): _e["operator_exempt"] = True logger.warning( "shot_conti_light %s: lane 실패 — 운영자 무콘티 예외로 " "계수 제외 (error=%s)", _t, (_e.get("error") or "")[:120], ) lane_applicable = len(_map_entries) + structure_delta["applicable"] lane_completed = ( sum( 1 for e in _map_entries if e.get("status") == "ok" or e.get("operator_exempt") ) + structure_delta["completed"] ) lane_failed = ( sum( 1 for e in _map_entries if e.get("status") == "failed" and not e.get("operator_exempt") ) + structure_delta["failed"] ) data: Dict[str, Any] = { "contis": result["contis"], "map_conti": map_plates, } # BGFIRST2: 스틸(still_recipe_service)이 구(v1) 콘티와의 조용한 # 혼용을 fail-closed 로 막을 수 있게 팩 provenance 기록 — v2 일 # 때만 노출(OFF CP shape byte-identical). if _selector != PROMPT_VERSION: data["conti_pack"] = _conti_pack(_selector) # Codex Stage D HIGH-4: OFF 시 CP shape byte-identical — lane 키는 # lane pipe ON 일 때만 노출 if lane_pipe_on: data["lane_conti"] = lane_contis # R1: 플레이트 권위 기록 — 스틸(still_recipe)이 재판정 없이 소비. # ON 일 때만 노출 (OFF byte-identical 관례) if plate_select_on: data["plate_authority"] = plate_authority return { "applicable_count": result["applicable_count"] + lane_applicable, "completed_count": result["completed_count"] + lane_completed, "failed_count": result["failed_count"] + lane_failed, "schema_version": SCHEMA_VERSION, "config_hash": self._config_hash(), "data": data, }