video_format.py

"""Transcode a generated MP4 into an alternate delivery format (GIF / WebM).

Generation always produces an H.264 MP4 first — that is also what the watermark
pass (which re-encodes with libx264) emits — so the rest of the pipeline never
has to care about the requested container. When the caller asks for ``gif`` or
``webm`` via the ``output_format`` parameter, this module runs one ffmpeg pass to
transcode that MP4 to the target format. ``mp4`` is a no-op passthrough.

Every entry point degrades to "return the original mp4 path" on any failure, so a
bad or unsupported format request can never break generation — the caller just
gets the MP4 it already had.
"""
from __future__ import annotations

import os
import subprocess
import tempfile

# format key -> (file extension, MIME content-type)
_FORMATS: dict[str, tuple[str, str]] = {
    "mp4": (".mp4", "video/mp4"),
    "gif": (".gif", "image/gif"),
    "webm": (".webm", "video/webm"),
}

DEFAULT_FORMAT = "mp4"

# GIF target frame rate. GIFs balloon in size at high fps, and a palette pass
# already smooths the result, so 15 fps is a sane delivery default.
_GIF_FPS = 15


def supported_formats() -> list[str]:
    """The format keys this module can produce, mp4 first."""
    return list(_FORMATS.keys())


def normalize_format(fmt) -> str:
    """Return a supported format key, defaulting to mp4 for unknown/empty input."""
    if not fmt:
        return DEFAULT_FORMAT
    key = str(fmt).strip().lower().lstrip(".")
    return key if key in _FORMATS else DEFAULT_FORMAT


def ext_for(fmt) -> str:
    """File extension (with leading dot) for a format, e.g. ``.webm``."""
    return _FORMATS[normalize_format(fmt)][0]


def content_type_for(fmt) -> str:
    """MIME content-type for a format, e.g. ``video/webm``."""
    return _FORMATS[normalize_format(fmt)][1]


def _ffmpeg_exe() -> str:
    import imageio_ffmpeg

    return imageio_ffmpeg.get_ffmpeg_exe()


def convert(in_path: str, fmt, out_path: str | None = None) -> str:
    """Transcode ``in_path`` (an MP4) to ``fmt``; return the resulting path.

    ``mp4`` (and any unsupported/empty format) returns ``in_path`` unchanged. On
    any ffmpeg failure the original ``in_path`` is returned so delivery still
    proceeds with the MP4.
    """
    key = normalize_format(fmt)
    if key == "mp4" or not in_path:
        return in_path

    own_out = out_path is None
    try:
        ffmpeg = _ffmpeg_exe()
        if out_path is None:
            fd, out_path = tempfile.mkstemp(suffix=_FORMATS[key][0])
            os.close(fd)

        if key == "gif":
            # Two-pass palette (palettegen/paletteuse) for high-quality GIF.
            cmd = [
                ffmpeg, "-y", "-i", in_path,
                "-filter_complex",
                f"[0:v] fps={_GIF_FPS},split [a][b];"
                "[a] palettegen [p];[b][p] paletteuse",
                out_path,
            ]
        else:  # webm — VP9 video, Opus audio (audio mapping is a no-op if absent)
            cmd = [
                ffmpeg, "-y", "-i", in_path,
                "-c:v", "libvpx-vp9", "-b:v", "0", "-crf", "32",
                "-pix_fmt", "yuv420p", "-c:a", "libopus",
                out_path,
            ]
        subprocess.run(cmd, check=True, capture_output=True)
        return out_path
    except Exception as exc:  # noqa: BLE001 - never break generation
        detail = (
            exc.stderr.decode("utf-8", "ignore")[-400:]
            if isinstance(exc, subprocess.CalledProcessError)
            else exc
        )
        print(f"[video_format] {key} transcode failed: {detail}")
        # Drop the half-written output we created before falling back.
        if own_out and out_path and os.path.exists(out_path):
            try:
                os.unlink(out_path)
            except OSError:
                pass
        return in_path