services/ffmpeg_automation/app.py

import base64
import binascii
import os
import shutil
import sqlite3
import subprocess
import textwrap
import threading
import time
import uuid
from dataclasses import dataclass
from typing import Callable, Dict, List, Optional

import gradio as gr
import magic
import uvicorn
import yt_dlp
from fastapi import BackgroundTasks, FastAPI, File, Form, HTTPException, Request, UploadFile
from fastapi.responses import FileResponse

app = FastAPI(title="Basyx FFmpeg Automation Hub")

TEMP_DIR = os.getenv("TEMP_DIR", "temp")
FONT_PATH = os.getenv(
    "FONT_PATH", "/usr/share/fonts/truetype/dejavu/DejaVuSans-Bold.ttf"
)
MAX_UPLOAD_MB = int(os.getenv("MAX_UPLOAD_MB", "500"))
MAX_UPLOAD_BYTES = MAX_UPLOAD_MB * 1024 * 1024
FILE_TTL_SECONDS = int(os.getenv("FILE_TTL_SECONDS", "3600"))
HISTORY_LIMIT = int(os.getenv("HISTORY_LIMIT", "50"))
FFMPEG_TIMEOUT_SECONDS = int(os.getenv("FFMPEG_TIMEOUT_SECONDS", "1800"))
MAX_URL_DOWNLOAD_MB = int(os.getenv("MAX_URL_DOWNLOAD_MB", str(MAX_UPLOAD_MB)))
MAX_URL_DOWNLOAD_BYTES = MAX_URL_DOWNLOAD_MB * 1024 * 1024
ALLOW_URL_INPUTS = os.getenv("ALLOW_URL_INPUTS", "true").lower() in {"1", "true", "yes"}
STATE_DB_PATH = os.getenv("STATE_DB_PATH", os.path.join(TEMP_DIR, "state.sqlite3"))
OPTION_FIELDS = [
    "text",
    "start_time",
    "end_time",
    "duration",
    "aspect_ratio",
    "resolution",
    "crf",
    "preset",
    "audio_bitrate",
    "volume",
    "position",
    "opacity",
    "fps",
    "width",
    "speed",
    "timestamp",
    "image_duration",
    "frame_rate",
    "font_size",
    "wave_color",
]

os.makedirs(TEMP_DIR, exist_ok=True)


@dataclass(frozen=True)
class TaskDefinition:
    label: str
    description: str
    category: str
    output_ext: str
    min_files: int
    max_files: Optional[int]
    accepted_types: List[str]
    builder: Optional[Callable[[List[str], Dict[str, str], str], List[str]]] = None
    runner: Optional[Callable[[List[str], Dict[str, str], str], str]] = None
    file_types: Optional[List[List[str]]] = None


JOBS: Dict[str, Dict[str, object]] = {}
HISTORY: List[Dict[str, object]] = []
STATE_LOCK = threading.Lock()


def db_connect():
    connection = sqlite3.connect(STATE_DB_PATH, timeout=30)
    connection.row_factory = sqlite3.Row
    return connection


def init_state_store() -> None:
    with db_connect() as connection:
        connection.execute(
            """
            CREATE TABLE IF NOT EXISTS jobs (
                job_id TEXT PRIMARY KEY,
                task_id TEXT NOT NULL,
                status TEXT NOT NULL,
                message TEXT NOT NULL,
                output TEXT,
                created_at INTEGER NOT NULL,
                updated_at INTEGER NOT NULL
            )
            """
        )
        connection.execute(
            """
            CREATE TABLE IF NOT EXISTS history (
                history_id TEXT PRIMARY KEY,
                job_id TEXT NOT NULL,
                task_id TEXT NOT NULL,
                task TEXT NOT NULL,
                output TEXT NOT NULL,
                filename TEXT NOT NULL,
                size INTEGER NOT NULL,
                created_at INTEGER NOT NULL
            )
            """
        )
        connection.execute(
            "CREATE INDEX IF NOT EXISTS idx_history_created_at ON history(created_at DESC)"
        )


def load_state_store() -> None:
    with STATE_LOCK, db_connect() as connection:
        JOBS.clear()
        for row in connection.execute("SELECT * FROM jobs ORDER BY created_at DESC"):
            JOBS[row["job_id"]] = dict(row)
        HISTORY.clear()
        for row in connection.execute(
            "SELECT * FROM history ORDER BY created_at DESC LIMIT ?", (HISTORY_LIMIT,)
        ):
            item = dict(row)
            item["download_url"] = f"/history/{item['history_id']}/download"
            HISTORY.append(item)


def persist_job(job: Dict[str, object]) -> None:
    now = int(time.time())
    with db_connect() as connection:
        connection.execute(
            """
            INSERT INTO jobs (job_id, task_id, status, message, output, created_at, updated_at)
            VALUES (?, ?, ?, ?, ?, ?, ?)
            ON CONFLICT(job_id) DO UPDATE SET
                status=excluded.status,
                message=excluded.message,
                output=excluded.output,
                updated_at=excluded.updated_at
            """,
            (
                str(job["job_id"]),
                str(job["task_id"]),
                str(job["status"]),
                str(job["message"]),
                str(job.get("output") or ""),
                int(job.get("created_at") or now),
                now,
            ),
        )


def persist_history(item: Dict[str, object]) -> None:
    with db_connect() as connection:
        connection.execute(
            """
            INSERT OR REPLACE INTO history
            (history_id, job_id, task_id, task, output, filename, size, created_at)
            VALUES (?, ?, ?, ?, ?, ?, ?, ?)
            """,
            (
                str(item["history_id"]),
                str(item["job_id"]),
                str(item["task_id"]),
                str(item["task"]),
                str(item["output"]),
                str(item["filename"]),
                int(item["size"]),
                int(item["created_at"]),
            ),
        )
        stale_ids = [
            row["history_id"]
            for row in connection.execute(
                """
                SELECT history_id FROM history
                ORDER BY created_at DESC
                LIMIT -1 OFFSET ?
                """,
                (HISTORY_LIMIT,),
            )
        ]
        if stale_ids:
            connection.executemany(
                "DELETE FROM history WHERE history_id = ?",
                [(history_id,) for history_id in stale_ids],
            )


init_state_store()
load_state_store()


def ffmpeg_escape(value: str) -> str:
    return (
        value.replace("\\", "\\\\")
        .replace(":", "\\:")
        .replace("'", "\\'")
        .replace("\n", "\\n")
        .replace("%", "\\%")
    )


def filter_path(path: str) -> str:
    return ffmpeg_escape(os.path.abspath(path))


def ffconcat_path(path: str) -> str:
    return os.path.abspath(path).replace("\\", "\\\\").replace("'", "\\'")


def drawtext_file_path(path: str) -> str:
    return filter_path(path)


def remove_file_quietly(path: str) -> None:
    try:
        os.remove(path)
    except FileNotFoundError:
        return
    except OSError:
        return


def clamp_int(value: str, default: int, minimum: int, maximum: int) -> int:
    try:
        parsed = int(float(value))
    except (TypeError, ValueError):
        parsed = default
    return max(minimum, min(maximum, parsed))


def clamp_float(value: str, default: float, minimum: float, maximum: float) -> float:
    try:
        parsed = float(value)
    except (TypeError, ValueError):
        parsed = default
    return max(minimum, min(maximum, parsed))


def dimensions(preset: str) -> tuple[int, int]:
    presets = {
        "9:16": (1080, 1920),
        "16:9": (1920, 1080),
        "1:1": (1080, 1080),
        "4:5": (1080, 1350),
    }
    return presets.get(preset, presets["9:16"])


def scaled_crop_filter(preset: str) -> str:
    width, height = dimensions(preset)
    return (
        f"scale={width}:{height}:force_original_aspect_ratio=increase,"
        f"crop={width}:{height}"
    )


def output_scale_filter(size: str) -> str:
    sizes = {"480p": 480, "720p": 720, "1080p": 1080}
    height = sizes.get(size, 720)
    return f"scale=-2:{height}"


def write_wrapped_text_file(text: str, width: int = 24, max_lines: Optional[int] = None) -> str:
    cleaned = " ".join((text or "").split())
    if not cleaned:
        cleaned = "Your faceless video"
    lines = textwrap.wrap(cleaned, width=width, break_long_words=False)
    if max_lines:
        lines = lines[:max_lines]
    path = os.path.abspath(os.path.join(TEMP_DIR, f"text_{uuid.uuid4().hex[:8]}.txt"))
    with open(path, "w", encoding="utf-8") as handle:
        handle.write("\n".join(lines))
    return path


def split_story_pages(text: str, words_per_page: int = 22) -> List[str]:
    words = (text or "").split()
    if not words:
        words = ["Add", "story", "text", "to", "generate", "faceless", "video", "slides."]
    pages = [
        " ".join(words[index : index + words_per_page])
        for index in range(0, len(words), words_per_page)
    ]
    return pages[:20]


def production_checks() -> Dict[str, Dict[str, object]]:
    checks = {
        "ffmpeg": {"ok": shutil.which("ffmpeg") is not None},
        "ffprobe": {"ok": shutil.which("ffprobe") is not None},
        "font": {"ok": os.path.exists(FONT_PATH), "path": FONT_PATH},
        "temp_dir": {
            "ok": os.path.isdir(TEMP_DIR) and os.access(TEMP_DIR, os.W_OK),
            "path": TEMP_DIR,
        },
        "state_db": {
            "ok": os.path.exists(STATE_DB_PATH) and os.access(STATE_DB_PATH, os.W_OK),
            "path": STATE_DB_PATH,
        },
    }
    checks["url_inputs"] = {"ok": True, "enabled": ALLOW_URL_INPUTS}
    return checks


def production_ready() -> bool:
    required = ["ffmpeg", "ffprobe", "font", "temp_dir", "state_db"]
    checks = production_checks()
    return all(bool(checks[name]["ok"]) for name in required)


def add_video_quality(cmd: List[str], opts: Dict[str, str]) -> List[str]:
    crf = str(clamp_int(opts.get("crf", "23"), 23, 12, 35))
    preset = opts.get("preset") or "fast"
    if preset not in {"ultrafast", "veryfast", "fast", "medium", "slow"}:
        preset = "fast"
    return cmd + ["-c:v", "libx264", "-preset", preset, "-crf", crf, "-c:a", "aac"]


