"""SceneStillWriter — PlannedStill 리스트 → scene_still DB UPSERT. 책임: 1) 기존 scene_still row 로드 → UPSERT (ID 보존) 2) 계획에 없는 기존 row는 stale 마킹 (이미지 연결 보존 — DELETE 금지) 3) shot_dependency 또는 scene_dependency(legacy)에서 dependent_scene_id 갱신 4) scene_summary / shot_type 컬럼 전파 UPSERT 정책 (baseline 동작): - key = (scene_index, shot_index) — shot-based (v4/unselected) - 기존 row가 shot_index IS NULL이면 key = (still_index, None) — legacy 호환 """ from __future__ import annotations import logging import uuid from typing import Any, Dict, List, Optional, Tuple from sqlalchemy import text as sql_text from sqlalchemy.orm import Session as OrmSession from app.services.checkpoint_sync._scene_still_contracts import ( CheckpointBundle, PlannedStill, StillKey, ) class SceneStillWriter: def __init__(self, db: OrmSession, project_id: str, episode_id: str, now: str): self.db = db self.project_id = project_id self.episode_id = episode_id self.now = now self.logger = logging.getLogger(__name__) def write(self, planned: List[PlannedStill], bundle: CheckpointBundle) -> int: from app.models.project import SceneStill existing_by_key = self._load_existing(SceneStill) # shot_dep_completed이면 기존 dependent_scene_id 초기화 (전 씬 재계산 원칙). # Codex P1-2: partial scene_detail 기반 sync 시 dep 초기화 skip — 미포함 씬의 # dep을 누락으로 보고 NULL로 초기화하면 그 씬의 dep 정보가 손실됨. if bundle.shot_dep_completed and not bundle.sd_partial: self.db.execute(sql_text( "UPDATE scene_still SET dependent_scene_id = NULL " "WHERE project_id = :pid AND episode_id = :eid" ), {"pid": self.project_id, "eid": self.episode_id}) new_keys: set = set() shot_key_to_id: Dict[StillKey, str] = {} for p in planned: new_keys.add(p.key) if p.key in existing_by_key: ss = existing_by_key[p.key] ss.still_index = p.still_index ss.scene_index = p.scene_index # Codex P3-1 High: baseline `cb4e832`는 UPDATE 경로에서 # t2i_composer_version을 덮어쓰지 않음 (INSERT 시에만 설정). # shot-more에서 선택↔미선택 토글 시 버전이 바뀌는 회귀 방지. for col, val in p.columns.items(): setattr(ss, col, val) shot_key_to_id[p.key] = ss.id else: new_id = str(uuid.uuid4()) self.db.add(SceneStill( id=new_id, project_id=self.project_id, episode_id=self.episode_id, still_index=p.still_index, scene_index=p.scene_index, t2i_composer_version=p.t2i_composer_version, created_at=self.now, **p.columns, )) shot_key_to_id[p.key] = new_id # 계획에 없는 기존 row는 stale — 이미지 연결 보존 위해 삭제 금지. # Codex P1-2: partial scene_detail이면 stale 마킹 skip — 부분 cp의 missing # 씬을 정상 누락으로 보고 stale 처리하면 다른 씬의 row 전부 손상 (validate 거부). if not bundle.sd_partial: for key, ss in existing_by_key.items(): if key not in new_keys: ss.still_index = -1 ss.status = "stale" elif existing_by_key: self.logger.info( "scene_still sync: partial scene_detail — stale marking deferred " "(awaiting completed checkpoint)" ) # dep UPDATE(raw SQL)는 INSERT/stale 결과가 DB에 반영된 후 실행돼야 함. # 오리지널 baseline은 autoflush에 의존했으나 테스트 격리성(SQLite)을 위해 명시적 flush. self.db.flush() # dependent_scene_id 갱신 — shot_deps 우선, 없으면 legacy if bundle.shot_deps: self._apply_shot_deps(bundle.shot_deps, shot_key_to_id) else: self._apply_legacy_deps(bundle.scenes, bundle.dep_map, shot_key_to_id) if planned: self.logger.info( "Synced %d scene_stills (shot-based) to DB", len(planned) ) # scene_summary + shot_type 전파 — plan과 독립 self._sync_scene_summary(bundle.scene_summaries) self._sync_shot_type(bundle.shot_cine_shots, bundle.scene_cine_scenes) return len(planned) def _load_existing(self, SceneStill) -> Dict[StillKey, Any]: rows = self.db.query(SceneStill).filter( SceneStill.project_id == self.project_id, SceneStill.episode_id == self.episode_id, ).all() out: Dict[StillKey, Any] = {} for ss in rows: if ss.scene_index is not None and ss.shot_index is not None: out[(ss.scene_index, ss.shot_index)] = ss else: out[(ss.still_index, None)] = ss return out def _apply_shot_deps( self, shot_deps: List[Dict[str, Any]], shot_key_to_id: Dict[StillKey, str], ) -> None: for dep in shot_deps: loc_refs = dep.get("location_refs", []) if not loc_refs: continue ref = loc_refs[0] dep_still = shot_key_to_id.get( (ref.get("scene_index"), ref.get("shot_index")) ) still_id = shot_key_to_id.get( (dep.get("scene_index"), dep.get("shot_index")) ) if dep_still and still_id: self.db.execute(sql_text( "UPDATE scene_still SET dependent_scene_id = :dep WHERE id = :sid" ), {"dep": dep_still, "sid": still_id}) def _apply_legacy_deps( self, scenes: List[Dict[str, Any]], dep_map: Dict[str, Dict[str, Any]], shot_key_to_id: Dict[StillKey, str], ) -> None: for s in scenes: si = s.get("scene_index", 0) dep_info = dep_map.get(str(si), dep_map.get(si, {})) if not dep_info: continue loc_refs = dep_info.get("location_refs", []) prev_raw = loc_refs[0] if loc_refs else dep_info.get("prev_ref") prev_ref = prev_raw.get("scene_index") if isinstance(prev_raw, dict) else prev_raw dep_id = shot_key_to_id.get((prev_ref, None)) if prev_ref is not None else None still_id = shot_key_to_id.get((si, None)) if dep_id and still_id: self.db.execute(sql_text( "UPDATE scene_still SET dependent_scene_id = :dep WHERE id = :sid" ), {"dep": dep_id, "sid": still_id}) def _sync_scene_summary(self, summaries: List[Dict[str, Any]]) -> None: if not summaries: return for item in summaries: si = item.get("scene_index") text = item.get("scene_summary", "") self.db.execute(sql_text( "UPDATE scene_still SET scene_summary = :summary " "WHERE project_id = :pid AND episode_id = :eid AND scene_index = :si" ), {"pid": self.project_id, "eid": self.episode_id, "si": si, "summary": text}) self.db.flush() self.logger.info("Synced scene_summary from checkpoint") def _sync_shot_type( self, shot_cine_shots: List[Dict[str, Any]], scene_cine_scenes: List[Dict[str, Any]], ) -> None: if shot_cine_shots: for sc in shot_cine_shots: si = sc.get("scene_index") shot_idx = sc.get("shot_index") t1 = sc.get("technique_1", {}).get("name", "") t2 = sc.get("technique_2", {}).get("name", "") self.db.execute(sql_text( "UPDATE scene_still SET shot_type_1 = :s1, shot_type_2 = :s2 " "WHERE project_id = :pid AND episode_id = :eid " "AND scene_index = :si AND shot_index = :shot_idx" ), {"pid": self.project_id, "eid": self.episode_id, "si": si, "shot_idx": shot_idx, "s1": t1, "s2": t2}) self.db.flush() self.logger.info("Synced shot_type from shot_cinematography checkpoint") return if scene_cine_scenes: for cs in scene_cine_scenes: si = cs.get("scene_index") shots = cs.get("shots", []) s1 = shots[0].get("name", "") if shots else cs.get("shot_1", "") s2 = shots[1].get("name", "") if len(shots) > 1 else cs.get("shot_2", "") self.db.execute(sql_text( "UPDATE scene_still SET shot_type_1 = :s1, shot_type_2 = :s2 " "WHERE project_id = :pid AND episode_id = :eid AND scene_index = :si" ), {"pid": self.project_id, "eid": self.episode_id, "si": si, "s1": s1, "s2": s2}) self.db.flush() self.logger.info("Synced shot_type from scene_cinematography checkpoint (legacy)")