from __future__ import annotations import json import shutil import uuid import zipfile from pathlib import Path from typing import Any from fastapi import Depends, FastAPI, File, Form, Header, HTTPException, Query, UploadFile from fastapi.responses import FileResponse from pydantic import BaseModel, Field from renderer.core.config import Settings from renderer.core.ingest import stage_upload from renderer.core.models import AIReelsRequest, RenderRequest, Scene from renderer.core.security import verify_download_token from renderer.core.utils import safe_filename from renderer.jobs import JobManager from renderer.platform import PlatformProcessor, supported_toolkit_tasks from renderer.scenes import Timeline from renderer.studio import ( ProjectStore, StudioTaskProcessor, add_effect, add_filter, add_keyframe, add_transition, capability_catalog, ) from renderer import RenderEngine from renderer.subtitles import SubtitleEvent, SubtitleGenerator from renderer.templates import ( apply_creative_style, apply_preset, creative_style_metadata, get_creative_style, list_creative_styles, list_platform_profiles, list_presets, list_scene_effects, list_templates, platform_profile_metadata, scene_effect_metadata, ) from renderer.transitions import TransitionBuilder settings = Settings() settings.ensure_dirs() job_manager = JobManager(settings) project_manager = ProjectStore(settings) api = FastAPI(title="Ava2lon Studio AI", version="2.0.0") API_KEY_HEADER = "X-API-Key" def _verify_api_key(api_key: str | None = Header(default=None, alias=API_KEY_HEADER)) -> None: configured = getattr(settings, "api_key", "") if configured and api_key != configured: raise HTTPException(status_code=401, detail="Invalid or missing API key") def _submission_response(job_id: str) -> dict[str, str]: download_url = f"/download/{job_id}" try: record = job_manager.get(job_id) except KeyError: token = None else: token = record.download_token if token: download_url = f"{download_url}?token={token}" return {"job_id": job_id, "status_url": f"/status/{job_id}", "download_url": download_url} class ScenePayload(BaseModel): start: float = Field(ge=0) duration: float = Field(gt=0) media: str caption: str = "" transition: str = "fade" background: str = "blur" layout: str = "fill" effect: str | None = None class RenderPayload(BaseModel): scenes: list[ScenePayload] template: str = "tiktok_classic" preset: str | None = None creative_style: str | None = None platform: str | None = None output_name: str = "render.mp4" voiceover: str | None = None background_music: str | None = None music_volume: float = Field(default=0.316, ge=0, le=2) music_fade_in: float = Field(default=0.0, ge=0) music_fade_out: float = Field(default=0.0, ge=0) music_loop: bool = True music_start: float = Field(default=0.0, ge=0) music_ducking: bool = True voice_volume: float = Field(default=1.0, ge=0, le=2) subtitle_format: str = "ass" auto_subtitles: bool = False subtitle_language: str | None = None whisper_model_size: str | None = None preview: bool = False audio_normalize: bool = False watermark: str | None = None watermark_position: str = "bottom-right" intro: str | None = None outro: str | None = None callback_url: str | None = None export_target: str | None = None priority: int = 0 scheduled_at: float | None = None normalize: bool = True metadata: dict[str, Any] = Field(default_factory=dict) class AIReelsPayload(BaseModel): script: str voiceover: str assets: list[str] template: str = "tiktok_classic" creative_style: str | None = None platform: str | None = None output_name: str = "ai_reel.mp4" background_music: str | None = None music_volume: float = Field(default=0.316, ge=0, le=2) music_fade_in: float = Field(default=0.0, ge=0) music_fade_out: float = Field(default=0.0, ge=0) music_loop: bool = True music_start: float = Field(default=0.0, ge=0) music_ducking: bool = True voice_volume: float = Field(default=1.0, ge=0, le=2) class BatchPayload(BaseModel): jobs: list[RenderPayload] class UploadedAsset(BaseModel): filename: str path: str reference: str kind: str = "other" metadata: dict[str, Any] | None = None class TranscribePayload(BaseModel): audio: str model_size: str | None = None language: str | None = None task: str = "transcribe" beam_size: int = Field(default=5, ge=1, le=10) vad_filter: bool = True word_timestamps: bool = True class SubtitlePayload(BaseModel): events: list[dict[str, Any]] format: str = "srt" template: str = "tiktok_classic" class SceneBuildPayload(BaseModel): script: str assets: list[str] duration: float | None = None transition: str = "fade" creative_style: str | None = None class IngestSourcePayload(BaseModel): url: str type: str | None = None name: str | None = None class IngestPayload(BaseModel): sources: list[IngestSourcePayload] callback_url: str | None = None class AnalyzePayload(BaseModel): media: str transcript: str = "" platform: str | None = None callback_url: str | None = None class ClipsPayload(BaseModel): media: str clips: list[dict[str, Any]] | None = None callback_url: str | None = None class ToolkitPayload(BaseModel): task: str input: str | None = None media: str | None = None output_name: str | None = None params: dict[str, Any] = Field(default_factory=dict) callback_url: str | None = None export_target: str | None = None model_config = {"extra": "allow"} class ThumbnailPayload(BaseModel): media: str text: str = "" timestamp: float | None = None template: str = "bold" callback_url: str | None = None class MetadataPayload(BaseModel): topic: str = "" transcript: str = "" platform: str | None = None callback_url: str | None = None class PublishPayload(BaseModel): media: str | None = None asset: str | None = None title: str | None = None description: str | None = None platforms: list[str] = Field(default_factory=list) platform: str | None = None scheduled_at: str | None = None draft: bool = True callback_url: str | None = None class ProjectPayload(BaseModel): name: str metadata: dict[str, Any] = Field(default_factory=dict) export_settings: dict[str, Any] = Field(default_factory=dict) template: dict[str, Any] | None = None class ProjectSavePayload(BaseModel): project_id: str | None = None project: dict[str, Any] class ProjectAssetPayload(BaseModel): project_id: str asset: dict[str, Any] class TimelineAddPayload(BaseModel): project_id: str track_type: str = "video" track_id: str | None = None item: dict[str, Any] = Field(default_factory=dict) class TimelineOperationPayload(BaseModel): project_id: str operation: str = "drag" item_id: str | None = None params: dict[str, Any] = Field(default_factory=dict) class EffectApplyPayload(BaseModel): project_id: str | None = None target_id: str | None = None item_id: str | None = None effect: str params: dict[str, Any] = Field(default_factory=dict) callback_url: str | None = None class FilterApplyPayload(BaseModel): project_id: str | None = None target_id: str | None = None item_id: str | None = None filter: str params: dict[str, Any] = Field(default_factory=dict) lut: str | None = None callback_url: str | None = None class TransitionAddPayload(BaseModel): project_id: str | None = None from_item_id: str | None = None to_item_id: str | None = None target_id: str | None = None transition: str duration: float = Field(default=0.45, gt=0) params: dict[str, Any] = Field(default_factory=dict) callback_url: str | None = None class KeyframePayload(BaseModel): project_id: str target_id: str property: str time: float = Field(ge=0) value: Any easing: str = "linear" class GenerationPayload(BaseModel): prompt: str | None = None text: str | None = None media: str | None = None provider: str | None = None callback_url: str | None = None export_target: str | None = None params: dict[str, Any] = Field(default_factory=dict) model_config = {"extra": "allow"} class CaptionGeneratePayload(BaseModel): media: str | None = None audio: str | None = None text: str | None = None transcript: str | None = None events: list[dict[str, Any]] | None = None template: str = "capcut" language: str | None = None engine: str = "whisper" word_timestamps: bool = True emoji_insertion: bool = False speaker_detection: bool = False karaoke: bool = True animated: bool = True callback_url: str | None = None model_config = {"extra": "allow"} class AIToolPayload(BaseModel): project_id: str | None = None media: str | None = None transcript: str | None = None text: str | None = None platform: str | None = None callback_url: str | None = None params: dict[str, Any] = Field(default_factory=dict) model_config = {"extra": "allow"} class ProjectRenderPayload(BaseModel): output_name: str = "project_render.mp4" template: str | None = None preset: str | None = None creative_style: str | None = None platform: str | None = None callback_url: str | None = None export_target: str | None = None preview: bool = False normalize: bool = True metadata: dict[str, Any] = Field(default_factory=dict) @api.get("/health") def health() -> dict[str, str]: return {"status": "ok"} @api.get("/monitor") def monitor() -> dict[str, Any]: return _monitor_payload() @api.get("/queue") def queue() -> dict[str, Any]: return job_manager.summary() @api.get("/workers") def workers() -> dict[str, Any]: return { "max_workers": settings.max_workers, "ffmpeg_timeout_seconds": settings.ffmpeg_timeout_seconds, "whisper_device": settings.whisper_device, "whisper_model_size": settings.whisper_model_size, "toolkit_tasks": supported_toolkit_tasks(), } @api.get("/presets") def presets() -> dict[str, Any]: return { "presets": list_presets(), "caption_templates": list_templates(), "platforms": list_platform_profiles(), "creative_styles": list_creative_styles(), "scene_effects": list_scene_effects(), "transitions": TransitionBuilder().list_transitions(), "creative_style_metadata": creative_style_metadata(), "scene_effect_metadata": scene_effect_metadata(), "toolkit_tasks": supported_toolkit_tasks(), } @api.get("/platforms") def platforms() -> dict[str, dict[str, Any]]: return {"platforms": platform_profile_metadata()} @api.get("/toolkit/tasks") def toolkit_tasks() -> dict[str, list[str]]: return {"tasks": supported_toolkit_tasks()} @api.get("/capabilities") def capabilities() -> dict[str, Any]: catalog = capability_catalog() catalog["runtime"] = { "toolkit_tasks": supported_toolkit_tasks(), "caption_templates": list_templates(), "creative_styles": list_creative_styles(), "scene_effects": list_scene_effects(), "platforms": list_platform_profiles(), "transitions": TransitionBuilder().list_transitions(), } return catalog @api.get("/effects") def effects() -> dict[str, Any]: catalog = capability_catalog() return {"effects": catalog["effects"], "scene_effects": scene_effect_metadata()} @api.get("/filters") def filters() -> dict[str, Any]: return {"filters": capability_catalog()["filters"]} @api.get("/transitions") def transitions() -> dict[str, Any]: return {"families": capability_catalog()["transitions"], "ffmpeg": TransitionBuilder().list_transitions()} @api.get("/templates/catalog") def templates_catalog() -> dict[str, Any]: return { "categories": capability_catalog()["templates"], "caption_templates": list_templates(), "render_presets": list_presets(), "creative_styles": creative_style_metadata(), } @api.get("/projects") def list_projects() -> dict[str, list[dict[str, Any]]]: return {"projects": project_manager.list()} @api.post("/projects", dependencies=[Depends(_verify_api_key)]) def create_project(payload: ProjectPayload) -> dict[str, Any]: project = project_manager.create(payload.name, metadata=payload.metadata, template=payload.template) if payload.export_settings: project["export_settings"].update(payload.export_settings) project = project_manager.save(project["id"], project) return {"project": project} @api.post("/project/create", dependencies=[Depends(_verify_api_key)]) def project_create(payload: ProjectPayload) -> dict[str, Any]: return create_project(payload) @api.get("/project/{project_id}") def project_get(project_id: str) -> dict[str, Any]: try: return {"project": project_manager.get(project_id)} except KeyError as exc: raise HTTPException(status_code=404, detail="Project not found") from exc @api.post("/project/save", dependencies=[Depends(_verify_api_key)]) def project_save(payload: ProjectSavePayload) -> dict[str, Any]: project_id = payload.project_id or str(payload.project.get("id") or safe_filename(payload.project.get("name", "project"))) return {"project": project_manager.save(project_id, payload.project)} @api.post("/project/assets/add", dependencies=[Depends(_verify_api_key)]) def project_asset_add(payload: ProjectAssetPayload) -> dict[str, Any]: try: return {"project": project_manager.add_asset(payload.project_id, payload.asset)} except KeyError as exc: raise HTTPException(status_code=404, detail="Project not found") from exc @api.post("/timeline/add", dependencies=[Depends(_verify_api_key)]) def timeline_add(payload: TimelineAddPayload) -> dict[str, Any]: try: project = project_manager.add_to_timeline(payload.project_id, payload.item, payload.track_type, payload.track_id) return {"project": project} except KeyError as exc: raise HTTPException(status_code=404, detail="Project not found") from exc except ValueError as exc: raise HTTPException(status_code=400, detail=str(exc)) from exc @api.post("/timeline/operation", dependencies=[Depends(_verify_api_key)]) def timeline_operation(payload: TimelineOperationPayload) -> dict[str, Any]: try: project = project_manager.timeline_operation(payload.project_id, payload.operation, payload.item_id, payload.params) return {"project": project} except KeyError as exc: raise HTTPException(status_code=404, detail="Timeline item or project not found") from exc except ValueError as exc: raise HTTPException(status_code=400, detail=str(exc)) from exc @api.post("/timeline/split", dependencies=[Depends(_verify_api_key)]) def timeline_split(payload: TimelineOperationPayload) -> dict[str, Any]: payload.operation = "split" return timeline_operation(payload) @api.post("/timeline/trim", dependencies=[Depends(_verify_api_key)]) def timeline_trim(payload: TimelineOperationPayload) -> dict[str, Any]: payload.operation = "trim" return timeline_operation(payload) @api.post("/timeline/ripple-delete", dependencies=[Depends(_verify_api_key)]) def timeline_ripple_delete(payload: TimelineOperationPayload) -> dict[str, Any]: payload.operation = "ripple_delete" return timeline_operation(payload) @api.post("/timeline/insert", dependencies=[Depends(_verify_api_key)]) def timeline_insert(payload: TimelineOperationPayload) -> dict[str, Any]: payload.operation = "insert" return timeline_operation(payload) @api.post("/timeline/replace", dependencies=[Depends(_verify_api_key)]) def timeline_replace(payload: TimelineOperationPayload) -> dict[str, Any]: payload.operation = "replace" return timeline_operation(payload) @api.post("/timeline/group", dependencies=[Depends(_verify_api_key)]) def timeline_group(payload: TimelineOperationPayload) -> dict[str, Any]: payload.operation = "group" return timeline_operation(payload) @api.post("/timeline/lock", dependencies=[Depends(_verify_api_key)]) def timeline_lock(payload: TimelineOperationPayload) -> dict[str, Any]: payload.operation = "lock" return timeline_operation(payload) @api.post("/timeline/hide", dependencies=[Depends(_verify_api_key)]) def timeline_hide(payload: TimelineOperationPayload) -> dict[str, Any]: payload.operation = "hide" return timeline_operation(payload) @api.post("/timeline/duplicate", dependencies=[Depends(_verify_api_key)]) def timeline_duplicate(payload: TimelineOperationPayload) -> dict[str, Any]: payload.operation = "duplicate" return timeline_operation(payload) @api.post("/effect/apply", dependencies=[Depends(_verify_api_key)]) def effect_apply(payload: EffectApplyPayload) -> dict[str, Any]: target_id = payload.target_id or payload.item_id if payload.project_id and target_id: try: project = project_manager.get(payload.project_id) record = add_effect(project, target_id, payload.effect, payload.params) project = project_manager.save(payload.project_id, project) return {"effect": record, "project": project} except KeyError as exc: raise HTTPException(status_code=404, detail="Project or target item not found") from exc data = {"task": payload.effect, "input": target_id, "params": payload.params, "callback_url": payload.callback_url} job_id = job_manager.submit_task( lambda task_id, log: StudioTaskProcessor(settings, log=log).ai_tool("auto_edit", data, task_id), callback_url=payload.callback_url, ) return _submission_response(job_id) @api.post("/filter/apply", dependencies=[Depends(_verify_api_key)]) def filter_apply(payload: FilterApplyPayload) -> dict[str, Any]: target_id = payload.target_id or payload.item_id params = dict(payload.params) if payload.lut: params["lut"] = payload.lut if payload.project_id and target_id: try: project = project_manager.get(payload.project_id) record = add_filter(project, target_id, payload.filter, params) project = project_manager.save(payload.project_id, project) return {"filter": record, "project": project} except KeyError as exc: raise HTTPException(status_code=404, detail="Project or target item not found") from exc job_id = job_manager.submit_task( lambda task_id, log: StudioTaskProcessor(settings, log=log).ai_tool("auto_color_match", payload.model_dump(), task_id), callback_url=payload.callback_url, ) return _submission_response(job_id) @api.post("/transition/add", dependencies=[Depends(_verify_api_key)]) def transition_add(payload: TransitionAddPayload) -> dict[str, Any]: if payload.project_id and payload.from_item_id and payload.to_item_id: try: project = project_manager.get(payload.project_id) record = add_transition(project, payload.from_item_id, payload.to_item_id, payload.transition, payload.duration) project = project_manager.save(payload.project_id, project) return {"transition": record, "project": project} except KeyError as exc: raise HTTPException(status_code=404, detail="Project not found") from exc job_id = job_manager.submit_task( lambda task_id, log: StudioTaskProcessor(settings, log=log).ai_tool("auto_edit", payload.model_dump(), task_id), callback_url=payload.callback_url, ) return _submission_response(job_id) @api.post("/keyframe/add", dependencies=[Depends(_verify_api_key)]) def keyframe_add(payload: KeyframePayload) -> dict[str, Any]: try: project = project_manager.get(payload.project_id) record = add_keyframe(project, payload.target_id, payload.property, payload.time, payload.value, payload.easing) project = project_manager.save(payload.project_id, project) return {"keyframe": record, "project": project} except KeyError as exc: raise HTTPException(status_code=404, detail="Project or target item not found") from exc @api.post("/caption/generate", dependencies=[Depends(_verify_api_key)]) def caption_generate(payload: CaptionGeneratePayload) -> dict[str, str]: data = payload.model_dump() job_id = job_manager.submit_task( lambda task_id, log: StudioTaskProcessor(settings, log=log).caption_generate(data, task_id), callback_url=payload.callback_url, ) return _submission_response(job_id) @api.post("/music/generate", dependencies=[Depends(_verify_api_key)]) def music_generate(payload: GenerationPayload) -> dict[str, str]: data = _generation_data(payload) job_id = job_manager.submit_task( lambda task_id, log: StudioTaskProcessor(settings, log=log).music_generate(data, task_id), callback_url=payload.callback_url, export_target=payload.export_target, ) return _submission_response(job_id) @api.post("/voice/generate", dependencies=[Depends(_verify_api_key)]) def voice_generate(payload: GenerationPayload) -> dict[str, str]: data = _generation_data(payload) job_id = job_manager.submit_task( lambda task_id, log: StudioTaskProcessor(settings, log=log).voice_generate(data, task_id), callback_url=payload.callback_url, export_target=payload.export_target, ) return _submission_response(job_id) @api.post("/image/generate", dependencies=[Depends(_verify_api_key)]) def image_generate(payload: GenerationPayload) -> dict[str, str]: data = _generation_data(payload) job_id = job_manager.submit_task( lambda task_id, log: StudioTaskProcessor(settings, log=log).image_generate(data, task_id), callback_url=payload.callback_url, export_target=payload.export_target, ) return _submission_response(job_id) @api.post("/video/generate", dependencies=[Depends(_verify_api_key)]) def video_generate(payload: GenerationPayload) -> dict[str, str]: data = _generation_data(payload) job_id = job_manager.submit_task( lambda task_id, log: StudioTaskProcessor(settings, log=log).video_generate(data, task_id), callback_url=payload.callback_url, export_target=payload.export_target, ) return _submission_response(job_id) @api.post("/ai/{tool}", dependencies=[Depends(_verify_api_key)]) def ai_tool(tool: str, payload: AIToolPayload) -> dict[str, str]: data = payload.model_dump() if payload.model_extra: data.update(payload.model_extra) job_id = job_manager.submit_task( lambda task_id, log: StudioTaskProcessor(settings, log=log).ai_tool(tool, data, task_id), callback_url=payload.callback_url, ) return _submission_response(job_id) @api.post("/assistant/{tool}", dependencies=[Depends(_verify_api_key)]) def assistant_tool(tool: str, payload: AIToolPayload) -> dict[str, str]: data = payload.model_dump() if payload.model_extra: data.update(payload.model_extra) job_id = job_manager.submit_task( lambda task_id, log: StudioTaskProcessor(settings, log=log).assistant_tool(tool, data, task_id), callback_url=payload.callback_url, ) return _submission_response(job_id) @api.post("/render") def render(payload: RenderPayload | AIReelsPayload) -> dict[str, str]: if isinstance(payload, AIReelsPayload): job_id = job_manager.submit_ai_reels(_ai_reels_request(payload)) else: job_id = job_manager.submit_render(_render_request(payload)) return _submission_response(job_id) @api.post("/render/ai-reels") def render_ai_reels(payload: AIReelsPayload) -> dict[str, str]: job_id = job_manager.submit_ai_reels(_ai_reels_request(payload)) return _submission_response(job_id) @api.post("/render/batch") def render_batch(payload: BatchPayload) -> dict[str, list[str]]: job_ids = job_manager.submit_batch([_render_request(job) for job in payload.jobs]) return {"job_ids": job_ids} @api.post("/automation/batch", dependencies=[Depends(_verify_api_key)]) def automation_batch(payload: BatchPayload) -> dict[str, list[str]]: return render_batch(payload) @api.post("/project/{project_id}/render", dependencies=[Depends(_verify_api_key)]) def project_render(project_id: str, payload: ProjectRenderPayload) -> dict[str, str]: try: project = project_manager.get(project_id) request = _project_render_request(project, payload) job_id = job_manager.submit_render(request) return _submission_response(job_id) except KeyError as exc: raise HTTPException(status_code=404, detail="Project not found") from exc except ValueError as exc: raise HTTPException(status_code=400, detail=str(exc)) from exc @api.post("/render/upload") async def render_upload(request_json: str = Form(...), files: list[UploadFile] = File(default=[])) -> dict[str, str]: uploads = await _stage_uploads(files) payload_data = _replace_upload_refs(json.loads(request_json), uploads) payload = RenderPayload.model_validate(payload_data) job_id = job_manager.submit_render(_render_request(payload)) return _submission_response(job_id) @api.post("/render/ai-reels/upload") async def render_ai_reels_upload(request_json: str = Form(...), files: list[UploadFile] = File(default=[])) -> dict[str, str]: uploads = await _stage_uploads(files) payload_data = _replace_upload_refs(json.loads(request_json), uploads) payload = AIReelsPayload.model_validate(payload_data) job_id = job_manager.submit_ai_reels(_ai_reels_request(payload)) return _submission_response(job_id) @api.post("/assets/upload") async def upload_assets(files: list[UploadFile] = File(...)) -> dict[str, list[UploadedAsset]]: uploads = await _stage_uploads(files) assets = [ UploadedAsset(filename=filename, path=path, reference=f"upload://{filename}", kind=_asset_kind(filename)) for filename, path in uploads.items() ] return {"assets": [asset.model_dump() for asset in assets]} @api.post("/upload") async def upload(files: list[UploadFile] = File(...), expand_zip: bool = Form(default=True)) -> dict[str, list[UploadedAsset]]: uploads = await _stage_uploads(files) expanded: dict[str, str] = {} for filename, path in uploads.items(): if expand_zip and filename.lower().endswith(".zip"): expanded.update(_extract_zip(Path(path))) else: expanded[filename] = path assets = [ UploadedAsset(filename=filename, path=path, reference=f"upload://{filename}", kind=_asset_kind(filename)) for filename, path in expanded.items() ] return {"assets": [asset.model_dump() for asset in assets]} @api.post("/ingest", dependencies=[Depends(_verify_api_key)]) def ingest(payload: IngestPayload) -> dict[str, str]: job_id = job_manager.submit_task( lambda task_id, log: PlatformProcessor(settings, log=log).ingest_sources([source.model_dump() for source in payload.sources], task_id), callback_url=payload.callback_url, ) return _submission_response(job_id) @api.post("/analyze", dependencies=[Depends(_verify_api_key)]) def analyze(payload: AnalyzePayload) -> dict[str, str]: job_id = job_manager.submit_task( lambda task_id, log: PlatformProcessor(settings, log=log).analyze( payload.media, task_id, transcript=payload.transcript, platform=payload.platform, ), callback_url=payload.callback_url, ) return _submission_response(job_id) @api.post("/clips", dependencies=[Depends(_verify_api_key)]) def clips(payload: ClipsPayload) -> dict[str, str]: job_id = job_manager.submit_task( lambda task_id, log: PlatformProcessor(settings, log=log).clips(payload.media, task_id, payload.clips), callback_url=payload.callback_url, ) return _submission_response(job_id) @api.post("/thumbnail", dependencies=[Depends(_verify_api_key)]) def thumbnail(payload: ThumbnailPayload) -> dict[str, str]: job_id = job_manager.submit_task( lambda task_id, log: PlatformProcessor(settings, log=log).thumbnail( payload.media, task_id, text=payload.text, timestamp=payload.timestamp, template=payload.template, ), callback_url=payload.callback_url, ) return _submission_response(job_id) @api.post("/thumbnail/create", dependencies=[Depends(_verify_api_key)]) def thumbnail_create(payload: ThumbnailPayload) -> dict[str, str]: return thumbnail(payload) @api.post("/metadata", dependencies=[Depends(_verify_api_key)]) def metadata(payload: MetadataPayload) -> dict[str, str]: job_id = job_manager.submit_task( lambda task_id, log: PlatformProcessor(settings, log=log).metadata( task_id, topic=payload.topic, transcript=payload.transcript, platform=payload.platform, ), callback_url=payload.callback_url, ) return _submission_response(job_id) @api.post("/publish", dependencies=[Depends(_verify_api_key)]) def publish(payload: PublishPayload) -> dict[str, str]: job_id = job_manager.submit_task( lambda task_id, log: PlatformProcessor(settings, log=log).publish(payload.model_dump(), task_id), callback_url=payload.callback_url, ) return _submission_response(job_id) @api.post("/toolkit", dependencies=[Depends(_verify_api_key)]) def toolkit(payload: ToolkitPayload) -> dict[str, str]: data = payload.model_dump() if payload.model_extra: data.update(payload.model_extra) job_id = job_manager.submit_task( lambda task_id, log: PlatformProcessor(settings, log=log).toolkit(data, task_id), callback_url=payload.callback_url, export_target=payload.export_target, ) return _submission_response(job_id) @api.post("/edit", dependencies=[Depends(_verify_api_key)]) def edit(payload: ToolkitPayload) -> dict[str, str]: return toolkit(payload) @api.post("/transcribe") def transcribe(payload: TranscribePayload) -> dict: from renderer.core.ingest import AssetIngestor from renderer.core.utils import temp_workdir try: with temp_workdir(settings.temp_dir, "transcribe") as work: audio = AssetIngestor(settings).resolve(payload.audio, Path(work) / "inputs", "audio") return RenderEngine(settings).transcribe( audio, model_size=payload.model_size, language=payload.language, task=payload.task, beam_size=payload.beam_size, vad_filter=payload.vad_filter, word_timestamps=payload.word_timestamps, ) except Exception as exc: raise HTTPException(status_code=400, detail=str(exc)) from exc @api.post("/subtitles") def subtitles(payload: SubtitlePayload) -> FileResponse: try: events = [SubtitleEvent(float(event["start"]), float(event["end"]), str(event["text"])) for event in payload.events] path = settings.temp_dir / f"subtitles_{uuid.uuid4().hex}.{payload.format}" generator = SubtitleGenerator() if payload.format == "ass": generator.write_ass(events, path, payload.template) media_type = "text/x-ssa" else: generator.write_srt(events, path) media_type = "application/x-subrip" return FileResponse(path, media_type=media_type, filename=path.name) except Exception as exc: raise HTTPException(status_code=400, detail=str(exc)) from exc @api.post("/scene-builder") def scene_builder(payload: SceneBuildPayload) -> dict[str, Any]: if not payload.assets: raise HTTPException(status_code=400, detail="At least one asset is required") style = get_creative_style(payload.creative_style) words = payload.script.split() total_duration = payload.duration or max(style.scene_duration * len(payload.assets), len(words) * 0.35, 3.0) per_scene = total_duration / len(payload.assets) captions = _split_words_for_assets(words, len(payload.assets)) scenes = [ { "start": round(index * per_scene, 3), "duration": round(per_scene, 3), "media": asset, "caption": captions[index] if index < len(captions) else "", "transition": _style_transition(payload.transition, style.transition_sequence, index), "effect": style.scene_effect_sequence[index % len(style.scene_effect_sequence)], "background": "blur", "layout": "fill", } for index, asset in enumerate(payload.assets) ] return {"scenes": scenes, "creative_style": style.metadata_payload()} @api.post("/transcribe/upload") async def transcribe_upload( file: UploadFile = File(...), model_size: str | None = Form(default=None), language: str | None = Form(default=None), task: str = Form(default="transcribe"), beam_size: int = Form(default=5), vad_filter: bool = Form(default=True), word_timestamps: bool = Form(default=True), ) -> dict: try: uploads = await _stage_uploads([file]) audio = next(iter(uploads.values())) return RenderEngine(settings).transcribe( audio, model_size=model_size, language=language, task=task, beam_size=beam_size, vad_filter=vad_filter, word_timestamps=word_timestamps, ) except Exception as exc: raise HTTPException(status_code=400, detail=str(exc)) from exc @api.get("/status") def status_query(job_id: str = Query(...)) -> dict: return status(job_id) @api.get("/status/{job_id}") def status(job_id: str) -> dict: try: return job_manager.get(job_id).__dict__ except KeyError as exc: raise HTTPException(status_code=404, detail="Job not found") from exc @api.post("/cancel/{job_id}") def cancel(job_id: str) -> dict: try: return job_manager.cancel(job_id).__dict__ except KeyError as exc: raise HTTPException(status_code=404, detail="Job not found") from exc @api.post("/admin/cleanup") def cleanup(older_than_seconds: int | None = None) -> dict[str, int]: return job_manager.cleanup(older_than_seconds) @api.get("/download") def download_query(job_id: str = Query(...), token: str | None = Query(default=None)) -> FileResponse: return download(job_id, token) @api.get("/download/{job_id}") def download(job_id: str, token: str | None = Query(default=None)) -> FileResponse: try: record = job_manager.get(job_id) except KeyError as exc: raise HTTPException(status_code=404, detail="Job not found") from exc if record.state != "COMPLETED" or not record.output_path: raise HTTPException(status_code=409, detail=f"Job is {record.state}") if record.download_token and not verify_download_token(settings.signing_secret, job_id, token): raise HTTPException(status_code=403, detail="Invalid or missing download token") path = Path(record.output_path) if not path.exists(): raise HTTPException(status_code=404, detail="Output file is missing") return FileResponse(path, media_type=_media_type(path), filename=path.name) @api.post("/inspect") def inspect_asset(path: str) -> dict: try: return RenderEngine(settings).inspect_asset(path) except Exception as exc: raise HTTPException(status_code=400, detail=str(exc)) from exc def _render_request(payload: RenderPayload) -> RenderRequest: payload = RenderPayload.model_validate(apply_creative_style(apply_preset(payload.model_dump(exclude_unset=True)))) request = RenderRequest( scenes=[Scene(**scene.model_dump()) for scene in payload.scenes], template=payload.template, preset=payload.preset, creative_style=payload.creative_style, platform=payload.platform, output_name=payload.output_name, voiceover=payload.voiceover, background_music=payload.background_music, music_volume=payload.music_volume, music_fade_in=payload.music_fade_in, music_fade_out=payload.music_fade_out, music_loop=payload.music_loop, music_start=payload.music_start, music_ducking=payload.music_ducking, voice_volume=payload.voice_volume, subtitle_format=payload.subtitle_format, # type: ignore[arg-type] auto_subtitles=payload.auto_subtitles, subtitle_language=payload.subtitle_language, whisper_model_size=payload.whisper_model_size, preview=payload.preview, audio_normalize=payload.audio_normalize, watermark=payload.watermark, watermark_position=payload.watermark_position, intro=payload.intro, outro=payload.outro, callback_url=payload.callback_url, export_target=payload.export_target, priority=payload.priority, scheduled_at=payload.scheduled_at, normalize=payload.normalize, metadata=payload.metadata, ) Timeline(request.scenes) return request def _project_render_request(project: dict[str, Any], payload: ProjectRenderPayload) -> RenderRequest: scenes: list[dict[str, Any]] = [] for tracks in project.get("timeline", {}).get("tracks", {}).values(): for track in tracks: if track.get("hidden") or track.get("type") not in {"video", "overlay"}: continue for item in track.get("items", []): if item.get("hidden"): continue media = _item_media(project, item) if not media: continue scenes.append( { "start": float(item.get("start", 0.0)), "duration": float(item.get("duration", 1.0)), "media": media, "caption": item.get("caption") or item.get("text") or "", "transition": item.get("transition", "fade"), "background": item.get("background", "blur"), "layout": item.get("layout", "fill"), "effect": _first_named(item.get("effects")), } ) if not scenes: raise ValueError("Project has no renderable video or overlay timeline items") export_settings = project.get("export_settings", {}) data = { "scenes": sorted(scenes, key=lambda scene: scene["start"]), "template": payload.template or export_settings.get("template", "tiktok_classic"), "preset": payload.preset, "creative_style": payload.creative_style or project.get("metadata", {}).get("creative_style"), "platform": payload.platform or export_settings.get("platform"), "output_name": payload.output_name, "callback_url": payload.callback_url, "export_target": payload.export_target, "preview": payload.preview, "normalize": payload.normalize, "metadata": project.get("metadata", {}) | payload.metadata | {"project_id": project.get("id"), "project_name": project.get("name")}, } return _render_request(RenderPayload.model_validate(data)) def _ai_reels_request(payload: AIReelsPayload) -> AIReelsRequest: payload = AIReelsPayload.model_validate(apply_creative_style(apply_preset(payload.model_dump(exclude_unset=True)))) return AIReelsRequest(**payload.model_dump()) def _generation_data(payload: GenerationPayload) -> dict[str, Any]: data = payload.model_dump() if payload.model_extra: data.update(payload.model_extra) params = data.pop("params", {}) or {} if isinstance(params, dict): data.update(params) return data def _item_media(project: dict[str, Any], item: dict[str, Any]) -> str | None: direct = item.get("media") or item.get("path") or item.get("source") if direct: return str(direct) asset_id = item.get("asset_id") if not asset_id: return None for asset in project.get("assets", []): if asset.get("id") == asset_id: return str(asset.get("path") or asset.get("url") or asset.get("source") or "") return None def _first_named(records: Any) -> str | None: if not isinstance(records, list) or not records: return None first = records[0] if isinstance(first, dict): return first.get("effect") or first.get("name") return str(first) async def _stage_uploads(files: list[UploadFile]) -> dict[str, str]: upload_dir = settings.temp_dir / "uploads" / uuid.uuid4().hex staged: dict[str, str] = {} total_bytes = 0 for upload in files: filename = safe_filename(upload.filename or f"asset_{len(staged)}") if not _allowed_upload(filename): raise HTTPException(status_code=415, detail=f"Unsupported asset type: {filename}") target = upload_dir / filename target.parent.mkdir(parents=True, exist_ok=True) with target.open("wb") as output: while True: chunk = await upload.read(1024 * 1024) if not chunk: break total_bytes += len(chunk) if total_bytes > settings.max_download_bytes: shutil.rmtree(upload_dir, ignore_errors=True) raise HTTPException(status_code=413, detail="Uploaded assets exceed MAX_DOWNLOAD_BYTES") output.write(chunk) staged[filename] = str(stage_upload(target, upload_dir, filename)) return staged def _monitor_payload() -> dict[str, Any]: disk = shutil.disk_usage(settings.base_dir) return { "health": "ok", "queue": job_manager.summary(), "workers": { "max_workers": settings.max_workers, "ffmpeg_timeout_seconds": settings.ffmpeg_timeout_seconds, "whisper_model_size": settings.whisper_model_size, "whisper_device": settings.whisper_device, }, "disk": { "total_bytes": disk.total, "used_bytes": disk.used, "free_bytes": disk.free, }, "directories": { "temp": str(settings.temp_dir), "exports": str(settings.exports_dir), "jobs": str(settings.jobs_dir), "storage": str(settings.storage_dir), }, } def _extract_zip(path: Path) -> dict[str, str]: output_dir = settings.temp_dir / "uploads" / f"zip_{uuid.uuid4().hex}" output_dir.mkdir(parents=True, exist_ok=True) extracted: dict[str, str] = {} with zipfile.ZipFile(path) as archive: for member in archive.infolist(): if member.is_dir(): continue name = safe_filename(Path(member.filename).name) if not _allowed_upload(name): continue target = output_dir / name resolved = target.resolve() if output_dir.resolve() not in resolved.parents and resolved != output_dir.resolve(): raise HTTPException(status_code=400, detail="Unsafe ZIP member path") with archive.open(member) as source, target.open("wb") as destination: shutil.copyfileobj(source, destination) extracted[name] = str(target) return extracted def _asset_kind(filename: str) -> str: suffix = Path(filename).suffix.lower() if suffix in {".mp4", ".mov", ".m4v", ".webm", ".mkv", ".avi", ".gif"}: return "video" if suffix in {".mp3", ".wav", ".m4a", ".aac", ".flac", ".ogg"}: return "audio" if suffix in {".jpg", ".jpeg", ".png", ".webp", ".avif"}: return "image" if suffix in {".srt", ".ass", ".vtt"}: return "subtitle" return "other" def _allowed_upload(filename: str) -> bool: suffix = Path(filename).suffix.lower() return suffix in { ".mp4", ".mov", ".m4v", ".webm", ".mkv", ".avi", ".gif", ".mp3", ".wav", ".m4a", ".aac", ".flac", ".ogg", ".jpg", ".jpeg", ".png", ".webp", ".avif", ".srt", ".ass", ".vtt", ".zip", } def _media_type(path: Path) -> str: suffix = path.suffix.lower() if suffix == ".json": return "application/json" if suffix == ".zip": return "application/zip" if suffix == ".jpg" or suffix == ".jpeg": return "image/jpeg" if suffix == ".png": return "image/png" if suffix == ".gif": return "image/gif" if suffix == ".mp3": return "audio/mpeg" if suffix in {".srt", ".vtt", ".ass"}: return "text/plain" return "video/mp4" def _replace_upload_refs(value: Any, uploads: dict[str, str]) -> Any: if isinstance(value, dict): return {key: _replace_upload_refs(item, uploads) for key, item in value.items()} if isinstance(value, list): return [_replace_upload_refs(item, uploads) for item in value] if isinstance(value, str) and value.startswith("upload://"): name = safe_filename(value.removeprefix("upload://")) if name not in uploads: raise HTTPException(status_code=400, detail=f"Missing uploaded file for reference: upload://{name}") return uploads[name] return value def _split_words_for_assets(words: list[str], count: int) -> list[str]: if count <= 0: return [] if not words: return [""] * count base, remainder = divmod(len(words), count) captions: list[str] = [] cursor = 0 for index in range(count): size = base + (1 if index < remainder else 0) size = max(1, size) captions.append(" ".join(words[cursor : cursor + size])) cursor += size return captions def _style_transition(requested: str, sequence: tuple[str, ...], index: int) -> str: if requested and requested != "fade": return requested return sequence[index % len(sequence)] if sequence else "fade"