def run_command(cmd: List[str]) -> subprocess.CompletedProcess:
    try:
        return subprocess.run(
            cmd,
            check=True,
            capture_output=True,
            text=True,
            timeout=FFMPEG_TIMEOUT_SECONDS,
        )
    except subprocess.TimeoutExpired as exc:
        raise HTTPException(
            status_code=504,
            detail=f"FFmpeg timed out after {FFMPEG_TIMEOUT_SECONDS} seconds.",
        ) from exc


def media_duration_seconds(path: str) -> float:
    result = run_command(
        [
            "ffprobe",
            "-v",
            "error",
            "-show_entries",
            "format=duration",
            "-of",
            "default=noprint_wrappers=1:nokey=1",
            path,
        ]
    )
    try:
        duration = float(result.stdout.strip())
    except ValueError as exc:
        raise HTTPException(status_code=500, detail="Could not read media duration.") from exc
    if duration <= 0:
        raise HTTPException(status_code=500, detail="Media duration is not usable.")
    return duration


def build_normalize(paths: List[str], opts: Dict[str, str], out: str) -> List[str]:
    cmd = ["ffmpeg", "-i", paths[0]]
    return add_video_quality(cmd, opts) + ["-y", out]


def build_extract_audio(paths: List[str], opts: Dict[str, str], out: str) -> List[str]:
    bitrate = str(clamp_int(opts.get("audio_bitrate", "192"), 192, 64, 320))
    return [
        "ffmpeg",
        "-i",
        paths[0],
        "-vn",
        "-acodec",
        "libmp3lame",
        "-ar",
        "44100",
        "-ac",
        "2",
        "-b:a",
        f"{bitrate}k",
        "-y",
        out,
    ]


def build_resize(paths: List[str], opts: Dict[str, str], out: str) -> List[str]:
    vf = scaled_crop_filter(opts.get("aspect_ratio", "9:16"))
    return ["ffmpeg", "-i", paths[0], "-vf", vf, "-c:a", "copy", "-y", out]


def build_subtitles(paths: List[str], opts: Dict[str, str], out: str) -> List[str]:
    subtitle_filter = f"subtitles='{filter_path(paths[1])}'"
    return ["ffmpeg", "-i", paths[0], "-vf", subtitle_filter, "-c:a", "copy", "-y", out]


def build_burn_lyrics(paths: List[str], opts: Dict[str, str], out: str) -> List[str]:
    style = (
        "Fontname=DejaVu Sans,Fontsize=24,PrimaryColour=&H00FFFF,"
        "OutlineColour=&H000000,BorderStyle=3,Outline=1,Shadow=1,Alignment=2"
    )
    vf = f"subtitles='{filter_path(paths[1])}':force_style='{style}'"
    return ["ffmpeg", "-i", paths[0], "-vf", vf, "-c:a", "copy", "-y", out]


def build_text_overlay(paths: List[str], opts: Dict[str, str], out: str) -> List[str]:
    text = ffmpeg_escape(opts.get("text", ""))
    font_size = clamp_int(opts.get("font_size", "60"), 60, 18, 160)
    vf = (
        f"drawtext=fontfile='{filter_path(FONT_PATH)}':text='{text}':"
        f"fontcolor=white:fontsize={font_size}:box=1:boxcolor=black@0.45:"
        "boxborderw=16:x=(w-text_w)/2:y=h-200"
    )
    return ["ffmpeg", "-i", paths[0], "-vf", vf, "-c:a", "copy", "-y", out]


def build_merge_music(paths: List[str], opts: Dict[str, str], out: str) -> List[str]:
    volume = clamp_float(opts.get("volume", "0.3"), 0.3, 0.0, 2.0)
    return [
        "ffmpeg",
        "-i",
        paths[0],
        "-i",
        paths[1],
        "-filter_complex",
        f"[1:a]volume={volume}[a1]",
        "-map",
        "0:v",
        "-map",
        "[a1]",
        "-c:v",
        "copy",
        "-shortest",
        "-y",
        out,
    ]


def build_thumbnail(paths: List[str], opts: Dict[str, str], out: str) -> List[str]:
    timestamp = opts.get("timestamp") or "00:00:02"
    return ["ffmpeg", "-i", paths[0], "-ss", timestamp, "-vframes", "1", "-y", out]


def build_watermark(paths: List[str], opts: Dict[str, str], out: str) -> List[str]:
    positions = {
        "bottom-right": "W-w-20:H-h-20",
        "bottom-left": "20:H-h-20",
        "top-right": "W-w-20:20",
        "top-left": "20:20",
        "center": "(W-w)/2:(H-h)/2",
    }
    position = positions.get(opts.get("position"), positions["bottom-right"])
    opacity = clamp_float(opts.get("opacity", "1"), 1, 0.1, 1.0)
    overlay = f"[1:v]format=rgba,colorchannelmixer=aa={opacity}[wm];[0:v][wm]overlay={position}"
    return ["ffmpeg", "-i", paths[0], "-i", paths[1], "-filter_complex", overlay, "-y", out]


def build_compress(paths: List[str], opts: Dict[str, str], out: str) -> List[str]:
    vf = output_scale_filter(opts.get("resolution", "720p"))
    cmd = ["ffmpeg", "-i", paths[0], "-vf", vf]
    return add_video_quality(cmd, opts) + ["-y", out]


def build_gif(paths: List[str], opts: Dict[str, str], out: str) -> List[str]:
    duration = str(clamp_int(opts.get("duration", "5"), 5, 1, 60))
    fps = str(clamp_int(opts.get("fps", "10"), 10, 5, 30))
    width = str(clamp_int(opts.get("width", "480"), 480, 240, 1080))
    vf = f"fps={fps},scale={width}:-1:flags=lanczos"
    return ["ffmpeg", "-i", paths[0], "-t", duration, "-vf", vf, "-y", out]


def build_tiktok_lyrics(paths: List[str], opts: Dict[str, str], out: str) -> List[str]:
    text = ffmpeg_escape(opts.get("text", ""))
    duration = str(clamp_int(opts.get("duration", "15"), 15, 1, 180))
    vf = (
        scaled_crop_filter("9:16")
        + ","
        + f"drawtext=fontfile='{filter_path(FONT_PATH)}':text='{text}':"
        "fontcolor=white:fontsize=50:box=1:boxcolor=black@0.5:"
        "boxborderw=20:line_spacing=15:x=(w-text_w)/2:y=(h-text_h)/2"
    )
    return ["ffmpeg", "-i", paths[0], "-vf", vf, "-c:a", "copy", "-t", duration, "-y", out]


def build_tiktok_pro(paths: List[str], opts: Dict[str, str], out: str) -> List[str]:
    text = ffmpeg_escape(opts.get("text", ""))
    vf = (
        scaled_crop_filter("9:16")
        + ","
        + f"drawtext=fontfile='{filter_path(FONT_PATH)}':text='{text}':"
        "fontcolor=white:fontsize=48:box=1:boxcolor=black@0.4:"
        "boxborderw=14:x=(w-text_w)/2:y=h-200,"
        "eq=contrast=1.15:brightness=0.04"
    )
    return ["ffmpeg", "-i", paths[0], "-vf", vf, "-c:a", "copy", "-y", out]


def build_reels_blur_fit(paths: List[str], opts: Dict[str, str], out: str) -> List[str]:
    filter_complex = (
        "[0:v]scale=1080:1920:force_original_aspect_ratio=increase,"
        "crop=1080:1920,gblur=sigma=24,eq=brightness=-0.08[bg];"
        "[0:v]scale=1080:1920:force_original_aspect_ratio=decrease[fg];"
        "[bg][fg]overlay=(W-w)/2:(H-h)/2,setsar=1[v]"
    )
    return [
        "ffmpeg",
        "-i",
        paths[0],
        "-filter_complex",
        filter_complex,
        "-map",
        "[v]",
        "-map",
        "0:a?",
        "-c:v",
        "libx264",
        "-preset",
        opts.get("preset") or "fast",
        "-crf",
        str(clamp_int(opts.get("crf", "23"), 23, 12, 35)),
        "-c:a",
        "aac",
        "-y",
        out,
    ]


def build_reels_safe_caption(paths: List[str], opts: Dict[str, str], out: str) -> List[str]:
    text = ffmpeg_escape(opts.get("text", ""))
    font_size = clamp_int(opts.get("font_size", "58"), 58, 24, 140)
    vf = (
        scaled_crop_filter("9:16")
        + ","
        + f"drawtext=fontfile='{filter_path(FONT_PATH)}':text='{text}':"
        f"fontcolor=white:fontsize={font_size}:line_spacing=10:"
        "box=1:boxcolor=black@0.55:boxborderw=22:"
        "x=(w-text_w)/2:y=h-430"
    )
    return ["ffmpeg", "-i", paths[0], "-vf", vf, "-c:a", "copy", "-y", out]


def build_reels_hook_title(paths: List[str], opts: Dict[str, str], out: str) -> List[str]:
    text = ffmpeg_escape(opts.get("text", ""))
    font_size = clamp_int(opts.get("font_size", "64"), 64, 28, 150)
    vf = (
        scaled_crop_filter("9:16")
        + ",drawbox=x=0:y=95:w=iw:h=210:color=black@0.62:t=fill,"
        + f"drawtext=fontfile='{filter_path(FONT_PATH)}':text='{text}':"
        f"fontcolor=white:fontsize={font_size}:line_spacing=8:"
        "x=(w-text_w)/2:y=145"
    )
    return ["ffmpeg", "-i", paths[0], "-vf", vf, "-c:a", "copy", "-y", out]


def build_reels_progress_bar(paths: List[str], opts: Dict[str, str], out: str) -> List[str]:
    duration = clamp_float(opts.get("duration", "30"), 30, 1, 180)
    vf = (
        scaled_crop_filter("9:16")
        + ",drawbox=x=0:y=0:w=iw:h=10:color=black@0.35:t=fill,"
        + f"drawbox=x=0:y=0:w='min(iw,iw*t/{duration:.3f})':h=10:"
        "color=white@0.95:t=fill"
    )
    return ["ffmpeg", "-i", paths[0], "-vf", vf, "-c:a", "copy", "-t", f"{duration:.3f}", "-y", out]


