"""Asset readiness preflight — Group 1 #2 (visual_pipeline_contracts_plan.md). scene_image_pipeline 진입 전 expected vs registered vs disk_found 3-way 검증. fail-fast 정책 — 참조 이미지가 없는데 silent text-only 로 fallback 하지 않는다. 핵심 원칙 (memory: feedback_no_silent_fallback): - 참조 없으면 만들지 않는다. - continuity 모르면 조용히 비우지 않는다. - 배경 소유 객체를 scene prompt가 다시 그리지 않는다. 본 모듈은 첫 원칙을 강제한다. ENV `ALLOW_TEXT_ONLY_WITHOUT_REFS=true` 또는 ``settings.allow_text_only_without_refs=True`` 시에만 missing 을 무시 (디버깅용 opt-in). 기본값은 block. 본 모듈은 시나리오 의존성 0 — entity short_id / asset_type 같은 generic 표기만 검사한다. 어떤 작품 고유명사도 hardcode 안 한다. """ from __future__ import annotations import logging import os from dataclasses import dataclass, field from pathlib import Path from typing import Any, Dict, List, Optional, Tuple from sqlalchemy.orm import Session as OrmSession from app.core.subject_state import is_immobilized_state logger = logging.getLogger(__name__) _TRUTHY_VALUES = ("1", "true", "yes", "on") _FALSY_VALUES = ("0", "false", "no", "off", "") def _is_text_only_allowed() -> bool: """디버깅 opt-in flag — ENV 우선, settings fallback, default False (block). ENV ``ALLOW_TEXT_ONLY_WITHOUT_REFS`` 인식 못 하는 값 (typo) 은 silent True 가 아니라 settings default 로 fallback (fail-fast 원칙 — typo 가 safety 우회 X). Codex review IMPORTANT (Group 1 #2): settings field 가 ``bool`` 타입이면 Pydantic 이 import 단계에서 ENV ``garbage`` 를 ValidationError 로 raise 해 앱 부팅이 실패. settings field 를 ``str`` 로 두고 본 helper 가 직접 parse. """ raw = os.environ.get("ALLOW_TEXT_ONLY_WITHOUT_REFS") if raw is not None: v = raw.strip().lower() if v in _TRUTHY_VALUES: return True if v in _FALSY_VALUES: return False logger.warning( "ALLOW_TEXT_ONLY_WITHOUT_REFS=%r unrecognized — falling back to " "settings default. Use one of %s.", raw, _TRUTHY_VALUES + _FALSY_VALUES, ) try: from app.core.config import settings raw_setting = getattr(settings, "allow_text_only_without_refs", "") v = (raw_setting or "").strip().lower() if v in _TRUTHY_VALUES: return True # _FALSY_VALUES + 그 외 모두 default block (fail-fast). return False except Exception: return False @dataclass class AssetMissEntry: """단일 expected asset 의 missing 사유. 상태 분류: - "no_db_row": ImageAsset row 자체가 없음 (preceding step 누락 or skip) - "no_file_path": ImageAsset row 있으나 file_path 비어 있음 (저장 실패) - "disk_missing": file_path 있으나 실제 disk 에 파일 없음 (이동/삭제됨) """ entity_id: str short_id: str name: str asset_type: str # "reference" / "composite" / "state_variant" / etc. reason: str file_path: str = "" @dataclass class AssetReadinessCard: """scene_image_pipeline preflight 결과. expected/registered/disk_found 3-way: - expected: 본 episode 가 필요로 하는 모든 ref asset (entity scope 기반) - registered: 그 중 ImageAsset row 가 존재하는 것 - disk_found: 그 중 file_path 가 실제 disk 에 존재하는 것 invariant: ``len(disk_found_ids) <= len(registered_ids) <= len(expected_ids)``. missing 은 위 invariant 의 빈자리를 reason 별로 분류. """ project_id: str episode_id: str expected_ids: List[str] = field(default_factory=list) registered_ids: List[str] = field(default_factory=list) disk_found_ids: List[str] = field(default_factory=list) missing: List[AssetMissEntry] = field(default_factory=list) @property def is_ready(self) -> bool: """missing 0 이고 expected ≥ 1.""" return not self.missing and bool(self.expected_ids) @property def missing_count(self) -> int: return len(self.missing) def summary(self) -> str: """logger / error message 용 한 줄 요약.""" return ( f"AssetReadiness episode={self.episode_id[:8]}: " f"expected={len(self.expected_ids)} " f"registered={len(self.registered_ids)} " f"disk={len(self.disk_found_ids)} " f"missing={self.missing_count}" ) def missing_summary(self, max_items: int = 5) -> str: """missing 들의 reason-별 sample 요약.""" if not self.missing: return "(no missing)" head = self.missing[:max_items] more = self.missing_count - len(head) items = ", ".join( f"{m.short_id or m.entity_id[:8]}({m.asset_type}/{m.reason})" for m in head ) return items + (f" (+{more} more)" if more > 0 else "") def compute_episode_asset_readiness( db: OrmSession, project_id: str, episode_id: str, ) -> AssetReadinessCard: """scene_image_pipeline 의 expected ref asset 들이 모두 준비됐는지 검사. Expected scope: - 본 episode 의 모든 entity (location/outlook 제외, low_freq skip 제외) 의 primary reference asset. - 본 episode 의 모든 (character, outlook) composite asset (O00 + cid/oid skip 제외). - 본 episode 의 character_state_variant ref (shot_staging cp 의 dead/injured/ unconscious 인물 — Codex review BLOCKING 보강). 각 expected 에 대해: - DB 에 ImageAsset row 가 있는가? (asset_type="reference", is_primary=1) - file_path 가 비어 있지 않은가? - 실제 disk 에 파일이 존재하는가? 어느 단계가 빠지면 ``AssetMissEntry`` 로 missing 에 add. silent skip 0. """ from pathlib import Path as _Path from app.core.file_paths import resolve_image_path from app.core.low_freq_skip import load_low_freq_skip_ids from app.models.project import ( CharacterOutlook, EntityCanon, EntityEpisodeLink, ImageAsset, ) card = AssetReadinessCard(project_id=project_id, episode_id=episode_id) # 본 episode 의 entity scope (location/outlook 제외, low_freq skip 제외). skipped_ids = load_low_freq_skip_ids(project_id, episode_id) entity_rows = ( db.query(EntityCanon) .join(EntityEpisodeLink, EntityEpisodeLink.canon_id == EntityCanon.id) .filter( EntityEpisodeLink.project_id == project_id, EntityEpisodeLink.episode_id == episode_id, EntityCanon.entity_type.notin_(["location", "outlook"]), ) .all() ) expected_entities = [ {"id": e.id, "short_id": e.short_id or "", "name": e.name or "", "type": e.entity_type} for e in entity_rows if e.id not in skipped_ids ] # 1) base reference per entity (Phase 1 — character + prop). for ent in expected_entities: eid = ent["id"] card.expected_ids.append(f"reference:{eid}") primary = ( db.query(ImageAsset) .filter( ImageAsset.project_id == project_id, ImageAsset.entity_id == eid, ImageAsset.asset_type == "reference", ImageAsset.is_primary == 1, ) .order_by(ImageAsset.created_at.desc()) .first() ) if not primary: card.missing.append(AssetMissEntry( entity_id=eid, short_id=ent["short_id"], name=ent["name"], asset_type="reference", reason="no_db_row", )) continue card.registered_ids.append(f"reference:{eid}") if not primary.file_path: card.missing.append(AssetMissEntry( entity_id=eid, short_id=ent["short_id"], name=ent["name"], asset_type="reference", reason="no_file_path", )) continue path = resolve_image_path(primary.file_path) if not (path and path.exists()): card.missing.append(AssetMissEntry( entity_id=eid, short_id=ent["short_id"], name=ent["name"], asset_type="reference", reason="disk_missing", file_path=primary.file_path, )) continue card.disk_found_ids.append(f"reference:{eid}") # 2) composite (character × outlook) — O00 + skip 제외. char_ids = {ent["id"] for ent in expected_entities if ent["type"] == "character"} o00_ids = { e.id for e in db.query(EntityCanon).filter( EntityCanon.project_id == project_id, EntityCanon.short_id == "O00", EntityCanon.entity_type == "outlook", ).all() } if char_ids: all_combos = ( db.query(CharacterOutlook) .filter( CharacterOutlook.project_id == project_id, CharacterOutlook.character_id.in_(char_ids), ) .all() ) # composite 는 prompt_used 패턴 으로 식별 (alembic 010 이전 schema 호환). composite_assets = ( db.query(ImageAsset) .filter( ImageAsset.project_id == project_id, ImageAsset.asset_type == "reference", ImageAsset.prompt_used.like("%composite:%"), ) .all() ) import re as _re existing_pairs: Dict[Tuple[str, str], ImageAsset] = {} for ca in composite_assets: m = _re.search(r"composite:([a-f0-9-]+):([a-f0-9-]+)", ca.prompt_used or "") if m: existing_pairs[(m.group(1), m.group(2))] = ca # entity_lookup for descriptive missing entries. ent_by_id = {ent["id"]: ent for ent in expected_entities} for combo in all_combos: cid, oid = combo.character_id, combo.outlook_id # Claude review IMPORTANT (Group 1 #2): outlook 도 low_freq_skip 대상이면 # composite_image_gen 이 skip — false-positive block 방지로 expected scope 외. if oid in o00_ids or cid in skipped_ids or oid in skipped_ids: continue key = f"composite:{cid}:{oid}" card.expected_ids.append(key) asset = existing_pairs.get((cid, oid)) if not asset: card.missing.append(AssetMissEntry( entity_id=cid, short_id=ent_by_id.get(cid, {}).get("short_id", ""), name=ent_by_id.get(cid, {}).get("name", ""), asset_type="composite", reason="no_db_row", )) continue card.registered_ids.append(key) if not asset.file_path: card.missing.append(AssetMissEntry( entity_id=cid, short_id=ent_by_id.get(cid, {}).get("short_id", ""), name=ent_by_id.get(cid, {}).get("name", ""), asset_type="composite", reason="no_file_path", )) continue path = resolve_image_path(asset.file_path) if not (path and path.exists()): card.missing.append(AssetMissEntry( entity_id=cid, short_id=ent_by_id.get(cid, {}).get("short_id", ""), name=ent_by_id.get(cid, {}).get("name", ""), asset_type="composite", reason="disk_missing", file_path=asset.file_path, )) continue card.disk_found_ids.append(key) # 3) character_state_variant — shot_staging 에서 immobilized subject_state # (dead/severely_injured/unconscious) 로 등록된 인물의 state ref. # Codex review BLOCKING (Group 1 #2): scene_reference_service.py:230,721 가 # state_variant ref 를 사용하는데 disk/DB 누락 시 silent 으로 composite/base 로 # fallback 하던 path 를 차단. immobilized enum 판정은 SOT helper # ``app.core.subject_state.is_immobilized_state`` 단일 path (Area #2 W5). expected_state_pairs = _collect_expected_state_variants( project_id, episode_id, expected_entities, ) if expected_state_pairs: # producer step 의 prompt_used 패턴: ``state_variant:{char_uuid}:{state_type}``. sv_assets = ( db.query(ImageAsset) .filter( ImageAsset.project_id == project_id, ImageAsset.asset_type == "character_state_variant", ) .all() ) # entity_type 컬럼 표기가 producer 와 다를 수 있어 prompt_used 패턴으로도 검사. if not sv_assets: sv_assets = ( db.query(ImageAsset) .filter( ImageAsset.project_id == project_id, ImageAsset.prompt_used.like("%state_variant:%"), ) .all() ) sv_by_pair: Dict[Tuple[str, str], ImageAsset] = {} for sa in sv_assets: m = _re.search(r"state_variant:([a-f0-9-]+):(\w+)", sa.prompt_used or "") if m: sv_by_pair[(m.group(1), m.group(2))] = sa ent_by_id = {ent["id"]: ent for ent in expected_entities} for cid, state_type in expected_state_pairs: # Area #2 W5: literal tuple 폐기 — is_immobilized_state SOT 단일 path. # _collect_expected_state_variants 가 이미 immobilized 만 emit 하므로 본 가드는 # corrupt CP / 미래 SUBJECT_STATES 추가 시 defense in depth. if not is_immobilized_state(state_type): continue key = f"state_variant:{cid}:{state_type}" card.expected_ids.append(key) asset = sv_by_pair.get((cid, state_type)) if not asset: card.missing.append(AssetMissEntry( entity_id=cid, short_id=ent_by_id.get(cid, {}).get("short_id", ""), name=ent_by_id.get(cid, {}).get("name", ""), asset_type=f"state_variant/{state_type}", reason="no_db_row", )) continue card.registered_ids.append(key) if not asset.file_path: card.missing.append(AssetMissEntry( entity_id=cid, short_id=ent_by_id.get(cid, {}).get("short_id", ""), name=ent_by_id.get(cid, {}).get("name", ""), asset_type=f"state_variant/{state_type}", reason="no_file_path", )) continue path = resolve_image_path(asset.file_path) if not (path and path.exists()): card.missing.append(AssetMissEntry( entity_id=cid, short_id=ent_by_id.get(cid, {}).get("short_id", ""), name=ent_by_id.get(cid, {}).get("name", ""), asset_type=f"state_variant/{state_type}", reason="disk_missing", file_path=asset.file_path, )) continue card.disk_found_ids.append(key) return card def _collect_expected_state_variants( project_id: str, episode_id: str, expected_entities: List[Dict[str, Any]], ) -> List[Tuple[str, str]]: """shot_staging cp 에서 dead/injured/unconscious 로 매핑된 (cid, state_type) 추출. CharacterStateVariantStep 와 동일 source + 동일 name normalization (Codex review Critical: producer 가 ``build_name_index/lookup_name`` 으로 normalized name 매칭. preflight 가 exact match 하면 LLM 의 name 변형 (괄호/공백/캐릭터 변형 표기) 이 drift 되어 false-positive block). shot_staging cp 부재 시 빈 리스트 (해당 step 미실행 — silent skip 아닌 expected 0 으로 처리). cp shape 두 가지 모두 지원: ``{"data": {"shots": [...]}}`` (pipeline-v4 manifest) + ``{"shots": [...]}`` (legacy / direct). """ import json as _json from pathlib import Path as _Path from app.core.config import settings as _settings from app.core.name_matcher import build_name_index, lookup_name cp_path = ( _Path(_settings.projects_dir) / project_id / "checkpoints" / "episodes" / episode_id / "shot_staging" / "manifest.json" ) if not cp_path.exists(): return [] try: cp = _json.loads(cp_path.read_text(encoding="utf-8")) except Exception as exc: logger.warning("shot_staging cp parse failed (%s) — state_variant scope=0", exc) return [] if not isinstance(cp, dict): return [] # Producer 와 동일 normalization. expected_entities 는 character 만. char_entities = [ent for ent in expected_entities if ent.get("type") == "character"] name_to_id = build_name_index( char_entities, key_fn=lambda e: e.get("name", ""), value_fn=lambda e: e.get("id", ""), ) # cp shape — {"data": {...}} wrapper / {"shots": [...]} flat 모두 처리. # 중간 결과가 dict 아니면 (e.g. {"data": null}) 빈 결과 (Claude review IMPORTANT). inner = cp.get("data", cp) if not isinstance(inner, dict): return [] shots = inner.get("shots", []) if not isinstance(shots, list): return [] pairs: set = set() for shot in shots: if not isinstance(shot, dict): continue for ca in shot.get("character_angles", []) or []: # ``isinstance(ca, dict)`` 가드는 corrupt CP 방어 (JSON loader 가 list 안 # 비-dict 반환 시 KeyError 차단). v13 producer 는 정상 dict emit. if not isinstance(ca, dict): continue state = ca["subject_state"] # required by v13 schema, KeyError = schema violation (Gate 4 fail-fast) if is_immobilized_state(state): cname = ca.get("character", "") cid = lookup_name(name_to_id, cname) if cname else None if cid: pairs.add((cid, state)) return sorted(pairs) def assert_episode_asset_readiness( db: OrmSession, project_id: str, episode_id: str, ) -> AssetReadinessCard: """compute_episode_asset_readiness + fail-fast block 합쳐서 호출. missing 0 이면 card 반환. missing > 0 이면: - settings.allow_text_only_without_refs=True (or ENV) 시: logger.warning 후 card 반환 (옛 silent path 유사하지만 명시 opt-in 만) - 기본 (block): logger.error + AppError("image.assets_not_ready") raise. 호출자: scene_image_service.generate_images / generate_single_scene_image. """ from app.core.errors import AppError card = compute_episode_asset_readiness(db, project_id, episode_id) if card.is_ready: logger.info("AssetReadiness OK: %s", card.summary()) return card if _is_text_only_allowed(): logger.warning( "AssetReadiness opt-in bypass (ALLOW_TEXT_ONLY_WITHOUT_REFS=true): %s — missing: %s", card.summary(), card.missing_summary(), ) return card # fail-fast block. logger.error( "AssetReadiness BLOCK: %s — missing: %s", card.summary(), card.missing_summary(), ) raise AppError( code="image.assets_not_ready", message=( f"참조 이미지 자산이 누락됐습니다 ({card.missing_count}건). " f"누락: {card.missing_summary()}. ref_image_gen / composite_image_gen 단계를 " f"먼저 실행하세요. (디버깅용 우회: ALLOW_TEXT_ONLY_WITHOUT_REFS=true)" ), status_code=400, )