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 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) api = FastAPI(title="Basyx FFmpeg Rendering Engine", version="1.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) @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("/projects") def list_projects() -> dict[str, list[dict[str, Any]]]: projects_dir = settings.storage_dir / "projects" projects_dir.mkdir(parents=True, exist_ok=True) projects: list[dict[str, Any]] = [] for manifest in sorted(projects_dir.glob("*/project.json")): try: projects.append(json.loads(manifest.read_text(encoding="utf-8"))) except Exception: continue return {"projects": projects} @api.post("/projects", dependencies=[Depends(_verify_api_key)]) def create_project(payload: ProjectPayload) -> dict[str, Any]: slug = safe_filename(payload.name) project_dir = settings.storage_dir / "projects" / slug project_dir.mkdir(parents=True, exist_ok=True) manifest = { "name": payload.name, "slug": slug, "path": str(project_dir), "metadata": payload.metadata, } (project_dir / "project.json").write_text(json.dumps(manifest, indent=2), encoding="utf-8") return {"project": manifest} @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("/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("/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/{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/{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 _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()) 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"