def build_reels_loop(paths: List[str], opts: Dict[str, str], out: str) -> List[str]:
    duration = clamp_float(opts.get("duration", "15"), 15, 1, 180)
    return [
        "ffmpeg",
        "-stream_loop",
        "-1",
        "-i",
        paths[0],
        "-t",
        f"{duration:.3f}",
        "-vf",
        scaled_crop_filter("9:16"),
        "-c:v",
        "libx264",
        "-preset",
        opts.get("preset") or "fast",
        "-crf",
        str(clamp_int(opts.get("crf", "23"), 23, 12, 35)),
        "-c:a",
        "aac",
        "-y",
        out,
    ]


def build_reels_subtitle_safe(paths: List[str], opts: Dict[str, str], out: str) -> List[str]:
    vf = (
        scaled_crop_filter("9:16")
        + ",subtitles='"
        + filter_path(paths[1])
        + "':force_style='Fontname=DejaVu Sans,Fontsize=28,"
        + "PrimaryColour=&H00FFFFFF,OutlineColour=&H00000000,"
        + "BorderStyle=3,Outline=1,Shadow=0,Alignment=2,MarginV=250'"
    )
    return ["ffmpeg", "-i", paths[0], "-vf", vf, "-c:a", "copy", "-y", out]


def build_reels_reaction_stack(paths: List[str], opts: Dict[str, str], out: str) -> List[str]:
    filter_complex = (
        "[0:v]scale=1080:960:force_original_aspect_ratio=increase,"
        "crop=1080:960,setsar=1[top];"
        "[1:v]scale=1080:960:force_original_aspect_ratio=increase,"
        "crop=1080:960,setsar=1[bottom];"
        "[top][bottom]vstack=inputs=2[v]"
    )
    return [
        "ffmpeg",
        "-i",
        paths[0],
        "-i",
        paths[1],
        "-filter_complex",
        filter_complex,
        "-map",
        "[v]",
        "-map",
        "0:a?",
        "-shortest",
        "-c:v",
        "libx264",
        "-preset",
        opts.get("preset") or "fast",
        "-crf",
        str(clamp_int(opts.get("crf", "23"), 23, 12, 35)),
        "-c:a",
        "aac",
        "-y",
        out,
    ]


def build_reels_audio_duck(paths: List[str], opts: Dict[str, str], out: str) -> List[str]:
    music_volume = clamp_float(opts.get("volume", "0.18"), 0.18, 0.0, 1.0)
    filter_complex = (
        f"[1:a]volume={music_volume}[music];"
        "[0:a][music]amix=inputs=2:duration=first:dropout_transition=2[a]"
    )
    return [
        "ffmpeg",
        "-i",
        paths[0],
        "-i",
        paths[1],
        "-filter_complex",
        filter_complex,
        "-map",
        "0:v",
        "-map",
        "[a]",
        "-c:v",
        "copy",
        "-c:a",
        "aac",
        "-shortest",
        "-y",
        out,
    ]


def build_faceless_quote_card(paths: List[str], opts: Dict[str, str], out: str) -> List[str]:
    duration = clamp_float(opts.get("duration", "12"), 12, 3, 180)
    font_size = clamp_int(opts.get("font_size", "68"), 68, 28, 140)
    text_path = write_wrapped_text_file(opts.get("text", ""), width=22, max_lines=9)
    vf = (
        "drawbox=x=0:y=0:w=iw:h=ih:color=0x111827@1:t=fill,"
        "drawbox=x=70:y=190:w=940:h=1540:color=0x0f766e@0.22:t=fill,"
        f"drawtext=fontfile='{filter_path(FONT_PATH)}':"
        f"textfile='{drawtext_file_path(text_path)}':fontcolor=white:"
        f"fontsize={font_size}:line_spacing=14:x=(w-text_w)/2:y=(h-text_h)/2"
    )
    return [
        "ffmpeg",
        "-f",
        "lavfi",
        "-i",
        f"color=c=0x111827:s=1080x1920:r=30:d={duration:.3f}",
        "-f",
        "lavfi",
        "-i",
        "anullsrc=channel_layout=stereo:sample_rate=44100",
        "-vf",
        vf,
        "-map",
        "0:v",
        "-map",
        "1:a",
        "-t",
        f"{duration:.3f}",
        "-c:v",
        "libx264",
        "-pix_fmt",
        "yuv420p",
        "-c:a",
        "aac",
        "-shortest",
        "-y",
        out,
    ]


def build_faceless_image_narration(paths: List[str], opts: Dict[str, str], out: str) -> List[str]:
    text_path = write_wrapped_text_file(opts.get("text", ""), width=24, max_lines=6)
    font_size = clamp_int(opts.get("font_size", "54"), 54, 24, 110)
    vf = (
        "scale=1200:-1,zoompan=z='min(zoom+0.0009,1.12)':"
        "d=1800:s=1080x1920:fps=30,"
        f"drawtext=fontfile='{filter_path(FONT_PATH)}':"
        f"textfile='{drawtext_file_path(text_path)}':fontcolor=white:"
        f"fontsize={font_size}:line_spacing=10:box=1:boxcolor=black@0.52:"
        "boxborderw=22:x=(w-text_w)/2:y=h-520"
    )
    return [
        "ffmpeg",
        "-loop",
        "1",
        "-i",
        paths[0],
        "-i",
        paths[1],
        "-vf",
        vf,
        "-map",
        "0:v",
        "-map",
        "1:a",
        "-c:v",
        "libx264",
        "-preset",
        opts.get("preset") or "fast",
        "-crf",
        str(clamp_int(opts.get("crf", "23"), 23, 12, 35)),
        "-c:a",
        "aac",
        "-shortest",
        "-pix_fmt",
        "yuv420p",
        "-y",
        out,
    ]


def build_faceless_video_narration(paths: List[str], opts: Dict[str, str], out: str) -> List[str]:
    text_path = write_wrapped_text_file(opts.get("text", ""), width=24, max_lines=6)
    font_size = clamp_int(opts.get("font_size", "52"), 52, 24, 110)
    filter_complex = (
        "[0:v]scale=1080:1920:force_original_aspect_ratio=increase,"
        "crop=1080:1920,setsar=1,"
        f"drawtext=fontfile='{filter_path(FONT_PATH)}':"
        f"textfile='{drawtext_file_path(text_path)}':fontcolor=white:"
        f"fontsize={font_size}:line_spacing=10:box=1:boxcolor=black@0.5:"
        "boxborderw=20:x=(w-text_w)/2:y=h-500[v]"
    )
    return [
        "ffmpeg",
        "-i",
        paths[0],
        "-i",
        paths[1],
        "-filter_complex",
        filter_complex,
        "-map",
        "[v]",
        "-map",
        "1:a",
        "-c:v",
        "libx264",
        "-preset",
        opts.get("preset") or "fast",
        "-crf",
        str(clamp_int(opts.get("crf", "23"), 23, 12, 35)),
        "-c:a",
        "aac",
        "-shortest",
        "-y",
        out,
    ]


def run_faceless_story_pages(paths: List[str], options: Dict[str, str], out: str) -> str:
    pages = split_story_pages(options.get("text", ""), words_per_page=24)
    seconds = clamp_float(options.get("image_duration", "3"), 3, 1, 12)
    work_dir = os.path.join(TEMP_DIR, f"story_{uuid.uuid4().hex[:8]}")
    os.makedirs(work_dir, exist_ok=True)
    list_path = os.path.join(work_dir, "concat.txt")
    try:
        with open(list_path, "w", encoding="utf-8") as concat_file:
            for index, page in enumerate(pages, start=1):
                clip_path = os.path.join(work_dir, f"page_{index:03d}.mp4")
                text_path = write_wrapped_text_file(page, width=24, max_lines=8)
                vf = (
                    "drawbox=x=0:y=0:w=iw:h=ih:color=0x0f172a@1:t=fill,"
                    f"drawtext=fontfile='{filter_path(FONT_PATH)}':"
                    f"textfile='{drawtext_file_path(text_path)}':fontcolor=white:"
                    "fontsize=62:line_spacing=12:x=(w-text_w)/2:y=(h-text_h)/2,"
                    f"drawtext=fontfile='{filter_path(FONT_PATH)}':text='{index}/{len(pages)}':"
                    "fontcolor=white@0.62:fontsize=34:x=(w-text_w)/2:y=h-150"
                )
                cmd = [
                    "ffmpeg",
                    "-f",
                    "lavfi",
                    "-i",
                    f"color=c=0x0f172a:s=1080x1920:r=30:d={seconds:.3f}",
                    "-f",
                    "lavfi",
                    "-i",
                    "anullsrc=channel_layout=stereo:sample_rate=44100",
                    "-vf",
                    vf,
                    "-map",
                    "0:v",
                    "-map",
                    "1:a",
                    "-t",
                    f"{seconds:.3f}",
                    "-c:v",
                    "libx264",
                    "-pix_fmt",
                    "yuv420p",
                    "-c:a",
                    "aac",
                    "-y",
                    clip_path,
                ]
                run_command(cmd)
                concat_file.write(f"file '{ffconcat_path(clip_path)}'\n")
        run_command(["ffmpeg", "-f", "concat", "-safe", "0", "-i", list_path, "-c", "copy", "-y", out])
        return out
    finally:
        shutil.rmtree(work_dir, ignore_errors=True)


def run_faceless_broll_montage(paths: List[str], options: Dict[str, str], out: str) -> str:
    clip_seconds = clamp_float(options.get("image_duration", "2.5"), 2.5, 1, 8)
    max_clips = clamp_int(options.get("duration", "12"), 12, 1, 60)
    work_dir = os.path.join(TEMP_DIR, f"broll_{uuid.uuid4().hex[:8]}")
    os.makedirs(work_dir, exist_ok=True)
    list_path = os.path.join(work_dir, "concat.txt")
    try:
        with open(list_path, "w", encoding="utf-8") as concat_file:
            for index, path in enumerate(paths[:max_clips], start=1):
                clip_path = os.path.join(work_dir, f"clip_{index:03d}.mp4")
                cmd = [
                    "ffmpeg",
                    "-i",
                    path,
                    "-t",
                    f"{clip_seconds:.3f}",
                    "-vf",
                    scaled_crop_filter("9:16") + ",setsar=1",
                    "-an",
                    "-c:v",
                    "libx264",
                    "-preset",
                    options.get("preset") or "fast",
                    "-crf",
                    str(clamp_int(options.get("crf", "23"), 23, 12, 35)),
                    "-pix_fmt",
                    "yuv420p",
                    "-y",
                    clip_path,
                ]
                run_command(cmd)
                concat_file.write(f"file '{ffconcat_path(clip_path)}'\n")
        run_command(["ffmpeg", "-f", "concat", "-safe", "0", "-i", list_path, "-c", "copy", "-y", out])
        return out
    finally:
        shutil.rmtree(work_dir, ignore_errors=True)


