"""Scene persistence service — scene_image_service에서 분리된 DB 저장 전용. W5 F22 Phase B.9.1~B.19 (2026-04-22): scene_image_service.generate_images()의 ImageAsset 저장/복원 + WorldGuide 해소 로직을 점진 이관. ## 공개 API | 메서드 | 역할 | |--------|------| | `save_scene_variations(var_results_list, scene_lineage)` | N variation ImageAsset bulk 저장 + commit | | `set_primary_asset(asset_id)` | 단일 asset을 is_primary=1로 업데이트 + commit | | `save_single_scene_asset(scene_result, lineage)` | 단일 scene asset 저장 + auto_set_primary + commit | | `save_fal_angle_asset(...)` | fal.ai angle-edited asset 저장 + commit | | `build_resume_state(stills, already_done_stills, entity_lookup)` | resume 모드에서 기존 scene asset들을 읽어 scene_paths/scene_paths_by_id/scene_results/location_history 4-map 복원 | | `resolve_world_guide(...)` | WorldGuide 해소 (hash 재사용 또는 재생성) + project style 주입 | | `load_episode_entity_dicts(episode_id)` | 에피소드에 연결된 EntityCanon을 dict 리스트로 로드 | | `load_episode_still_dicts(episode_id)` | 에피소드의 선택된 SceneStill을 dict 리스트 + ORM 리스트 tuple로 로드 | | `validate_episode_ready(episode_id)` | pipeline_gate + scene_count>0 검증 (AppError raise) | | `fetch_still_for_generation(still_id)` | still 조회 + pipeline_gate + gemini_key 검증 (single scene 생성용) | | `mark_asset_as_original(asset_id)` | 지정 asset의 variant_type='original' 마킹 + flush | 각 save/update 메서드는 self._db.commit()을 포함한다 (호출자 commit 불필요). build_resume_state는 read-only. resolve_world_guide는 재생성 시 self._db.flush()만 수행 (commit은 상위 호출자 책임 — 기존 generate_images 흐름 보존). """ from __future__ import annotations import hashlib import json import logging import uuid from datetime import datetime, timezone from pathlib import Path from typing import Any, Dict, List, Optional, Tuple from sqlalchemy.orm import Session as OrmSession from app.core.config import settings from app.core.errors import AppError from app.core.file_paths import to_relative_image_path from app.services.image_capture.annotate import annotate_generated_asset from app.i18n.loader import t from app.modules.llm.gemini_key_pool import key_count as gemini_key_count from app.models.project import ( EntityCanon, EntityEpisodeLink, ImageAsset, LLMCallLog, ProjectSettings, SceneStill, WorldGuide, ) from app.modules.llm.openai_client import OpenAIClient from app.modules.world_guide_generator import WorldGuideGenerator from app.services.image_service_helpers import auto_set_primary __all__ = ["ScenePersistenceService"] logger = logging.getLogger(__name__) class ScenePersistenceService: """씬 이미지 persistence 전담 서비스 (W5 F22 Phase B.9).""" def __init__(self, db: OrmSession, project_id: str) -> None: self._db = db self._project_id = project_id def save_scene_variations( self, var_results_list: List[Dict[str, Any]], scene_lineage: Dict[str, Any], ) -> Tuple[List[str], List[Path]]: """Save N variation ImageAssets for a scene (bulk insert + commit). Each var_result must include the keys populated by the scene generation pipeline (id, asset_type, entity_id, still_id, episode_id, file_path, prompt_used, generation_model, width, height, status, review_notes, created_at) plus optional (validation_score, validation_result, variant_type, theme_label, sanitization_strategy, sanitization_note). `scene_lineage` must provide prompt_type, code_version, prompt_file_version, reference_image_ids. Returns (saved_asset_ids, saved_paths) in input order. """ saved_asset_ids: List[str] = [] saved_paths: List[Path] = [] for vr in var_results_list: asset = ImageAsset( id=vr["id"], project_id=self._project_id, asset_type=vr["asset_type"], entity_id=vr["entity_id"], still_id=vr["still_id"], episode_id=vr["episode_id"], file_path=to_relative_image_path(vr["file_path"]), prompt_used=vr["prompt_used"], generation_model=vr["generation_model"], width=vr["width"], height=vr["height"], status=vr["status"], review_notes=vr["review_notes"], validation_score=vr.get("validation_score"), validation_result=vr.get("validation_result"), variant_type=vr.get("variant_type", "base"), theme_label=vr.get("theme_label"), prompt_type=scene_lineage["prompt_type"], code_version=scene_lineage["code_version"], prompt_file_version=scene_lineage["prompt_file_version"], reference_image_ids=scene_lineage["reference_image_ids"], is_primary=0, created_at=vr["created_at"], ) self._db.add(asset) # P0 (2026-07-01, Codex 합의): 씬 이미지 생성에 **실제 첨부된 모든** # image asset UUID(배경 plate / 캐릭터 face·composite·state / 마네킹· # 구도 가이드 / 소품)를 input_image_ids 에 실어 canvas/모달 generated_input # 엣지를 진실화한다(라벨 파싱 금지, 구조 metadata SOT). char/prop 도 포함 — # reference_image_ids(의도된 entity lineage)와 의미가 다르며 공존한다. # actual_attached_image_ids(coordinator 수집) 우선, 없으면 legacy # pose_guide_asset_ids fallback (backward-compat). _input_ids, _gen_call_id, _meta = self._build_lineage_annotation( vr, vr.get("episode_id"), "scene_image_gen", ) if _gen_call_id: asset.generation_call_id = _gen_call_id annotate_generated_asset( asset, pipeline_role="scene_still", stage="scene_image", input_image_ids=(_input_ids or None), pipeline_metadata=_meta, ) saved_asset_ids.append(vr["id"]) saved_paths.append(Path(vr["file_path"])) self._db.commit() return saved_asset_ids, saved_paths def _build_lineage_annotation( self, source: Dict[str, Any], episode_id: Any, operation_type: str, ) -> Tuple[List[str], Optional[str], Dict[str, Any]]: """P0: var_result/scene_result 에서 실제 첨부 UUID lineage annotation 구성. 배치(save_scene_variations)/단건(save_single_scene_asset) 공유 — 동일 정책. actual_attached_image_ids 우선(없으면 legacy pose_guide_asset_ids) + unresolved 중 registered_pose_guide post-hoc resolve(구조키) 병합 + generation_call_id. Returns (input_image_ids, generation_call_id, pipeline_metadata). """ _actual_ids = list(source.get("actual_attached_image_ids") or []) _fallback_ids = source.get("pose_guide_asset_ids") or [] _input_ids: List[str] = [] for _gid in (_actual_ids or _fallback_ids): if _gid and _gid not in _input_ids: _input_ids.append(_gid) _actual_refs = list(source.get("actual_attached_refs") or []) _unresolved = list(source.get("unresolved_attached_refs") or []) # registered_pose_guide post-hoc resolve(worker-thread 라 attach 시 asset_id 부재). _r_ids, _r_refs, _remaining = self._resolve_registered_guide_asset_ids( _unresolved, episode_id, ) for _rid in _r_ids: if _rid not in _input_ids: _input_ids.append(_rid) _actual_refs.extend(_r_refs) # A5 (2026-07-02): immobilized prev-frame ref post-hoc resolve — 구조키 # source_still_id 로 anchor still 의 scene primary UUID 를 붙인다 # (registered_pose_guide 와 동일 패턴, bytes/UUID 미혼합 P0 계약). _p_ids, _p_refs, _remaining = self._resolve_prev_frame_asset_ids( _remaining, episode_id, ) for _pid in _p_ids: if _pid not in _input_ids: _input_ids.append(_pid) _actual_refs.extend(_p_refs) # #108 (Codex R2 BLOCK-1): 호출자가 키를 **명시**했으면 그 값을 # 그대로 존중한다 — 명시 None(=exact miss, 링크 없음 확정)까지 # `or` 로 broad resolve 에 되살리면, 같은 still 의 과거 무관 # single_scene 호출이 최종본에 거짓 연결된다. legacy fallback 은 # 키 부재 호출자(배치·단건 API 경로)에만 유지. if "generation_call_id" in source: _gen_call_id = source.get("generation_call_id") else: _gen_call_id = self._resolve_generation_call_id( source.get("still_id"), episode_id, operation_type=operation_type, ) meta = { "reference_lineage_source": "input_image_ids_actual_attached", # 표시/감사용 상세(asset_id + role + label). edge 복원 SOT 아님. "actual_attached_refs": _actual_refs, "unresolved_attached_refs": _remaining, } # A5: prev-frame chaining 진단(부착/미부착 사유, Codex 합의 필드) — # 캔버스/모달/후속 SQL 재현용. 없으면 키 자체 미기록 (기존 meta 불변). if source.get("prev_frame_chain"): meta["prev_frame_chain"] = source["prev_frame_chain"] # shot_run_uid (2026-08-24) — Opik trace·records.json 과 자산을 잇는 값. # ★여기가 **실제 sink** 다. 호출자가 scene_result 최상위에 넣어도 # ImageAsset 생성자에는 그런 칸이 없고 annotate 는 이 meta 만 # merge 하므로, 여기서 안 실으면 pipeline_metadata_json 에 # 한 글자도 안 남는다(2026-08-24 Codex BLOCK 1 — 내 시험이 ternary # 만 재연해 실제 경로를 안 탔다). # ★truthy 일 때만 — 산출을 재사용한 방문은 호출자가 None 을 준다 # (그 자산을 만든 것은 이전 주행이다). 키 자체를 안 만들어 # 기존 meta 와 바이트 동일하다. if source.get("shot_run_uid"): meta["shot_run_uid"] = source["shot_run_uid"] return _input_ids, _gen_call_id, meta def _resolve_generation_call_id( self, still_id: Any, episode_id: Any, operation_type: str = "scene_image_gen", multiroll_tag: str = "", ) -> str | None: """P0 best-effort: 이 still 의 image_gen llm_call_log 감사 링크 회수. 구조키(metadata_json 의 still_id UUID) 로만 매칭 — 라벨/프롬프트 파싱 아님. 한 still 에 variation 별 call 이 여럿이면 최신 success 1개(감사 수준 링크, variation 단위 정합은 v1 미보장·문서화). still_id 부재 시 None. operation_type: 배치="scene_image_gen", 단건="single_scene_image_gen". multiroll_tag(Codex 8e70d4c0 HIGH-2 재리뷰): selected roll/fix 의 exact 호출 태그 — 지정 시 exact 매칭만 허용하고 miss 는 None (broad still_id fallback 은 패자 branch/다른 롤을 거짓 링크할 수 있어 금지). legacy fallback 은 multiroll_tag 미지정 호출에만. """ if not still_id: return None try: base = ( self._db.query(LLMCallLog.id) .filter( LLMCallLog.project_id == self._project_id, LLMCallLog.operation_type == operation_type, LLMCallLog.status == "success", LLMCallLog.metadata_json.like(f'%"still_id": "{still_id}"%'), ) ) if multiroll_tag: exact = ( base.filter( LLMCallLog.metadata_json.like( f'%"multiroll_tag": "{multiroll_tag}"%') ) .order_by(LLMCallLog.created_at.desc()) .first() ) return exact[0] if exact else None q = base.order_by(LLMCallLog.created_at.desc()).first() return q[0] if q else None except Exception as exc: # 감사 링크 실패는 비치명(핵심 lineage=input_image_ids) logger.warning("generation_call_id resolve 실패 still=%s: %s", still_id, exc) return None def safety_ladder_call_provenance( self, call_id: Any, ) -> Optional[Dict[str, Any]]: """(Codex safety-ladder BLOCK-2) fallback 성공 자산의 provenance SOT. 해당 호출의 metadata_json 에 safety_ladder(비 primary 단계명)가 있을 때만 {stage, model_name, user_prompt} 를 돌려준다 — 사다리 미발화·플래그 OFF 호출은 키 자체가 없어 None(기존 경로 byte-identical). 소비자는 이것으로 자산의 generation_model· prompt_used 를 실제 성공 호출과 정렬한다(설정 플래그 추정 금지). """ if not call_id: return None try: row = ( self._db.query( LLMCallLog.model_name, LLMCallLog.user_prompt, LLMCallLog.metadata_json) .filter(LLMCallLog.id == call_id) .first() ) if not row: return None meta = json.loads(row[2] or "{}") stage = meta.get("safety_ladder") if not stage: return None return { "stage": str(stage), "model_name": row[0], "user_prompt": row[1], } except Exception as exc: # noqa: BLE001 — provenance 보정 실패 비치명 logger.warning( "safety_ladder provenance 조회 실패 call=%s: %s", call_id, exc) return None def call_model_name(self, call_id: Any) -> Optional[str]: """그 호출이 **실제로 쓴 모델** — 없으면 None (2026-08-26 감사 0-B). ★설정값으로 내려가지 않는다. 제공자를 모를 때 설정값을 적으면 빈 칸 보다 **더 확정적인 거짓 기록**이 된다(Codex BLOCK-3). 변환 자산의 `generation_model` 은 이 호출 기록이 SOT 이고, backfill 스크립트도 같은 값을 쓴다. """ if not call_id: return None try: row = ( self._db.query(LLMCallLog.model_name) .filter(LLMCallLog.id == call_id) .first() ) return (str(row[0]).strip() or None) if row and row[0] else None except Exception as exc: # noqa: BLE001 — provenance 조회 실패 비치명 logger.warning( "호출 모델 조회 실패 call=%s: %s", call_id, exc) return None def _resolve_registered_guide_asset_ids( self, unresolved: List[Dict[str, Any]], episode_id: Any, ) -> Tuple[List[str], List[Dict[str, Any]], List[Dict[str, Any]]]: """P0 (2026-07-01, Codex 합의): unresolved 중 registered_pose_guide 를 구조키(group_id+bg_key+guide_hash)로 post-hoc resolve. attach_registered_pose_guide_ref 는 worker-thread(DB접근0) 라 asset_id 를 못 실어 unresolved 로 떨어진다. 여기서 저장 시점(DB session 有)에 라벨없이 구조키로 실제 registered_pose_guide ImageAsset UUID 를 붙인다. 우선순위: (group_id,bg_key,guide_hash) exact → 1개면 resolve / fallback (group_id,bg_key) → 정확히 1개면 resolve, 2+ 면 ambiguous(unresolved 유지 + reason) / still_id fallback 은 그룹캐시 cache_hit 시 still 이 다를 수 있어 여기선 안 씀(coordinator 가 이미 shot 매핑). resolve 실패는 그대로 unresolved. Returns: (resolved_ids, resolved_refs[resolved_from_unresolved 마커], remaining_unresolved). """ resolved_ids: List[str] = [] resolved_refs: List[Dict[str, Any]] = [] remaining: List[Dict[str, Any]] = [] for u in unresolved: if not isinstance(u, dict) or u.get("pipeline_role") != "registered_pose_guide": remaining.append(u) continue gid = u.get("group_id") bg_key = u.get("bg_key") gh = u.get("guide_hash") rid: Optional[str] = None ambiguous = False try: base = self._db.query(ImageAsset.id).filter( ImageAsset.project_id == self._project_id, ImageAsset.episode_id == episode_id, ImageAsset.pipeline_role == "registered_pose_guide", ImageAsset.is_intermediate.is_(True), ) if gid and bg_key and gh: rows = base.filter( ImageAsset.pipeline_metadata_json.like(f'%"group_id": "{gid}"%'), ImageAsset.pipeline_metadata_json.like(f'%"bg_key": "{bg_key}"%'), ImageAsset.pipeline_metadata_json.like(f'%"guide_hash": "{gh}"%'), ).all() if len(rows) == 1: rid = rows[0][0] if rid is None and gid and bg_key: rows = base.filter( ImageAsset.pipeline_metadata_json.like(f'%"group_id": "{gid}"%'), ImageAsset.pipeline_metadata_json.like(f'%"bg_key": "{bg_key}"%'), ).all() if len(rows) == 1: rid = rows[0][0] elif len(rows) > 1: ambiguous = True except Exception as exc: # resolve 실패는 비치명 → unresolved 유지 logger.warning("registered_pose_guide resolve 실패 group=%s: %s", gid, exc) remaining.append(u) continue if rid: if rid not in resolved_ids: resolved_ids.append(rid) resolved_refs.append({ "asset_id": rid, "role": u.get("role") or "immobilized_pose_guide", "label": u.get("label", ""), "resolved_from_unresolved": True, }) else: _u = dict(u) if ambiguous: _u["reason"] = "registered_pose_guide_ambiguous_pair" remaining.append(_u) return resolved_ids, resolved_refs, remaining def _resolve_prev_frame_asset_ids( self, unresolved: List[Dict[str, Any]], episode_id: Any, ) -> Tuple[List[str], List[Dict[str, Any]], List[Dict[str, Any]]]: """A5 (2026-07-02): unresolved 중 immobilized_prev_frame 을 구조키 ``source_still_id`` 로 post-hoc resolve (registered_pose_guide 패턴). attach_immobilized_prev_frame_ref 는 worker-thread(DB 접근 0) 라 anchor 프레임의 asset UUID 를 못 싣는다 — 저장 시점에 anchor still 의 **scene primary** ImageAsset UUID 를 라벨 파싱 없이 붙인다. primary 마킹이 아직 없거나(동일 batch 내 예외 경로) 조회 실패면 그대로 unresolved 유지 (비치명 — bytes ref 는 이미 첨부됨, lineage 만 부분). Returns: (resolved_ids, resolved_refs[resolved_from_unresolved 마커], remaining_unresolved). """ resolved_ids: List[str] = [] resolved_refs: List[Dict[str, Any]] = [] remaining: List[Dict[str, Any]] = [] for u in unresolved: # B (2026-07-02): outdoor prev_shot ref(scene_prev_frame)도 동일 구조키 # (source_still_id) resolve 대상 — immobilized 와 같은 primary lookup. if not isinstance(u, dict) or u.get("pipeline_role") not in ( "immobilized_prev_frame", "scene_prev_frame"): remaining.append(u) continue src_sid = u.get("source_still_id") rid: Optional[str] = None if src_sid: try: row = ( self._db.query(ImageAsset.id) .filter( ImageAsset.project_id == self._project_id, ImageAsset.episode_id == episode_id, ImageAsset.still_id == src_sid, ImageAsset.asset_type == "scene", ImageAsset.is_primary == 1, ) .order_by(ImageAsset.created_at.desc()) .first() ) rid = row[0] if row else None except Exception as exc: # resolve 실패는 비치명 → unresolved 유지 logger.warning( "immobilized_prev_frame resolve 실패 source_still=%s: %s", src_sid, exc) remaining.append(u) continue if rid: if rid not in resolved_ids: resolved_ids.append(rid) resolved_refs.append({ "asset_id": rid, "role": u.get("role") or "previous_shot_same_room", "label": u.get("label", ""), "pipeline_role": u.get("pipeline_role"), "resolved_from_unresolved": True, }) else: _u = dict(u) if not src_sid: _u["reason"] = "prev_frame_source_still_id_missing" elif "reason" not in _u: _u["reason"] = "prev_frame_primary_not_found" remaining.append(_u) return resolved_ids, resolved_refs, remaining def set_primary_asset(self, asset_id: str) -> None: """Mark a single ImageAsset as is_primary=1 (commit included).""" self._db.query(ImageAsset).filter( ImageAsset.id == asset_id, ).update({"is_primary": 1}) self._db.commit() def scan_completed_scene_stills(self, episode_id: str, scene_cp: Any) -> set: """Scan DB + checkpoint to determine already-completed still_ids (resume mode). W5 F22 Phase B.15: generate_images의 resume 스캔 루프 이관. 하나의 still은 다음 중 하나를 만족하면 "완료"로 간주한다: 1) DB에 해당 still의 scene primary ImageAsset이 있고 file_path가 디스크에 실제 존재 2) scene_cp.get_completed_ids()에 등록되어 있고 primary_path가 디스크에 존재 scene_cp은 ImageCheckpointManager 인터페이스 (`get_completed_ids()` + `_data["completed"][cp_id]["primary_path"]` 접근)로 쓰인다. """ done: set = set() existing_stills = ( self._db.query(ImageAsset.still_id) .filter( ImageAsset.project_id == self._project_id, ImageAsset.episode_id == episode_id, ImageAsset.asset_type == "scene", ) .distinct() .all() ) for r in existing_stills: sid = r[0] if not sid: continue primary = self._db.query(ImageAsset).filter( ImageAsset.project_id == self._project_id, ImageAsset.still_id == sid, ImageAsset.asset_type == "scene", ImageAsset.is_primary == 1, ).first() if primary and Path(primary.file_path).exists(): done.add(sid) # 체크포인트 병합 for cp_id in scene_cp.get_completed_ids(): cp_data = scene_cp._data.get("completed", {}).get(cp_id, {}) cp_path = cp_data.get("primary_path", "") if cp_path and Path(cp_path).exists(): done.add(cp_id) return done def delete_existing_scene_assets(self, episode_id: str) -> None: """Hard-fenced: 씬 이미지 일괄 DELETE 절대 금지. 절대 규칙 (feedback_never_delete_images): 이미지 step 재실행 시 기존 이미지 절대 삭제 금지 — 이전/새 이미지 비교 필수. 대안: - mode='resume' + force: UPDATE is_primary=0 으로 row 보존, 새 row 가 is_primary=1. PNG 파일도 그대로 유지. Orphan 정리는 별도 메서드 ``delete_orphan_scene_assets`` 사용 (scene_still 부모 row 가 사라진 경우만 — 분석 단계 재실행 시). """ raise RuntimeError( "delete_existing_scene_assets is hard-fenced (feedback_never_delete_images). " "Image step re-run MUST preserve existing rows via is_primary=0 pattern. " "For analysis-step orphan cleanup, use delete_orphan_scene_assets()." ) def delete_orphan_scene_assets(self, episode_id: str) -> int: """Orphan 정리: scene_still 부모 row 가 없는 image_asset 만 삭제. 분석 step 재실행 (scene_save/shot_extract/entity_merge 등) 으로 scene_still 의 still_id 가 reshuffle 되어 image_asset.still_id 가 존재하지 않는 still 을 가리키게 된 경우 호출. 부모 still 이 살아 있는 image_asset 은 절대 건드리지 않음. Returns: 삭제된 orphan row 수. """ from sqlalchemy import select valid_still_ids = select(SceneStill.id).where( SceneStill.episode_id == episode_id ) orphan_q = self._db.query(ImageAsset).filter( ImageAsset.project_id == self._project_id, ImageAsset.episode_id == episode_id, ImageAsset.asset_type == "scene", ImageAsset.still_id.isnot(None), ~ImageAsset.still_id.in_(valid_still_ids), ) count = orphan_q.delete(synchronize_session="fetch") self._db.commit() if count: logger.info( "delete_orphan_scene_assets: removed %d orphan rows (project=%s episode=%s)", count, self._project_id, episode_id, ) return count def delete_orphan_entity_assets(self) -> int: """Orphan 정리: entity_canon 부모 row 가 없는 reference/composite 자산 삭제. Entity 분석 step 재실행 (entity_extract/entity_merge 등) 으로 entity_id 가 reshuffle 되어 image_asset.entity_id 가 존재하지 않는 entity 를 가리키는 경우 호출. 부모 entity 가 살아있는 image_asset 은 보존. 프로젝트 범위 (episode 무관) — entity 는 프로젝트 전체에서 공유. Returns: 삭제된 orphan row 수. """ from sqlalchemy import select from app.models.project import EntityCanon valid_entity_ids = select(EntityCanon.id).where( EntityCanon.project_id == self._project_id ) orphan_q = self._db.query(ImageAsset).filter( ImageAsset.project_id == self._project_id, ImageAsset.asset_type.in_(["reference", "composite"]), ImageAsset.entity_id.isnot(None), ~ImageAsset.entity_id.in_(valid_entity_ids), ) count = orphan_q.delete(synchronize_session="fetch") self._db.commit() if count: logger.info( "delete_orphan_entity_assets: removed %d orphan rows (project=%s)", count, self._project_id, ) return count def save_single_scene_asset( self, scene_result: Dict[str, Any], lineage: Dict[str, Any], ) -> ImageAsset: """Save a single scene ImageAsset (used by generate_single_scene_image). scene_result fields: id, asset_type, entity_id, still_id, episode_id, file_path, prompt_used, generation_model, width, height, status, review_notes, created_at, plus optional (sanitization_strategy, sanitization_note, original_prompt). lineage fields: prompt_type, code_version, prompt_file_version, reference_image_ids. `auto_set_primary` promotes this asset to is_primary=1 if no other primary exists for the same still. Commit is included. Returns the saved ImageAsset so caller can call image_to_dict(). """ asset = ImageAsset( id=scene_result["id"], project_id=self._project_id, asset_type=scene_result["asset_type"], entity_id=scene_result["entity_id"], still_id=scene_result["still_id"], episode_id=scene_result["episode_id"], file_path=to_relative_image_path(scene_result["file_path"]), prompt_used=scene_result["prompt_used"], generation_model=scene_result["generation_model"], width=scene_result["width"], height=scene_result["height"], status=scene_result["status"], review_notes=scene_result["review_notes"], sanitization_strategy=scene_result.get("sanitization_strategy"), original_prompt=scene_result.get("original_prompt"), sanitization_note=scene_result.get("sanitization_note"), prompt_type=lineage["prompt_type"], code_version=lineage["code_version"], prompt_file_version=lineage["prompt_file_version"], reference_image_ids=lineage["reference_image_ids"], created_at=scene_result["created_at"], ) self._db.add(asset) # P0 (2026-07-01): 단건 재생성도 실제 첨부 UUID lineage 영속화(배치 # save_scene_variations 와 동일 헬퍼). single_scene_image_gen operation_type. _input_ids, _gen_call_id, _meta = self._build_lineage_annotation( scene_result, scene_result.get("episode_id"), "single_scene_image_gen", ) if _gen_call_id: asset.generation_call_id = _gen_call_id annotate_generated_asset( asset, pipeline_role="scene_still", stage="scene_image", input_image_ids=(_input_ids or None), pipeline_metadata=_meta, ) auto_set_primary(self._db, self._project_id, asset) self._db.commit() return asset def build_resume_state( self, stills: List[Dict[str, Any]], already_done_stills: set, entity_lookup: Dict[str, Dict[str, Any]], fallback_location_uuid_by_scene: Optional[Dict[int, str]] = None, ) -> Dict[str, Any]: """Restore per-still resume state from existing scene ImageAssets. W5 F22 Phase B.13 (2026-04-22): scene_image_service.generate_images의 resume 복원 루프를 이관. 각 already_done still에 대해 primary asset (없으면 최신) 조회 + 파일 존재 확인 후 path map들과 location history map을 populate한다. 2026-04-27: sd_shot_path_map 반환 제거 (downstream 미사용). Returns a dict with 4 populated maps: - scene_paths_by_index: {scene_index → Path} - scene_paths_by_index_by_id: {still_id → Path} - scene_results_by_index: {scene_index → dict(id/file_path/...)} - location_scene_history: {location_id → (bytes, still_data)} already_done_stills가 비어 있으면 4개 빈 dict를 반환. """ scene_paths_by_index: Dict[int, Path] = {} scene_paths_by_index_by_id: Dict[str, Path] = {} scene_results_by_index: Dict[int, Dict[str, Any]] = {} location_scene_history: Dict[str, Any] = {} if not already_done_stills: return { "scene_paths_by_index": scene_paths_by_index, "scene_paths_by_index_by_id": scene_paths_by_index_by_id, "scene_results_by_index": scene_results_by_index, "location_scene_history": location_scene_history, } for si, still_data in enumerate(stills): still_id = still_data.get("id") if still_id not in already_done_stills: continue # DB에서 기존 씬 이미지 조회 (v5: primary 우선, 없으면 최신) existing_asset = ( self._db.query(ImageAsset) .filter( ImageAsset.project_id == self._project_id, ImageAsset.still_id == still_id, ImageAsset.asset_type == "scene", ImageAsset.is_primary == 1, ) .order_by(ImageAsset.created_at.desc()) .first() ) if not existing_asset: existing_asset = ( self._db.query(ImageAsset) .filter( ImageAsset.project_id == self._project_id, ImageAsset.still_id == still_id, ImageAsset.asset_type == "scene", ) .order_by(ImageAsset.created_at.desc()) .first() ) if not existing_asset: continue # Fix-B (2026-06-11 fresh full E2E 발견#8, Codex APPROVED_NARROW): # ImageAsset.file_path 는 상대('projects/...') 저장 — cwd(backend) # 기준 exists() 는 항상 False 라 단일 still 재생성 경로에서 # scene_paths/dep bytes 가 silent 하게 빈 상태였다 (batch run 은 # in-memory 누적이라 가려짐). 표준 helper 로 절대화. from app.core.file_paths import resolve_image_path fp = resolve_image_path(existing_asset.file_path) if not fp or not fp.exists(): continue scene_paths_by_index[si] = fp scene_paths_by_index_by_id[still_id] = fp scene_results_by_index[si] = { "id": existing_asset.id, "file_path": existing_asset.file_path, "prompt_used": existing_asset.prompt_used or "", "status": existing_asset.status, "review_notes": existing_asset.review_notes or "", } # location history 복원 try: vis_ids = json.loads(still_data.get("visible_entities_json", "[]")) except json.JSONDecodeError: vis_ids = [] _loc_recorded = False for v in vis_ids: if isinstance(v, dict): eid = v.get("id") or v.get("entity_id", "") if eid in entity_lookup and entity_lookup[eid].get("entity_type") == "location": location_scene_history[eid] = (fp.read_bytes(), still_data) _loc_recorded = True # B (2026-07-02): VE 에 location 이 없는 still — scene_director # primary_location UUID(호출자 계산, outdoor 한정 맵) 로 복원해 # 단건/resume 경로의 outdoor prev-frame lookup 소스를 살린다. # 파라미터 None(기존 caller) = byte-identical. _fb_si = still_data.get("scene_index") if (not _loc_recorded and fallback_location_uuid_by_scene and isinstance(_fb_si, int)): _fb_u = fallback_location_uuid_by_scene.get(_fb_si) if _fb_u: location_scene_history[_fb_u] = (fp.read_bytes(), still_data) return { "scene_paths_by_index": scene_paths_by_index, "scene_paths_by_index_by_id": scene_paths_by_index_by_id, "scene_results_by_index": scene_results_by_index, "location_scene_history": location_scene_history, } def save_fal_angle_asset( self, *, asset_id: str, file_path: str, still_id: str, episode_id: str, prompt_used: str, source_image_id: str, created_at: str, ) -> None: """Save a fal.ai angle-edited scene asset (single insert + commit). asset_type/generation_model/variant_type/status/is_primary are fixed to the fal.ai angle pipeline contract. """ asset = ImageAsset( id=asset_id, project_id=self._project_id, asset_type="scene", still_id=still_id, episode_id=episode_id, file_path=to_relative_image_path(file_path), prompt_used=prompt_used, generation_model="fal-ai/qwen-image-edit-2511-multiple-angles", status="generated", is_primary=0, variant_type="angle_fal", source_image_id=source_image_id, created_at=created_at, ) self._db.add(asset) self._db.commit() def resolve_world_guide( self, episode_id: str, episode: Any, entities: List[Dict[str, Any]], stills: List[Dict[str, Any]], mode: str, provenance: Any, language: str, ) -> Dict[str, Any]: """Resolve world guide: hash-based reuse or regenerate + save. W5 F22 Phase B.19 (2026-04-22): scene_image_service.generate_images의 WorldGuide 블록을 이관. 처리: 1. fulltext[:500] + len(entities) + len(stills) 해시 계산 2. mode != "full"이면 기존 WorldGuide 조회, source_hash 일치 시 재사용 3. 불일치/없음/force 이면 OpenAIClient + WorldGuideGenerator로 재생성 후 DB 저장 (flush만) 4. ProjectSettings.style_rules_json이 있으면 world_guide에 주입 - WorldGuide의 style_rules.must_maintain이 있으면 project_style로 추가 - 없으면 style_rules 자리에 그대로 삽입 commit은 하지 않음 (caller의 후속 DB 작업과 함께 flush). """ fulltext = episode.fulltext or "" wg_hash = hashlib.md5( f"{fulltext[:500]}:{len(entities)}:{len(stills)}".encode() ).hexdigest() existing_wg = None if mode != "full": existing_wg = ( self._db.query(WorldGuide) .filter( WorldGuide.project_id == self._project_id, WorldGuide.episode_id == episode_id, ) .order_by(WorldGuide.created_at.desc()) .first() ) if existing_wg and existing_wg.source_hash == wg_hash: world_guide = json.loads(existing_wg.guide_json) else: openai_client = OpenAIClient() wg_gen = WorldGuideGenerator(llm_client=openai_client) with provenance.start_operation( "image_generation", "world_guide_generator", episode_id=episode_id, ) as op: op.set_input({ "fulltext_chars": len(episode.fulltext), "entities": len(entities), "stills": len(stills), }) world_guide = wg_gen.generate( fulltext=episode.fulltext, language=language, source_file=episode.source_filename or "episode", entities=entities, stills=stills, ) op.set_output({ "world_setting_summary_len": len( world_guide.get("world_setting_summary", "") ), }) wg_record = WorldGuide( id=str(uuid.uuid4()), project_id=self._project_id, episode_id=episode_id, guide_json=json.dumps(world_guide, ensure_ascii=False), source_hash=wg_hash, created_at=datetime.now(timezone.utc).isoformat(), ) self._db.add(wg_record) self._db.flush() proj_settings = ( self._db.query(ProjectSettings) .filter(ProjectSettings.project_id == self._project_id) .first() ) if proj_settings and proj_settings.style_rules_json: proj_style = json.loads(proj_settings.style_rules_json) existing_sr = world_guide.get("style_rules", {}) if isinstance(existing_sr, dict) and existing_sr.get("must_maintain"): world_guide["project_style"] = proj_style else: world_guide["style_rules"] = proj_style return world_guide def validate_episode_ready(self, episode_id: str) -> None: """pipeline_gate 검사 + 선택된 씬 1개 이상 존재 검증. W5 F22 Phase B.22.11 (2026-04-23): scene_image_service.generate_images의 pipeline_gate + scene_count 검증 블록(~18 LOC)을 이관. 처리: 1. app.core.pipeline_gate.check_scene_images_ready (lazy import) 2. SceneStill count (project+episode+is_selected+still_index>=0+status!=stale) 3. scene_count == 0이면 image.no_scenes AppError (status 400) 예외 메시지는 원본 리터럴 '씬이 없습니다. 분석을 먼저 완료하세요.' 보존 (i18n 미적용 블록). """ from app.core.pipeline_gate import check_scene_images_ready check_scene_images_ready(self._db, self._project_id, episode_id) scene_count = self._db.query(SceneStill).filter( SceneStill.project_id == self._project_id, SceneStill.episode_id == episode_id, SceneStill.is_selected == True, # noqa: E712 SceneStill.still_index >= 0, SceneStill.status != "stale", ).count() if scene_count == 0: raise AppError( code="image.no_scenes", message="씬이 없습니다. 분석을 먼저 완료하세요.", status_code=400, ) def mark_asset_as_original(self, asset_id: str) -> None: """ImageAsset.variant_type='original'로 마킹 + flush. W5 F22 Phase B.24.3 (2026-04-23): scene_image_service.generate_scene_with_variations의 original asset 마킹 블록(~9 LOC)을 이관. 처리: 1. ImageAsset.id == asset_id 조회 (project_id 필터 없음 — 원본 유지) 2. 있으면 variant_type='original' + self._db.flush() 3. 없으면 no-op (원본도 동일) commit은 호출자 책임 (원본은 자체 commit 없이 flush만 — 이후 caller가 commit). """ asset = ( self._db.query(ImageAsset) .filter(ImageAsset.id == asset_id) .first() ) if asset: asset.variant_type = "original" self._db.flush() def fetch_still_for_generation(self, still_id: str) -> SceneStill: """single scene 생성을 위한 still 조회 + 파이프라인 준비성 검증. W5 F22 Phase B.23.1 (2026-04-23): scene_image_service.generate_single_scene_image의 입력 검증 블록(~28 LOC)을 이관. 처리 순서 (리터럴 보존): 1. still_check 쿼리 (project_id + still_id) 2. still_check가 있으면 check_scene_images_ready 호출 (pipeline_gate) 3. gemini_api_key 또는 key_pool.count가 없으면 image.gemini_key_missing AppError 4. 본 still 쿼리 재실행 (원본 중복 유지 — literal lift) 5. still 없으면 still.not_found AppError (status 404) 원본 중복 쿼리 패턴은 변경하지 않음 (검증 경로와 사용 경로가 별개로 실행). """ still_check = ( self._db.query(SceneStill) .filter(SceneStill.id == still_id, SceneStill.project_id == self._project_id) .first() ) if still_check: from app.core.pipeline_gate import check_scene_images_ready check_scene_images_ready(self._db, self._project_id, still_check.episode_id) 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, ) return still def load_episode_entity_dicts(self, episode_id: str) -> List[Dict[str, Any]]: """에피소드에 연결된 EntityCanon을 dict 리스트로 로드. W5 F22 Phase B.22.10 (2026-04-23): scene_image_service.generate_images의 entity ORM → dict 변환 블록(~28 LOC)을 이관. 처리: 1. EntityEpisodeLink 조회 (project_id + episode_id) 2. canon_ids 수집. 빈 리스트면 entities_orm=[] (2차 쿼리 스킵) 3. EntityCanon.id.in_(canon_ids) 로드 4. dict 변환: id/name/short_id/entity_type/description/stable_traits/t2i_prompt - short_id/description/t2i_prompt: None → "" fallback - stable_traits: None → "{}" fallback """ # ★보류(shelved) 제외 — 정본 하나 (2026-09-04). from app.core.entity_identity import active_episode_canon_ids canon_ids = active_episode_canon_ids( self._db, self._project_id, episode_id) entities_orm = ( self._db.query(EntityCanon) .filter(EntityCanon.id.in_(canon_ids)) .all() ) if canon_ids else [] return [ { "id": e.id, "name": e.name, "short_id": e.short_id or "", "entity_type": e.entity_type, "description": e.description or "", "stable_traits": e.stable_traits or "{}", "t2i_prompt": e.t2i_prompt or "", } for e in entities_orm ] def load_episode_still_dicts( self, episode_id: str, ) -> Tuple[List[Dict[str, Any]], List[SceneStill]]: """에피소드의 선택된 SceneStill을 dict + ORM tuple로 로드. W5 F22 Phase B.22.10 (2026-04-23): scene_image_service.generate_images의 still ORM → dict 변환 블록(~29 LOC)을 이관. 필터: - project_id + episode_id - is_selected == True (shot-more: 선택된 샷만) - still_index >= 0 - status != "stale" 정렬: still_index ASC dict 변환 키: id / still_index / scene_index / shot_index / screenplay_scene_heading / beat_title / still_frame_prompt / camera_json / lighting_json / visible_entities_json / dependent_scene_id None fallback: - screenplay_scene_heading / beat_title / still_frame_prompt: "" - camera_json / lighting_json: "{}" - visible_entities_json: "[]" - dependent_scene_id: None 그대로 Returns: (stills_dict_list, stills_orm_list) — populate_t2i_prompts가 ORM 리스트 입력을 요구하므로 tuple로 반환. """ stills_orm = ( self._db.query(SceneStill) .filter( SceneStill.project_id == self._project_id, SceneStill.episode_id == episode_id, SceneStill.is_selected == True, # noqa: E712 SceneStill.still_index >= 0, SceneStill.status != "stale", ) .order_by(SceneStill.still_index) .all() ) stills = [ { "id": s.id, "still_index": s.still_index, "scene_index": s.scene_index, "shot_index": s.shot_index, "screenplay_scene_heading": s.screenplay_scene_heading or "", "beat_title": s.beat_title or "", "still_frame_prompt": s.still_frame_prompt or "", "camera_json": s.camera_json or "{}", "lighting_json": s.lighting_json or "{}", "visible_entities_json": s.visible_entities_json or "[]", "dependent_scene_id": s.dependent_scene_id, } for s in stills_orm ] return stills, stills_orm