from __future__ import annotations import time from pathlib import Path from typing import Literal from fastapi import APIRouter, HTTPException from fastapi.responses import FileResponse from pydantic import BaseModel, Field from capabilities import CapabilityService from data_paths import DataPaths, make_id, read_json_file, write_json_file from media_composer import build_slideshow_video class TimelineItem(BaseModel): id: str title: str source_video_project_id: str | None = None asset_filename: str | None = None duration_seconds: float = 3.0 class PostPreview(BaseModel): id: str filename: str created_at: int duration_seconds: float class PostExport(BaseModel): id: str filename: str created_at: int duration_seconds: float class PostProject(BaseModel): id: str name: str aspect_ratio: Literal["9:16", "16:9"] = "9:16" width: int = 1080 height: int = 1920 source_video_project_id: str | None = None source_audio_project_id: str | None = None audio_mixdown_filename: str | None = None timeline: list[TimelineItem] previews: list[PostPreview] = Field(default_factory=list) exports: list[PostExport] = Field(default_factory=list) created_at: int updated_at: int class PostProjectCreateRequest(BaseModel): name: str = "Post Production Project" source_video_project_id: str | None = None source_audio_project_id: str | None = None class PostProjectUpdateRequest(BaseModel): name: str | None = None aspect_ratio: Literal["9:16", "16:9"] | None = None width: int | None = None height: int | None = None audio_mixdown_filename: str | None = None timeline: list[TimelineItem] | None = None class PostRenderRequest(BaseModel): fps: int = 24 class PostProjectSummary(BaseModel): id: str name: str timeline_count: int updated_at: int def create_post_router( capabilities: CapabilityService, data_paths: DataPaths, ) -> APIRouter: router = APIRouter(prefix="/api/post", tags=["post"]) store = PostProjectStore(data_paths) @router.get("/projects") async def list_projects() -> list[dict[str, object]]: capabilities.require_feature("post_production") return [summary.model_dump() for summary in store.list_projects()] @router.post("/projects") async def create_project(req: PostProjectCreateRequest) -> dict[str, object]: capabilities.require_feature("post_production") return store.create_project(req).model_dump() @router.get("/projects/{project_id}") async def get_project(project_id: str) -> dict[str, object]: capabilities.require_feature("post_production") return store.get_project(project_id).model_dump() @router.patch("/projects/{project_id}") async def update_project( project_id: str, req: PostProjectUpdateRequest, ) -> dict[str, object]: capabilities.require_feature("post_production") project = store.get_project(project_id) payload = project.model_dump() if req.name is not None: payload["name"] = req.name if req.aspect_ratio is not None: payload["aspect_ratio"] = req.aspect_ratio if req.width is not None: payload["width"] = req.width if req.height is not None: payload["height"] = req.height if req.audio_mixdown_filename is not None: payload["audio_mixdown_filename"] = req.audio_mixdown_filename if req.timeline is not None: payload["timeline"] = [item.model_dump() for item in req.timeline] updated = PostProject.model_validate(payload) store.save_project(updated) return updated.model_dump() @router.post("/projects/{project_id}/preview") async def build_preview( project_id: str, req: PostRenderRequest, ) -> dict[str, object]: capabilities.require_feature("post_production") preview = store.build_preview(project_id, req.fps) return preview.model_dump() @router.post("/projects/{project_id}/export") async def export_project( project_id: str, req: PostRenderRequest, ) -> dict[str, object]: capabilities.require_feature("post_production") capabilities.require_feature("video_export") export = store.export_project(project_id, req.fps) return export.model_dump() @router.get("/projects/{project_id}/files/{category}/{filename}") async def get_project_file( project_id: str, category: Literal["previews", "exports"], filename: str, ) -> FileResponse: capabilities.require_feature("post_production") file_path = store.resolve_project_file(project_id, category, filename) if not file_path.exists(): raise HTTPException(status_code=404, detail="Post-production file not found.") return FileResponse(file_path, filename=file_path.name) return router class PostProjectStore: def __init__(self, data_paths: DataPaths) -> None: self.data_paths = data_paths def list_projects(self) -> list[PostProjectSummary]: summaries: list[PostProjectSummary] = [] for manifest_path in sorted(self.data_paths.post_root.glob("*/project.json")): project = PostProject.model_validate(read_json_file(manifest_path, {})) summaries.append( PostProjectSummary( id=project.id, name=project.name, timeline_count=len(project.timeline), updated_at=project.updated_at, ) ) summaries.sort(key=lambda item: item.updated_at, reverse=True) return summaries def create_project(self, req: PostProjectCreateRequest) -> PostProject: now = int(time.time()) timeline = self._build_timeline_from_video(req.source_video_project_id) audio_mixdown_filename = self._latest_audio_mixdown(req.source_audio_project_id) project = PostProject( id=make_id("post"), name=req.name, source_video_project_id=req.source_video_project_id, source_audio_project_id=req.source_audio_project_id, audio_mixdown_filename=audio_mixdown_filename, timeline=timeline, created_at=now, updated_at=now, ) self.save_project(project) return project def get_project(self, project_id: str) -> PostProject: manifest_path = self._project_manifest_path(project_id) payload = read_json_file(manifest_path, None) if payload is None: raise HTTPException(status_code=404, detail="Post-production project not found.") return PostProject.model_validate(payload) def save_project(self, project: PostProject) -> None: project.updated_at = int(time.time()) root = self._project_root(project.id) (root / "previews").mkdir(parents=True, exist_ok=True) (root / "exports").mkdir(parents=True, exist_ok=True) write_json_file(self._project_manifest_path(project.id), project.model_dump()) def build_preview(self, project_id: str, fps: int) -> PostPreview: project = self.get_project(project_id) preview = PostPreview( id=make_id("preview"), filename=f"{make_id('preview')}.mp4", created_at=int(time.time()), duration_seconds=round(sum(item.duration_seconds for item in project.timeline), 2), ) preview_path = self.resolve_project_file(project_id, "previews", preview.filename) build_slideshow_video( items=[(self._resolve_timeline_asset(item), item.duration_seconds) for item in project.timeline], output_path=preview_path, width=project.width, height=project.height, fps=fps, audio_path=None, ) project.previews.insert(0, preview) self.save_project(project) return preview def export_project(self, project_id: str, fps: int) -> PostExport: project = self.get_project(project_id) export = PostExport( id=make_id("export"), filename=f"{make_id('master')}.mp4", created_at=int(time.time()), duration_seconds=round(sum(item.duration_seconds for item in project.timeline), 2), ) export_path = self.resolve_project_file(project_id, "exports", export.filename) build_slideshow_video( items=[(self._resolve_timeline_asset(item), item.duration_seconds) for item in project.timeline], output_path=export_path, width=project.width, height=project.height, fps=fps, audio_path=self._resolve_audio_mixdown(project), ) project.exports.insert(0, export) self.save_project(project) return export def resolve_project_file( self, project_id: str, category: Literal["previews", "exports"], filename: str, ) -> Path: return self._project_root(project_id) / category / Path(filename).name def _project_root(self, project_id: str) -> Path: return self.data_paths.post_root / project_id def _project_manifest_path(self, project_id: str) -> Path: return self._project_root(project_id) / "project.json" def _build_timeline_from_video(self, video_project_id: str | None) -> list[TimelineItem]: if not video_project_id: return [ TimelineItem( id=make_id("timeline"), title="Scene 1", duration_seconds=3.0, ) ] manifest = read_json_file(self.data_paths.video_root / video_project_id / "project.json", None) if manifest is None: return [ TimelineItem( id=make_id("timeline"), title="Scene 1", duration_seconds=3.0, ) ] timeline: list[TimelineItem] = [] for scene in manifest.get("scenes", []): timeline.append( TimelineItem( id=make_id("timeline"), title=scene.get("title", "Scene"), source_video_project_id=video_project_id, asset_filename=self._resolve_video_asset_filename(manifest, scene.get("asset_id")), duration_seconds=float(scene.get("duration_seconds", 3.0)), ) ) return timeline or [ TimelineItem( id=make_id("timeline"), title="Scene 1", duration_seconds=3.0, ) ] def _resolve_video_asset_filename(self, manifest: dict[str, object], asset_id: str | None) -> str | None: if not asset_id: return None for asset in manifest.get("assets", []): if asset.get("id") == asset_id: return asset.get("filename") return None def _latest_audio_mixdown(self, audio_project_id: str | None) -> str | None: if not audio_project_id: return None manifest = read_json_file(self.data_paths.audio_root / audio_project_id / "project.json", None) if manifest is None: return None mixdowns = manifest.get("mixdowns", []) if not mixdowns: return None return mixdowns[0].get("filename") def _resolve_timeline_asset(self, item: TimelineItem) -> Path | None: if not item.source_video_project_id or not item.asset_filename: return None asset_path = self.data_paths.video_root / item.source_video_project_id / "assets" / item.asset_filename if asset_path.exists(): return asset_path return None def _resolve_audio_mixdown(self, project: PostProject) -> Path | None: if not project.source_audio_project_id or not project.audio_mixdown_filename: return None audio_path = ( self.data_paths.audio_root / project.source_audio_project_id / "mixdowns" / project.audio_mixdown_filename ) if audio_path.exists(): return audio_path return None