def build_series_episode_badge(paths: List[str], opts: Dict[str, str], out: str) -> List[str]:
    title = ffmpeg_escape(opts.get("text", "Mini Series"))
    font_size = clamp_int(opts.get("font_size", "48"), 48, 24, 110)
    vf = (
        scaled_crop_filter("9:16")
        + ",drawbox=x=36:y=86:w=360:h=92:color=black@0.68:t=fill,"
        + f"drawtext=fontfile='{filter_path(FONT_PATH)}':text='{title}':"
        f"fontcolor=white:fontsize={font_size}:x=60:y=106,"
        + "drawbox=x=36:y=h-238:w=1008:h=116:color=black@0.44:t=fill"
    )
    return ["ffmpeg", "-i", paths[0], "-vf", vf, "-c:a", "copy", "-y", out]


def build_series_recap_card(paths: List[str], opts: Dict[str, str], out: str) -> List[str]:
    duration = clamp_float(opts.get("duration", "5"), 5, 2, 30)
    text_path = write_wrapped_text_file(opts.get("text", "Previously in this series"), width=22, max_lines=7)
    vf = (
        "drawbox=x=0:y=0:w=iw:h=ih:color=0x111827@1:t=fill,"
        "drawbox=x=70:y=260:w=940:h=1180:color=0x7c2d12@0.32:t=fill,"
        f"drawtext=fontfile='{filter_path(FONT_PATH)}':text='PREVIOUSLY':"
        "fontcolor=white@0.72:fontsize=46:x=(w-text_w)/2:y=350,"
        f"drawtext=fontfile='{filter_path(FONT_PATH)}':textfile='{drawtext_file_path(text_path)}':"
        "fontcolor=white:fontsize=68:line_spacing=14:x=(w-text_w)/2:y=(h-text_h)/2"
    )
    return [
        "ffmpeg",
        "-f",
        "lavfi",
        "-i",
        f"color=c=0x111827:s=1080x1920:r=30:d={duration:.3f}",
        "-f",
        "lavfi",
        "-i",
        "anullsrc=channel_layout=stereo:sample_rate=44100",
        "-vf",
        vf,
        "-map",
        "0:v",
        "-map",
        "1:a",
        "-t",
        f"{duration:.3f}",
        "-c:v",
        "libx264",
        "-pix_fmt",
        "yuv420p",
        "-c:a",
        "aac",
        "-shortest",
        "-y",
        out,
    ]


