"""用 FFmpeg 合成平台兼容的无音轨商品轮播视频。""" from __future__ import annotations import asyncio import json import os import tempfile from collections.abc import Awaitable, Callable from dataclasses import dataclass from pathlib import Path from config import get_settings from services.storage import ( ResourceSizeLimitError, download_bytes, image_extension, local_path, normalize_image_bytes, read_local_bytes, ) class VideoRenderError(RuntimeError): """FFmpeg 无法生成符合预设的商品视频。""" class VideoRenderCancelled(VideoRenderError): """任务在 FFmpeg 运行期间被用户取消。""" @dataclass(frozen=True) class RenderedVideo: data: bytes duration_seconds: float width: int height: int def slide_seconds_for(count: int) -> float: """按图片数量维持约 9–13 秒成片，并确保两图时也不会过短。""" by_count = {2: 4.5, 3: 3.5, 4: 3.0, 5: 2.8} try: return by_count[count] except KeyError as exc: raise VideoRenderError("轮播视频只支持 2–5 张图片") from exc async def _read_image(url: str) -> tuple[bytes, str]: path = local_path(url) if path is not None: return read_local_bytes(path), "" return await download_bytes(url, max_bytes=get_settings().image_download_max_bytes) async def _run_process( args: list[str], *, should_cancel: Callable[[], Awaitable[bool]], timeout_seconds: int, ) -> None: try: process = await asyncio.create_subprocess_exec( *args, stdout=asyncio.subprocess.DEVNULL, stderr=asyncio.subprocess.PIPE, ) except FileNotFoundError as exc: raise VideoRenderError( f"未找到视频合成工具「{args[0]}」。请安装 FFmpeg，并配置 VIDEO_FFMPEG_PATH 为其可执行文件的绝对路径。" ) from exc try: waited = 0.0 while process.returncode is None: if await should_cancel(): process.terminate() try: await asyncio.wait_for(process.wait(), timeout=5) except TimeoutError: process.kill() await process.wait() raise VideoRenderCancelled("视频任务已取消") await asyncio.sleep(0.25) waited += 0.25 if waited > timeout_seconds: process.kill() await process.wait() raise VideoRenderError(f"视频渲染超过 {timeout_seconds} 秒，请减少图片后重试") stderr = (await process.stderr.read()).decode("utf-8", errors="replace") if process.stderr else "" if process.returncode != 0: raise VideoRenderError(f"FFmpeg 渲染失败：{stderr[-800:] or '未知错误'}") finally: if process.returncode is None: process.kill() await process.wait() async def _probe(path: Path) -> dict: settings = get_settings() try: process = await asyncio.create_subprocess_exec( settings.video_ffprobe_path, "-v", "error", "-show_entries", "format=duration,size:stream=codec_name,width,height,pix_fmt", "-of", "json", str(path), stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, ) except FileNotFoundError as exc: raise VideoRenderError( f"未找到视频校验工具「{settings.video_ffprobe_path}」。请安装 FFmpeg，并配置 VIDEO_FFPROBE_PATH 为其可执行文件的绝对路径。" ) from exc stdout, stderr = await process.communicate() if process.returncode != 0: raise VideoRenderError(f"无法校验视频输出：{stderr.decode('utf-8', errors='replace')[-400:]}") try: return json.loads(stdout) except json.JSONDecodeError as exc: raise VideoRenderError("FFprobe 未返回可解析的视频信息") from exc async def render_slideshow( source_urls: list[str], *, should_cancel: Callable[[], Awaitable[bool]], ) -> RenderedVideo: """将已入库图片缩放留白后串为 MP4，严禁接收调用方任意命令参数。""" settings = get_settings() if not settings.video_ffmpeg_path or not settings.video_ffprobe_path: raise VideoRenderError("未配置 FFmpeg 或 FFprobe 可执行路径") count = len(source_urls) seconds = slide_seconds_for(count) transition = settings.video_transition_seconds with tempfile.TemporaryDirectory(prefix="osk-video-") as temp_dir: workdir = Path(temp_dir) inputs: list[Path] = [] for index, url in enumerate(source_urls): if await should_cancel(): raise VideoRenderCancelled("视频任务已取消") data, content_type = await _read_image(url) normalized = normalize_image_bytes(data, max_bytes=settings.image_download_max_bytes) suffix = image_extension(normalized.data, content_type or normalized.content_type) image_path = workdir / f"input-{index}{suffix}" image_path.write_bytes(normalized.data) inputs.append(image_path) output_path = workdir / "result.mp4" args = [settings.video_ffmpeg_path, "-y", "-hide_banner", "-loglevel", "error"] for image_path in inputs: args.extend(["-loop", "1", "-framerate", str(settings.video_fps), "-t", str(seconds), "-i", str(image_path)]) filters: list[str] = [] for index in range(count): filters.append( f"[{index}:v]scale={settings.video_width}:{settings.video_height}:" "force_original_aspect_ratio=decrease," f"pad={settings.video_width}:{settings.video_height}:(ow-iw)/2:(oh-ih):color=white," "setsar=1,format=yuv420p" f"[slide{index}]" ) previous = "slide0" accumulated = seconds for index in range(1, count): output = f"fade{index}" filters.append( f"[{previous}][slide{index}]xfade=transition=fade:duration={transition}:" f"offset={accumulated - transition:.3f}[{output}]" ) previous = output accumulated += seconds - transition filters.append(f"[{previous}]fps={settings.video_fps},format=yuv420p[video]") args.extend([ "-filter_complex", ";".join(filters), "-map", "[video]", "-an", "-c:v", "libx264", "-profile:v", "high", "-pix_fmt", "yuv420p", "-preset", "medium", "-crf", str(settings.video_crf), "-maxrate", settings.video_max_rate, "-bufsize", settings.video_buffer_size, "-movflags", "+faststart", str(output_path), ]) await _run_process(args, should_cancel=should_cancel, timeout_seconds=settings.video_render_timeout_seconds) if not output_path.is_file(): raise VideoRenderError("FFmpeg 未生成视频文件") file_size = output_path.stat().st_size if file_size > settings.video_max_bytes: raise ResourceSizeLimitError(settings.video_max_bytes, file_size) probe = await _probe(output_path) stream = next((item for item in probe.get("streams", []) if item.get("codec_name") == "h264"), None) if stream is None: raise VideoRenderError("输出视频不是 H.264 编码") width, height = int(stream.get("width", 0)), int(stream.get("height", 0)) duration = float((probe.get("format") or {}).get("duration", 0)) if (width, height) != (settings.video_width, settings.video_height): raise VideoRenderError(f"输出分辨率异常：{width} × {height}") if not 0 < duration <= settings.video_max_duration_seconds: raise VideoRenderError(f"输出视频时长异常：{duration:.1f} 秒") data = output_path.read_bytes() if len(data) != file_size: raise VideoRenderError("读取视频输出失败") return RenderedVideo(data=data, duration_seconds=duration, width=width, height=height)