用户CI Test
9fc82df6b9
CI/CD Pipeline / Validate Code Quality And Tests (pull_request) Failing after 10s
CI/CD Pipeline / Frontend Lint (pull_request) Successful in 1m32s
CI/CD Pipeline / Build & Push Staging (Watchtower auto-deploy) (pull_request) Has been skipped
CI/CD Pipeline / Build Production Runtime Images (pull_request) Has been skipped
CI/CD Pipeline / Staging E2E Tests (pull_request) Has been skipped
CI/CD Pipeline / Staging API Integration Tests (pull_request) Has been skipped
CI/CD Pipeline / Deploy Production (pull_request) Has been skipped
CI/CD Pipeline / Production Browser E2E (pull_request) Has been skipped
- build_xfade_filter_chain: 加 safe_td = min(safe_td, clip_durations[i]) - 新增 test_middle_clip_shorter_than_td 测试用例 [5.0, 0.3, 5.0] - 压力测试增加 P1 场景 + td ≤ second_input 断言 - 修复 PR #207 Comment #1282 审计发现
325 lines
10 KiB
Python
325 lines
10 KiB
Python
"""FFmpeg 工具函数 — 从 editing_modes.py / video_compose_service.py 提取的共享原语.
|
||
|
||
提供 FFmpeg / FFprobe 调用、视频信息探测、视频标准化、xfade 转场滤镜构建
|
||
等底层能力，供 EditingModeProcessor、VideoComposeService、UnifiedRenderService
|
||
共同复用。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
import shutil
|
||
import subprocess # nosec B404
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# ── 常量 ──────────────────────────────────────────────────────────────────────
|
||
|
||
FFMPEG_BIN: str = shutil.which("ffmpeg") or "ffmpeg"
|
||
FFPROBE_BIN: str = shutil.which("ffprobe") or "ffprobe"
|
||
|
||
DEFAULT_OUTPUT_WIDTH = 1280
|
||
DEFAULT_OUTPUT_HEIGHT = 720
|
||
DEFAULT_FPS = 25
|
||
|
||
# xfade 转场映射：transition_effect 名称 → FFmpeg xfade transition 名称
|
||
# 键同时支持 TransitionEffect 枚举值和字符串名称（向后兼容）
|
||
XFADE_TRANSITION_MAP: dict[str, str] = {
|
||
"fade": "fade",
|
||
"slideleft": "slideleft",
|
||
"slide_left": "slideleft",
|
||
"slideright": "slideright",
|
||
"slide_right": "slideright",
|
||
"dissolve": "dissolve",
|
||
"wipe": "wipeleft",
|
||
"wipeleft": "wipeleft",
|
||
}
|
||
|
||
DEFAULT_TRANSITION_DURATION = 0.5
|
||
|
||
|
||
# ── FFmpeg 执行 ───────────────────────────────────────────────────────────────
|
||
|
||
|
||
def run_ffmpeg(
|
||
command: list[str],
|
||
*,
|
||
capture_output: bool = True,
|
||
) -> tuple[str, str]:
|
||
"""执行 FFmpeg 命令。
|
||
|
||
Args:
|
||
command: 完整的 ffmpeg 命令列表（含 "ffmpeg" 本身）
|
||
capture_output: 是否捕获 stdout/stderr
|
||
|
||
Returns:
|
||
(stdout, stderr) 元组
|
||
|
||
Raises:
|
||
subprocess.CalledProcessError: 命令执行失败时抛出，
|
||
异常信息包含完整 stderr 以便排查。
|
||
"""
|
||
try:
|
||
result = subprocess.run( # nosec B603
|
||
command,
|
||
check=True,
|
||
stdout=subprocess.PIPE if capture_output else None,
|
||
stderr=subprocess.PIPE if capture_output else None,
|
||
text=True,
|
||
)
|
||
return (result.stdout or "", result.stderr or "")
|
||
except subprocess.CalledProcessError as e:
|
||
# 把完整 stderr 打到日志，方便排查 exit code 183 等问题
|
||
stderr_text = (e.stderr or "").strip()
|
||
logger.error(
|
||
"FFmpeg 命令失败: exit_code=%d command=%s\nstderr:\n%s",
|
||
e.returncode,
|
||
" ".join(str(c) for c in command[:20]), # 截断过长的命令
|
||
stderr_text[:5000], # 截断过长的 stderr
|
||
)
|
||
raise
|
||
|
||
|
||
def probe_duration(local_path: str | Path) -> float:
|
||
"""用 ffprobe 获取视频时长（秒）。
|
||
|
||
失败时返回默认值 5.0 秒。
|
||
"""
|
||
try:
|
||
result = subprocess.run( # nosec B603
|
||
[
|
||
FFPROBE_BIN,
|
||
"-v",
|
||
"error",
|
||
"-show_entries",
|
||
"format=duration",
|
||
"-of",
|
||
"default=noprint_wrappers=1:nokey=1",
|
||
str(local_path),
|
||
],
|
||
check=True,
|
||
stdout=subprocess.PIPE,
|
||
stderr=subprocess.PIPE,
|
||
text=True,
|
||
)
|
||
return round(float(result.stdout.strip()), 3)
|
||
except Exception:
|
||
return 5.0
|
||
|
||
|
||
def probe_video_info(video_path: str) -> dict[str, Any]:
|
||
"""获取视频信息（宽、高、时长、fps）。
|
||
|
||
Returns:
|
||
{"width": int, "height": int, "duration": float, "fps": float}
|
||
失败时返回默认值。
|
||
"""
|
||
try:
|
||
result = subprocess.run( # nosec B603
|
||
[
|
||
FFPROBE_BIN,
|
||
"-v",
|
||
"error",
|
||
"-select_streams",
|
||
"v:0",
|
||
"-show_entries",
|
||
"stream=width,height,r_frame_rate,duration",
|
||
"-show_entries",
|
||
"format=duration",
|
||
"-of",
|
||
"json",
|
||
video_path,
|
||
],
|
||
check=True,
|
||
stdout=subprocess.PIPE,
|
||
stderr=subprocess.PIPE,
|
||
text=True,
|
||
)
|
||
|
||
import json
|
||
|
||
info = json.loads(result.stdout)
|
||
stream = info.get("streams", [{}])[0]
|
||
fmt = info.get("format", {})
|
||
|
||
width = int(stream.get("width", DEFAULT_OUTPUT_WIDTH))
|
||
height = int(stream.get("height", DEFAULT_OUTPUT_HEIGHT))
|
||
|
||
# 解析帧率
|
||
fps_str = stream.get("r_frame_rate", "25/1")
|
||
if "/" in fps_str:
|
||
num, den = fps_str.split("/")
|
||
fps = float(num) / float(den) if float(den) > 0 else DEFAULT_FPS
|
||
else:
|
||
fps = float(fps_str) if fps_str else DEFAULT_FPS
|
||
|
||
# 时长
|
||
duration = float(fmt.get("duration", 0)) or float(stream.get("duration", 0))
|
||
|
||
return {
|
||
"width": width,
|
||
"height": height,
|
||
"duration": duration,
|
||
"fps": round(fps, 2),
|
||
}
|
||
except Exception as e:
|
||
logger.warning("获取视频信息失败: %s, error: %s", video_path, e)
|
||
return {
|
||
"width": DEFAULT_OUTPUT_WIDTH,
|
||
"height": DEFAULT_OUTPUT_HEIGHT,
|
||
"duration": 0.0,
|
||
"fps": DEFAULT_FPS,
|
||
}
|
||
|
||
|
||
def normalize_video(
|
||
input_path: str,
|
||
output_path: str,
|
||
*,
|
||
width: int = DEFAULT_OUTPUT_WIDTH,
|
||
height: int = DEFAULT_OUTPUT_HEIGHT,
|
||
fps: int = DEFAULT_FPS,
|
||
) -> dict[str, Any]:
|
||
"""标准化视频（缩放 + 恒定帧率）。
|
||
|
||
使用 scale + pad 保持宽高比，黑边填充到目标分辨率。
|
||
|
||
Returns:
|
||
{"width": int, "height": int, "path": str}
|
||
"""
|
||
command = [
|
||
FFMPEG_BIN,
|
||
"-y",
|
||
"-i",
|
||
input_path,
|
||
"-vf",
|
||
f"scale={width}:{height}:force_original_aspect_ratio=decrease,"
|
||
f"pad={width}:{height}:(ow-iw)/2:(oh-ih)/2:black,"
|
||
f"fps={fps}",
|
||
"-c:v",
|
||
"libx264",
|
||
"-crf",
|
||
"23",
|
||
"-preset",
|
||
"medium",
|
||
"-c:a",
|
||
"aac",
|
||
"-b:a",
|
||
"128k",
|
||
"-movflags",
|
||
"+faststart",
|
||
output_path,
|
||
]
|
||
run_ffmpeg(command)
|
||
return {"width": width, "height": height, "path": output_path}
|
||
|
||
|
||
# ── xfade / concat 滤镜构建 ──────────────────────────────────────────────────
|
||
|
||
|
||
def chain_filters(filters: list[str], output_label: str, *, input_label: str = "0:v") -> str:
|
||
"""将滤镜列表串联为 FFmpeg 滤镜字符串。
|
||
|
||
例：chain_filters(["scale=1280:720", "fps=25"], "v0")
|
||
→ "[0:v]scale=1280:720,fps=25[v0]"
|
||
"""
|
||
filter_body = ",".join(filters)
|
||
return f"[{input_label}]{filter_body}[{output_label}]"
|
||
|
||
|
||
def resolve_xfade_transition(transition_name: str) -> str:
|
||
"""将转场效果名称映射为 FFmpeg xfade transition 名称。
|
||
|
||
支持 TransitionEffect 枚举值和字符串名称，未知值回退到 "fade"。
|
||
"""
|
||
# 兼容 TransitionEffect 枚举（有 .value 属性）
|
||
if hasattr(transition_name, "value"):
|
||
transition_name = transition_name.value
|
||
return XFADE_TRANSITION_MAP.get(transition_name, "fade")
|
||
|
||
|
||
def build_xfade_filter_chain(
|
||
clip_durations: list[float],
|
||
clip_video_labels: list[str],
|
||
transitions: list[str],
|
||
*,
|
||
transition_duration: float = DEFAULT_TRANSITION_DURATION,
|
||
output_label: str = "outv",
|
||
) -> tuple[str, float]:
|
||
"""构建 xfade 转场滤镜链。
|
||
|
||
对每步 xfade 自动钳制 transition duration，确保
|
||
``offset + td ≤ first_input_duration``，避免 FFmpeg exit 234。
|
||
|
||
Args:
|
||
clip_durations: 每个片段的时长（必须与 trim 后的实际时长一致）
|
||
clip_video_labels: 每个片段的视频流标签（如 "v0", "v1"）
|
||
transitions: 每个片段对应的转场效果（第一个片段的转场被忽略）
|
||
transition_duration: 转场时长（秒）
|
||
output_label: 最终输出标签
|
||
|
||
Returns:
|
||
(filter_string, estimated_total_duration)
|
||
"""
|
||
n = len(clip_durations)
|
||
parts: list[str] = []
|
||
|
||
if n == 0:
|
||
return "", 0.0
|
||
|
||
if n == 1:
|
||
parts.append(f"[{clip_video_labels[0]}]copy[{output_label}]")
|
||
return ";".join(parts), clip_durations[0]
|
||
|
||
# xfade 链 — 每步动态钳制 td，防止 offset + td > first_input_duration
|
||
cumulative = 0.0
|
||
prev_label = clip_video_labels[0]
|
||
total_transition = 0.0 # 累计已使用的转场时长
|
||
|
||
for i in range(1, n):
|
||
cumulative += clip_durations[i - 1]
|
||
|
||
# 当前 xfade 的第一个输入时长
|
||
if i == 1:
|
||
first_input_dur = clip_durations[0]
|
||
else:
|
||
first_input_dur = cumulative - total_transition
|
||
|
||
# 原始 offset 计算
|
||
offset = max(0.0, cumulative - transition_duration * i)
|
||
|
||
# 安全钳制：offset + td 不能超过第一个输入的时长
|
||
available = max(0.0, first_input_dur - offset)
|
||
safe_td = min(transition_duration, available)
|
||
|
||
# 同时不能超过剩余总时长
|
||
remaining = max(0.0, sum(clip_durations) - cumulative)
|
||
safe_td = min(safe_td, remaining)
|
||
# 同时不能超过当前第二个输入（单个片段）的时长
|
||
safe_td = min(safe_td, clip_durations[i])
|
||
safe_td = max(0.001, safe_td) # 至少 1ms，避免 td=0
|
||
|
||
transition = transitions[i] if i < len(transitions) else "cut"
|
||
xfade_transition = resolve_xfade_transition(transition)
|
||
|
||
if i == n - 1:
|
||
out_label = output_label
|
||
else:
|
||
out_label = f"xf{i}"
|
||
|
||
parts.append(
|
||
f"[{prev_label}][{clip_video_labels[i]}]"
|
||
f"xfade=transition={xfade_transition}"
|
||
f":duration={safe_td:.3f}"
|
||
f":offset={offset:.3f}"
|
||
f"[{out_label}]"
|
||
)
|
||
prev_label = out_label
|
||
total_transition += safe_td
|
||
|
||
# 总时长减去转场重叠部分
|
||
total_duration = sum(clip_durations) - total_transition
|
||
return ";".join(parts), max(0.0, total_duration)
|