"""ShotSelectionService — shot_selection 체크포인트 토글 로직. Phase 2.2 (architecture-refactor-final/02-final-roadmap.md §Phase 2.2). 기존 `steps.py:toggle_shot_selection`에서 이관. 트랜잭션 순서 + 원자 쓰기 + warnings 응답 보존. """ from __future__ import annotations import hashlib import json import logging from pathlib import Path from typing import Any, Dict, List from sqlalchemy import text as sql_text from sqlalchemy.orm import Session as OrmSession from app.core.checkpoint_io import atomic_write_json, read_json_safe # noqa: F401 — read_json_safe는 resume_sensitive 체크포인트에서만 from app.core.config import settings from app.core.errors import AppError from app.core.step_catalog import ( get_all_downstream_recursive, get_resume_sensitive_step_ids, ) class ShotSelectionService: def __init__(self, db: OrmSession, project_id: str, episode_id: str): self.db = db self.project_id = project_id self.episode_id = episode_id self.logger = logging.getLogger(__name__) @property def _cp_root(self) -> Path: return Path(settings.projects_dir) / self.project_id / "checkpoints" / "episodes" / self.episode_id def toggle(self, scene_index: int, shot_index: int) -> Dict[str, Any]: """Shot 선택/해제 토글. 순서: (1) advisory lock → (2) 체크포인트 로드 & validate → (3) cap 검증 → (4) DB 커밋 (lock 자동 해제) → (5) 파일 쓰기. 원자 쓰기는 checkpoint_io.atomic_write_json. baseline(json.loads 직접 호출) 동작 보존 — 손상 시 raw JSONDecodeError 전파. """ cp_path = self._cp_root / "shot_selection" / "manifest.json" if not cp_path.exists(): raise AppError(code="step.no_checkpoint", message="shot_selection 체크포인트 없음", status_code=404) # 동시 toggle 직렬화 — transaction-scope advisory lock. # 같은 episode의 동시 요청을 순차 처리하여 read-modify-write 경합 방지. # commit 또는 rollback 시 자동 해제. self._acquire_toggle_lock() cp = json.loads(cp_path.read_text(encoding="utf-8")) scenes = cp.get("data", {}).get("scenes", []) target = next((s for s in scenes if s["scene_index"] == scene_index), None) if not target: raise AppError(code="step.scene_not_found", message=f"씬 {scene_index} 없음", status_code=404) selected: List[int] = list(target.get("selected_shot_indices", [])) is_deselect = shot_index in selected # shot_extract 체크포인트에서 유효한 shot_index 검증 (선택 추가 시에만). # deselect는 항상 허용 — 스냅샷 복원 후 불일치 상태 수정 가능. if not is_deselect: self._validate_shot_index(scene_index, shot_index) # EPISODE_MAX_SHOTS 초과 방지 (add 시에만). 초과 예상이면 400으로 거절 — # toggle은 단일 씬 수정이므로 자동 비례 삭감은 다른 씬 선택을 임의 변경해 위험. if not is_deselect and settings.episode_max_shots > 0: current_total = sum(len(s.get("selected_shot_indices", [])) for s in scenes) projected_total = current_total + 1 if projected_total > settings.episode_max_shots: raise AppError( code="step.episode_max_shots_exceeded", message=( f"에피소드 최대 shot 수({settings.episode_max_shots}) 초과 예상: " f"{projected_total}. 먼저 다른 선택을 해제하세요." ), status_code=400, ) # v4: selected_shots (with reason) 동기화. v3 체크포인트는 selected_shots가 # 없으므로 selected_shot_indices로부터 backfill — 그렇지 않으면 기존 선택이 유실됨. sel_shots: List[Dict[str, Any]] = list(target.get("selected_shots", [])) if not sel_shots and selected: sel_shots = [{"shot_index": i, "reason": "migrated from v3"} for i in selected] if is_deselect: selected.remove(shot_index) sel_shots = [s for s in sel_shots if s.get("shot_index") != shot_index] action = "deselected" else: selected.append(shot_index) selected.sort() if not any(s.get("shot_index") == shot_index for s in sel_shots): sel_shots.append({"shot_index": shot_index, "reason": "user toggled"}) sel_shots.sort(key=lambda s: s.get("shot_index", 0)) action = "selected" target["selected_shot_indices"] = selected target["selected_shots"] = sel_shots target["selected_count"] = len(selected) cp["data"]["total_selected"] = sum(s.get("selected_count", 0) for s in scenes) # 1) DB 업데이트 + downstream stale self._update_db(scene_index, shot_index, is_deselect) downstream = self._mark_downstream_stale() self.db.commit() # 2) DB 커밋 성공 후 파일 쓰기 file_warnings = self._write_checkpoint_files(cp_path, cp, scene_index, shot_index) self.logger.info("Shot selection toggle: invalidated %d downstream steps", len(downstream)) response: Dict[str, Any] = { "ok": True, "action": action, "scene_index": scene_index, "shot_index": shot_index, "selected_shot_indices": selected, "total_selected": cp["data"]["total_selected"], "invalidated_steps": downstream, } if file_warnings: response["warnings"] = file_warnings return response def _acquire_toggle_lock(self) -> None: """Transaction-scope advisory lock — 같은 episode 동시 toggle 직렬화. PostgreSQL pg_advisory_xact_lock. commit/rollback 시 자동 해제. key는 (project_id, episode_id) 해시 기반 signed bigint. """ key_str = f"shot_selection:{self.project_id}:{self.episode_id}" digest = hashlib.sha256(key_str.encode()).digest()[:8] lock_key = int.from_bytes(digest, "big", signed=True) self.db.execute(sql_text("SELECT pg_advisory_xact_lock(:k)"), {"k": lock_key}) def _validate_shot_index(self, scene_index: int, shot_index: int) -> None: shot_extract_path = self._cp_root / "shot_extract" / "manifest.json" if not shot_extract_path.exists(): raise AppError( code="step.no_checkpoint", message="shot_extract 체크포인트 없음 — shot_index 검증 불가", status_code=404, ) # baseline 동작 보존: 손상 manifest는 raw JSONDecodeError 전파 (Codex Phase 2 Item 6). shot_cp = json.loads(shot_extract_path.read_text(encoding="utf-8")) shot_scene = next( (s for s in shot_cp.get("data", {}).get("scenes", []) if s["scene_index"] == scene_index), None, ) valid_indices = {sh["shot_index"] for sh in (shot_scene or {}).get("shots", [])} if shot_index not in valid_indices: raise AppError( code="step.invalid_shot_index", message=f"shot_index {shot_index}은(는) 씬 {scene_index}에 존재하지 않음 (유효: {sorted(valid_indices)})", status_code=400, ) def _update_db(self, scene_index: int, shot_index: int, is_deselect: bool) -> None: upd = self.db.execute(sql_text( "UPDATE scene_still SET is_selected = :is_sel " "WHERE project_id = :pid AND episode_id = :eid " "AND scene_index = :si AND shot_index = :shi" ), { "pid": self.project_id, "eid": self.episode_id, "si": scene_index, "shi": shot_index, "is_sel": not is_deselect, }) if upd.rowcount == 0: self.logger.warning( "Shot selection toggle: no scene_still row for scene=%d shot=%d — " "DB will reflect change after next scene_detail sync", scene_index, shot_index, ) def _mark_downstream_stale(self) -> List[str]: downstream = get_all_downstream_recursive("shot_selection") for sid in downstream: self.db.execute(sql_text( "UPDATE step_run SET status = 'stale', updated_at = NOW() " "WHERE project_id = :pid AND episode_id = :eid AND step_id = :sid " "AND status NOT IN ('pending', 'stale')" ), {"pid": self.project_id, "eid": self.episode_id, "sid": sid}) return downstream def _write_checkpoint_files( self, cp_path: Path, cp: Dict[str, Any], scene_index: int, shot_index: int, ) -> List[str]: """shot_selection manifest 쓰기 + resume_sensitive step의 체크포인트 stale 표시. 파일 쓰기 실패 시 DB는 이미 반영되었으므로 warnings만 수집. """ file_warnings: List[str] = [] try: atomic_write_json(cp_path, cp) resume_sensitive = get_resume_sensitive_step_ids() for sid in resume_sensitive: cp_file = self._cp_root / sid / "manifest.json" if not cp_file.exists(): continue try: cp_data = read_json_safe(cp_file) if cp_data is None: file_warnings.append(f"{sid}: corrupted manifest, skipped") continue cp_data["status"] = "stale" atomic_write_json(cp_file, cp_data) except Exception as exc: self.logger.warning("Failed to mark %s checkpoint stale: %s", sid, exc) file_warnings.append(f"{sid}: {exc}") except Exception as write_exc: self.logger.error( "Shot toggle DB committed but checkpoint file write failed for scene=%d shot=%d: %s", scene_index, shot_index, write_exc, ) file_warnings.append(f"shot_selection: {write_exc}") return file_warnings