def run_series_split_pack(paths: List[str], options: Dict[str, str], out: str) -> str:
    segment_seconds = clamp_float(options.get("duration", "60"), 60, 15, 180)
    title = options.get("text", "Mini Series")
    duration = media_duration_seconds(paths[0])
    episode_count = min(100, int((duration + segment_seconds - 0.001) // segment_seconds))
    work_dir = os.path.join(TEMP_DIR, f"series_{uuid.uuid4().hex[:8]}")
    os.makedirs(work_dir, exist_ok=True)
    manifest_path = os.path.join(work_dir, "manifest.txt")
    try:
        with open(manifest_path, "w", encoding="utf-8") as manifest:
            manifest.write(f"series_title={title}\n")
            manifest.write(f"episode_count={episode_count}\n")
            manifest.write(f"episode_seconds={segment_seconds:.3f}\n\n")
            for episode in range(1, episode_count + 1):
                start = (episode - 1) * segment_seconds
                remaining = max(0.1, duration - start)
                clip_duration = min(segment_seconds, remaining)
                episode_out = os.path.join(work_dir, f"episode_{episode:03d}_of_{episode_count:03d}.mp4")
                label = ffmpeg_escape(f"PART {episode}/{episode_count}")
                caption = ffmpeg_escape(title)
                vf = (
                    scaled_crop_filter("9:16")
                    + ",drawbox=x=36:y=86:w=390:h=92:color=black@0.68:t=fill,"
                    + f"drawtext=fontfile='{filter_path(FONT_PATH)}':text='{label}':"
                    "fontcolor=white:fontsize=46:x=58:y=108,"
                    + "drawbox=x=36:y=h-238:w=1008:h=116:color=black@0.48:t=fill,"
                    + f"drawtext=fontfile='{filter_path(FONT_PATH)}':text='{caption}':"
                    "fontcolor=white:fontsize=42:x=(w-text_w)/2:y=h-205"
                )
                cmd = [
                    "ffmpeg",
                    "-ss",
                    f"{start:.3f}",
                    "-i",
                    paths[0],
                    "-t",
                    f"{clip_duration:.3f}",
                    "-vf",
                    vf,
                    "-c:v",
                    "libx264",
                    "-preset",
                    options.get("preset") or "fast",
                    "-crf",
                    str(clamp_int(options.get("crf", "23"), 23, 12, 35)),
                    "-c:a",
                    "aac",
                    "-y",
                    episode_out,
                ]
                run_command(cmd)
                manifest.write(f"{episode_out}\n")
        shutil.make_archive(out[:-4], "zip", work_dir)
        return out
    finally:
        shutil.rmtree(work_dir, ignore_errors=True)


def run_series_batch_pack(paths: List[str], options: Dict[str, str], out: str) -> str:
    title = options.get("text", "Mini Series")
    work_dir = os.path.join(TEMP_DIR, f"series_batch_{uuid.uuid4().hex[:8]}")
    os.makedirs(work_dir, exist_ok=True)
    manifest_path = os.path.join(work_dir, "manifest.txt")
    try:
        with open(manifest_path, "w", encoding="utf-8") as manifest:
            manifest.write(f"series_title={title}\n")
            manifest.write(f"episode_count={len(paths)}\n\n")
            for episode, path in enumerate(paths, start=1):
                episode_out = os.path.join(work_dir, f"episode_{episode:03d}_of_{len(paths):03d}.mp4")
                label = ffmpeg_escape(f"PART {episode}/{len(paths)}")
                caption = ffmpeg_escape(title)
                vf = (
                    scaled_crop_filter("9:16")
                    + ",drawbox=x=36:y=86:w=390:h=92:color=black@0.68:t=fill,"
                    + f"drawtext=fontfile='{filter_path(FONT_PATH)}':text='{label}':"
                    "fontcolor=white:fontsize=46:x=58:y=108,"
                    + "drawbox=x=36:y=h-238:w=1008:h=116:color=black@0.48:t=fill,"
                    + f"drawtext=fontfile='{filter_path(FONT_PATH)}':text='{caption}':"
                    "fontcolor=white:fontsize=42:x=(w-text_w)/2:y=h-205"
                )
                cmd = [
                    "ffmpeg",
                    "-i",
                    path,
                    "-vf",
                    vf,
                    "-c:v",
                    "libx264",
                    "-preset",
                    options.get("preset") or "fast",
                    "-crf",
                    str(clamp_int(options.get("crf", "23"), 23, 12, 35)),
                    "-c:a",
                    "aac",
                    "-y",
                    episode_out,
                ]
                run_command(cmd)
                manifest.write(f"{episode_out}\n")
        shutil.make_archive(out[:-4], "zip", work_dir)
        return out
    finally:
        shutil.rmtree(work_dir, ignore_errors=True)


def build_concat(paths: List[str], opts: Dict[str, str], out: str) -> List[str]:
    job_id = uuid.uuid4().hex[:8]
    list_path = os.path.join(TEMP_DIR, f"list_{job_id}.txt")
    with open(list_path, "w", encoding="utf-8") as handle:
        for path in paths:
            quoted = ffconcat_path(path)
            handle.write(f"file '{quoted}'\n")
    return ["ffmpeg", "-f", "concat", "-safe", "0", "-i", list_path, "-c", "copy", "-y", out]


def build_slideshow(paths: List[str], opts: Dict[str, str], out: str) -> List[str]:
    duration = str(clamp_int(opts.get("image_duration", "3"), 3, 1, 15))
    job_id = uuid.uuid4().hex[:8]
    list_path = os.path.join(TEMP_DIR, f"slides_{job_id}.txt")
    with open(list_path, "w", encoding="utf-8") as handle:
        for path in paths:
            quoted = ffconcat_path(path)
            handle.write(f"file '{quoted}'\nduration {duration}\n")
        quoted = ffconcat_path(paths[-1])
        handle.write(f"file '{quoted}'\n")
    return [
        "ffmpeg",
        "-f",
        "concat",
        "-safe",
        "0",
        "-i",
        list_path,
        "-vsync",
        "vfr",
        "-pix_fmt",
        "yuv420p",
        "-y",
        out,
    ]


def build_trim(paths: List[str], opts: Dict[str, str], out: str) -> List[str]:
    start = opts.get("start_time") or "00:00:00"
    end = opts.get("end_time") or ""
    cmd = ["ffmpeg", "-ss", start, "-i", paths[0]]
    if end:
        cmd += ["-to", end]
    return cmd + ["-c", "copy", "-y", out]


def build_crop(paths: List[str], opts: Dict[str, str], out: str) -> List[str]:
    vf = scaled_crop_filter(opts.get("aspect_ratio", "1:1"))
    return ["ffmpeg", "-i", paths[0], "-vf", vf, "-c:a", "copy", "-y", out]


def build_waveform(paths: List[str], opts: Dict[str, str], out: str) -> List[str]:
    color = opts.get("wave_color") or "00e5ff"
    if not all(ch in "0123456789abcdefABCDEF" for ch in color) or len(color) != 6:
        color = "00e5ff"
    return [
        "ffmpeg",
        "-i",
        paths[0],
        "-filter_complex",
        f"[0:a]showwaves=s=1280x720:mode=line:colors=#{color}[v]",
        "-map",
        "[v]",
        "-map",
        "0:a",
        "-c:v",
        "libx264",
        "-pix_fmt",
        "yuv420p",
        "-c:a",
        "aac",
        "-shortest",
        "-y",
        out,
    ]


def build_extract_frames(paths: List[str], opts: Dict[str, str], out: str) -> List[str]:
    fps = str(clamp_float(opts.get("frame_rate", "1"), 1, 0.1, 30))
    pattern = out.replace(".zip", "_%04d.jpg")
    return ["ffmpeg", "-i", paths[0], "-vf", f"fps={fps}", "-y", pattern]


def build_add_intro_outro(paths: List[str], opts: Dict[str, str], out: str) -> List[str]:
    clips = paths[:]
    return build_concat(clips, opts, out)


def build_speed(paths: List[str], opts: Dict[str, str], out: str) -> List[str]:
    speed = clamp_float(opts.get("speed", "1.25"), 1.25, 0.25, 4.0)
    setpts = 1 / speed
    atempo_parts = []
    remaining = speed
    while remaining > 2.0:
        atempo_parts.append("atempo=2.0")
        remaining /= 2.0
    while remaining < 0.5:
        atempo_parts.append("atempo=0.5")
        remaining /= 0.5
    atempo_parts.append(f"atempo={remaining:.4f}")
    filter_complex = f"[0:v]setpts={setpts:.6f}*PTS[v];[0:a]{','.join(atempo_parts)}[a]"
    return [
        "ffmpeg",
        "-i",
        paths[0],
        "-filter_complex",
        filter_complex,
        "-map",
        "[v]",
        "-map",
        "[a]",
        "-y",
        out,
    ]


def build_remove_audio(paths: List[str], opts: Dict[str, str], out: str) -> List[str]:
    return ["ffmpeg", "-i", paths[0], "-c:v", "copy", "-an", "-y", out]


def build_replace_audio(paths: List[str], opts: Dict[str, str], out: str) -> List[str]:
    return [
        "ffmpeg",
        "-i",
        paths[0],
        "-i",
        paths[1],
        "-map",
        "0:v",
        "-map",
        "1:a",
        "-c:v",
        "copy",
        "-c:a",
        "aac",
        "-shortest",
        "-y",
        out,
    ]


def run_batch_compress(paths: List[str], options: Dict[str, str], out: str) -> str:
    archive_dir = os.path.join(TEMP_DIR, f"batch_{uuid.uuid4().hex[:8]}")
    os.makedirs(archive_dir, exist_ok=True)
    try:
        for index, path in enumerate(paths, start=1):
            name, _ext = os.path.splitext(os.path.basename(path))
            item_out = os.path.join(archive_dir, f"{index:02d}_{name}_compressed.mp4")
            cmd = build_compress([path], options, item_out)
            run_command(cmd)
        shutil.make_archive(out[:-4], "zip", archive_dir)
        return out
    finally:
        shutil.rmtree(archive_dir, ignore_errors=True)


TASKS: Dict[str, TaskDefinition] = {
    "normalize": TaskDefinition("Normalize MP4", "Convert to H.264/AAC MP4.", "Video", "mp4", 1, 1, ["video"], build_normalize),
    "extract_audio": TaskDefinition("Extract Audio", "Export MP3 audio from a video.", "Audio", "mp3", 1, 1, ["video", "audio"], build_extract_audio),
    "resize_916": TaskDefinition("Resize 9:16", "Crop video for vertical platforms.", "Social", "mp4", 1, 1, ["video"], build_resize),
    "add_subtitles": TaskDefinition("Add Subtitles", "Burn an SRT/ASS subtitle file into video.", "Subtitles", "mp4", 2, 2, ["video", "text"], build_subtitles, file_types=[["video"], ["text"]]),
    "burn_lyrics": TaskDefinition("Burn Lyrics", "Burn styled lyric subtitles into video.", "Subtitles", "mp4", 2, 2, ["video", "text"], build_burn_lyrics, file_types=[["video"], ["text"]]),
    "text_overlay": TaskDefinition("Text Overlay", "Add caption text to video.", "Social", "mp4", 1, 1, ["video"], build_text_overlay),
    "merge_music": TaskDefinition("Merge Music", "Add background music to video.", "Audio", "mp4", 2, 2, ["video", "audio"], build_merge_music, file_types=[["video"], ["audio"]]),
    "thumbnail": TaskDefinition("Thumbnail", "Export one JPG frame.", "Image", "jpg", 1, 1, ["video"], build_thumbnail),
    "watermark": TaskDefinition("Watermark", "Overlay a logo/image on video.", "Branding", "mp4", 2, 2, ["video", "image"], build_watermark, file_types=[["video"], ["image"]]),
    "compress": TaskDefinition("Compress", "Reduce video size using CRF and scale.", "Video", "mp4", 1, 1, ["video"], build_compress),
    "batch_compress": TaskDefinition("Batch Compress", "Compress multiple videos and return a ZIP.", "Video", "zip", 1, None, ["video"], runner=run_batch_compress),
    "make_gif": TaskDefinition("Make GIF", "Create a short GIF clip.", "Image", "gif", 1, 1, ["video"], build_gif),
    "tiktok_lyrics": TaskDefinition("TikTok Lyrics", "Vertical lyric clip with boxed center text.", "Social", "mp4", 1, 1, ["video"], build_tiktok_lyrics),
    "tiktok_pro_reframer": TaskDefinition("TikTok Pro Reframer", "Vertical crop, text, and contrast pass.", "Social", "mp4", 1, 1, ["video"], build_tiktok_pro),
    "reels_blur_fit": TaskDefinition("Reels Blur Fit", "Fit any video into 9:16 with a blurred background.", "Social", "mp4", 1, 1, ["video"], build_reels_blur_fit),
    "reels_safe_caption": TaskDefinition("Reels Safe Caption", "Add lower-third caption text inside Reels and Shorts safe zones.", "Social", "mp4", 1, 1, ["video"], build_reels_safe_caption),
    "reels_hook_title": TaskDefinition("Reels Hook Title", "Add a bold top hook/title banner to a vertical clip.", "Social", "mp4", 1, 1, ["video"], build_reels_hook_title),
    "reels_progress_bar": TaskDefinition("Reels Progress Bar", "Add a top progress bar and optional duration trim.", "Social", "mp4", 1, 1, ["video"], build_reels_progress_bar),
    "reels_loop": TaskDefinition("Reels Loop", "Loop a clip to a target vertical Shorts/Reels duration.", "Social", "mp4", 1, 1, ["video"], build_reels_loop),
    "reels_subtitle_safe": TaskDefinition("Reels Subtitle Safe", "Burn subtitles with mobile-safe lower margins.", "Social", "mp4", 2, 2, ["video", "text"], build_reels_subtitle_safe, file_types=[["video"], ["text"]]),
    "reels_reaction_stack": TaskDefinition("Reels Reaction Stack", "Stack two videos vertically for reaction-style Shorts.", "Social", "mp4", 2, 2, ["video"], build_reels_reaction_stack),
    "reels_audio_duck": TaskDefinition("Reels Audio Duck", "Mix background music under the original video audio.", "Social", "mp4", 2, 2, ["video", "audio"], build_reels_audio_duck, file_types=[["video"], ["audio"]]),
    "faceless_quote_card": TaskDefinition("Faceless Quote Card", "Generate a vertical text-only quote short from caption text.", "Faceless", "mp4", 0, 0, [], build_faceless_quote_card),
    "faceless_story_pages": TaskDefinition("Faceless Story Pages", "Split long text into timed vertical story slides.", "Faceless", "mp4", 0, 0, [], runner=run_faceless_story_pages),
    "faceless_image_narration": TaskDefinition("Faceless Image Narration", "Create a Ken Burns image short with narration audio and caption text.", "Faceless", "mp4", 2, 2, ["image", "audio"], build_faceless_image_narration, file_types=[["image"], ["audio"]]),
    "faceless_video_narration": TaskDefinition("Faceless Video Narration", "Create a vertical b-roll short with narration audio and caption text.", "Faceless", "mp4", 2, 2, ["video", "audio"], build_faceless_video_narration, file_types=[["video"], ["audio"]]),
    "faceless_broll_montage": TaskDefinition("Faceless B-roll Montage", "Create a silent vertical montage from multiple b-roll videos.", "Faceless", "mp4", 2, None, ["video"], runner=run_faceless_broll_montage),
    "series_split_pack": TaskDefinition("Mini Series Split Pack", "Split one long video into numbered vertical TikTok/Reels episodes and return a ZIP.", "Series", "zip", 1, 1, ["video"], runner=run_series_split_pack),
    "series_episode_badge": TaskDefinition("Mini Series Episode Badge", "Add a series title and episode-safe badge area to one vertical clip.", "Series", "mp4", 1, 1, ["video"], build_series_episode_badge),
    "series_batch_pack": TaskDefinition("Mini Series Batch Pack", "Number multiple clips as a TikTok/Reels mini-series and return a ZIP.", "Series", "zip", 2, None, ["video"], runner=run_series_batch_pack),
    "series_recap_card": TaskDefinition("Mini Series Recap Card", "Generate a short text-only recap card for the next episode.", "Series", "mp4", 0, 0, [], build_series_recap_card),
    "concat": TaskDefinition("Concat Videos", "Join clips in upload order.", "Video", "mp4", 2, None, ["video"], build_concat),
    "slideshow": TaskDefinition("Slideshow", "Create a video from images.", "Image", "mp4", 2, None, ["image"], build_slideshow),
    "trim": TaskDefinition("Trim", "Cut a clip by start/end time.", "Video", "mp4", 1, 1, ["video", "audio"], build_trim),
    "crop_aspect": TaskDefinition("Crop Aspect", "Crop to 1:1, 4:5, 16:9, or 9:16.", "Video", "mp4", 1, 1, ["video"], build_crop),
    "waveform": TaskDefinition("Waveform Video", "Create a waveform video from audio.", "Audio", "mp4", 1, 1, ["audio", "video"], build_waveform),
    "extract_frames": TaskDefinition("Extract Frames", "Export JPG frames and return a ZIP.", "Image", "zip", 1, 1, ["video"], build_extract_frames),
    "add_intro_outro": TaskDefinition("Add Intro/Outro", "Join intro, main clip, and outro.", "Video", "mp4", 2, 3, ["video"], build_add_intro_outro),
    "speed": TaskDefinition("Speed Change", "Speed up or slow down video and audio.", "Video", "mp4", 1, 1, ["video"], build_speed),
    "remove_audio": TaskDefinition("Remove Audio", "Export video without audio.", "Audio", "mp4", 1, 1, ["video"], build_remove_audio),
    "replace_audio": TaskDefinition("Replace Audio", "Replace video audio with another file.", "Audio", "mp4", 2, 2, ["video", "audio"], build_replace_audio, file_types=[["video"], ["audio"]]),
}


def cleanup_worker():
    while True:
        now = time.time()
        expired_outputs = []
        for name in os.listdir(TEMP_DIR):
            path = os.path.join(TEMP_DIR, name)
            try:
                if os.path.abspath(path) == os.path.abspath(STATE_DB_PATH):
                    continue
                if os.path.isfile(path) and now - os.path.getmtime(path) > FILE_TTL_SECONDS:
                    expired_outputs.append(os.path.abspath(path))
                    remove_file_quietly(path)
            except OSError:
                continue
        if expired_outputs:
            with db_connect() as connection:
                connection.executemany(
                    "DELETE FROM history WHERE output = ?",
                    [(path,) for path in expired_outputs],
                )
            load_state_store()
        time.sleep(600)


threading.Thread(target=cleanup_worker, daemon=True).start()


def file_category(path: str) -> str:
    try:
        mime_type = magic.from_file(path, mime=True) or ""
    except Exception:
        mime_type = ""
    if mime_type.startswith("video/"):
        return "video"
    if mime_type.startswith("audio/"):
        return "audio"
    if mime_type.startswith("image/"):
        return "image"
    if mime_type in {"text/plain", "application/x-subrip"} or path.lower().endswith(
        (".srt", ".ass", ".vtt")
    ):
        return "text"
    return "unknown"


def validate_files(task_id: str, paths: List[str]) -> None:
    task = get_task(task_id)
    if len(paths) < task.min_files:
        raise HTTPException(
            status_code=400,
            detail=f"{task.label} requires at least {task.min_files} file(s).",
        )
    if task.max_files is not None and len(paths) > task.max_files:
        raise HTTPException(
            status_code=400,
            detail=f"{task.label} accepts at most {task.max_files} file(s).",
        )
    for index, path in enumerate(paths):
        if not os.path.exists(path):
            raise HTTPException(status_code=400, detail=f"Input not found: {path}")
        if os.path.getsize(path) > MAX_UPLOAD_BYTES:
            raise HTTPException(
                status_code=413,
                detail=f"{os.path.basename(path)} exceeds {MAX_UPLOAD_MB} MB.",
            )
        category = file_category(path)
        expected_types = (
            task.file_types[index]
            if task.file_types and index < len(task.file_types)
            else task.accepted_types
        )
        if category not in expected_types:
            raise HTTPException(
                status_code=400,
                detail=(
                    f"{os.path.basename(path)} is {category}; "
                    f"{task.label} expects {', '.join(expected_types)}."
                ),
            )


def get_task(task_id: str) -> TaskDefinition:
    task = TASKS.get(task_id)
    if not task:
        raise HTTPException(status_code=404, detail=f"Unknown task: {task_id}")
    return task


def output_path(task_id: str, ext: str, job_id: Optional[str] = None) -> str:
    safe_job_id = job_id or uuid.uuid4().hex[:8]
    return os.path.abspath(os.path.join(TEMP_DIR, f"{task_id}_{safe_job_id}.{ext}"))


def zip_frames(pattern_prefix: str, zip_path: str) -> str:
    frame_dir = os.path.dirname(pattern_prefix)
    prefix = os.path.basename(pattern_prefix).split("_%04d")[0]
    archive_base = zip_path[:-4]
    frame_paths = [
        os.path.join(frame_dir, name)
        for name in os.listdir(frame_dir)
        if name.startswith(prefix) and name.endswith(".jpg")
    ]
    if not frame_paths:
        raise HTTPException(status_code=500, detail="No frames were generated.")
    temp_archive_dir = os.path.join(TEMP_DIR, f"frames_{uuid.uuid4().hex[:8]}")
    os.makedirs(temp_archive_dir, exist_ok=True)
    try:
        for frame_path in frame_paths:
            shutil.copy2(frame_path, os.path.join(temp_archive_dir, os.path.basename(frame_path)))
        shutil.make_archive(archive_base, "zip", temp_archive_dir)
    finally:
        shutil.rmtree(temp_archive_dir, ignore_errors=True)
    return zip_path


def run_ffmpeg(task_id: str, paths: List[str], options: Optional[Dict[str, str]] = None) -> str:
    task = get_task(task_id)
    opts = options or {}
    validate_files(task_id, paths)
    job_id = uuid.uuid4().hex[:8]
    out = output_path(task_id, task.output_ext, job_id)
    if task.runner:
        try:
            return task.runner(paths, opts, out)
        except subprocess.CalledProcessError as exc:
            stderr = exc.stderr.strip()[-2000:] or "FFmpeg failed without stderr."
            raise HTTPException(status_code=500, detail=stderr)
    if not task.builder:
        raise HTTPException(status_code=500, detail=f"{task_id} is not executable.")
    cmd = task.builder(paths, opts, out)
    try:
        run_command(cmd)
        if task_id == "extract_frames":
            out = zip_frames(out.replace(".zip", "_%04d.jpg"), out)
        return out
    except subprocess.CalledProcessError as exc:
        stderr = exc.stderr.strip()[-2000:] or "FFmpeg failed without stderr."
        raise HTTPException(status_code=500, detail=stderr)


def record_history(job_id: str, task_id: str, output: str) -> None:
    task = get_task(task_id)
    history_id = uuid.uuid4().hex[:12]
    item = {
        "history_id": history_id,
        "job_id": job_id,
        "task_id": task_id,
        "task": task.label,
        "output": output,
        "filename": os.path.basename(output),
        "size": os.path.getsize(output) if os.path.exists(output) else 0,
        "created_at": int(time.time()),
        "download_url": f"/history/{history_id}/download",
    }
    with STATE_LOCK:
        HISTORY.insert(0, item)
        del HISTORY[HISTORY_LIMIT:]
    persist_history(item)


def update_job(job_id: str, **updates: object) -> None:
    with STATE_LOCK:
        job = JOBS.get(job_id)
        if not job:
            return
        job.update(updates)
        snapshot = dict(job)
    persist_job(snapshot)


def execute_job(job_id: str, task_id: str, paths: List[str], options: Dict[str, str]) -> None:
    update_job(job_id, status="running", message="Running FFmpeg")
    try:
        output = run_ffmpeg(task_id, paths, options)
        record_history(job_id, task_id, output)
        update_job(job_id, status="complete", message="Complete", output=output)
    except HTTPException as exc:
        update_job(job_id, status="failed", message=exc.detail)
    except Exception as exc:
        update_job(job_id, status="failed", message=str(exc))


async def save_uploads(files: Optional[List[UploadFile]]) -> List[str]:
    saved = []
    if not files:
        return saved
    for upload in files:
        filename = os.path.basename(upload.filename or "upload.bin")
        path = os.path.join(TEMP_DIR, f"{uuid.uuid4().hex}_{filename}")
        size = 0
        with open(path, "wb") as handle:
            while chunk := await upload.read(1024 * 1024):
                size += len(chunk)
                if size > MAX_UPLOAD_BYTES:
                    handle.close()
                    remove_file_quietly(path)
                    raise HTTPException(
                        status_code=413,
                        detail=f"{filename} exceeds {MAX_UPLOAD_MB} MB.",
                    )
                handle.write(chunk)
        saved.append(path)
    return saved


def safe_upload_name(filename: str, fallback: str = "upload.bin") -> str:
    cleaned = os.path.basename(filename or fallback).strip()
    return cleaned or fallback


def save_bytes_file(content: bytes, filename: str) -> str:
    if len(content) > MAX_UPLOAD_BYTES:
        raise HTTPException(
            status_code=413,
            detail=f"{safe_upload_name(filename)} exceeds {MAX_UPLOAD_MB} MB.",
        )
    path = os.path.abspath(
        os.path.join(TEMP_DIR, f"{uuid.uuid4().hex}_{safe_upload_name(filename)}")
    )
    with open(path, "wb") as handle:
        handle.write(content)
    return path


def decode_base64_content(value: str) -> bytes:
    if "," in value and value.lstrip().startswith("data:"):
        value = value.split(",", 1)[1]
    try:
        return base64.b64decode(value, validate=True)
    except (binascii.Error, ValueError) as exc:
        raise HTTPException(status_code=400, detail="Invalid base64 file content.") from exc


def collect_base64_files(payload: object) -> List[str]:
    saved = []

    def walk(value: object, inherited_name: str = "n8n-upload.bin") -> None:
        if isinstance(value, list):
            for item in value:
                walk(item, inherited_name)
            return
        if not isinstance(value, dict):
            return

        filename = str(
            value.get("fileName")
            or value.get("filename")
            or value.get("name")
            or inherited_name
        )
        content = (
            value.get("content_base64")
            or value.get("base64")
            or value.get("content")
            or value.get("data")
        )
        if isinstance(content, str) and (
            value.get("mimeType")
            or value.get("mime_type")
            or value.get("fileName")
            or value.get("filename")
            or value.get("base64")
            or value.get("content_base64")
        ):
            saved.append(save_bytes_file(decode_base64_content(content), filename))
            return

        for key, nested in value.items():
            walk(nested, str(key))

    walk(payload)
    return saved


def split_url_values(value: object) -> List[str]:
    if not value:
        return list()
    if isinstance(value, list):
        urls = []
        for item in value:
            urls.extend(split_url_values(item))
        return urls
    if not isinstance(value, str):
        return list()
    chunks = value.replace(",", "\n").splitlines()
    return [chunk.strip() for chunk in chunks if chunk.strip()]


def options_from_mapping(values: Dict[str, object]) -> Dict[str, str]:
    nested_options = values.get("options")
    merged = dict(values)
    if isinstance(nested_options, dict):
        merged.update(nested_options)
    return {
        field: "" if merged.get(field) is None else str(merged.get(field, ""))
        for field in OPTION_FIELDS
    }


async def collect_n8n_inputs(request: Request) -> tuple[List[str], Dict[str, str]]:
    content_type = request.headers.get("content-type", "").lower()
    saved: List[str] = []
    values: Dict[str, object] = {}

    if "multipart/form-data" in content_type:
        form = await request.form()
        uploads = []
        urls = []
        for key, value in form.multi_items():
            if hasattr(value, "filename") and hasattr(value, "read"):
                uploads.append(value)
            else:
                values[key] = value
                if key in {"url", "urls", "file_url", "source_url", "download_url"}:
                    urls.extend(split_url_values(value))
        saved.extend(await save_uploads(uploads))
        for url in urls:
            saved.append(download_url(url))
        return saved, options_from_mapping(values)

    if "application/json" in content_type:
        payload = await request.json()
        if not isinstance(payload, dict):
            raise HTTPException(status_code=400, detail="JSON body must be an object.")
        values.update(payload)
        for key in ("url", "urls", "file_url", "source_url", "download_url"):
            for url in split_url_values(payload.get(key)):
                saved.append(download_url(url))
        saved.extend(collect_base64_files(payload.get("files", [])))
        saved.extend(collect_base64_files(payload.get("binary", {})))
        saved.extend(collect_base64_files(payload.get("data", {})))
        if not saved:
            saved.extend(collect_base64_files(payload))
        return saved, options_from_mapping(values)

    body = await request.body()
    if body:
        filename = (
            request.headers.get("x-filename")
            or request.headers.get("x-file-name")
            or "n8n-upload.bin"
        )
        saved.append(save_bytes_file(body, filename))
    return saved, options_from_mapping(dict(request.query_params))


def download_url(url: str) -> str:
    if not ALLOW_URL_INPUTS:
        raise HTTPException(status_code=403, detail="URL inputs are disabled.")
    outtmpl = os.path.join(TEMP_DIR, f"{uuid.uuid4().hex}.%(ext)s")
    ydl_opts = {
        "outtmpl": outtmpl,
        "format": "bestvideo+bestaudio/best",
        "max_filesize": MAX_URL_DOWNLOAD_BYTES,
        "noplaylist": True,
        "quiet": True,
        "no_warnings": True,
    }
    with yt_dlp.YoutubeDL(ydl_opts) as ydl:
        info = ydl.extract_info(url, download=True)
        path = ydl.prepare_filename(info)
    if not os.path.exists(path):
        raise HTTPException(status_code=502, detail="URL download did not produce a file.")
    if os.path.getsize(path) > MAX_URL_DOWNLOAD_BYTES:
        remove_file_quietly(path)
        raise HTTPException(
            status_code=413,
            detail=f"Downloaded media exceeds {MAX_URL_DOWNLOAD_MB} MB.",
        )
    return path


def options_from_form(
    text: str,
    start_time: str,
    end_time: str,
    duration: str,
    aspect_ratio: str,
    resolution: str,
    crf: str,
    preset: str,
    audio_bitrate: str,
    volume: str,
    position: str,
    opacity: str,
    fps: str,
    width: str,
    speed: str,
    timestamp: str,
    image_duration: str,
    frame_rate: str,
    font_size: str,
    wave_color: str,
) -> Dict[str, str]:
    return {
        "text": text or "",
        "start_time": start_time or "",
        "end_time": end_time or "",
        "duration": duration or "",
        "aspect_ratio": aspect_ratio or "",
        "resolution": resolution or "",
        "crf": crf or "",
        "preset": preset or "",
        "audio_bitrate": audio_bitrate or "",
        "volume": volume or "",
        "position": position or "",
        "opacity": opacity or "",
        "fps": fps or "",
        "width": width or "",
        "speed": speed or "",
        "timestamp": timestamp or "",
        "image_duration": image_duration or "",
        "frame_rate": frame_rate or "",
        "font_size": font_size or "",
        "wave_color": wave_color or "",
    }


@app.get("/healthz")
def healthz():
    return {"status": "ok", "checks": production_checks()}


@app.get("/readyz")
def readyz():
    checks = production_checks()
    if not production_ready():
        raise HTTPException(status_code=503, detail={"status": "not_ready", "checks": checks})
    return {"status": "ready", "checks": checks}


@app.get("/tasks")
def list_tasks():
    return {
        task_id: {
            "label": task.label,
            "description": task.description,
            "category": task.category,
            "output_ext": task.output_ext,
            "min_files": task.min_files,
            "max_files": task.max_files,
            "accepted_types": task.accepted_types,
            "file_types": task.file_types,
        }
        for task_id, task in TASKS.items()
    }


@app.post("/n8n/execute/{task_id}")
async def n8n_execute(task_id: str, request: Request):
    task = get_task(task_id)
    saved, options = await collect_n8n_inputs(request)
    if not saved and task.min_files > 0:
        raise HTTPException(status_code=400, detail="No n8n binary, base64 file, raw body, or URL input provided.")
    output = run_ffmpeg(task_id, saved, options)
    record_history(uuid.uuid4().hex[:8], task_id, output)
    return FileResponse(output, filename=os.path.basename(output))


@app.post("/n8n/jobs/{task_id}")
async def n8n_create_job(task_id: str, request: Request, background_tasks: BackgroundTasks):
    task = get_task(task_id)
    saved, options = await collect_n8n_inputs(request)
    if not saved and task.min_files > 0:
        raise HTTPException(status_code=400, detail="No n8n binary, base64 file, raw body, or URL input provided.")
    job_id = uuid.uuid4().hex[:12]
    job = {
        "job_id": job_id,
        "task_id": task_id,
        "status": "queued",
        "message": "Queued",
        "output": None,
        "created_at": int(time.time()),
    }
    with STATE_LOCK:
        JOBS[job_id] = job
    persist_job(job)
    background_tasks.add_task(execute_job, job_id, task_id, saved, options)
    return {"job_id": job_id, "status_url": f"/status/{job_id}"}


@app.post("/execute/{task_id}")
async def api_call(
    task_id: str,
    files: List[UploadFile] = File(None),
    url: Optional[str] = Form(None),
    text: str = Form(""),
    start_time: str = Form(""),
    end_time: str = Form(""),
    duration: str = Form(""),
    aspect_ratio: str = Form(""),
    resolution: str = Form(""),
    crf: str = Form(""),
    preset: str = Form(""),
    audio_bitrate: str = Form(""),
    volume: str = Form(""),
    position: str = Form(""),
    opacity: str = Form(""),
    fps: str = Form(""),
    width: str = Form(""),
    speed: str = Form(""),
    timestamp: str = Form(""),
    image_duration: str = Form(""),
    frame_rate: str = Form(""),
    font_size: str = Form(""),
    wave_color: str = Form(""),
):
    task = get_task(task_id)
    saved = []
    if url:
        saved.append(download_url(url))
    saved.extend(await save_uploads(files))
    if not saved and task.min_files > 0:
        raise HTTPException(status_code=400, detail="No input provided")
    options = options_from_form(
        text,
        start_time,
        end_time,
        duration,
        aspect_ratio,
        resolution,
        crf,
        preset,
        audio_bitrate,
        volume,
        position,
        opacity,
        fps,
        width,
        speed,
        timestamp,
        image_duration,
        frame_rate,
        font_size,
        wave_color,
    )
    output = run_ffmpeg(task_id, saved, options)
    record_history(uuid.uuid4().hex[:8], task_id, output)
    return FileResponse(output, filename=os.path.basename(output))


@app.post("/jobs/{task_id}")
async def create_job(
    task_id: str,
    background_tasks: BackgroundTasks,
    files: List[UploadFile] = File(None),
    url: Optional[str] = Form(None),
    text: str = Form(""),
    start_time: str = Form(""),
    end_time: str = Form(""),
    duration: str = Form(""),
    aspect_ratio: str = Form(""),
    resolution: str = Form(""),
    crf: str = Form(""),
    preset: str = Form(""),
    audio_bitrate: str = Form(""),
    volume: str = Form(""),
    position: str = Form(""),
    opacity: str = Form(""),
    fps: str = Form(""),
    width: str = Form(""),
    speed: str = Form(""),
    timestamp: str = Form(""),
    image_duration: str = Form(""),
    frame_rate: str = Form(""),
    font_size: str = Form(""),
    wave_color: str = Form(""),
):
    task = get_task(task_id)
    saved = []
    if url:
        saved.append(download_url(url))
    saved.extend(await save_uploads(files))
    if not saved and task.min_files > 0:
        raise HTTPException(status_code=400, detail="No input provided")
    options = options_from_form(
        text,
        start_time,
        end_time,
        duration,
        aspect_ratio,
        resolution,
        crf,
        preset,
        audio_bitrate,
        volume,
        position,
        opacity,
        fps,
        width,
        speed,
        timestamp,
        image_duration,
        frame_rate,
        font_size,
        wave_color,
    )
    job_id = uuid.uuid4().hex[:12]
    job = {
        "job_id": job_id,
        "task_id": task_id,
        "status": "queued",
        "message": "Queued",
        "output": None,
        "created_at": int(time.time()),
    }
    with STATE_LOCK:
        JOBS[job_id] = job
    persist_job(job)
    background_tasks.add_task(execute_job, job_id, task_id, saved, options)
    return {"job_id": job_id, "status_url": f"/status/{job_id}"}


@app.get("/status/{job_id}")
def job_status(job_id: str):
    with STATE_LOCK:
        job = JOBS.get(job_id)
        if not job:
            raise HTTPException(status_code=404, detail="Job not found")
        data = dict(job)
    if data.get("output"):
        data["download_url"] = f"/download/{job_id}"
    return data


@app.get("/download/{job_id}")
def download_job(job_id: str):
    with STATE_LOCK:
        job = JOBS.get(job_id)
        if not job:
            raise HTTPException(status_code=404, detail="Job not found")
        output = job.get("output")
    if not output or not os.path.exists(str(output)):
        raise HTTPException(status_code=404, detail="Output not available")
    return FileResponse(str(output), filename=os.path.basename(str(output)))


@app.get("/history")
def history():
    with STATE_LOCK:
        return [
            {key: value for key, value in item.items() if key != "output"}
            for item in HISTORY
        ]


@app.get("/history/{history_id}/download")
def download_history_item(history_id: str):
    with STATE_LOCK:
        item = next(
            (entry for entry in HISTORY if entry.get("history_id") == history_id), None
        )
    if not item:
        raise HTTPException(status_code=404, detail="History item not found")
    output = str(item["output"])
    if not os.path.exists(output):
        raise HTTPException(status_code=404, detail="Output expired")
    return FileResponse(output, filename=os.path.basename(output))


def task_choices() -> List[str]:
    return [f"{task.label} ({task_id})" for task_id, task in TASKS.items()]


def selected_task_id(choice: str) -> str:
    if not choice:
        return "normalize"
    if "(" in choice and choice.endswith(")"):
        return choice.rsplit("(", 1)[1][:-1]
    return choice if choice in TASKS else "normalize"


def describe_task(choice: str) -> str:
    task_id = selected_task_id(choice)
    task = get_task(task_id)
    max_files = "unlimited" if task.max_files is None else str(task.max_files)
    return (
        f"{task.description}\n\n"
        f"Files: {task.min_files}-{max_files}. "
        f"Accepted: {', '.join(task.accepted_types)}. "
        f"Output: .{task.output_ext}"
    )


def option_visibility(choice: str):
    task_id = selected_task_id(choice)
    text_tasks = {
        "text_overlay",
        "tiktok_lyrics",
        "tiktok_pro_reframer",
        "reels_safe_caption",
        "reels_hook_title",
        "faceless_quote_card",
        "faceless_story_pages",
        "faceless_image_narration",
        "faceless_video_narration",
        "series_split_pack",
        "series_episode_badge",
        "series_batch_pack",
        "series_recap_card",
    }
    duration_tasks = {
        "make_gif",
        "tiktok_lyrics",
        "reels_progress_bar",
        "reels_loop",
        "faceless_quote_card",
        "faceless_broll_montage",
        "series_split_pack",
        "series_recap_card",
    }
    quality_tasks = {
        "normalize",
        "compress",
        "batch_compress",
        "reels_blur_fit",
        "reels_loop",
        "reels_reaction_stack",
        "faceless_image_narration",
        "faceless_video_narration",
        "faceless_broll_montage",
        "series_split_pack",
        "series_batch_pack",
    }
    audio_tasks = {"extract_audio", "merge_music", "reels_audio_duck"}
    font_tasks = {
        "text_overlay",
        "tiktok_lyrics",
        "tiktok_pro_reframer",
        "reels_safe_caption",
        "reels_hook_title",
        "faceless_quote_card",
        "faceless_story_pages",
        "faceless_image_narration",
        "faceless_video_narration",
        "series_episode_badge",
        "series_recap_card",
    }
    visible = {
        "text": task_id in text_tasks,
        "trim": task_id == "trim",
        "duration": task_id in duration_tasks,
        "timestamp": task_id == "thumbnail",
        "aspect": task_id in {"resize_916", "crop_aspect"},
        "resolution": task_id == "compress",
        "quality": task_id in quality_tasks,
        "audio": task_id in audio_tasks,
        "watermark": task_id == "watermark",
        "gif": task_id == "make_gif",
        "speed": task_id == "speed",
        "slideshow": task_id in {"slideshow", "faceless_story_pages", "faceless_broll_montage"},
        "frames": task_id == "extract_frames",
        "font": task_id in font_tasks,
        "wave": task_id == "waveform",
    }
    return [
        describe_task(choice),
        gr.update(visible=visible["text"]),
        gr.update(visible=visible["trim"]),
        gr.update(visible=visible["duration"]),
        gr.update(visible=visible["timestamp"]),
        gr.update(visible=visible["aspect"]),
        gr.update(visible=visible["resolution"]),
        gr.update(visible=visible["quality"]),
        gr.update(visible=visible["audio"]),
        gr.update(visible=visible["watermark"]),
        gr.update(visible=visible["gif"]),
        gr.update(visible=visible["speed"]),
        gr.update(visible=visible["slideshow"]),
        gr.update(visible=visible["frames"]),
        gr.update(visible=visible["font"]),
        gr.update(visible=visible["wave"]),
    ]


def ui_handler(
    files,
    url,
    task_choice,
    text,
    start_time,
    end_time,
    duration,
    aspect_ratio,
    resolution,
    crf,
    preset,
    audio_bitrate,
    volume,
    position,
    opacity,
    fps,
    width,
    speed,
    timestamp,
    image_duration,
    frame_rate,
    font_size,
    wave_color,
):
    task_id = selected_task_id(task_choice)
    task = get_task(task_id)
    paths = []
    if url:
        paths.append(download_url(url))
    if files:
        paths.extend([file.name for file in files])
    if not paths and task.min_files > 0:
        raise gr.Error("Upload files or provide a URL.")
    options = options_from_form(
        text,
        start_time,
        end_time,
        duration,
        aspect_ratio,
        resolution,
        crf,
        preset,
        audio_bitrate,
        volume,
        position,
        opacity,
        fps,
        width,
        speed,
        timestamp,
        image_duration,
        frame_rate,
        font_size,
        wave_color,
    )
    try:
        output = run_ffmpeg(task_id, paths, options)
        record_history(uuid.uuid4().hex[:8], task_id, output)
        return output, history_table()
    except HTTPException as exc:
        raise gr.Error(str(exc.detail))


def history_table():
    with STATE_LOCK:
        rows = [
            [
                item["task"],
                item["filename"],
                item["size"],
                time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(item["created_at"])),
            ]
            for item in HISTORY[:10]
        ]
    return rows


with gr.Blocks(title="Basyx FFmpeg Automation Hub") as ui:
    gr.Markdown("# Basyx FFmpeg Automation Hub")
    gr.Markdown("Upload media, choose a task, tune the options, and download the output.")
    with gr.Row():
        with gr.Column(scale=1):
            task_input = gr.Dropdown(
                choices=task_choices(),
                value=task_choices()[0],
                label="Task",
            )
            task_info = gr.Textbox(
                label="Task requirements", value=describe_task(task_choices()[0]), lines=4
            )
            file_input = gr.File(label="Files", file_count="multiple")
            url_input = gr.Textbox(label="Media URL")
        with gr.Column(scale=1):
            text_input = gr.Textbox(label="Text / Caption / Lyrics", lines=3, visible=False)
            with gr.Row(visible=False) as trim_row:
                start_input = gr.Textbox(label="Start", value="00:00:00")
                end_input = gr.Textbox(label="End")
            with gr.Row(visible=False) as duration_row:
                duration_input = gr.Number(label="Duration seconds", value=5, precision=0)
            with gr.Row(visible=False) as timestamp_row:
                timestamp_input = gr.Textbox(label="Thumbnail timestamp", value="00:00:02")
            with gr.Row(visible=False) as aspect_row:
                aspect_input = gr.Dropdown(["9:16", "16:9", "1:1", "4:5"], label="Aspect", value="9:16")
            with gr.Row(visible=False) as resolution_row:
                resolution_input = gr.Dropdown(["480p", "720p", "1080p"], label="Resolution", value="720p")
            with gr.Row(visible=True) as quality_row:
                crf_input = gr.Slider(12, 35, value=23, step=1, label="CRF")
                preset_input = gr.Dropdown(
                    ["ultrafast", "veryfast", "fast", "medium", "slow"],
                    label="Preset",
                    value="fast",
                )
            with gr.Row(visible=False) as audio_row:
                audio_bitrate_input = gr.Slider(64, 320, value=192, step=16, label="Audio kbps")
                volume_input = gr.Slider(0, 2, value=0.3, step=0.05, label="Music volume")
            with gr.Row(visible=False) as watermark_row:
                position_input = gr.Dropdown(
                    ["bottom-right", "bottom-left", "top-right", "top-left", "center"],
                    label="Watermark position",
                    value="bottom-right",
                )
                opacity_input = gr.Slider(0.1, 1, value=1, step=0.05, label="Watermark opacity")
            with gr.Row(visible=False) as gif_row:
                fps_input = gr.Slider(5, 30, value=10, step=1, label="GIF FPS")
                width_input = gr.Slider(240, 1080, value=480, step=20, label="GIF width")
            with gr.Row(visible=False) as speed_row:
                speed_input = gr.Slider(0.25, 4, value=1.25, step=0.05, label="Speed")
            with gr.Row(visible=False) as slideshow_row:
                image_duration_input = gr.Slider(1, 15, value=3, step=1, label="Slide seconds")
            with gr.Row(visible=False) as frames_row:
                frame_rate_input = gr.Slider(0.1, 30, value=1, step=0.1, label="Frame FPS")
            with gr.Row(visible=False) as font_row:
                font_size_input = gr.Slider(18, 160, value=60, step=1, label="Font size")
            wave_color_input = gr.Textbox(label="Wave color hex", value="00e5ff", visible=False)
    run_button = gr.Button("Run")
    output_file = gr.File(label="Output")
    history_output = gr.Dataframe(
        headers=["Task", "Filename", "Bytes", "Created"],
        datatype=["str", "str", "number", "str"],
        label="Recent outputs",
        value=history_table,
    )

    task_input.change(
        option_visibility,
        inputs=task_input,
        outputs=[
            task_info,
            text_input,
            trim_row,
            duration_row,
            timestamp_row,
            aspect_row,
            resolution_row,
            quality_row,
            audio_row,
            watermark_row,
            gif_row,
            speed_row,
            slideshow_row,
            frames_row,
            font_row,
            wave_color_input,
        ],
    )
    run_button.click(
        ui_handler,
        inputs=[
            file_input,
            url_input,
            task_input,
            text_input,
            start_input,
            end_input,
            duration_input,
            aspect_input,
            resolution_input,
            crf_input,
            preset_input,
            audio_bitrate_input,
            volume_input,
            position_input,
            opacity_input,
            fps_input,
            width_input,
            speed_input,
            timestamp_input,
            image_duration_input,
            frame_rate_input,
            font_size_input,
            wave_color_input,
        ],
        outputs=[output_file, history_output],
    )

app = gr.mount_gradio_app(app, ui, path="/")


if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=7860)