"""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