"""씬 이미지 생성 서비스 — ImageService에서 분리된 scene 전용. Phase 3b.3 (2026-04-18): ImageService의 씬 이미지 관련 12개 메서드를 이관. Phase 3b.5b (2026-04-18): ImageService wrapper 전부 제거. 호출자 (api/v1/images.py, core/steps/image_steps.py)는 본 서비스를 직접 인스턴스화. W5 F22 Phase A (2026-04-22): **Facade freeze**. 이 모듈의 공개 API 9개는 본 파일의 facade 책임이다. 파라미터 이름/순서는 `tests/services/test_scene_image_service.py`의 9개 signature 테스트로 고정. Phase B에서 내부 구현을 5개 모듈로 분리해도 아래 9개 메서드의 외부 계약은 바뀌지 않는다. (호출자: api/v1/images.py + core/steps/image_steps.py) ## 공개 API (freeze — 변경 시 signature 테스트 실패) | 메서드 | 역할 | 주요 호출자 | |--------|------|-------------| | `get_image(image_id)` | ImageAsset 조회 | api/v1/images.py | | `generate_images(episode_id, ip, mode)` | 전체 씬 이미지 생성 (background) | api/v1/images.py, image_steps.py | | `generate_variations_only(episode_id)` | variation만 추가 생성 | api/v1/images.py | | `generate_single_scene_image(still_id, custom_prompt, ip)` | 단일 still 이미지 재생성 | api/v1/images.py | | `generate_scene_with_variations(still_id, ip)` | 단일 still + variation 전체 | api/v1/images.py | | `recommend_variations(still_id, ip)` | LVM 기반 variation 추천 | api/v1/images.py | | `regenerate_variation(image_id, angle_json, color_prompt, ip)` | **현재는 fal.ai angle 편집만 구현**; `color_prompt`는 reserved — images.py color route는 angle_json 없으면 400 반환. Phase B에서 color 구현 예정. | api/v1/images.py | | `select_variant(still_id, variant, ip)` | variation 대표 이미지 선택 | api/v1/images.py | | `select_original(still_id, image_id, ip)` | 원본 이미지 복원 선택 | api/v1/images.py | ## Phase B 분해 대상 (내부 모듈, 공개 API 유지) 1. `scene_variation_service` — variation 생성/선택 로직 2. `scene_validation_service` — T2I 검증 + retry 전략 3. `scene_persistence_service` — DB UPSERT + still 동기화 4. `scene_provenance_service` — PNG metadata + 이력 추적 5. `scene_image_service` — facade (얇은 래퍼, ≤300 LOC 목표) """ from __future__ import annotations import json import logging import uuid from concurrent.futures import ThreadPoolExecutor, as_completed from datetime import datetime, timezone from pathlib import Path from typing import Any, Dict, List, Optional from sqlalchemy.orm import Session as OrmSession from app.core.config import settings from app.core.file_paths import to_relative_image_path from app.core.errors import AppError, StaleUpstreamError from app.core.image_call_budget import bind_current_budget from app.core.ref_contract_validator import RefContractError from app.i18n.loader import t from app.logging.activity_logger import ActivityLogger from app.models.project import ( Episode, ImageAsset, SceneStill, ) from app.modules.llm.gemini_image_client import ModerationError # noqa: F401 — caller 경로용 from app.modules.llm.gemini_key_pool import key_count as gemini_key_count, get_next_key from app.modules.progress_tracker import ProgressTracker from app.modules.provenance import ProvenanceRecorder from app.modules.entity_dependency import ( build_scene_dependency_graph, topological_sort_scenes, ) from app.services.image_service_helpers import ( build_lineage_fields, fill_missing_t2i_prompts, get_latest_world_guide, load_project_llm_config, populate_t2i_prompts, ) from app.services.scene_reference_service import SceneReferenceService from app.services.scene_checkpoint_loaders import ( load_background_chain_bg_map, load_outdoor_direct_context, load_shot_dependency_map, load_shot_staging_map, ) from app.modules.semantic_contract_router import build_semantic_contract from app.services.scene_generation_coordinator import ( SceneGenerationCoordinator, lookup_render_prompt_card, ) from app.services.scene_persistence_service import ScenePersistenceService from app.services.scene_provenance_service import SceneProvenanceService from app.services.scene_validation_service import SceneValidationService from app.services.scene_variation_service import SceneVariationService # W5 F22 Phase B.22.5: final best 선택은 scene_variation_service로 이관 from app.services.scene_variation_service import select_best_variation_idx as _select_best_variation_idx # W5 F22 Phase B.22.7: validation score override도 scene_variation_service로 이관 from app.services.scene_variation_service import apply_validation_override as _apply_validation_override __all__ = ["SceneImageService"] logger = logging.getLogger(__name__) def _now() -> str: return datetime.now(timezone.utc).isoformat() def _new_id() -> str: return str(uuid.uuid4()) # ────────────────────────────────────────────────────────────────────── # 2026-05-10 — Fix C — typed failure summary # 24 shot deterministic 실패에서 RefContractError detail 이 cp.failed 의 # generic "all variations failed" 로 swallow 되던 결함 fix. # coordinator._generate_variation_in_loop / _await_variation_results 가 # typed failure dict (`_failure_reason`/`_failure_detail`) 를 반환 → # generate_images 흐름이 본 helper 로 cp.failed 메시지 직렬화. # ────────────────────────────────────────────────────────────────────── _FAILURE_MSG_MAX = 800 def _summarize_failed_variations( var_results_list: List[Dict[str, Any]], scene_index: int, ) -> Optional[str]: """variation result list 의 typed failure 들을 cp.failed 메시지로 직렬화. - 성공 dict 가 1개라도 있으면 (mark_failed 안 함) None 반환. - 모두 typed failure → reason 들을 join 한 메시지 반환 (운영자 가시화). - 빈 list → "unknown" generic msg (이전 silent path 회피). - 길이 ``_FAILURE_MSG_MAX`` cap (cp.failed 손상 방지). """ successes = [ r for r in var_results_list if isinstance(r, dict) and not r.get("_failure_reason") ] if successes: return None failures = [ r for r in var_results_list if isinstance(r, dict) and r.get("_failure_reason") ] head = f"all variations failed (scene_index={scene_index})" if not failures: return f"{head}; unknown — no failure reason captured" parts: List[str] = [] for f in failures[:5]: reason = f.get("_failure_reason", "UNKNOWN") detail = (f.get("_failure_detail") or "")[:200] parts.append(f"{reason}: {detail}" if detail else reason) msg = head + "; " + " | ".join(parts) if len(msg) > _FAILURE_MSG_MAX: msg = msg[: _FAILURE_MSG_MAX - 3] + "..." return msg class SceneImageService: """씬 이미지 생성 전담 서비스 (Phase 3b.3 · W5 F22 facade). 본 클래스의 9개 public 메서드는 W5 F22 Phase A에서 facade로 freeze됐다. Phase B에서 내부 구현을 5개 모듈로 분리할 때도 이 9개의 외부 계약 (파라미터 이름/순서/키워드 호출 가능성)은 변경하지 않는다. 공개 API 및 Phase B 분해 계획은 모듈 docstring 참조. """ def __init__( self, db: OrmSession, project_id: str, actor_id: str, ) -> None: self._db = db self._logger = ActivityLogger(db) self._project_id = project_id self._actor_id = actor_id # W5 F22 Phase B.1: variant selection/primary 로직은 scene_variation_service # 로 이관됨. facade 공개 메서드는 delegation만 수행. self._variation_svc = SceneVariationService(db, project_id, actor_id) # W5 F22 Phase B.6: T2I 검증 로직은 scene_validation_service로 이관됨. self._validation_svc = SceneValidationService(db, project_id) # W5 F22 Phase B.8: reference/entity 해결 로직은 scene_reference_service로 이관됨. self._reference_svc = SceneReferenceService(db, project_id) # W5 F22 Phase B.9.1: ImageAsset bulk 저장 + primary 지정은 # scene_persistence_service로 이관됨. self._persistence_svc = ScenePersistenceService(db, project_id) # W5 F22 Phase B.10: PNG 메타데이터 임베딩은 # scene_provenance_service로 이관됨. self._provenance_svc = SceneProvenanceService(db, project_id) # W5 F22 Phase B.25: scene 생성 bound method들을 SceneGenerationCoordinator로 이관 self._coord = SceneGenerationCoordinator( db=db, project_id=project_id, actor_id=actor_id, activity_logger=self._logger, persistence_svc=self._persistence_svc, reference_svc=self._reference_svc, validation_svc=self._validation_svc, variation_svc=self._variation_svc, provenance_svc=self._provenance_svc, ) # ------------------------------------------------------------------ # Internal helpers (HEAD byte-for-byte — Phase 3b.5에서 module 승격 검토) # ------------------------------------------------------------------ def _get_episode(self, episode_id: str) -> Episode: ep = ( self._db.query(Episode) .filter(Episode.id == episode_id, Episode.project_id == self._project_id) .first() ) if not ep: raise AppError( code="episode.not_found", message=t("episode.not_found"), status_code=404, ) return ep def _get_project_dir(self) -> Path: return Path(settings.projects_dir) / self._project_id # W5 F22 Phase B.25: _get_style_context → SceneGenerationCoordinator로 이관. # facade 호출자는 self._coord._get_style_context(...) 사용. # ------------------------------------------------------------------ # Phase 3b.3 — ImageService에서 이관된 scene 경로 필수 helper # ------------------------------------------------------------------ def get_image(self, image_id: str) -> ImageAsset: """image_id로 ImageAsset ORM 조회 (없으면 image.not_found).""" img = self._db.query(ImageAsset).filter( ImageAsset.id == image_id, ImageAsset.project_id == self._project_id, ).first() if not img: raise AppError( code="image.not_found", message=t("image.not_found"), status_code=404, ) return img # W5 F22 Phase B.25: _init_scene_gen_clients / _generate_scene_in_loop / # _run_fal_angle_pipeline / _finalize_and_track_primary / # _finalize_scene_with_variations / _maybe_generate_variation_slot / # _build_single_scene_prompt_and_refs / _finalize_single_scene / # _generate_variation_in_loop 9개 bound method는 SceneGenerationCoordinator로 이관. # facade는 self._coord 참조로 호출 (generate_images / generate_single_scene_image / generate_scene_with_variations 내부). def generate_images(self, episode_id: str, ip: Optional[str] = None, mode: str = "resume") -> None: """Full image generation pipeline for an episode. mode: "resume" — 이미 씬 이미지 있는 still_id 스킵 (기본). 과거 'full' 모드 (전체 삭제 후 재생성) 는 feedback_never_delete_images 규칙 위반으로 제거됨. 강제 재생성은 SceneImagePipelineStep mode='force' 를 사용 (UPDATE is_primary=0 + checkpoint 재시작 — DB row/PNG 보존). """ if mode != "resume": raise ValueError( f"SceneImageService.generate_images: mode={mode!r} unsupported. " "Only 'resume' is allowed (feedback_never_delete_images). " "Use SceneImagePipelineStep mode='force' for non-destructive regeneration." ) episode = self._get_episode(episode_id) if not episode.fulltext: raise AppError( code="analysis.no_text", message=t("analysis.no_text"), status_code=400, ) if not settings.gemini_api_key and gemini_key_count() == 0: raise AppError( code="image.gemini_key_missing", message=t("image.gemini_key_missing"), status_code=400, ) # W5 F22 Phase B.22.11: pipeline_gate + scene_count 검증을 persistence_svc로 이관 self._persistence_svc.validate_episode_ready(episode_id) # Group 1 #2 (visual_pipeline_contracts_plan): asset readiness preflight — # 참조 이미지 (entity reference + composite) 가 모두 준비됐는지 3-way 검증. # 누락 시 기본은 fail-fast block. silent text-only fallback 차단. # ENV ALLOW_TEXT_ONLY_WITHOUT_REFS=true 시 디버깅용 우회 (warning log). from app.core.asset_readiness import assert_episode_asset_readiness assert_episode_asset_readiness(self._db, self._project_id, episode_id) language = episode.language or "ko" project_dir = self._get_project_dir() images_dir = project_dir / "images" / episode_id scene_dir = images_dir / "scene" # 프로젝트 LLM 설정 _plc = load_project_llm_config(self._db, self._project_id) # 씬 이미지 체크포인트 from app.modules.image_checkpoint import ImageCheckpointManager cp_dir = project_dir / "checkpoints" / "images" / episode_id scene_cp = ImageCheckpointManager(cp_dir, "scene") # W5 F22 Phase B.22.10: entity/still ORM→dict 로드를 persistence_svc로 이관 entities = self._persistence_svc.load_episode_entity_dicts(episode_id) stills, stills_orm = self._persistence_svc.load_episode_still_dicts(episode_id) # 'full' 모드 destructive path 제거됨 (feedback_never_delete_images). # mode='resume' 만 지원 — 위에서 mode 검증으로 진입 차단. # resume 모드: 이미 씬 이미지 있는 still_id 파악 (W5 F22 Phase B.15) already_done_stills: set = set() if mode == "resume": already_done_stills = self._persistence_svc.scan_completed_scene_stills( episode_id, scene_cp, ) # ── 표적 씬 슬라이스 (2026-07-23, Codex 설계 합의) ─────────── # 전체 stills 는 컨텍스트 SOT 로 그대로 유지(월드 가이드·프롬프트 # population·ref map·groupbg 지문 불변, BLOCKING-1) — 실행 루프만 # 표적 씬+의존 클로저 allowlist 로 제한(recipe=서비스 내부 prev # 체인 클로저 / legacy=최종 합산 그래프 클로저 아래). 빈 설정= # 전체 생성 byte-identical. 표적 씬에 still 0개=fail-fast(HIGH-3). from app.modules.pipeline.scene_image_scope import ( parse_target_scenes, requested_still_ids, ) target_scenes = parse_target_scenes( getattr(settings, "scene_image_target_scenes", "")) target_requested: Optional[List[str]] = None if target_scenes: target_requested = requested_still_ids(stills, target_scenes) # Initialize provenance recorder and progress tracker provenance = ProvenanceRecorder(self._db, self._project_id) stills_to_gen = len(stills) - len(already_done_stills) if target_requested is not None: # 표적 모드 잠정 카운트(requested 기준) — legacy 경로는 아래 # 클로저 확정 후 effective 기준으로 재계산 (Codex HIGH-4) stills_to_gen = len(set(target_requested) - already_done_stills) total_items = stills_to_gen + 1 # +1 for world guide progress = ProgressTracker(self._db, episode_id, "image_generation", self._project_id) # 1. Generate world guide (W5 F22 Phase B.19: scene_persistence_service로 이관) progress.update("세계관 가이드 생성 중", 0, total_items) world_guide = self._persistence_svc.resolve_world_guide( episode_id=episode_id, episode=episode, entities=entities, stills=stills, mode=mode, provenance=provenance, language=language, ) # Build entity lookup by ID entity_lookup = {e["id"]: e for e in entities} # W5 F22 Phase B.23.2: 4-client 초기화 공통 helper (gemini/openai/sanitizer/tracker) gemini_client, openai_client, sanitizer, tracker = self._coord._init_scene_gen_clients() max_concurrent = settings.max_concurrent_image_gen progress.update("기존 참조 이미지 로드 중", 1, total_items) # W5 F22 Phase B.12: scene_reference_service로 이관 ref_image_map = self._reference_svc.load_entity_reference_images(entities) # 참조 이미지가 하나도 없으면 씬 생성 차단 if not ref_image_map: raise AppError( code="image.no_reference_images", message=t("image.no_reference_images") if t("image.no_reference_images") != "image.no_reference_images" else "참조 이미지가 없습니다. 엔티티 참조 이미지를 먼저 생성하세요.", status_code=400, ) # Initialize validator for scene image validation validator = self._validation_svc.create_validator() # T2I 프롬프트 로드 + 런타임 fallback (W5 F22 Phase B.20: image_service_helpers로 이관) populate_t2i_prompts(self._db, stills, stills_orm, entities, openai_client) # Fallback: t2i_variations_json에서 T2I 프롬프트 추출 # (W5 F22 Phase B.18: image_service_helpers로 이관) fill_missing_t2i_prompts(stills) # 3. Generate scene images — dependency-ordered, concurrent within batches # P0 (2026-07-01): scene_ref_asset_id_map = 동일 key → 실제 ImageAsset UUID # (실제 첨부 lineage SOT). _generate_scene_in_loop → resolve_refs 로 thread. scene_ref_asset_id_map: Dict[str, str] = {} scene_ref_image_map = self._reference_svc.build_scene_ref_image_map( ref_image_map, entity_lookup, out_asset_id_map=scene_ref_asset_id_map, ) # ── 체크포인트 3종 로드 (W5 F22 Phase B.11: scene_checkpoint_loaders로 이관) ── staging_map = load_shot_staging_map(settings.projects_dir, self._project_id, episode_id) # ── s40/s41 레시피 v1 분기 (2026-07-13, opt-in) ────────────── # still_recipe_mode="v1" 이면 배치 경로 대신 스토리 순서 순차 생성 # (prev 앵커 체인 + N롤 Gemini 선정 + 결함 i2i 수정). 프롬프트는 # 레시피 자체 조립(t2i_variations 미사용) — s39 텍스트 안전망 전체. # scene_ref_image_map(composite/outlook/state)+asset UUID map+staging # 을 통과시켜 기존 인물 조합·lineage 계약 유지 (Codex BLOCKING-2). # OFF(default) = 아래 배치 경로 byte-identical. if getattr(settings, "still_recipe_mode", "off") == "v1": from app.services.still_recipe_service import ( run_still_recipe_generation, ) generated = run_still_recipe_generation( db=self._db, project_id=self._project_id, episode_id=episode_id, stills=stills, stills_orm=stills_orm, entity_lookup=entity_lookup, ref_image_map=ref_image_map, reference_svc=self._reference_svc, scene_ref_image_map=scene_ref_image_map, scene_ref_asset_id_map=scene_ref_asset_id_map, staging_map=staging_map, scene_cp=scene_cp, persistence_svc=self._persistence_svc, progress=progress, project_config=_plc, scene_dir=scene_dir, already_done_stills=already_done_stills, # 표적 씬 슬라이스: 클로저(effective prev 체인)는 recipe # 서비스가 classify/share_plan 실사용 맵으로 계산 — 여기 # 재구현 금지 (드리프트 차단) target_scenes=target_scenes or None, ) progress.complete() if target_requested is None: logger.info( "still_recipe v1: %d/%d 샷 생성 완료 (episode=%s)", generated, len(stills), episode_id, ) else: # 표적 모드: 분모=전체 에피소드 수가 아니라 표적 scope — # 정확 집계는 audit(effective)이 SOT 라 개수만 로그 logger.info( "still_recipe v1(표적): %d 샷 생성 완료 (episode=%s, " "target_scenes=%s)", generated, episode_id, ",".join(map(str, target_scenes)), ) return dep_detail_map = load_shot_dependency_map(settings.projects_dir, self._project_id, episode_id) background_chain_bg_map = load_background_chain_bg_map( settings.projects_dir, self._project_id, episode_id, ) # W22 야외 직행 컨텍스트 (2026-07-10) — flag OFF/cp 부재 = {} (no-op) outdoor_direct_ctx = load_outdoor_direct_context( settings.projects_dir, self._project_id, episode_id, ) # W5 F22 Phase B.22.12: build_resume_state가 4-map을 새 dict로 반환하므로 직접 사용. # resume이 아니면 4개 모두 빈 dict로 초기화되어 반환됨 (기존 update 패턴과 등가). # B (2026-07-02): VE 에 location 이 없는 outdoor still 의 history 복원/기록용 # primary_location fallback 맵 — flag OFF → {} = 기존과 byte-identical. from app.core.steps.outdoor_site_layout_step import ( load_outdoor_prev_frame_context, ) from app.modules.pipeline.outdoor_site_layout_plan import ( primary_location_uuid_by_scene, ) _opf_ctx_main = load_outdoor_prev_frame_context(self._project_id, episode_id) _opf_uuid_by_scene_main = primary_location_uuid_by_scene( _opf_ctx_main.get("primary_location_by_scene") or {}, _opf_ctx_main.get("outdoor_loc_sids") or set(), entity_lookup, ) if _opf_ctx_main else {} _rs = self._persistence_svc.build_resume_state( stills, already_done_stills, entity_lookup, fallback_location_uuid_by_scene=_opf_uuid_by_scene_main or None, ) scene_paths_by_index: Dict[int, Path] = _rs["scene_paths_by_index"] scene_paths_by_index_by_id: Dict[str, Path] = _rs["scene_paths_by_index_by_id"] scene_results_by_index: Dict[int, Dict[str, Any]] = _rs["scene_results_by_index"] location_scene_history: Dict[str, tuple] = _rs["location_scene_history"] # Phase C: zoom crop intermediate 의 source lineage UUID 를 batch worker # (worker 내 DB 금지) 에 주입하기 위한 메인스레드 맵. scene_paths_by_index_by_id # 평행 — _finalize_and_track_primary 가 primary 영속 시 still_id→primary_asset_id # 를 갱신하고 worker 는 읽기 전용으로 source 의 UUID 를 조회한다. resume 으로 # 직전 run 에 완료된 source 는 이 맵에 없어 source_asset_id None(부분 lineage 허용). scene_primary_asset_id_by_still_id: Dict[str, str] = {} # Build scene dependency graph and topological sort into batches scene_deps = build_scene_dependency_graph(stills, entity_lookup) # W21B-W7 W-C2b (2026-06-12): zoom continuity dep edge — zoom still 은 # source_wide still 에 의존 (edge 방향 zoom→source = source 먼저, Codex # guard 2). batch 경계가 barrier (batch 결과는 다음 batch 전에 일괄 # 영속화 — primary 포함) 라 ThreadPool still 병렬에서도 race 없음. # flag OFF/anchor 부재 → 빈 dict = 그래프 불변 (byte-identical). from app.core.steps.zoom_continuity_anchor_step import ( load_zoom_continuity_context, ) _zoom_ctx = load_zoom_continuity_context(self._project_id, episode_id) # W21B-W8 (2026-06-12): outdoor site layout 의 위치 구절 override 를 # zoom override 와 **명시적 priority merge** (custom > zoom > site > # original — custom 은 별도 분기 선행 승리, original 은 override 부재). # 두 flag OFF → 두 맵 다 빈 dict = merge 도 빈 dict (byte-identical). from app.core.steps.outdoor_site_layout_step import ( load_outdoor_site_layout_context, ) from app.modules.pipeline.outdoor_site_layout_plan import ( merge_prompt_overrides, ) _site_ctx = load_outdoor_site_layout_context(self._project_id, episode_id) _merged_overrides = merge_prompt_overrides( _zoom_ctx.get("source_overrides") or {}, _site_ctx.get("prompt_overrides") or {}, ) if _merged_overrides: _zoom_ctx = {**_zoom_ctx, "source_overrides": _merged_overrides} if _zoom_ctx: _zc_key_to_idx = { (s_.get("scene_index"), s_.get("shot_index")): i_ for i_, s_ in enumerate(stills) } for _zkey, _zt in (_zoom_ctx.get("zoom_targets") or {}).items(): _zi = _zc_key_to_idx.get(_zkey) _src_i = _zc_key_to_idx.get(tuple(_zt.get("source") or ())) if _zi is not None and _src_i is not None and _zi != _src_i: scene_deps[_zi].add(_src_i) logger.info( "zoom_continuity(batch): dep edge 추가 — still[%d](zoom %s) → " "still[%d](source %s)", _zi, _zkey, _src_i, _zt.get("source"), ) # W21B-W8 composition continuity (option C, 2026-06-15): continuity_anchor # 샷은 같은 그룹 직전 admitted 샷(anchor_source)의 **현재-run 완성 프레임**을 # 명시 참조한다 → source 가 먼저 생성되도록 dep edge(target→source) 추가 # (zoom_continuity 패턴 동일). batch barrier(batch 결과는 다음 batch 전 일괄 # 영속 — primary 포함)가 source primary 영속을 보장해 target 의 resolver 가 # current-run bytes 를 읽는다. 두 flag OFF/guide 부재 시 빈 dict = 그래프 불변. from app.core.steps.outdoor_site_layout_step import ( load_composition_guide_context, ) _cg_ctx = load_composition_guide_context(self._project_id, episode_id) if _cg_ctx: _cg_key_to_idx = { (s_.get("scene_index"), s_.get("shot_index")): i_ for i_, s_ in enumerate(stills) } for _ckey, _centry in _cg_ctx.items(): if _centry.get("mode") != "continuity_anchor": continue _src = _centry.get("anchor_source") if not (isinstance(_src, (list, tuple)) and len(_src) == 2): continue _ti = _cg_key_to_idx.get(_ckey) _si = _cg_key_to_idx.get((_src[0], _src[1])) if _ti is not None and _si is not None and _ti != _si: scene_deps[_ti].add(_si) logger.info( "composition_continuity(batch): dep edge 추가 — still[%d]" "(anchor %s) → still[%d](source %s)", _ti, _ckey, _si, (_src[0], _src[1]), ) # A5 (2026-07-02): immobilized prev-frame chaining — 그룹 후속 멤버가 선행 # environment 멤버의 현재-run 완성 프레임을 참조하므로 anchor 가 먼저 # 생성되도록 dep edge(target→source) 추가 (composition_continuity 패턴 동일, # batch barrier 가 source primary 영속 보장). flag OFF / 그룹 부재 → 그래프 # 불변 (byte-identical). if bool(getattr(settings, "immobilized_prev_frame_chain_enabled", False)): from app.core.steps.visual_continuity_anchor_step import ( load_immobilized_subject_anchor_context, ) from app.modules.pipeline.visual_continuity_anchor_plan import ( build_immobilized_prev_frame_plan, ) _ipf_ctx = load_immobilized_subject_anchor_context( self._project_id, episode_id) _ipf_plan = build_immobilized_prev_frame_plan( (_ipf_ctx or {}).get("members_by_group") or {}) if _ipf_plan: _ipf_key_to_idx = { (s_.get("scene_index"), s_.get("shot_index")): i_ for i_, s_ in enumerate(stills) } for _tkey, _tentry in _ipf_plan.items(): _ti2 = _ipf_key_to_idx.get(_tkey) _si2 = _ipf_key_to_idx.get(tuple(_tentry["anchor_source"])) if _ti2 is not None and _si2 is not None and _ti2 != _si2: scene_deps[_ti2].add(_si2) logger.info( "immobilized_prev_frame(batch): dep edge 추가 — still[%d]" "(target %s) → still[%d](env anchor %s)", _ti2, _tkey, _si2, _tentry["anchor_source"], ) # ── 표적 씬 슬라이스 클로저 (legacy 경로) — base+zoom+composition+ # immobilized edge 가 **전부 합쳐진 최종 그래프**의 재귀 고정점 # (Codex BLOCKING-2). audit=스텝 카운트·verify 와 공유 단일 SOT. target_effective: Optional[set] = None if target_requested is not None: from app.modules.pipeline.scene_image_scope import ( build_scope_audit, closure_over_index_graph, save_scope_audit, scope_audit_path, ) _eff_ids, _dep_added = closure_over_index_graph( stills, target_requested, scene_deps) save_scope_audit( scope_audit_path( settings.projects_dir, self._project_id, episode_id), build_scope_audit( target_scenes=target_scenes, requested_ids=target_requested, dependency_added_ids=_dep_added, path_kind="batch", ), ) target_effective = set(_eff_ids) # Codex HIGH-4: 카운트=effective ∩ 미완료 (기존 full run 뒤 # 표적 실행에서 음수/오집계 차단) stills_to_gen = len(target_effective - already_done_stills) total_items = stills_to_gen + 1 logger.info( "scene_image 표적 scope(batch): scenes=%s requested=%d " "dep_added=%d effective=%d to_gen=%d", list(target_scenes), len(target_requested), len(_dep_added), len(target_effective), stills_to_gen, ) scene_batches = topological_sort_scenes(len(stills), scene_deps) # Wave5 (2026-06-30): 실내 shared-model pose 가이드 — episode당 1회 main-thread # precompute(candidate→indoor 게이트→judge→admit→멤버별 guide+QC). flag OFF → # {} no-op(byte-identical). worker 는 결과 ctx 를 attach_indoor_pose_guide_ref 로 # lookup 만(DB/VLM/생성 0). 생성 가이드/underlay 는 generation_context scope 안에서 # capture(indoor_pose_guide / indoor_pose_underlay). ★bg plate underlay 필수 — # bg 부재 그룹은 no-guide degrade(흰배경 경로 없음). from app.core.steps.indoor_shared_pose_guide_context import ( load_indoor_shared_pose_guides, ) # ★openai_client 미전달(None 유지) — _init_scene_gen_clients 의 OpenAIClient 는 # 텍스트/LVM 래퍼라 gpt-image edit(.images) 계약 불충족('object has no attribute # images'로 guide 생성 전멸, E2E 0e07d5e3). None 이면 guide 서비스가 raw OpenAI # SDK 클라이언트를 lazy resolve 한다(registered_pose_guide 경로와 동일 계약). _indoor_pose_ctx = load_indoor_shared_pose_guides( project_id=self._project_id, episode_id=episode_id, stills=stills, staging_map=staging_map, background_chain_bg_map=background_chain_bg_map, zoom_ctx=_zoom_ctx, db=self._db, already_done_still_ids=already_done_stills, # 표적 씬 슬라이스 (Codex 재리뷰 HIGH-4): guide/underlay **생성** # 도 effective allowlist 로 제한 — 전체 stills 는 그룹 형성· # 연속성 신호(컨텍스트)로 그대로 전달. None=기존 byte-identical. execution_allowlist_still_ids=target_effective, ) logger.info( "Scene generation: %d scenes in %d batches (max_concurrent=%d)", len(stills), len(scene_batches), max_concurrent, ) scenes_generated = 0 # 스타일 컨텍스트를 메인 스레드에서 한번만 로드 (스레드 안에서 DB 접근 방지) _cached_style_context = self._coord._get_style_context() # ref image 없는 엔티티의 텍스트 설명 맵 (short_id + C##O## composite) # (W5 F22 Phase B.16: scene_reference_service로 이관) _cached_entity_text_map = self._reference_svc.build_entity_text_map(entities) # 동시성 레이스 fix (2026-07-02): chain_bg UUID 를 메인 스레드에서 일괄 # pre-resolve. resolve_chain_bg_asset_id 는 (episode_id, bg_id) 키별 첫 # 조회가 공유 sync Session 을 치는데, 배치 워커(max_concurrent)가 동시에 # 첫 resolve 를 하면 Session 커넥션 체크아웃이 겹쳐 # IllegalStateChangeError("This session is provisioning a new connection; # concurrent operations are not permitted")로 해당 씬이 통째로 실패한다 # (E2E b0ad5c18: 배치 시작 직후 4씬). 여기서 캐시를 채워 워커는 캐시 # 히트만 하게 한다 — 위 style/entity_text 캐시와 동일한 # "스레드 안에서 DB 접근 방지" 원칙. for _bg_entry in (background_chain_bg_map or {}).values(): _warm_bg_id = (_bg_entry or {}).get("bg_id") if _warm_bg_id: self._reference_svc.resolve_chain_bg_asset_id(episode_id, _warm_bg_id) for batch_idx, scene_batch in enumerate(scene_batches): # resume: 이미 완료된 still_id 제외 scene_batch = [si for si in scene_batch if stills[si].get("id") not in already_done_stills] # 표적 씬 슬라이스: 실행 루프만 effective allowlist 제한 — # 그래프/배치 구성은 전체 stills 기준 그대로 (BLOCKING-1) if target_effective is not None: scene_batch = [ si for si in scene_batch if stills[si].get("id") in target_effective ] if not scene_batch: continue progress.update( f"씬 이미지 {scenes_generated}/{stills_to_gen}", scenes_generated + 1, total_items, ) # W5 F22 Phase B.22.4: _generate_one_scene nested → bound method wrapper def _generate_one_scene(si: int) -> tuple: return self._coord._generate_scene_in_loop( si=si, stills=stills, entity_lookup=entity_lookup, scene_paths_by_index_by_id=scene_paths_by_index_by_id, scene_primary_asset_id_by_still_id=scene_primary_asset_id_by_still_id, location_scene_history=location_scene_history, staging_map=staging_map, scene_ref_image_map=scene_ref_image_map, scene_ref_asset_id_map=scene_ref_asset_id_map, background_chain_bg_map=background_chain_bg_map, dep_detail_map=dep_detail_map, gemini_client=gemini_client, sanitizer=sanitizer, validator=validator, scene_dir=scene_dir, cached_style_context=_cached_style_context, cached_entity_text_map=_cached_entity_text_map, world_guide=world_guide, episode_id=episode_id, zoom_ctx=_zoom_ctx, indoor_pose_ctx=_indoor_pose_ctx, outdoor_direct_ctx=outdoor_direct_ctx, # W22 직행 ) # Run batch concurrently with ThreadPoolExecutor workers = min(max_concurrent, len(scene_batch)) batch_results: List[tuple] = [] with ThreadPoolExecutor(max_workers=workers) as executor: # W20E5 Codex B1 — propagate parent-thread image-call # budget into pool workers (each scene generation may hit # gpt-image-2 / fal under the cap). _submit_generate = bind_current_budget(_generate_one_scene) futures = { executor.submit(_submit_generate, si): si for si in scene_batch } for future in as_completed(futures): si = futures[future] try: result = future.result() batch_results.append(result) scenes_generated += 1 except StaleUpstreamError: # D6 T9-fix BLOCKING: scene-wide stale (catalog/render # manifest) — 전체 batch abort + structured propagate. # 다른 scene 도 같은 chain_bg/master_plan 의존이라 모두 # 같은 stale 신호. mark_failed 의 string-only 기록으로는 # 운영 신호 손실 → 상위 background job 가 structured # error 로 capture (logging + retry guidance). raise except Exception as exc: logger.error( "Scene generation failed: scene_index=%d, error=%s", si, exc, ) # 체크포인트에 실패 기록 still_id_failed = stills[si].get("id", "") if si < len(stills) else "" scene_cp.mark_failed(still_id_failed, str(exc)[:500]) scenes_generated += 1 # Post-process batch results: save assets, GPT select best, update location history # Process in scene_index order for deterministic DB writes for result in sorted(batch_results, key=lambda r: r[0]): si, var_results_list, visible_entities, current_location_ids = result still_data = stills[si] # Fix C (2026-05-10): typed failure dict 들 분리. 모두 실패면 # _summarize_failed_variations 가 reason+detail 직렬화한 cp 메시지 # 산출 (RefContractError 등 root cause 가시화). 성공 dict 만 # downstream 으로 전달. _all_failed_msg = _summarize_failed_variations(var_results_list, si) if _all_failed_msg is not None: still_id_failed = still_data.get("id", "") scene_cp.mark_failed(still_id_failed, _all_failed_msg) logger.error( "Scene %d still %s: %s", si, still_id_failed[:8], _all_failed_msg, ) continue # _failure_reason 마커가 있는 항목 제외 후 downstream 처리. var_results_list = [ r for r in var_results_list if isinstance(r, dict) and not r.get("_failure_reason") ] scene_ref_ids = [e["id"] for e in visible_entities] scene_lineage = build_lineage_fields(self._db, self._project_id, "scene_image_generator", ref_entity_ids=scene_ref_ids, prompt_type="cinematic", ) # Save all N variation images to DB saved_asset_ids, saved_paths = self._persistence_svc.save_scene_variations( var_results_list, scene_lineage, ) # W5 F22 Phase B.22.8: PNG metadata 컨텍스트 계산+embed를 provenance_svc로 이관 self._provenance_svc.compute_and_embed_scene_metadata( var_results_list, still_data, stills, si, scene_paths_by_index, scene_paths_by_index_by_id, ) # ── v6: 새 파이프라인 ── # 1) N개 변형 저장 (is_primary=0) — 위에서 이미 완료 # 2) Gemini Vision: 앵글 적용할 이미지 선택 + 앵글 추천 (최소 20도) # 3) fal.ai 앵글 적용 → N+1번째 이미지 # 4) Gemini Vision: 전체 N+1개 중 최종 대표 이미지 선택 → is_primary=1 # 3) fal.ai 앵글 적용 → N+1번째 이미지 # W5 F22 Phase B.22.6: _run_fal_angle_pipeline bound method로 이관 fal_generated = self._coord._run_fal_angle_pipeline( var_results_list=var_results_list, saved_asset_ids=saved_asset_ids, still_data=still_data, stills=stills, scene_dir=scene_dir, episode_id=episode_id, si=si, ) # 4) Gemini Vision: 전체 N(+1) 이미지 중 최종 대표 선택 # W5 F22 Phase B.22.5: scene_variation_service.select_best_variation_idx로 이관 best_idx, _best_was_selected = _select_best_variation_idx( var_results_list, still_data.get("beat_title", ""), ) if _best_was_selected: logger.info( "Scene %d: final best selection — image %d/%d (fal=%s) for still %s", si, best_idx + 1, len(var_results_list), fal_generated, still_data.get("id"), ) # W5 F22 Phase B.22.7: validation score override를 scene_variation_service로 이관 best_idx, _override = _apply_validation_override( var_results_list, best_idx, len(saved_asset_ids), fal_generated, ) if _override: _old_score, _new_score = _override logger.info( "Scene %d: overriding selection with higher-scored alternative (score %s->%s)", si, _old_score, _new_score, ) # W5 F22 Phase B.22.9: primary 마킹 + 추적 + 체크포인트를 bound method로 이관 primary_path = self._coord._finalize_and_track_primary( best_idx=best_idx, saved_asset_ids=saved_asset_ids, var_results_list=var_results_list, still_data=still_data, si=si, scene_cp=scene_cp, scene_results_by_index=scene_results_by_index, scene_paths_by_index=scene_paths_by_index, scene_paths_by_index_by_id=scene_paths_by_index_by_id, scene_primary_asset_id_by_still_id=scene_primary_asset_id_by_still_id, ) progress.update( f"씬 이미지 {scenes_generated}/{stills_to_gen}: {still_data.get('beat_title', '')}", scenes_generated + 1, total_items, ) # Update location_scene_history for same-background linking if primary_path.exists(): scene_bytes = primary_path.read_bytes() for loc_id in current_location_ids: location_scene_history[loc_id] = (scene_bytes, still_data) # B (2026-07-02): VE 에 location 이 없는 outdoor 샷 — primary_ # location UUID 로 기록해 same-location 후속 샷의 prev 프레임 # 소스를 살린다 (flag OFF → 맵 {} = 불변). _opf_si = still_data.get("scene_index") if (not current_location_ids and _opf_uuid_by_scene_main and isinstance(_opf_si, int)): _opf_u = _opf_uuid_by_scene_main.get(_opf_si) if _opf_u: location_scene_history[_opf_u] = (scene_bytes, still_data) progress.complete() self._logger.log( actor_id=self._actor_id, action="episode.generate_images", resource_type="episode", resource_id=episode_id, project_id=self._project_id, detail={ "reference_loaded": len(ref_image_map), "scene_count": len(stills), }, ip_address=ip, ) def generate_variations_only(self, episode_id: str) -> None: """기존 원본 이미지 기반 A/B 변형 추천 + 생성 (별도 실행). W5 F22 Phase B.4: 구현이 scene_variation_service.SceneVariationService로 이관됨. """ return self._variation_svc.generate_variations_only(episode_id) def _single_scene_generate_and_build_result( self, *, gemini_client: Any, t2i_prompt: str, beat_title: str, output_dir: Path, reference_images: Optional[List], still_id: str, episode_id: str, prompt_used: str, scene_index: Optional[int] = None, shot_index: Optional[int] = None, semantic_constraints: Optional[Dict[str, Any]] = None, attached_payload: Any = None, ) -> Dict[str, Any]: """Generate scene image via pipeline + build scene_result dict. W5 F22 Phase B.21.4 (2026-04-22): generate_single_scene_image의 두 경로 (custom_prompt / _build_final_scene_prompt)가 공유하던 try/except + scene_result dict 블록을 통합. ModerationError는 AppError로 re-raise. prompt_used만 path별로 다름 (custom_prompt 원문 vs _full_prompt 빌드 결과). Phase 4 iter 7 follow-up I1: scene_index / shot_index 인자 추가 — targeted mode 의 _decide_scene_lvm 가 trace_meta.scene_index/shot_index 로 매칭 키 생성. 이 인자가 없으면 single-scene/UI 재생성 경로가 항상 targeted_no_meta 로 silent skip (Codex+Claude review I1 합의). """ from app.modules.pipeline.scene_image_pipeline import generate_and_validate_scene from app.services.scene_generation_coordinator import ( _attached_lineage_fields as _single_attached_lineage_fields, ) try: pipe_result = generate_and_validate_scene( gemini_client=gemini_client, t2i_prompt=t2i_prompt, beat_title=beat_title, output_dir=output_dir, reference_images=reference_images if reference_images else None, previous_scene_bytes=None, # W3 observability — llm_call_log PID/EID NULL 차단 (single-scene path). # I1: scene_index/shot_index 도 forward — targeted mode 매칭용. trace_meta={ "project_id": self._project_id, "episode_id": episode_id, "operation_type": "single_scene_image_gen", "still_id": still_id, "scene_index": scene_index, "shot_index": shot_index, }, semantic_constraints=semantic_constraints, ) except ModerationError as exc: raise AppError( code="image.generation_blocked", message=f"Image generation blocked: {exc.block_reason}", status_code=400, ) return { "id": _new_id(), "asset_type": "scene", "entity_id": None, "still_id": still_id, "episode_id": episode_id, "file_path": pipe_result["file_path"], "prompt_used": prompt_used, "generation_model": pipe_result.get("generation_model", settings.gemini_image_model), "width": None, "height": None, "status": "generated", "review_notes": json.dumps(pipe_result.get("validation", {}), ensure_ascii=False), # P0 (2026-07-01): 실제 첨부 UUID lineage — save_single_scene_asset 가 # input_image_ids/actual_attached_refs 로 영속화(배치 var_result 와 대칭). **(_single_attached_lineage_fields(attached_payload) if attached_payload is not None else {}), "created_at": _now(), } def generate_single_scene_image( self, still_id: str, custom_prompt: Optional[str] = None, ip: Optional[str] = None, ) -> Dict[str, Any]: """Generate a single scene image for a still.""" # W5 F22 Phase B.23.1: still 조회 + 파이프라인 준비성 검증을 persistence_svc로 이관 still = self._persistence_svc.fetch_still_for_generation(still_id) episode_id = still.episode_id project_dir = self._get_project_dir() scene_dir = project_dir / "images" / episode_id / "scene" # Group 1 #2 (visual_pipeline_contracts_plan): asset readiness preflight — # 단일 still 재생성 경로도 동일 정책 적용. ENV ALLOW_TEXT_ONLY_WITHOUT_REFS # 우회 가능. 기본 block. from app.core.asset_readiness import assert_episode_asset_readiness assert_episode_asset_readiness(self._db, self._project_id, episode_id) # batch path (scene_persistence_service) 의 still_data shape 와 정합 — # scene_index/shot_index/dependent_scene_id 누락 시 build_scene_attached_refs # 의 _bc_key 가 "0_0" 으로 silent miss → chain_bg / prev_shot ref 첨부 실패 # (2026-05-09 단건 regen 결함 보정). still_data = { "id": still.id, "still_index": still.still_index, "scene_index": still.scene_index, "shot_index": still.shot_index, "dependent_scene_id": still.dependent_scene_id, "screenplay_scene_heading": still.screenplay_scene_heading or "", "beat_title": still.beat_title or "", "still_frame_prompt": custom_prompt or still.still_frame_prompt or "", "camera_json": still.camera_json or "{}", "lighting_json": still.lighting_json or "{}", "visible_entities_json": still.visible_entities_json or "[]", } # Get world guide world_guide = get_latest_world_guide(self._db, self._project_id, episode_id) # Get visible entities visible_entities = self._reference_svc.get_visible_entities(still.visible_entities_json) # W5 F22 Phase B.23.3: ref_image_map location 필터링을 reference_svc로 이관 ref_image_map = self._reference_svc.get_ref_image_map_excluding_locations(visible_entities) ep = self._db.query(Episode).filter(Episode.id == episode_id).first() language = ep.language if ep else "ko" # W5 F22 Phase B.23.2: 4-client 초기화 공통 helper gemini_client, openai_client, sanitizer, tracker = self._coord._init_scene_gen_clients() # Patch B-min — moderation sanitize 시점의 polarity 보존 contract. # custom_prompt 경로도 sanitize 가 동일하게 작동 → 양쪽 모두 wire. # staging/rpc lookup 은 coordinator 의 helper 재사용 (D1 patch idiom). # _stg_key 포맷은 coordinator 와 동일 `"{scene_index}_{shot_index}"` # (load_shot_staging_map 의 key 포맷 — scene_checkpoint_loaders.py:53). _stg_map = load_shot_staging_map(settings.projects_dir, self._project_id, episode_id) _stg_key = ( f"{still.scene_index}_{still.shot_index}" if still.scene_index is not None and still.shot_index is not None else None ) _stg = _stg_map.get(_stg_key) if _stg_map and _stg_key else None try: _rpc = lookup_render_prompt_card( project_id=self._project_id, episode_id=episode_id, scene_index=still.scene_index, shot_index=still.shot_index, ) except RefContractError as _rpc_exc: logger.warning( "Patch B-min: render_prompt_card lookup failed for still %s " "(scene=%s shot=%s): %s", still_id, still.scene_index, still.shot_index, _rpc_exc, ) _rpc = None # I2: dedup downstream lookup — inject rpc back into still_data so # coordinator._build_single_scene_prompt_and_refs (L1262) skips re-fetch. # Failure case (_rpc=None): do NOT inject — coordinator will re-attempt # and surface RefContractError for non-custom_prompt path (preserved). if _rpc is not None: still_data["render_prompt_card"] = _rpc _semantic_contract = build_semantic_contract( shot_staging=_stg, render_prompt_card=_rpc, visible_entities=visible_entities, ) _semantic_constraints = _semantic_contract.sanitizer_constraints if custom_prompt: # raw_prompt 모드: 사용자가 편집한 최종 프롬프트를 래핑 없이 Gemini에 직접 전달 # 참조 이미지 구성 (W5 F22 Phase B.21.3: scene_reference_service로 이관) # Phase 3 정련 (2026-06-11 S10 sh5 v2 실측): custom 경로도 해당 shot 의 # bg map(space plate/chain) ref 를 부착 — 없으면 모델이 환경(벽 재질 등)을 # 텍스트만으로 발명. sentinel(no-plate)은 builder 가 걸러낸다. _bg_map = load_background_chain_bg_map( settings.projects_dir, self._project_id, episode_id, ) _bg_entry = _bg_map.get(_stg_key) if _stg_key else None labeled_refs = self._reference_svc.build_custom_labeled_refs( visible_entities, ref_image_map, chain_bg_entry=_bg_entry, ) # 생성 + scene_result (W5 F22 Phase B.21.4: _single_scene_generate_and_build_result) # I1: still ORM 에서 scene_index/shot_index forward — targeted mode 매칭. scene_result = self._single_scene_generate_and_build_result( gemini_client=gemini_client, t2i_prompt=custom_prompt, beat_title=still_data.get("beat_title", ""), output_dir=scene_dir, reference_images=labeled_refs, still_id=still_id, episode_id=episode_id, prompt_used=custom_prompt, scene_index=still.scene_index, shot_index=still.shot_index, semantic_constraints=_semantic_constraints, ) else: # W21B-W7 W-C2 (2026-06-12): zoom continuity 분기 — flag OFF/anchor # 부재 시 빈 dict = 기존 경로 byte-identical (Codex 판정 ②: 정식 # generation path 내부 분기, post-pass 금지). from app.core.steps.zoom_continuity_anchor_step import ( load_zoom_continuity_context, ) _zoom_ctx = load_zoom_continuity_context(self._project_id, episode_id) _zc_key = (still.scene_index, still.shot_index) # (a) image-phase prompt override — 단건 경로는 t2i_variations[0] # 을 쓰므로 revised[0] 매칭. scene_detail cp 불변, 실사용 프롬프트는 # prompt_used 로 기록 (Codex guard). W21B-W8: outdoor site layout # override 와 명시적 priority merge (custom > zoom > site > # original — custom 은 위 분기 선행 승리). from app.core.steps.outdoor_site_layout_step import ( load_outdoor_site_layout_context, ) from app.modules.pipeline.outdoor_site_layout_plan import ( merge_prompt_overrides, ) _site_ctx = load_outdoor_site_layout_context(self._project_id, episode_id) _merged_overrides = merge_prompt_overrides( _zoom_ctx.get("source_overrides") or {}, _site_ctx.get("prompt_overrides") or {}, ) _var_t2i_override = None _zc_override = _merged_overrides.get(_zc_key) if _zc_override and 0 in _zc_override["revised"]: _var_t2i_override = _zc_override["revised"][0] logger.info( "scene prompt override 적용: S%ssh%s " "(prompt_source=%s, group=%s, revised_hash=%s)", still.scene_index, still.shot_index, (_zc_override.get("prompt_source_by_index") or {}).get(0) or _zc_override.get("prompt_source"), _zc_override.get("group_id"), (_zc_override.get("provenance") or {}).get("0", {}).get("revised_prompt_hash"), ) # (b) zoom 멤버 — source_wide primary 에서 crop/i2i-fill. 실패는 # 비차단 fallback (기존 T2I) + artifacts 의 fallback reason — # canary acceptance 에선 실패 간주 (Codex guard 4). _zc_target = (_zoom_ctx.get("zoom_targets") or {}).get(_zc_key) scene_result = None if _zc_target: from app.core.steps.visual_continuity_anchor_step import ( build_zoom_fill_prop_refs, ) from app.services.zoom_continuity_render_service import ( ZoomContinuityRenderError, generate_continuity_crop_png, ) # S12sh12 fix (2026-06-12): printed_prop anchor 계약(printed_content # 등)을 fill 의 object ref 라벨에 적용 — anchor 부재 시 기존 라벨 # + 프롬프트 모두 byte-identical (content-SOT 절은 anchor ref 한정). _prop_refs, _prop_anchor_applied = build_zoom_fill_prop_refs( self._project_id, episode_id, visible_entities, ref_image_map, scene_index=still.scene_index, shot_index=still.shot_index, ) try: _png, _fill_prompt, _zc_diag = generate_continuity_crop_png( db=self._db, project_id=self._project_id, episode_id=episode_id, scene_index=still.scene_index, shot_index=still.shot_index, zoom_target=_zc_target, gemini_client=gemini_client, prop_refs=_prop_refs, prop_refs_have_anchor_contract=_prop_anchor_applied, # P0c (2026-06-18): 단건 regen 도 로깅 — single 경로와 동일 # operation_type, still_id 포함. trace_meta={ "operation_type": "single_scene_image_gen", "still_id": still_id, }, # Phase C: zoom crop 중간물 영속화. source_asset_id 는 단건 path # 라 db(_load_source_primary_bytes) 안에서 내부 resolve. still_id=still_id, ) _out = scene_dir / f"{_new_id()}.png" _out.parent.mkdir(parents=True, exist_ok=True) _out.write_bytes(_png) scene_result = { "id": _new_id(), "asset_type": "scene", "entity_id": None, "still_id": still_id, "episode_id": episode_id, "file_path": to_relative_image_path(str(_out)), "prompt_used": _fill_prompt, "generation_model": settings.gemini_image_model, "width": None, "height": None, "status": "generated", "review_notes": json.dumps( {"zoom_continuity": _zc_diag}, ensure_ascii=False), "created_at": _now(), } logger.info( "zoom_continuity: S%ssh%s crop/i2i-fill 성공 (group=%s)", still.scene_index, still.shot_index, _zc_target.get("group_id"), ) except ZoomContinuityRenderError as _zc_exc: logger.warning( "zoom_continuity: S%ssh%s crop 경로 fallback → 기존 T2I (%s)", still.scene_index, still.shot_index, _zc_exc, ) if scene_result is not None: return self._coord._finalize_single_scene( scene_result=scene_result, visible_entities=visible_entities, still_id=still_id, ip=ip, ) # W5 F22 Phase B.23.5: _build_final_scene_prompt 경로 bound method lift. # D5 T2 §4.3: builder return 이 3-tuple. attached_meta 는 T3 까지 unused. _full_prompt, labeled_refs, _attached_meta, _attached_payload = self._coord._build_single_scene_prompt_and_refs( still=still, episode_id=episode_id, still_data=still_data, visible_entities=visible_entities, ref_image_map=ref_image_map, var_t2i_override=_var_t2i_override, ) # 생성 + scene_result (W5 F22 Phase B.21.4: _single_scene_generate_and_build_result) # I1: still ORM 에서 scene_index/shot_index forward — targeted mode 매칭. # P0 (2026-07-01): _attached_payload 전달 → scene_result 에 실제 첨부 # UUID lineage(input_image_ids/actual_attached_refs) 실어 영속화. scene_result = self._single_scene_generate_and_build_result( gemini_client=gemini_client, t2i_prompt=_full_prompt, beat_title=still_data.get("beat_title", ""), output_dir=scene_dir, reference_images=labeled_refs, still_id=still_id, episode_id=episode_id, prompt_used=_full_prompt, scene_index=still.scene_index, shot_index=still.shot_index, semantic_constraints=_semantic_constraints, attached_payload=_attached_payload, ) # A5 (2026-07-02): prev-frame chaining 진단 — build_scene_attached_refs 가 # still_data 에 기록(멤버 샷만). persistence 가 pipeline_metadata 로 영속. if still_data.get("_prev_frame_chain_diag"): scene_result["prev_frame_chain"] = still_data["_prev_frame_chain_diag"] # W5 F22 Phase B.23.4: 후처리(lineage + save + log + dict)를 bound method로 이관 return self._coord._finalize_single_scene( scene_result=scene_result, visible_entities=visible_entities, still_id=still_id, ip=ip, ) # ------------------------------------------------------------------ # Upload custom image # ------------------------------------------------------------------ def recommend_variations( self, still_id: str, ip: Optional[str] = None, ) -> Dict[str, Any]: """LLM에게 A/B 변형 추천 받기. W5 F22 Phase B.3: 구현이 scene_variation_service.SceneVariationService로 이관됨. """ return self._variation_svc.recommend_variations(still_id, ip=ip) def generate_scene_with_variations( self, still_id: str, ip: Optional[str] = None, ) -> Dict[str, Any]: """원본 + A변형 + B변형 생성. 1. Generate original (existing scene image generation) 2. If variation_a has angle/color -> i2i from original 3. If variation_b has angle/color -> i2i from original 4. Set recommended as PDF default """ if not settings.gemini_api_key and gemini_key_count() == 0: raise AppError( code="image.gemini_key_missing", message=t("image.gemini_key_missing"), status_code=400, ) still = ( self._db.query(SceneStill) .filter(SceneStill.id == still_id, SceneStill.project_id == self._project_id) .first() ) if not still: raise AppError( code="still.not_found", message=t("still.not_found"), status_code=404, ) episode_id = still.episode_id # Group 1 #2 (Claude review IMPORTANT): generate_scene_with_variations 도 # entry-level preflight 적용 — generate_single_scene_image 안 호출 전 동기 # check 로 일관성 유지. inner 호출이 idempotent fast path 이므로 redundant 영향 0. from app.core.asset_readiness import assert_episode_asset_readiness assert_episode_asset_readiness(self._db, self._project_id, episode_id) project_dir = self._get_project_dir() scene_dir = project_dir / "images" / episode_id / "scene" scene_dir.mkdir(parents=True, exist_ok=True) # Step 1: Generate original scene image original_result = self.generate_single_scene_image(still_id, ip=ip) original_id = original_result["id"] # W5 F22 Phase B.24.3: original variant_type 마킹을 persistence_svc로 이관 self._persistence_svc.mark_asset_as_original(original_id) # Read original image bytes for i2i original_path = Path(original_result["file_path"]) if not original_path.exists(): return { "original": original_result, "variant_a": None, "variant_b": None, "recommended": still.recommended_variant or "original", } original_bytes = original_path.read_bytes() from app.modules.gemini_i2i_editor import GeminiI2IEditor i2i_editor = GeminiI2IEditor( api_key=get_next_key(), model=settings.gemini_image_model, ) lineage = build_lineage_fields(self._db, self._project_id, "gemini_i2i_editor") # W5 F22 Phase B.24.1: A/B variation 슬롯 공통화 variant_a_result = self._coord._maybe_generate_variation_slot( slot="a", still=still, i2i_editor=i2i_editor, original_bytes=original_bytes, original_id=original_id, scene_dir=scene_dir, episode_id=episode_id, lineage=lineage, ) variant_b_result = self._coord._maybe_generate_variation_slot( slot="b", still=still, i2i_editor=i2i_editor, original_bytes=original_bytes, original_id=original_id, scene_dir=scene_dir, episode_id=episode_id, lineage=lineage, ) # W5 F22 Phase B.24.2: 최종화(recommended + commit + log + dict) bound method로 이관 return self._coord._finalize_scene_with_variations( still=still, still_id=still_id, original_result=original_result, original_id=original_id, variant_a_result=variant_a_result, variant_b_result=variant_b_result, ip=ip, ) def regenerate_variation( self, image_id: str, angle_json: Optional[str] = None, color_prompt: Optional[str] = None, ip: Optional[str] = None, ) -> Dict[str, Any]: """원본 기반 fal.ai i2i로 변형 재생성. W5 F22 Phase B.2: 구현이 scene_variation_service.SceneVariationService로 이관됨. color_prompt는 reserved (Phase A 기준 angle-only). 자세한 계약은 모듈 docstring 참조. """ return self._variation_svc.regenerate_variation( image_id, angle_json=angle_json, color_prompt=color_prompt, ip=ip, ) def select_variant( self, still_id: str, variant: str, ip: Optional[str] = None, ) -> Dict[str, Any]: """PDF 대표 이미지 선택 (original/A/B). W5 F22 Phase B.1: 구현이 scene_variation_service.SceneVariationService로 이관됨. facade 계약(still_id/variant/ip signature) 그대로 유지. """ return self._variation_svc.select_variant(still_id, variant, ip=ip) def select_original( self, still_id: str, image_id: str, ip: Optional[str] = None, ) -> Dict[str, Any]: """원본 이미지 선택 (여러 장 중). W5 F22 Phase B.1: 구현이 scene_variation_service.SceneVariationService로 이관됨. facade 계약(still_id/image_id/ip signature) 그대로 유지. """ return self._variation_svc.select_original(still_id, image_id, ip=ip) # ------------------------------------------------------------------ # Public validation endpoints # ------------------------------------------------------------------