"""파이프라인 Step API — 개별 단계 실행/상태 조회.""" import json import logging import re from pathlib import Path logger = logging.getLogger(__name__) from typing import Any, Dict, List, Optional from fastapi import APIRouter, Depends, Query from sqlalchemy import text from sqlalchemy.orm import Session as OrmSession from app.api.deps import api_endpoint, get_db, get_current_user, verify_project_access from app.core.checkpoint_io import atomic_write_json from app.core.errors import AppError from app.core.step_catalog import ( contains as _step_contains, get_all_downstream_recursive as catalog_get_all_downstream_recursive, get_consumers_of, get_manifest_dict, get_ordered_entries, get_resume_sensitive_step_ids, ) from app.models.catalog import UserAccount router = APIRouter(prefix="/api/v1/projects/{project_id}/episodes/{episode_id}/steps", tags=["steps"]) def _build_opik_context(db: OrmSession, project_id: str, episode_id: str) -> dict: """Opik thread 그룹핑용 context 생성 — dispatch_service로 위임.""" from app.services.analysis_dispatch_service import build_opik_context return build_opik_context(db, project_id, episode_id) @router.get("") def get_all_steps( episode_id: str, project_id: str = Depends(verify_project_access), db: OrmSession = Depends(get_db), current_user: UserAccount = Depends(get_current_user), ): """전체 단계 상태 + DAG 의존성 조회. W1-F11: 계산 로직은 `step_readmodel_service.get_all_steps_view`로 이관. """ from app.services.step_readmodel_service import get_all_steps_view return {"steps": get_all_steps_view(db, project_id, episode_id)} @router.get("/{step_id}/result") def get_step_result( episode_id: str, step_id: str, project_id: str = Depends(verify_project_access), db: OrmSession = Depends(get_db), current_user: UserAccount = Depends(get_current_user), ): """개별 단계 결과 조회 (체크포인트 데이터).""" import json from pathlib import Path from app.core.config import settings if not _step_contains(step_id): raise AppError(code="step.not_found", message=f"알 수 없는 단계: {step_id}", status_code=404) cp_path = ( Path(settings.projects_dir) / project_id / "checkpoints" / "episodes" / episode_id / step_id / "manifest.json" ) if not cp_path.exists(): return {"step_id": step_id, "status": "no_data", "result": None} try: data = json.loads(cp_path.read_text(encoding="utf-8")) return {"step_id": step_id, "status": data.get("status", "unknown"), "result": data} except Exception as exc: raise AppError(code="step.read_error", message=str(exc), status_code=500) @router.post("/run-all") def run_all_steps( episode_id: str, mode: str = Query("resume", pattern="^(resume|force)$"), category: str = Query("analysis", pattern="^(analysis|image|all)$"), image_call_cap: Optional[int] = Query( None, ge=0, description=( "W20E5: hard cap on provider image-generation calls for this " "run. Required (>= 1) when category in {image, all} and " "background_render_reference_mode == 'shot_aware_plan'. " "Ignored for category=analysis." ), ), approve_image_generation: bool = Query( False, description=( "W20E5: explicit operator approval to perform real image " "generation. Required for the shot-aware image path." ), ), project_id: str = Depends(verify_project_access), db: OrmSession = Depends(get_db), current_user: UserAccount = Depends(get_current_user), ): """전체 단계 순차 실행. category: analysis | image | all. Phase 4.1: dispatch 로직이 `analysis_dispatch_service`로 이관되어 `/episodes/{id}/analyze` 경로와 구현을 공유함. W20E5: ``image_call_cap`` + ``approve_image_generation`` 두 파라미터는 shot-aware 이미지 단계의 fail-closed 게이트에 연결된다. legacy 렌더 모드에서는 둘 다 옵션 (cap을 명시하면 advisory 캡으로 동작). """ from app.services.analysis_dispatch_service import dispatch_category_run return dispatch_category_run( project_id, episode_id, category, mode, db, image_call_cap=image_call_cap, approve_image_generation=approve_image_generation, ) @router.patch("/shot_selection/toggle") @api_endpoint def toggle_shot_selection( episode_id: str, scene_index: int = Query(...), shot_index: int = Query(...), project_id: str = Depends(verify_project_access), db: OrmSession = Depends(get_db), current_user: UserAccount = Depends(get_current_user), ): """Shot 선택/해제 토글 — Phase 2.2 ShotSelectionService로 위임.""" from app.services.shot_selection_service import ShotSelectionService return ShotSelectionService(db, project_id, episode_id).toggle(scene_index, shot_index) # ───────────────────────────────────────────────────────── # 체크포인트 스냅샷 — 저장 / 목록 / 복원 # (POST 라우트는 반드시 POST /{step_id} 위에 등록) # ───────────────────────────────────────────────────────── def _cp_base(project_id: str, episode_id: str) -> Path: from app.core.config import settings return Path(settings.projects_dir) / project_id / "checkpoints" / "episodes" / episode_id # 타임스탬프 추출용 (prerestore는 호출측에서 문자열 체크로 제외) _SNAP_TS_RE = re.compile( r"^manifest_(\d{8}_\d{6})(?:_[a-zA-Z0-9_-]+)?\.json$" ) @router.get("/snapshots") @api_endpoint def list_snapshots( episode_id: str, project_id: str = Depends(verify_project_access), step_id: Optional[str] = Query(None, description="특정 단계만 조회"), current_user: UserAccount = Depends(get_current_user), db: OrmSession = Depends(get_db), ): """체크포인트 스냅샷 목록 조회 — SnapshotService로 위임.""" from app.services.snapshot_service import SnapshotService return SnapshotService(db, project_id, episode_id).list_versions(step_id=step_id) @router.post("/snapshots") @api_endpoint def create_snapshot( episode_id: str, project_id: str = Depends(verify_project_access), step_id: Optional[str] = Query(None, description="특정 단계만 스냅샷. 없으면 전체."), label: Optional[str] = Query(None, pattern=r"^[a-zA-Z0-9_-]{0,32}$", description="스냅샷 라벨"), current_user: UserAccount = Depends(get_current_user), db: OrmSession = Depends(get_db), ): """현재 체크포인트를 수동 스냅샷으로 저장 — SnapshotService로 위임.""" from app.services.snapshot_service import SnapshotService return SnapshotService(db, project_id, episode_id).save_snapshot(step_id=step_id, label=label) @router.post("/snapshots/restore") @api_endpoint def restore_snapshot( episode_id: str, project_id: str = Depends(verify_project_access), version: str = Query(..., pattern=r"^\d{8}_\d{6}$", description="복원할 버전 (YYYYMMDD_HHMMSS)"), step_id: Optional[str] = Query(None, description="특정 단계만 복원. 없으면 해당 버전의 전체 단계."), current_user: UserAccount = Depends(get_current_user), db: OrmSession = Depends(get_db), ): """특정 버전의 스냅샷으로 체크포인트 복원 — SnapshotService로 위임.""" from app.services.snapshot_service import SnapshotService return SnapshotService(db, project_id, episode_id).restore(version=version, step_id=step_id) @router.post("/scene-detail/redo-shot") def redo_scene_detail_shot_endpoint( episode_id: str, scene_index: int = Query(..., ge=0), shot_index: int = Query(..., ge=0), mode: str = Query("force", pattern="^(force)$"), project_id: str = Depends(verify_project_access), db: OrmSession = Depends(get_db), current_user: UserAccount = Depends(get_current_user), ): """단건 shot scene_detail re-run — 운영 복구 도구. 2026-05-11 사용자 권장안 #2 축소형: scene_detail step 의 atomic-fail 대비. 1 shot 의 LLM 응답이 contract violation 으로 step 전체 abort 된 경우, 본 endpoint 로 그 shot 만 재실행. cp manifest 의 해당 shot 만 replace + checkpoint sync. UI 버튼 X — 운영 호출용. Returns: service.redo_scene_detail_shot 의 dict 그대로. """ from app.services.scene_detail_redo_service import redo_scene_detail_shot return redo_scene_detail_shot( db=db, project_id=project_id, episode_id=episode_id, scene_index=scene_index, shot_index=shot_index, mode=mode, ) @router.post("/{step_id}") def run_step( episode_id: str, step_id: str, mode: str = Query("resume", pattern="^(resume|force)$"), project_id: str = Depends(verify_project_access), db: OrmSession = Depends(get_db), current_user: UserAccount = Depends(get_current_user), ): """개별 단계 실행. mode: resume (기본) | force (재실행). 백그라운드 스레드. W1-F11: 실제 orchestration은 `step_execution_service.start_step`로 이관. route는 Opik context만 빌드한 뒤 서비스에 위임. """ from app.services.step_execution_service import start_step opik_context = _build_opik_context(db, project_id, episode_id) result = start_step( db=db, project_id=project_id, episode_id=episode_id, step_id=step_id, mode=mode, actor_id=current_user.id, opik_context=opik_context, ) # 기존 응답 호환: step_id 키 포함 result.setdefault("step_id", step_id) return result def _sync_t2i_appearance_counts(project_id: str, episode_id: str, db, logger=None, *, commit: bool = True) -> None: """T2I appearance count projection — Phase 2.1 EpisodeProjectionService로 이관. Legacy wrapper: 기존 호출자(entities.py) 호환 유지용. 신규 코드는 `app.services.checkpoint_sync.EpisodeProjectionService.sync_t2i_appearance_counts` 직접 사용. """ from app.services.checkpoint_sync import EpisodeProjectionService EpisodeProjectionService(db, project_id, episode_id).sync_t2i_appearance_counts(commit=commit) def _sync_checkpoints_to_db(project_id: str, episode_id: str, db) -> None: """분석 체크포인트 → DB 동기화 (Phase 2.1 Service 오케스트레이터로 위임). Legacy wrapper: 기존 호출자(image_steps.py, steps.py run_step 등) 호환 유지용. 실제 로직은 `app.services.checkpoint_sync.orchestrate_full_sync`에 5-way 분해됨. """ from app.services.checkpoint_sync import orchestrate_full_sync orchestrate_full_sync(project_id, episode_id, db) def _get_step_runner(step_id, project_id, episode_id, db, project_config, opik_context=None): """Step ID → StepRunner 인스턴스. Legacy wrapper: `app.services.analysis_dispatch_service.get_step_runner`로 위임. 기존 호출자(test_sync_v3.py, test_pipeline_v3_e2e.py, 내부 run_step) 호환 유지. """ from app.services.analysis_dispatch_service import get_step_runner return get_step_runner(step_id, project_id, episode_id, db, project_config, opik_context)