api.py

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"