File size: 3,982 Bytes
b3b2de2 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 | """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
|