R524809
188 lines
7.5 KiB
Python
188 lines
7.5 KiB
Python
"""用 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:
|
||
process = await asyncio.create_subprocess_exec(
|
||
*args,
|
||
stdout=asyncio.subprocess.DEVNULL,
|
||
stderr=asyncio.subprocess.PIPE,
|
||
)
|
||
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()
|
||
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,
|
||
)
|
||
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)
|