Compare commits
1 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 4db9353b18 |
@@ -29,47 +29,33 @@ from packages.adapters.sqlalchemy_impl.edit_plan_repository import (
|
||||
)
|
||||
from packages.domain.edit_plan import EditPlanStatus
|
||||
from packages.domain.edit_plan_clip import EditPlanClip, EditPlanClipStatus
|
||||
from packages.domain.template_clip_config import TransitionEffect
|
||||
from packages.domain.video_filter_builder import (
|
||||
DEFAULT_FPS,
|
||||
DEFAULT_OUTPUT_HEIGHT,
|
||||
DEFAULT_OUTPUT_WIDTH,
|
||||
DEFAULT_TRANSITION_DURATION,
|
||||
ClipFilterChain,
|
||||
build_clip_filter,
|
||||
build_concat_filter as _build_concat_filter_func,
|
||||
build_filter_complex as _build_filter_complex,
|
||||
build_xfade_filter as _build_xfade_filter_func,
|
||||
chain_filters as _chain_filters_func,
|
||||
has_audio as _has_audio_func,
|
||||
)
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# ── 常量 ──────────────────────────────────────────────────────────────────────
|
||||
# ── 常量(向后兼容别名) ──────────────────────────────────────────────────────
|
||||
# 实际定义已迁移至 packages/domain/video_filter_builder.py
|
||||
|
||||
DEFAULT_OUTPUT_WIDTH = 1280
|
||||
DEFAULT_OUTPUT_HEIGHT = 720
|
||||
DEFAULT_FPS = 25
|
||||
DEFAULT_CODEC = "libx264"
|
||||
DEFAULT_CRF = 23
|
||||
DEFAULT_PRESET = "medium"
|
||||
|
||||
# xfade 转场映射:TransitionEffect → FFmpeg xfade transition 名称
|
||||
_XFADE_TRANSITION_MAP: dict[str, str] = {
|
||||
TransitionEffect.FADE: "fade",
|
||||
TransitionEffect.SLIDE_LEFT: "slideleft",
|
||||
TransitionEffect.SLIDE_RIGHT: "slideright",
|
||||
TransitionEffect.DISSOLVE: "dissolve",
|
||||
TransitionEffect.WIPE: "wipeleft",
|
||||
}
|
||||
|
||||
# 转场默认时长(秒)
|
||||
DEFAULT_TRANSITION_DURATION = 0.5
|
||||
|
||||
|
||||
# ── 数据结构 ──────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ClipFilterChain:
|
||||
"""单个片段的滤镜链描述。"""
|
||||
|
||||
clip_id: str
|
||||
input_index: int
|
||||
video_label: str
|
||||
audio_label: str | None
|
||||
filters: list[str]
|
||||
duration: float
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ComposeCommand:
|
||||
"""完整的 FFmpeg 合成命令描述。"""
|
||||
@@ -401,62 +387,8 @@ class VideoComposeService:
|
||||
output_height: int,
|
||||
fps: int,
|
||||
) -> ClipFilterChain:
|
||||
"""为单个片段构建滤镜链。
|
||||
|
||||
滤镜顺序:
|
||||
1. scale — 等比缩放到目标分辨率(保证覆盖)
|
||||
2. crop — 居中裁剪到目标分辨率
|
||||
3. fps — 统一输出帧率(concat 要求所有输入帧率一致)
|
||||
4. setpts — 重置时间戳 + 偏移
|
||||
5. trim — 视频时长裁剪
|
||||
6. atrim — 音频时长裁剪(如有音频流)
|
||||
"""
|
||||
duration = clip.duration if clip.duration > 0 else 5.0 # 默认 5 秒
|
||||
start = clip.start_time
|
||||
|
||||
filters: list[str] = []
|
||||
|
||||
# 1. scale: 等比缩放(保持比例,不裁剪)
|
||||
filters.append(f"scale={output_width}:{output_height}" f":force_original_aspect_ratio=decrease")
|
||||
|
||||
# 2. pad: 居中+留黑边到目标分辨率(保持原始比例,不裁剪内容)
|
||||
filters.append(f"pad={output_width}:{output_height}:(ow-iw)/2:(oh-ih)/2:black")
|
||||
|
||||
# 3. format: 统一像素格式为 yuv420p(H.264 标准格式,concat 要求所有输入像素格式一致)
|
||||
# 不同素材可能是 yuv420p / yuv422p / yuv444p / nv12 等,必须统一
|
||||
filters.append("format=yuv420p")
|
||||
|
||||
# 4. fps: 统一帧率(concat 要求所有输入帧率一致)
|
||||
# 放在 pad 之后、setpts 之前,确保分辨率和帧率都已统一
|
||||
if fps and fps > 0:
|
||||
filters.append(f"fps={fps}")
|
||||
|
||||
# 3. setpts: 重置时间戳
|
||||
if start > 0:
|
||||
filters.append(f"setpts=PTS-STARTPTS+{start}/TB")
|
||||
else:
|
||||
filters.append("setpts=PTS-STARTPTS")
|
||||
|
||||
# 4. trim: 视频时长
|
||||
filters.append(f"trim=0:{duration}")
|
||||
filters.append("setpts=PTS-STARTPTS") # trim 后需要重置 PTS
|
||||
|
||||
video_label = f"v{input_index}"
|
||||
|
||||
# 5. 音频标签:仅当片段类型可能有音频时才设置
|
||||
# title/subtitle 是纯文字/图片卡片,没有音频流
|
||||
clip_type = clip.clip_type.lower() if clip.clip_type else ""
|
||||
has_audio_stream = clip_type not in ("title", "subtitle")
|
||||
audio_label = f"a{input_index}" if has_audio_stream else None
|
||||
|
||||
return ClipFilterChain(
|
||||
clip_id=clip.id,
|
||||
input_index=input_index,
|
||||
video_label=video_label,
|
||||
audio_label=audio_label,
|
||||
filters=filters,
|
||||
duration=duration,
|
||||
)
|
||||
"""向后兼容:委托给 video_filter_builder.build_clip_filter。"""
|
||||
return build_clip_filter(clip, input_index, output_width, output_height, fps)
|
||||
|
||||
@staticmethod
|
||||
def _build_filter_complex(
|
||||
@@ -466,102 +398,30 @@ class VideoComposeService:
|
||||
transition_duration: float,
|
||||
transitions: list[str],
|
||||
) -> tuple[str, float]:
|
||||
"""构建完整的 filter_complex 字符串。
|
||||
|
||||
策略:
|
||||
- 单片段:直接输出
|
||||
- 多片段 + 全 cut:使用 concat 滤镜(高效)
|
||||
- 多片段 + 有转场:使用 xfade 滤镜链
|
||||
|
||||
返回 (filter_complex_string, estimated_total_duration)。
|
||||
"""
|
||||
n = len(clip_chains)
|
||||
|
||||
if n == 0:
|
||||
return "", 0.0
|
||||
|
||||
# ── 单片段 ─────────────────────────────────────────────────────
|
||||
if n == 1:
|
||||
chain = clip_chains[0]
|
||||
filter_str = _chain_filters(chain.filters, chain.video_label)
|
||||
# 音频
|
||||
if chain.audio_label:
|
||||
filter_str += f";[0:a]{chain.audio_label}"
|
||||
total_duration = chain.duration
|
||||
return filter_str, total_duration
|
||||
|
||||
# ── 检查是否有转场 ─────────────────────────────────────────────
|
||||
has_transitions = any(t != TransitionEffect.CUT and t != "cut" for t in transitions)
|
||||
|
||||
if not has_transitions:
|
||||
return _build_concat_filter(clip_chains)
|
||||
|
||||
# ── 有转场:使用 xfade ─────────────────────────────────────────
|
||||
return _build_xfade_filter(
|
||||
clip_chains=clip_chains,
|
||||
transition_duration=transition_duration,
|
||||
transitions=transitions,
|
||||
)
|
||||
"""向后兼容:委托给 video_filter_builder.build_filter_complex。"""
|
||||
return _build_filter_complex(clip_chains, output_width, output_height, transition_duration, transitions)
|
||||
|
||||
@staticmethod
|
||||
def _has_audio(clip_chains: list[ClipFilterChain]) -> bool:
|
||||
"""是否有任何片段包含音频流。"""
|
||||
return any(c.audio_label is not None for c in clip_chains)
|
||||
"""向后兼容:委托给 video_filter_builder.has_audio。"""
|
||||
return _has_audio_func(clip_chains)
|
||||
|
||||
|
||||
# ── 模块级辅助函数 ────────────────────────────────────────────────────────────
|
||||
# ── 模块级辅助函数(向后兼容别名) ──────────────────────────────────────────
|
||||
# 实际实现已迁移至 packages/domain/video_filter_builder.py
|
||||
# 保留此处别名以兼容现有测试与调用方
|
||||
|
||||
|
||||
def _chain_filters(filters: list[str], output_label: str) -> str:
|
||||
"""将滤镜列表串联为 FFmpeg 滤镜字符串。"""
|
||||
filter_body = ",".join(filters)
|
||||
return f"[0:v]{filter_body}[{output_label}]"
|
||||
"""向后兼容:委托给 video_filter_builder.chain_filters。"""
|
||||
return _chain_filters_func(filters, output_label)
|
||||
|
||||
|
||||
def _build_concat_filter(
|
||||
clip_chains: list[ClipFilterChain],
|
||||
) -> tuple[str, float]:
|
||||
"""构建 concat 滤镜(无转场,高效拼接)。
|
||||
|
||||
格式:
|
||||
[0:v]filters[v0]; [1:v]filters[v1]; ...
|
||||
[v0][v1]...[vN]concat=n=N:v=1:a=0[outv]
|
||||
"""
|
||||
n = len(clip_chains)
|
||||
parts: list[str] = []
|
||||
total_duration = 0.0
|
||||
|
||||
# 每个片段的滤镜链
|
||||
for idx, chain in enumerate(clip_chains):
|
||||
filter_body = ",".join(chain.filters)
|
||||
parts.append(f"[{idx}:v]{filter_body}[{chain.video_label}]")
|
||||
total_duration += chain.duration
|
||||
|
||||
# concat 滤镜
|
||||
concat_inputs = "".join(f"[{c.video_label}]" for c in clip_chains)
|
||||
concat_filter = f"{concat_inputs}concat=n={n}:v=1:a=0[outv]"
|
||||
parts.append(concat_filter)
|
||||
|
||||
# 音频 concat(如果有)— 先统一音频格式再拼接,否则不同采样率/声道会导致concat失败
|
||||
audio_parts: list[str] = []
|
||||
for idx, chain in enumerate(clip_chains):
|
||||
if chain.audio_label:
|
||||
# aformat: 统一采样率48000Hz + 双声道stereo + fltp采样格式(AAC标准格式)
|
||||
audio_filters = [
|
||||
"aformat=sample_rates=48000:channel_layouts=stereo:sample_fmts=fltp",
|
||||
f"atrim=0:{chain.duration}",
|
||||
"asetpts=PTS-STARTPTS",
|
||||
]
|
||||
audio_parts.append(f"[{idx}:a]{','.join(audio_filters)}[{chain.audio_label}]")
|
||||
|
||||
if audio_parts:
|
||||
parts.extend(audio_parts)
|
||||
audio_inputs = "".join(f"[{c.audio_label}]" for c in clip_chains if c.audio_label)
|
||||
audio_count = sum(1 for c in clip_chains if c.audio_label)
|
||||
if audio_count > 0:
|
||||
parts.append(f"{audio_inputs}concat=n={audio_count}:v=0:a=1[outa]")
|
||||
|
||||
return ";".join(parts), total_duration
|
||||
"""向后兼容:委托给 video_filter_builder.build_concat_filter。"""
|
||||
return _build_concat_filter_func(clip_chains)
|
||||
|
||||
|
||||
def _build_xfade_filter(
|
||||
@@ -569,80 +429,5 @@ def _build_xfade_filter(
|
||||
transition_duration: float,
|
||||
transitions: list[str],
|
||||
) -> tuple[str, float]:
|
||||
"""构建 xfade 转场滤镜链。
|
||||
|
||||
每两个相邻片段之间插入 xfade 转场。
|
||||
offset = 前一个片段的累积时长 - 转场时长。
|
||||
|
||||
格式(2 片段):
|
||||
[0:v]filters[v0]; [1:v]filters[v1];
|
||||
[v0][v1]xfade=transition=fade:duration=0.5:offset=4.5[outv]
|
||||
|
||||
格式(3+ 片段):
|
||||
[v0][v1]xfade=...[tmp1]; [tmp1][v2]xfade=...[outv]
|
||||
"""
|
||||
n = len(clip_chains)
|
||||
parts: list[str] = []
|
||||
total_duration = 0.0
|
||||
|
||||
# 每个片段的滤镜链
|
||||
for idx, chain in enumerate(clip_chains):
|
||||
filter_body = ",".join(chain.filters)
|
||||
parts.append(f"[{idx}:v]{filter_body}[{chain.video_label}]")
|
||||
total_duration += chain.duration
|
||||
|
||||
# xfade 链
|
||||
if n == 1:
|
||||
# 单片段不需要 xfade
|
||||
parts.append(f"[{clip_chains[0].video_label}]copy[outv]")
|
||||
return ";".join(parts), total_duration
|
||||
|
||||
# 计算每个转场的 offset
|
||||
cumulative = 0.0
|
||||
prev_label = clip_chains[0].video_label
|
||||
|
||||
for i in range(1, n):
|
||||
cumulative += clip_chains[i - 1].duration
|
||||
offset = max(0.0, cumulative - transition_duration * i)
|
||||
|
||||
# 获取转场类型
|
||||
transition = transitions[i] if i < len(transitions) else "cut"
|
||||
xfade_transition = _XFADE_TRANSITION_MAP.get(transition, "fade")
|
||||
|
||||
if i == n - 1:
|
||||
# 最后一个转场,输出到 [outv]
|
||||
out_label = "outv"
|
||||
else:
|
||||
out_label = f"xf{i}"
|
||||
|
||||
parts.append(
|
||||
f"[{prev_label}][{clip_chains[i].video_label}]"
|
||||
f"xfade=transition={xfade_transition}"
|
||||
f":duration={transition_duration}"
|
||||
f":offset={offset:.3f}"
|
||||
f"[{out_label}]"
|
||||
)
|
||||
prev_label = out_label
|
||||
|
||||
# 总时长需要减去转场重叠部分
|
||||
total_duration -= transition_duration * (n - 1)
|
||||
|
||||
# 音频:先 aformat 归一化再 concat(不同采样率/声道/采样格式会导致concat失败)
|
||||
audio_chains_with_label = [(c, c.audio_label) for c in clip_chains if c.audio_label]
|
||||
if len(audio_chains_with_label) >= 2:
|
||||
normalized_audio_labels: list[str] = []
|
||||
for chain, _ in audio_chains_with_label:
|
||||
norm_label = f"anorm_{chain.video_label}"
|
||||
audio_filters = [
|
||||
"aformat=sample_rates=48000:channel_layouts=stereo:sample_fmts=fltp",
|
||||
f"atrim=0:{chain.duration}",
|
||||
"asetpts=PTS-STARTPTS",
|
||||
]
|
||||
parts.append(f"[{chain.audio_label}]{','.join(audio_filters)}[{norm_label}]")
|
||||
normalized_audio_labels.append(norm_label)
|
||||
audio_inputs = "".join(f"[{label}]" for label in normalized_audio_labels)
|
||||
parts.append(f"{audio_inputs}concat=n={len(normalized_audio_labels)}:v=0:a=1[outa]")
|
||||
elif len(audio_chains_with_label) == 1:
|
||||
parts.append(f"[{audio_chains_with_label[0][0].audio_label}]acopy[outa]")
|
||||
|
||||
return ";".join(parts), max(0.0, total_duration)
|
||||
"""向后兼容:委托给 video_filter_builder.build_xfade_filter。"""
|
||||
return _build_xfade_filter_func(clip_chains, transition_duration, transitions)
|
||||
|
||||
Executable
+375
@@ -0,0 +1,375 @@
|
||||
"""视频滤镜构建器 — FFmpeg filter_complex 纯逻辑层。
|
||||
|
||||
从 video_compose_service.py 抽离的纯函数集合,专门负责 FFmpeg 滤镜链的构建,
|
||||
不依赖数据库、不做 IO,便于单元测试。
|
||||
|
||||
主要职责:
|
||||
- 单片段滤镜链构建(scale / pad / format / fps / setpts / trim)
|
||||
- concat 滤镜构建(无转场高效拼接)
|
||||
- xfade 转场滤镜链构建(fade / slide / dissolve / wipe)
|
||||
- 完整 filter_complex 策略选择与组装
|
||||
- 音频流判断与音频滤镜归一化
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from packages.domain.template_clip_config import TransitionEffect
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from packages.domain.edit_plan_clip import EditPlanClip
|
||||
|
||||
|
||||
# ── 常量 ──────────────────────────────────────────────────────────────────────
|
||||
|
||||
DEFAULT_OUTPUT_WIDTH = 1280
|
||||
DEFAULT_OUTPUT_HEIGHT = 720
|
||||
DEFAULT_FPS = 25
|
||||
|
||||
# xfade 转场映射:TransitionEffect → FFmpeg xfade transition 名称
|
||||
XFADE_TRANSITION_MAP: dict[str, str] = {
|
||||
TransitionEffect.FADE: "fade",
|
||||
TransitionEffect.SLIDE_LEFT: "slideleft",
|
||||
TransitionEffect.SLIDE_RIGHT: "slideright",
|
||||
TransitionEffect.DISSOLVE: "dissolve",
|
||||
TransitionEffect.WIPE: "wipeleft",
|
||||
}
|
||||
|
||||
# 转场默认时长(秒)
|
||||
DEFAULT_TRANSITION_DURATION = 0.5
|
||||
|
||||
# 默认片段时长(当 clip.duration <= 0 时使用)
|
||||
DEFAULT_CLIP_DURATION = 5.0
|
||||
|
||||
|
||||
# ── 数据结构 ──────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ClipFilterChain:
|
||||
"""单个片段的滤镜链描述。"""
|
||||
|
||||
clip_id: str
|
||||
input_index: int
|
||||
video_label: str
|
||||
audio_label: str | None
|
||||
filters: list[str]
|
||||
duration: float
|
||||
|
||||
|
||||
# ── 单片段滤镜链 ──────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def build_clip_filter(
|
||||
clip: "EditPlanClip",
|
||||
input_index: int,
|
||||
output_width: int,
|
||||
output_height: int,
|
||||
fps: int,
|
||||
) -> ClipFilterChain:
|
||||
"""为单个片段构建滤镜链。
|
||||
|
||||
滤镜顺序:
|
||||
1. scale — 等比缩放到目标分辨率(保证覆盖,不裁剪内容)
|
||||
2. pad — 居中+留黑边到目标分辨率(保持原始比例)
|
||||
3. format — 统一像素格式为 yuv420p(concat 要求像素格式一致)
|
||||
4. fps — 统一帧率(concat 要求所有输入帧率一致)
|
||||
5. setpts — 重置时间戳 + 起始偏移
|
||||
6. trim — 视频时长裁剪 + 重置 PTS
|
||||
|
||||
Args:
|
||||
clip: 剪辑计划片段
|
||||
input_index: 输入流索引(对应第几个 -i)
|
||||
output_width: 输出宽度(像素)
|
||||
output_height: 输出高度(像素)
|
||||
fps: 输出帧率
|
||||
|
||||
Returns:
|
||||
ClipFilterChain 描述对象
|
||||
"""
|
||||
duration = clip.duration if clip.duration > 0 else DEFAULT_CLIP_DURATION
|
||||
start = clip.start_time
|
||||
|
||||
filters: list[str] = []
|
||||
|
||||
# 1. scale: 等比缩放(保持比例,不裁剪)
|
||||
filters.append(f"scale={output_width}:{output_height}" f":force_original_aspect_ratio=decrease")
|
||||
|
||||
# 2. pad: 居中+留黑边到目标分辨率
|
||||
filters.append(f"pad={output_width}:{output_height}:(ow-iw)/2:(oh-ih)/2:black")
|
||||
|
||||
# 3. format: 统一像素格式为 yuv420p
|
||||
filters.append("format=yuv420p")
|
||||
|
||||
# 4. fps: 统一帧率
|
||||
if fps and fps > 0:
|
||||
filters.append(f"fps={fps}")
|
||||
|
||||
# 5. setpts: 重置时间戳 + 偏移
|
||||
if start > 0:
|
||||
filters.append(f"setpts=PTS-STARTPTS+{start}/TB")
|
||||
else:
|
||||
filters.append("setpts=PTS-STARTPTS")
|
||||
|
||||
# 6. trim: 视频时长 + 重置 PTS
|
||||
filters.append(f"trim=0:{duration}")
|
||||
filters.append("setpts=PTS-STARTPTS")
|
||||
|
||||
video_label = f"v{input_index}"
|
||||
|
||||
# 音频标签:title/subtitle 是纯文字/图片卡片,没有音频流
|
||||
clip_type = clip.clip_type.lower() if clip.clip_type else ""
|
||||
has_audio_stream = clip_type not in ("title", "subtitle")
|
||||
audio_label = f"a{input_index}" if has_audio_stream else None
|
||||
|
||||
return ClipFilterChain(
|
||||
clip_id=clip.id,
|
||||
input_index=input_index,
|
||||
video_label=video_label,
|
||||
audio_label=audio_label,
|
||||
filters=filters,
|
||||
duration=duration,
|
||||
)
|
||||
|
||||
|
||||
# ── 滤镜串联工具 ──────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def chain_filters(filters: list[str], output_label: str, input_label: str = "0:v") -> str:
|
||||
"""将滤镜列表串联为 FFmpeg 滤镜字符串。
|
||||
|
||||
Args:
|
||||
filters: 滤镜表达式列表
|
||||
output_label: 输出标签名(不含方括号)
|
||||
input_label: 输入标签(默认 "0:v")
|
||||
|
||||
Returns:
|
||||
形如 "[0:v]scale=1280:720,fps=25[v0]" 的字符串
|
||||
"""
|
||||
filter_body = ",".join(filters)
|
||||
return f"[{input_label}]{filter_body}[{output_label}]"
|
||||
|
||||
|
||||
# ── 音频判断 ──────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def has_audio(clip_chains: list[ClipFilterChain]) -> bool:
|
||||
"""是否有任何片段包含音频流。"""
|
||||
return any(c.audio_label is not None for c in clip_chains)
|
||||
|
||||
|
||||
# ── concat 滤镜 ──────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def build_concat_filter(
|
||||
clip_chains: list[ClipFilterChain],
|
||||
) -> tuple[str, float]:
|
||||
"""构建 concat 滤镜(无转场,高效拼接)。
|
||||
|
||||
视频:每个片段先应用各自滤镜链,再用 concat 滤镜拼接
|
||||
音频:先 aformat 归一化(48000Hz/stereo/fltp)再 concat,
|
||||
避免不同采样率/声道导致 concat 失败
|
||||
|
||||
Args:
|
||||
clip_chains: 各片段的滤镜链描述
|
||||
|
||||
Returns:
|
||||
(filter_complex_string, estimated_total_duration)
|
||||
"""
|
||||
n = len(clip_chains)
|
||||
if n == 0:
|
||||
return "", 0.0
|
||||
|
||||
parts: list[str] = []
|
||||
total_duration = 0.0
|
||||
|
||||
# 每个片段的视频滤镜链
|
||||
for idx, chain in enumerate(clip_chains):
|
||||
filter_body = ",".join(chain.filters)
|
||||
parts.append(f"[{idx}:v]{filter_body}[{chain.video_label}]")
|
||||
total_duration += chain.duration
|
||||
|
||||
# 视频 concat 滤镜
|
||||
concat_inputs = "".join(f"[{c.video_label}]" for c in clip_chains)
|
||||
parts.append(f"{concat_inputs}concat=n={n}:v=1:a=0[outv]")
|
||||
|
||||
# 音频:归一化 + concat
|
||||
_append_audio_concat(parts, clip_chains)
|
||||
|
||||
return ";".join(parts), total_duration
|
||||
|
||||
|
||||
# ── xfade 转场滤镜 ────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def build_xfade_filter(
|
||||
clip_chains: list[ClipFilterChain],
|
||||
transition_duration: float,
|
||||
transitions: list[str],
|
||||
) -> tuple[str, float]:
|
||||
"""构建 xfade 转场滤镜链。
|
||||
|
||||
每两个相邻片段之间插入 xfade 转场。
|
||||
offset = 前一个片段的累积时长 - 转场时长。
|
||||
|
||||
视频转场支持:fade / slideleft / slideright / dissolve / wipeleft
|
||||
|
||||
Args:
|
||||
clip_chains: 各片段的滤镜链描述
|
||||
transition_duration: 转场时长(秒)
|
||||
transitions: 每个片段对应的转场效果列表(索引对应片段)
|
||||
|
||||
Returns:
|
||||
(filter_complex_string, estimated_total_duration)
|
||||
"""
|
||||
n = len(clip_chains)
|
||||
if n == 0:
|
||||
return "", 0.0
|
||||
|
||||
parts: list[str] = []
|
||||
total_duration = 0.0
|
||||
|
||||
# 每个片段的视频滤镜链
|
||||
for idx, chain in enumerate(clip_chains):
|
||||
filter_body = ",".join(chain.filters)
|
||||
parts.append(f"[{idx}:v]{filter_body}[{chain.video_label}]")
|
||||
total_duration += chain.duration
|
||||
|
||||
# 单片段:直接 copy 输出(无音频,与原实现保持一致)
|
||||
if n == 1:
|
||||
parts.append(f"[{clip_chains[0].video_label}]copy[outv]")
|
||||
return ";".join(parts), total_duration
|
||||
|
||||
# xfade 链式转场
|
||||
cumulative = 0.0
|
||||
prev_label = clip_chains[0].video_label
|
||||
|
||||
for i in range(1, n):
|
||||
cumulative += clip_chains[i - 1].duration
|
||||
offset = max(0.0, cumulative - transition_duration * i)
|
||||
|
||||
# 获取转场类型
|
||||
transition = transitions[i] if i < len(transitions) else "cut"
|
||||
xfade_transition = XFADE_TRANSITION_MAP.get(transition, "fade")
|
||||
|
||||
out_label = "outv" if i == n - 1 else f"xf{i}"
|
||||
|
||||
parts.append(
|
||||
f"[{prev_label}][{clip_chains[i].video_label}]"
|
||||
f"xfade=transition={xfade_transition}"
|
||||
f":duration={transition_duration}"
|
||||
f":offset={offset:.3f}"
|
||||
f"[{out_label}]"
|
||||
)
|
||||
prev_label = out_label
|
||||
|
||||
# 总时长减去转场重叠部分
|
||||
total_duration -= transition_duration * (n - 1)
|
||||
total_duration = max(0.0, total_duration)
|
||||
|
||||
# 音频:xfade 路径下的音频处理
|
||||
# 注意:使用 chain.audio_label 作为输入标签(与原实现保持一致)
|
||||
audio_chains = [c for c in clip_chains if c.audio_label]
|
||||
if len(audio_chains) >= 2:
|
||||
normalized_labels: list[str] = []
|
||||
for chain in audio_chains:
|
||||
norm_label = f"anorm_{chain.video_label}"
|
||||
audio_filters = [
|
||||
"aformat=sample_rates=48000:channel_layouts=stereo:sample_fmts=fltp",
|
||||
f"atrim=0:{chain.duration}",
|
||||
"asetpts=PTS-STARTPTS",
|
||||
]
|
||||
parts.append(f"[{chain.audio_label}]{','.join(audio_filters)}[{norm_label}]")
|
||||
normalized_labels.append(norm_label)
|
||||
audio_inputs = "".join(f"[{label}]" for label in normalized_labels)
|
||||
parts.append(f"{audio_inputs}concat=n={len(normalized_labels)}:v=0:a=1[outa]")
|
||||
elif len(audio_chains) == 1:
|
||||
parts.append(f"[{audio_chains[0].audio_label}]acopy[outa]")
|
||||
|
||||
return ";".join(parts), total_duration
|
||||
|
||||
|
||||
# ── 完整 filter_complex 构建(策略选择) ────────────────────────────────────
|
||||
|
||||
|
||||
def build_filter_complex(
|
||||
clip_chains: list[ClipFilterChain],
|
||||
output_width: int,
|
||||
output_height: int,
|
||||
transition_duration: float,
|
||||
transitions: list[str],
|
||||
) -> tuple[str, float]:
|
||||
"""构建完整的 filter_complex 字符串(策略自动选择)。
|
||||
|
||||
策略:
|
||||
- 空列表:返回空字符串 + 0 时长
|
||||
- 单片段:直接输出(scale+pad+fps+trim 单链)
|
||||
- 多片段 + 全 cut:使用 concat 滤镜(高效)
|
||||
- 多片段 + 有转场:使用 xfade 滤镜链
|
||||
|
||||
Args:
|
||||
clip_chains: 各片段的滤镜链描述
|
||||
output_width: 输出宽度(目前单片段策略不使用,保留参数一致性)
|
||||
output_height: 输出高度(同上)
|
||||
transition_duration: 转场时长(秒)
|
||||
transitions: 每个片段对应的转场效果列表
|
||||
|
||||
Returns:
|
||||
(filter_complex_string, estimated_total_duration)
|
||||
"""
|
||||
n = len(clip_chains)
|
||||
|
||||
if n == 0:
|
||||
return "", 0.0
|
||||
|
||||
# 单片段
|
||||
if n == 1:
|
||||
chain = clip_chains[0]
|
||||
filter_str = chain_filters(chain.filters, chain.video_label)
|
||||
# 音频直通
|
||||
if chain.audio_label:
|
||||
filter_str += f";[0:a]{chain.audio_label}"
|
||||
total_duration = chain.duration
|
||||
return filter_str, total_duration
|
||||
|
||||
# 检查是否有转场
|
||||
has_transitions = any(t != TransitionEffect.CUT and t != "cut" for t in transitions)
|
||||
|
||||
if not has_transitions:
|
||||
return build_concat_filter(clip_chains)
|
||||
|
||||
# 有转场:使用 xfade
|
||||
return build_xfade_filter(
|
||||
clip_chains=clip_chains,
|
||||
transition_duration=transition_duration,
|
||||
transitions=transitions,
|
||||
)
|
||||
|
||||
|
||||
# ── 内部辅助:音频处理 ───────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def _append_audio_concat(parts: list[str], clip_chains: list[ClipFilterChain]) -> None:
|
||||
"""追加音频归一化 + concat 滤镜链到 parts(concat 路径)。
|
||||
|
||||
与原实现保持一致:归一化输出标签复用 chain.audio_label,
|
||||
concat 直接使用 audio_label 作为输入。
|
||||
"""
|
||||
audio_chains = [c for c in clip_chains if c.audio_label]
|
||||
if not audio_chains:
|
||||
return
|
||||
|
||||
# 先 aformat 归一化,输出到 chain.audio_label
|
||||
for chain in audio_chains:
|
||||
audio_filters = [
|
||||
"aformat=sample_rates=48000:channel_layouts=stereo:sample_fmts=fltp",
|
||||
f"atrim=0:{chain.duration}",
|
||||
"asetpts=PTS-STARTPTS",
|
||||
]
|
||||
parts.append(f"[{chain.input_index}:a]{','.join(audio_filters)}[{chain.audio_label}]")
|
||||
|
||||
# concat 滤镜(使用 audio_label 作为输入)
|
||||
audio_inputs = "".join(f"[{c.audio_label}]" for c in audio_chains)
|
||||
parts.append(f"{audio_inputs}concat=n={len(audio_chains)}:v=0:a=1[outa]")
|
||||
Executable
+855
@@ -0,0 +1,855 @@
|
||||
"""video_filter_builder 单元测试 — FFmpeg 滤镜构建纯逻辑层。
|
||||
|
||||
覆盖:
|
||||
- ClipFilterChain 数据类
|
||||
- 常量与映射表
|
||||
- build_clip_filter:单片段滤镜链
|
||||
- chain_filters:滤镜串联工具
|
||||
- has_audio:音频流判断
|
||||
- build_concat_filter:concat 滤镜
|
||||
- build_xfade_filter:xfade 转场滤镜
|
||||
- build_filter_complex:策略选择(空/单片段/concat/xfade)
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import unittest
|
||||
|
||||
from packages.domain.edit_plan_clip import EditPlanClip, EditPlanClipStatus
|
||||
from packages.domain.template_clip_config import TransitionEffect
|
||||
from packages.domain.video_filter_builder import (
|
||||
DEFAULT_CLIP_DURATION,
|
||||
DEFAULT_FPS,
|
||||
DEFAULT_OUTPUT_HEIGHT,
|
||||
DEFAULT_OUTPUT_WIDTH,
|
||||
DEFAULT_TRANSITION_DURATION,
|
||||
ClipFilterChain,
|
||||
XFADE_TRANSITION_MAP,
|
||||
build_clip_filter,
|
||||
build_concat_filter,
|
||||
build_filter_complex,
|
||||
build_xfade_filter,
|
||||
chain_filters,
|
||||
has_audio,
|
||||
)
|
||||
|
||||
# ── 辅助:构造 EditPlanClip ──────────────────────────────────────────────────
|
||||
|
||||
|
||||
def _make_clip(
|
||||
clip_id: str = "clip-1",
|
||||
duration: float = 5.0,
|
||||
start_time: float = 0.0,
|
||||
clip_type: str = "video",
|
||||
asset_id: str | None = "asset-1",
|
||||
) -> EditPlanClip:
|
||||
"""构造一个测试用 EditPlanClip。"""
|
||||
return EditPlanClip(
|
||||
id=clip_id,
|
||||
plan_id="plan-1",
|
||||
asset_id=asset_id,
|
||||
clip_type=clip_type,
|
||||
duration=duration,
|
||||
start_time=start_time,
|
||||
order=0,
|
||||
status=EditPlanClipStatus.READY,
|
||||
)
|
||||
|
||||
|
||||
# ── ClipFilterChain 数据类测试 ──────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestClipFilterChain(unittest.TestCase):
|
||||
"""ClipFilterChain 数据类测试。"""
|
||||
|
||||
def test_immutable(self):
|
||||
"""ClipFilterChain 是 frozen dataclass,不可修改。"""
|
||||
chain = ClipFilterChain(
|
||||
clip_id="c1",
|
||||
input_index=0,
|
||||
video_label="v0",
|
||||
audio_label="a0",
|
||||
filters=["scale=1280:720"],
|
||||
duration=5.0,
|
||||
)
|
||||
with self.assertRaises(Exception):
|
||||
chain.duration = 10.0 # type: ignore[misc]
|
||||
|
||||
def test_fields(self):
|
||||
"""所有字段正确存储。"""
|
||||
chain = ClipFilterChain(
|
||||
clip_id="c1",
|
||||
input_index=2,
|
||||
video_label="v2",
|
||||
audio_label=None,
|
||||
filters=["fps=25", "trim=0:3"],
|
||||
duration=3.0,
|
||||
)
|
||||
self.assertEqual(chain.clip_id, "c1")
|
||||
self.assertEqual(chain.input_index, 2)
|
||||
self.assertEqual(chain.video_label, "v2")
|
||||
self.assertIsNone(chain.audio_label)
|
||||
self.assertEqual(chain.filters, ["fps=25", "trim=0:3"])
|
||||
self.assertEqual(chain.duration, 3.0)
|
||||
|
||||
|
||||
# ── 常量测试 ────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestConstants(unittest.TestCase):
|
||||
"""常量与映射表测试。"""
|
||||
|
||||
def test_default_output_size(self):
|
||||
"""默认输出分辨率 1280x720。"""
|
||||
self.assertEqual(DEFAULT_OUTPUT_WIDTH, 1280)
|
||||
self.assertEqual(DEFAULT_OUTPUT_HEIGHT, 720)
|
||||
|
||||
def test_default_fps(self):
|
||||
"""默认帧率 25。"""
|
||||
self.assertEqual(DEFAULT_FPS, 25)
|
||||
|
||||
def test_default_transition_duration(self):
|
||||
"""默认转场时长 0.5 秒。"""
|
||||
self.assertEqual(DEFAULT_TRANSITION_DURATION, 0.5)
|
||||
|
||||
def test_default_clip_duration(self):
|
||||
"""默认片段时长 5 秒。"""
|
||||
self.assertEqual(DEFAULT_CLIP_DURATION, 5.0)
|
||||
|
||||
def test_xfade_transition_map_keys(self):
|
||||
"""xfade 映射包含所有转场类型。"""
|
||||
self.assertIn(TransitionEffect.FADE, XFADE_TRANSITION_MAP)
|
||||
self.assertIn(TransitionEffect.SLIDE_LEFT, XFADE_TRANSITION_MAP)
|
||||
self.assertIn(TransitionEffect.SLIDE_RIGHT, XFADE_TRANSITION_MAP)
|
||||
self.assertIn(TransitionEffect.DISSOLVE, XFADE_TRANSITION_MAP)
|
||||
self.assertIn(TransitionEffect.WIPE, XFADE_TRANSITION_MAP)
|
||||
|
||||
def test_xfade_transition_map_values(self):
|
||||
"""xfade 映射值为 FFmpeg 合法 transition 名称。"""
|
||||
self.assertEqual(XFADE_TRANSITION_MAP[TransitionEffect.FADE], "fade")
|
||||
self.assertEqual(XFADE_TRANSITION_MAP[TransitionEffect.SLIDE_LEFT], "slideleft")
|
||||
self.assertEqual(XFADE_TRANSITION_MAP[TransitionEffect.SLIDE_RIGHT], "slideright")
|
||||
self.assertEqual(XFADE_TRANSITION_MAP[TransitionEffect.DISSOLVE], "dissolve")
|
||||
self.assertEqual(XFADE_TRANSITION_MAP[TransitionEffect.WIPE], "wipeleft")
|
||||
|
||||
|
||||
# ── chain_filters 测试 ──────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestChainFilters(unittest.TestCase):
|
||||
"""chain_filters 滤镜串联工具测试。"""
|
||||
|
||||
def test_single_filter(self):
|
||||
"""单个滤镜。"""
|
||||
result = chain_filters(["scale=1280:720"], "v0")
|
||||
self.assertEqual(result, "[0:v]scale=1280:720[v0]")
|
||||
|
||||
def test_multiple_filters(self):
|
||||
"""多个滤镜用逗号串联。"""
|
||||
result = chain_filters(["scale=1280:720", "fps=25", "trim=0:5"], "v1")
|
||||
self.assertEqual(result, "[0:v]scale=1280:720,fps=25,trim=0:5[v1]")
|
||||
|
||||
def test_empty_filters(self):
|
||||
"""空滤镜列表。"""
|
||||
result = chain_filters([], "v0")
|
||||
self.assertEqual(result, "[0:v][v0]")
|
||||
|
||||
def test_custom_input_label(self):
|
||||
"""自定义输入标签。"""
|
||||
result = chain_filters(["fps=30"], "out", input_label="v0")
|
||||
self.assertEqual(result, "[v0]fps=30[out]")
|
||||
|
||||
|
||||
# ── has_audio 测试 ──────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestHasAudio(unittest.TestCase):
|
||||
"""has_audio 音频流判断测试。"""
|
||||
|
||||
def test_all_have_audio(self):
|
||||
"""所有片段都有音频。"""
|
||||
chains = [
|
||||
ClipFilterChain("c1", 0, "v0", "a0", [], 5.0),
|
||||
ClipFilterChain("c2", 1, "v1", "a1", [], 3.0),
|
||||
]
|
||||
self.assertTrue(has_audio(chains))
|
||||
|
||||
def test_some_have_audio(self):
|
||||
"""部分片段有音频。"""
|
||||
chains = [
|
||||
ClipFilterChain("c1", 0, "v0", "a0", [], 5.0),
|
||||
ClipFilterChain("c2", 1, "v1", None, [], 3.0),
|
||||
]
|
||||
self.assertTrue(has_audio(chains))
|
||||
|
||||
def test_none_have_audio(self):
|
||||
"""没有片段有音频。"""
|
||||
chains = [
|
||||
ClipFilterChain("c1", 0, "v0", None, [], 5.0),
|
||||
ClipFilterChain("c2", 1, "v1", None, [], 3.0),
|
||||
]
|
||||
self.assertFalse(has_audio(chains))
|
||||
|
||||
def test_empty_list(self):
|
||||
"""空列表返回 False。"""
|
||||
self.assertFalse(has_audio([]))
|
||||
|
||||
|
||||
# ── build_clip_filter 测试 ──────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestBuildClipFilter(unittest.TestCase):
|
||||
"""build_clip_filter 单片段滤镜链测试。"""
|
||||
|
||||
def test_basic_video_clip(self):
|
||||
"""普通视频片段生成完整滤镜链。"""
|
||||
clip = _make_clip(duration=5.0, clip_type="video")
|
||||
chain = build_clip_filter(clip, 0, 1280, 720, 25)
|
||||
|
||||
self.assertEqual(chain.clip_id, "clip-1")
|
||||
self.assertEqual(chain.input_index, 0)
|
||||
self.assertEqual(chain.video_label, "v0")
|
||||
self.assertEqual(chain.audio_label, "a0")
|
||||
self.assertEqual(chain.duration, 5.0)
|
||||
# 应有 7 个滤镜:scale, pad, format, fps, setpts, trim, setpts
|
||||
self.assertEqual(len(chain.filters), 7)
|
||||
|
||||
def test_filter_order(self):
|
||||
"""滤镜顺序:scale → pad → format → fps → setpts → trim → setpts。"""
|
||||
clip = _make_clip(duration=3.0, clip_type="video")
|
||||
chain = build_clip_filter(clip, 0, 1280, 720, 25)
|
||||
|
||||
self.assertTrue(chain.filters[0].startswith("scale="))
|
||||
self.assertTrue(chain.filters[1].startswith("pad="))
|
||||
self.assertEqual(chain.filters[2], "format=yuv420p")
|
||||
self.assertTrue(chain.filters[3].startswith("fps="))
|
||||
self.assertTrue(chain.filters[4].startswith("setpts="))
|
||||
self.assertTrue(chain.filters[5].startswith("trim="))
|
||||
self.assertEqual(chain.filters[6], "setpts=PTS-STARTPTS")
|
||||
|
||||
def test_scale_force_original_aspect_ratio(self):
|
||||
"""scale 使用 decrease 保持比例。"""
|
||||
clip = _make_clip(duration=5.0)
|
||||
chain = build_clip_filter(clip, 0, 1280, 720, 25)
|
||||
self.assertIn("force_original_aspect_ratio=decrease", chain.filters[0])
|
||||
|
||||
def test_pad_centered_black(self):
|
||||
"""pad 居中 + 黑边。"""
|
||||
clip = _make_clip(duration=5.0)
|
||||
chain = build_clip_filter(clip, 0, 1280, 720, 25)
|
||||
self.assertIn("(ow-iw)/2:(oh-ih)/2:black", chain.filters[1])
|
||||
|
||||
def test_format_yuv420p(self):
|
||||
"""像素格式统一为 yuv420p。"""
|
||||
clip = _make_clip(duration=5.0)
|
||||
chain = build_clip_filter(clip, 0, 1280, 720, 25)
|
||||
self.assertEqual(chain.filters[2], "format=yuv420p")
|
||||
|
||||
def test_custom_resolution(self):
|
||||
"""自定义输出分辨率。"""
|
||||
clip = _make_clip(duration=5.0)
|
||||
chain = build_clip_filter(clip, 0, 1920, 1080, 30)
|
||||
self.assertIn("scale=1920:1080", chain.filters[0])
|
||||
self.assertIn("pad=1920:1080", chain.filters[1])
|
||||
self.assertEqual(chain.filters[3], "fps=30")
|
||||
|
||||
def test_zero_duration_uses_default(self):
|
||||
"""duration <= 0 时使用默认时长。"""
|
||||
clip = _make_clip(duration=0.0)
|
||||
chain = build_clip_filter(clip, 0, 1280, 720, 25)
|
||||
self.assertEqual(chain.duration, DEFAULT_CLIP_DURATION)
|
||||
self.assertIn(f"trim=0:{DEFAULT_CLIP_DURATION}", chain.filters[5])
|
||||
|
||||
def test_negative_duration_uses_default(self):
|
||||
"""负时长也使用默认时长。"""
|
||||
clip = _make_clip(duration=-1.0)
|
||||
chain = build_clip_filter(clip, 0, 1280, 720, 25)
|
||||
self.assertEqual(chain.duration, DEFAULT_CLIP_DURATION)
|
||||
|
||||
def test_start_time_offset(self):
|
||||
"""start_time > 0 时 setpts 带偏移。"""
|
||||
clip = _make_clip(duration=3.0, start_time=2.0)
|
||||
chain = build_clip_filter(clip, 0, 1280, 720, 25)
|
||||
self.assertIn("PTS-STARTPTS+2.0/TB", chain.filters[4])
|
||||
|
||||
def test_zero_start_time_no_offset(self):
|
||||
"""start_time = 0 时 setpts 不带偏移。"""
|
||||
clip = _make_clip(duration=3.0, start_time=0.0)
|
||||
chain = build_clip_filter(clip, 0, 1280, 720, 25)
|
||||
self.assertEqual(chain.filters[4], "setpts=PTS-STARTPTS")
|
||||
|
||||
def test_title_clip_no_audio(self):
|
||||
"""title 类型片段没有音频。"""
|
||||
clip = _make_clip(clip_type="title")
|
||||
chain = build_clip_filter(clip, 0, 1280, 720, 25)
|
||||
self.assertIsNone(chain.audio_label)
|
||||
|
||||
def test_subtitle_clip_no_audio(self):
|
||||
"""subtitle 类型片段没有音频。"""
|
||||
clip = _make_clip(clip_type="subtitle")
|
||||
chain = build_clip_filter(clip, 0, 1280, 720, 25)
|
||||
self.assertIsNone(chain.audio_label)
|
||||
|
||||
def test_video_clip_has_audio(self):
|
||||
"""video 类型片段有音频。"""
|
||||
clip = _make_clip(clip_type="video")
|
||||
chain = build_clip_filter(clip, 0, 1280, 720, 25)
|
||||
self.assertEqual(chain.audio_label, "a0")
|
||||
|
||||
def test_image_clip_has_audio(self):
|
||||
"""image 类型片段有音频标签(可能有BGM)。"""
|
||||
clip = _make_clip(clip_type="image")
|
||||
chain = build_clip_filter(clip, 0, 1280, 720, 25)
|
||||
self.assertIsNotNone(chain.audio_label)
|
||||
|
||||
def test_clip_type_case_insensitive(self):
|
||||
"""clip_type 大小写不敏感。"""
|
||||
clip = _make_clip(clip_type="TITLE")
|
||||
chain = build_clip_filter(clip, 0, 1280, 720, 25)
|
||||
self.assertIsNone(chain.audio_label)
|
||||
|
||||
def test_empty_clip_type_has_audio(self):
|
||||
"""空 clip_type 默认有音频。"""
|
||||
clip = _make_clip(clip_type="")
|
||||
chain = build_clip_filter(clip, 0, 1280, 720, 25)
|
||||
self.assertIsNotNone(chain.audio_label)
|
||||
|
||||
def test_input_index_reflected_in_labels(self):
|
||||
"""input_index 反映在 video_label 和 audio_label 中。"""
|
||||
clip = _make_clip()
|
||||
chain = build_clip_filter(clip, 3, 1280, 720, 25)
|
||||
self.assertEqual(chain.video_label, "v3")
|
||||
self.assertEqual(chain.audio_label, "a3")
|
||||
|
||||
def test_zero_fps_skipped(self):
|
||||
"""fps = 0 时跳过 fps 滤镜。"""
|
||||
clip = _make_clip(duration=5.0)
|
||||
chain = build_clip_filter(clip, 0, 1280, 720, 0)
|
||||
# 少了 fps 滤镜:scale, pad, format, setpts, trim, setpts = 6个
|
||||
self.assertEqual(len(chain.filters), 6)
|
||||
self.assertFalse(any(f.startswith("fps=") for f in chain.filters))
|
||||
|
||||
def test_negative_fps_skipped(self):
|
||||
"""fps < 0 时也跳过 fps 滤镜。"""
|
||||
clip = _make_clip(duration=5.0)
|
||||
chain = build_clip_filter(clip, 0, 1280, 720, -1)
|
||||
self.assertEqual(len(chain.filters), 6)
|
||||
|
||||
def test_trim_uses_duration(self):
|
||||
"""trim 时长等于 clip.duration。"""
|
||||
clip = _make_clip(duration=7.5)
|
||||
chain = build_clip_filter(clip, 0, 1280, 720, 25)
|
||||
self.assertIn("trim=0:7.5", chain.filters[5])
|
||||
|
||||
|
||||
# ── build_concat_filter 测试 ────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestBuildConcatFilter(unittest.TestCase):
|
||||
"""build_concat_filter 拼接滤镜测试。"""
|
||||
|
||||
def test_empty_clips(self):
|
||||
"""空列表返回空字符串和 0 时长。"""
|
||||
filter_str, duration = build_concat_filter([])
|
||||
self.assertEqual(filter_str, "")
|
||||
self.assertEqual(duration, 0.0)
|
||||
|
||||
def test_single_clip_no_audio(self):
|
||||
"""单片段无音频:视频滤镜 + concat(n=1)。"""
|
||||
chains = [
|
||||
ClipFilterChain("c1", 0, "v0", None, ["scale=1280:720"], 5.0),
|
||||
]
|
||||
filter_str, duration = build_concat_filter(chains)
|
||||
|
||||
self.assertIn("[0:v]scale=1280:720[v0]", filter_str)
|
||||
self.assertIn("[v0]concat=n=1:v=1:a=0[outv]", filter_str)
|
||||
self.assertEqual(duration, 5.0)
|
||||
# 没有音频相关
|
||||
self.assertNotIn("[outa]", filter_str)
|
||||
|
||||
def test_single_clip_with_audio(self):
|
||||
"""单片段有音频:视频 + 音频归一化 + concat。"""
|
||||
chains = [
|
||||
ClipFilterChain("c1", 0, "v0", "a0", ["scale=1280:720"], 5.0),
|
||||
]
|
||||
filter_str, duration = build_concat_filter(chains)
|
||||
|
||||
self.assertIn("[0:v]scale=1280:720[v0]", filter_str)
|
||||
self.assertIn("[v0]concat=n=1:v=1:a=0[outv]", filter_str)
|
||||
# 音频归一化
|
||||
self.assertIn("[0:a]aformat=sample_rates=48000", filter_str)
|
||||
self.assertIn("stereo:sample_fmts=fltp", filter_str)
|
||||
self.assertIn("atrim=0:5.0", filter_str)
|
||||
self.assertIn("[a0]", filter_str)
|
||||
# 音频 concat(n=1)
|
||||
self.assertIn("[a0]concat=n=1:v=0:a=1[outa]", filter_str)
|
||||
self.assertEqual(duration, 5.0)
|
||||
|
||||
def test_two_clips_no_audio(self):
|
||||
"""两片段无音频:两个视频滤镜 + concat(n=2)。"""
|
||||
chains = [
|
||||
ClipFilterChain("c1", 0, "v0", None, ["fps=25"], 5.0),
|
||||
ClipFilterChain("c2", 1, "v1", None, ["fps=25"], 3.0),
|
||||
]
|
||||
filter_str, duration = build_concat_filter(chains)
|
||||
|
||||
self.assertIn("[0:v]fps=25[v0]", filter_str)
|
||||
self.assertIn("[1:v]fps=25[v1]", filter_str)
|
||||
self.assertIn("[v0][v1]concat=n=2:v=1:a=0[outv]", filter_str)
|
||||
self.assertEqual(duration, 8.0)
|
||||
|
||||
def test_two_clips_with_audio(self):
|
||||
"""两片段都有音频:视频 concat + 音频归一化 + 音频 concat。"""
|
||||
chains = [
|
||||
ClipFilterChain("c1", 0, "v0", "a0", ["fps=25"], 5.0),
|
||||
ClipFilterChain("c2", 1, "v1", "a1", ["fps=25"], 3.0),
|
||||
]
|
||||
filter_str, duration = build_concat_filter(chains)
|
||||
|
||||
# 视频
|
||||
self.assertIn("[v0][v1]concat=n=2:v=1:a=0[outv]", filter_str)
|
||||
# 音频归一化
|
||||
self.assertIn(
|
||||
"[0:a]aformat=sample_rates=48000:channel_layouts=stereo:sample_fmts=fltp,atrim=0:5.0,asetpts=PTS-STARTPTS[a0]",
|
||||
filter_str,
|
||||
)
|
||||
self.assertIn(
|
||||
"[1:a]aformat=sample_rates=48000:channel_layouts=stereo:sample_fmts=fltp,atrim=0:3.0,asetpts=PTS-STARTPTS[a1]",
|
||||
filter_str,
|
||||
)
|
||||
# 音频 concat
|
||||
self.assertIn("[a0][a1]concat=n=2:v=0:a=1[outa]", filter_str)
|
||||
self.assertEqual(duration, 8.0)
|
||||
|
||||
def test_mixed_audio_some_none(self):
|
||||
"""部分有音频部分没有:只有有音频的片段参与音频 concat。"""
|
||||
chains = [
|
||||
ClipFilterChain("c1", 0, "v0", "a0", ["fps=25"], 5.0),
|
||||
ClipFilterChain("c2", 1, "v1", None, ["fps=25"], 3.0),
|
||||
ClipFilterChain("c3", 2, "v2", "a2", ["fps=25"], 4.0),
|
||||
]
|
||||
filter_str, duration = build_concat_filter(chains)
|
||||
|
||||
# 视频 concat 有 3 个输入
|
||||
self.assertIn("[v0][v1][v2]concat=n=3:v=1:a=0[outv]", filter_str)
|
||||
# 音频 concat 只有 2 个输入
|
||||
self.assertIn("[a0][a2]concat=n=2:v=0:a=1[outa]", filter_str)
|
||||
# 片段 1 没有音频归一化
|
||||
self.assertNotIn("[1:a]", filter_str)
|
||||
self.assertEqual(duration, 12.0)
|
||||
|
||||
def test_three_clips_total_duration(self):
|
||||
"""三片段总时长为各片段之和。"""
|
||||
chains = [
|
||||
ClipFilterChain("c1", 0, "v0", None, [], 2.0),
|
||||
ClipFilterChain("c2", 1, "v1", None, [], 3.0),
|
||||
ClipFilterChain("c3", 2, "v2", None, [], 4.0),
|
||||
]
|
||||
_, duration = build_concat_filter(chains)
|
||||
self.assertEqual(duration, 9.0)
|
||||
|
||||
def test_audio_format_normalization(self):
|
||||
"""音频归一化包含 aformat/atrim/asetpts。"""
|
||||
chains = [
|
||||
ClipFilterChain("c1", 0, "v0", "a0", [], 5.0),
|
||||
]
|
||||
filter_str, _ = build_concat_filter(chains)
|
||||
|
||||
self.assertIn("aformat=sample_rates=48000", filter_str)
|
||||
self.assertIn("channel_layouts=stereo", filter_str)
|
||||
self.assertIn("sample_fmts=fltp", filter_str)
|
||||
self.assertIn("atrim=0:5.0", filter_str)
|
||||
self.assertIn("asetpts=PTS-STARTPTS", filter_str)
|
||||
|
||||
def test_filter_parts_separated_by_semicolon(self):
|
||||
"""各滤镜部分用分号分隔。"""
|
||||
chains = [
|
||||
ClipFilterChain("c1", 0, "v0", "a0", ["fps=25"], 5.0),
|
||||
ClipFilterChain("c2", 1, "v1", "a1", ["fps=25"], 3.0),
|
||||
]
|
||||
filter_str, _ = build_concat_filter(chains)
|
||||
parts = filter_str.split(";")
|
||||
# 2 视频 + 2 音频归一化 + 1 视频 concat + 1 音频 concat = 6
|
||||
self.assertEqual(len(parts), 6)
|
||||
|
||||
|
||||
# ── build_xfade_filter 测试 ─────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestBuildXfadeFilter(unittest.TestCase):
|
||||
"""build_xfade_filter 转场滤镜测试。"""
|
||||
|
||||
def test_empty_clips(self):
|
||||
"""空列表返回空字符串和 0 时长。"""
|
||||
filter_str, duration = build_xfade_filter([], 0.5, [])
|
||||
self.assertEqual(filter_str, "")
|
||||
self.assertEqual(duration, 0.0)
|
||||
|
||||
def test_single_clip_no_audio(self):
|
||||
"""单片段无音频:视频滤镜 + copy。"""
|
||||
chains = [
|
||||
ClipFilterChain("c1", 0, "v0", None, ["scale=1280:720"], 5.0),
|
||||
]
|
||||
filter_str, duration = build_xfade_filter(chains, 0.5, ["fade"])
|
||||
|
||||
self.assertIn("[0:v]scale=1280:720[v0]", filter_str)
|
||||
self.assertIn("[v0]copy[outv]", filter_str)
|
||||
self.assertEqual(duration, 5.0)
|
||||
# 单片段 xfade 没有音频输出
|
||||
self.assertNotIn("[outa]", filter_str)
|
||||
|
||||
def test_single_clip_with_audio(self):
|
||||
"""单片段有音频:xfade 路径下单片段不输出音频(与原实现一致)。"""
|
||||
chains = [
|
||||
ClipFilterChain("c1", 0, "v0", "a0", ["scale=1280:720"], 5.0),
|
||||
]
|
||||
filter_str, _ = build_xfade_filter(chains, 0.5, ["fade"])
|
||||
# 单片段 xfade 没有音频输出
|
||||
self.assertNotIn("[outa]", filter_str)
|
||||
self.assertNotIn("acopy", filter_str)
|
||||
|
||||
def test_two_clips_fade_transition(self):
|
||||
"""两片段 fade 转场:xfade 滤镜结构正确。"""
|
||||
chains = [
|
||||
ClipFilterChain("c1", 0, "v0", None, ["fps=25"], 5.0),
|
||||
ClipFilterChain("c2", 1, "v1", None, ["fps=25"], 3.0),
|
||||
]
|
||||
filter_str, duration = build_xfade_filter(chains, 0.5, ["cut", "fade"])
|
||||
|
||||
# 两个视频滤镜链
|
||||
self.assertIn("[0:v]fps=25[v0]", filter_str)
|
||||
self.assertIn("[1:v]fps=25[v1]", filter_str)
|
||||
# xfade 转场
|
||||
self.assertIn("xfade=transition=fade", filter_str)
|
||||
self.assertIn(":duration=0.5", filter_str)
|
||||
self.assertIn("[outv]", filter_str)
|
||||
# 总时长 = 5 + 3 - 0.5 = 7.5
|
||||
self.assertAlmostEqual(duration, 7.5)
|
||||
|
||||
def test_two_clips_offset_calculation(self):
|
||||
"""转场 offset = 第一个片段时长 - 转场时长。"""
|
||||
chains = [
|
||||
ClipFilterChain("c1", 0, "v0", None, [], 5.0),
|
||||
ClipFilterChain("c2", 1, "v1", None, [], 3.0),
|
||||
]
|
||||
filter_str, _ = build_xfade_filter(chains, 0.5, ["cut", "fade"])
|
||||
|
||||
# offset = 5.0 - 0.5 * 1 = 4.5
|
||||
self.assertIn(":offset=4.500", filter_str)
|
||||
|
||||
def test_three_clips_chain(self):
|
||||
"""三片段链式转场:两个 xfade,中间用 xf1 标签。"""
|
||||
chains = [
|
||||
ClipFilterChain("c1", 0, "v0", None, [], 4.0),
|
||||
ClipFilterChain("c2", 1, "v1", None, [], 3.0),
|
||||
ClipFilterChain("c3", 2, "v2", None, [], 5.0),
|
||||
]
|
||||
filter_str, duration = build_xfade_filter(chains, 0.5, ["cut", "fade", "dissolve"])
|
||||
|
||||
# 第一个转场输出到 xf1
|
||||
self.assertIn("[xf1]", filter_str)
|
||||
# 第二个转场输出到 outv
|
||||
self.assertIn("[xf1][v2]xfade=transition=dissolve", filter_str)
|
||||
self.assertIn("[outv]", filter_str)
|
||||
# 总时长 = 4 + 3 + 5 - 0.5 * 2 = 11.0
|
||||
self.assertAlmostEqual(duration, 11.0)
|
||||
|
||||
def test_three_clips_offsets(self):
|
||||
"""三片段两个转场的 offset 计算正确。"""
|
||||
chains = [
|
||||
ClipFilterChain("c1", 0, "v0", None, [], 4.0),
|
||||
ClipFilterChain("c2", 1, "v1", None, [], 3.0),
|
||||
ClipFilterChain("c3", 2, "v2", None, [], 5.0),
|
||||
]
|
||||
filter_str, _ = build_xfade_filter(chains, 0.5, ["cut", "fade", "slideleft"])
|
||||
|
||||
# 第一个 offset = 4.0 - 0.5*1 = 3.5
|
||||
# 第二个 offset = (4.0+3.0) - 0.5*2 = 7.0 - 1.0 = 6.0
|
||||
self.assertIn(":offset=3.500", filter_str)
|
||||
self.assertIn(":offset=6.000", filter_str)
|
||||
|
||||
def test_all_transition_types(self):
|
||||
"""所有转场类型都能正确映射。"""
|
||||
transitions = [
|
||||
(TransitionEffect.FADE, "fade"),
|
||||
(TransitionEffect.SLIDE_LEFT, "slideleft"),
|
||||
(TransitionEffect.SLIDE_RIGHT, "slideright"),
|
||||
(TransitionEffect.DISSOLVE, "dissolve"),
|
||||
(TransitionEffect.WIPE, "wipeleft"),
|
||||
]
|
||||
for effect, expected_name in transitions:
|
||||
chains = [
|
||||
ClipFilterChain("c1", 0, "v0", None, [], 3.0),
|
||||
ClipFilterChain("c2", 1, "v1", None, [], 2.0),
|
||||
]
|
||||
filter_str, _ = build_xfade_filter(chains, 0.5, ["cut", effect])
|
||||
self.assertIn(
|
||||
f"xfade=transition={expected_name}",
|
||||
filter_str,
|
||||
f"Transition {effect} should map to {expected_name}",
|
||||
)
|
||||
|
||||
def test_unknown_transition_defaults_to_fade(self):
|
||||
"""未知转场类型默认使用 fade。"""
|
||||
chains = [
|
||||
ClipFilterChain("c1", 0, "v0", None, [], 3.0),
|
||||
ClipFilterChain("c2", 1, "v1", None, [], 2.0),
|
||||
]
|
||||
filter_str, _ = build_xfade_filter(chains, 0.5, ["cut", "nonexistent"])
|
||||
self.assertIn("xfade=transition=fade", filter_str)
|
||||
|
||||
def test_cut_still_uses_fade(self):
|
||||
"""cut 类型在 xfade 路径下也映射为 fade(因为走了 xfade 分支)。"""
|
||||
chains = [
|
||||
ClipFilterChain("c1", 0, "v0", None, [], 3.0),
|
||||
ClipFilterChain("c2", 1, "v1", None, [], 2.0),
|
||||
]
|
||||
# 只要有一个非 cut 就走 xfade,cut 的那个也用 fade 作为默认
|
||||
filter_str, _ = build_xfade_filter(chains, 0.5, ["fade", "cut"])
|
||||
# 第二个转场是 cut,默认用 fade
|
||||
self.assertIn("xfade=transition=fade", filter_str)
|
||||
|
||||
def test_transitions_shorter_than_clips(self):
|
||||
"""transitions 列表比 clip 短时,超出部分默认 cut→fade。"""
|
||||
chains = [
|
||||
ClipFilterChain("c1", 0, "v0", None, [], 2.0),
|
||||
ClipFilterChain("c2", 1, "v1", None, [], 2.0),
|
||||
ClipFilterChain("c3", 2, "v2", None, [], 2.0),
|
||||
]
|
||||
# 只给 1 个 transition(索引0),索引1和2会越界
|
||||
filter_str, _ = build_xfade_filter(chains, 0.5, ["fade"])
|
||||
# 应该有两个 xfade,都用 fade(第二个是默认值)
|
||||
self.assertEqual(filter_str.count("xfade=transition=fade"), 2)
|
||||
|
||||
def test_zero_transition_duration(self):
|
||||
"""转场时长为 0 时不减时长。"""
|
||||
chains = [
|
||||
ClipFilterChain("c1", 0, "v0", None, [], 5.0),
|
||||
ClipFilterChain("c2", 1, "v1", None, [], 3.0),
|
||||
]
|
||||
_, duration = build_xfade_filter(chains, 0.0, ["cut", "fade"])
|
||||
self.assertAlmostEqual(duration, 8.0)
|
||||
|
||||
def test_total_duration_not_negative(self):
|
||||
"""总时长不会为负数。"""
|
||||
chains = [
|
||||
ClipFilterChain("c1", 0, "v0", None, [], 0.1),
|
||||
ClipFilterChain("c2", 1, "v1", None, [], 0.1),
|
||||
]
|
||||
_, duration = build_xfade_filter(chains, 10.0, ["cut", "fade"])
|
||||
self.assertGreaterEqual(duration, 0.0)
|
||||
|
||||
def test_two_clips_with_audio_normalize_and_concat(self):
|
||||
"""两片段都有音频:归一化 + concat。"""
|
||||
chains = [
|
||||
ClipFilterChain("c1", 0, "v0", "a0", [], 5.0),
|
||||
ClipFilterChain("c2", 1, "v1", "a1", [], 3.0),
|
||||
]
|
||||
filter_str, _ = build_xfade_filter(chains, 0.5, ["cut", "fade"])
|
||||
|
||||
# 音频归一化(注意:xfade 路径用 audio_label 作为输入,与原实现一致)
|
||||
self.assertIn(
|
||||
"[a0]aformat=sample_rates=48000:channel_layouts=stereo:sample_fmts=fltp,atrim=0:5.0,asetpts=PTS-STARTPTS[anorm_v0]",
|
||||
filter_str,
|
||||
)
|
||||
self.assertIn(
|
||||
"[a1]aformat=sample_rates=48000:channel_layouts=stereo:sample_fmts=fltp,atrim=0:3.0,asetpts=PTS-STARTPTS[anorm_v1]",
|
||||
filter_str,
|
||||
)
|
||||
# 音频 concat
|
||||
self.assertIn("[anorm_v0][anorm_v1]concat=n=2:v=0:a=1[outa]", filter_str)
|
||||
|
||||
def test_single_audio_in_xfade_acopy(self):
|
||||
"""xfade 路径下只有一个音频片段时直接 acopy。"""
|
||||
chains = [
|
||||
ClipFilterChain("c1", 0, "v0", "a0", [], 5.0),
|
||||
ClipFilterChain("c2", 1, "v1", None, [], 3.0),
|
||||
]
|
||||
filter_str, _ = build_xfade_filter(chains, 0.5, ["cut", "fade"])
|
||||
|
||||
self.assertIn("[a0]acopy[outa]", filter_str)
|
||||
# 没有音频归一化
|
||||
self.assertNotIn("aformat", filter_str)
|
||||
self.assertNotIn("concat=n=", filter_str)
|
||||
|
||||
def test_no_audio_in_xfade(self):
|
||||
"""xfade 路径下都没有音频时没有 outa。"""
|
||||
chains = [
|
||||
ClipFilterChain("c1", 0, "v0", None, [], 5.0),
|
||||
ClipFilterChain("c2", 1, "v1", None, [], 3.0),
|
||||
]
|
||||
filter_str, _ = build_xfade_filter(chains, 0.5, ["cut", "fade"])
|
||||
self.assertNotIn("[outa]", filter_str)
|
||||
self.assertNotIn("acopy", filter_str)
|
||||
|
||||
|
||||
# ── build_filter_complex 测试 ───────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestBuildFilterComplex(unittest.TestCase):
|
||||
"""build_filter_complex 策略选择测试。"""
|
||||
|
||||
def _chain(self, idx: int, has_audio: bool = True) -> ClipFilterChain:
|
||||
return ClipFilterChain(
|
||||
clip_id=f"c{idx}",
|
||||
input_index=idx,
|
||||
video_label=f"v{idx}",
|
||||
audio_label=f"a{idx}" if has_audio else None,
|
||||
filters=["fps=25"],
|
||||
duration=3.0,
|
||||
)
|
||||
|
||||
def test_empty_clips(self):
|
||||
"""空列表返回空字符串和 0 时长。"""
|
||||
filter_str, duration = build_filter_complex([], 1280, 720, 0.5, [])
|
||||
self.assertEqual(filter_str, "")
|
||||
self.assertEqual(duration, 0.0)
|
||||
|
||||
def test_single_clip_direct_output(self):
|
||||
"""单片段:直接输出单链滤镜。"""
|
||||
chains = [self._chain(0)]
|
||||
filter_str, duration = build_filter_complex(chains, 1280, 720, 0.5, [])
|
||||
|
||||
self.assertIn("[0:v]fps=25[v0]", filter_str)
|
||||
self.assertNotIn("concat", filter_str)
|
||||
self.assertNotIn("xfade", filter_str)
|
||||
self.assertEqual(duration, 3.0)
|
||||
|
||||
def test_single_clip_audio_passthrough(self):
|
||||
"""单片段有音频:音频直通标签。"""
|
||||
chains = [self._chain(0, has_audio=True)]
|
||||
filter_str, _ = build_filter_complex(chains, 1280, 720, 0.5, [])
|
||||
# 单片段音频:[0:a]a0(直通标签)
|
||||
self.assertIn("[0:a]a0", filter_str)
|
||||
|
||||
def test_single_clip_no_audio(self):
|
||||
"""单片段无音频:没有音频部分。"""
|
||||
chains = [self._chain(0, has_audio=False)]
|
||||
filter_str, _ = build_filter_complex(chains, 1280, 720, 0.5, [])
|
||||
self.assertNotIn("[0:a]", filter_str)
|
||||
self.assertNotIn("[outa]", filter_str)
|
||||
|
||||
def test_multiple_all_cut_uses_concat(self):
|
||||
"""多片段 + 全 cut:使用 concat 滤镜。"""
|
||||
chains = [self._chain(0), self._chain(1), self._chain(2)]
|
||||
transitions = [TransitionEffect.CUT, TransitionEffect.CUT, TransitionEffect.CUT]
|
||||
filter_str, duration = build_filter_complex(chains, 1280, 720, 0.5, transitions)
|
||||
|
||||
self.assertIn("concat=n=3:v=1:a=0[outv]", filter_str)
|
||||
self.assertNotIn("xfade", filter_str)
|
||||
self.assertEqual(duration, 9.0)
|
||||
|
||||
def test_multiple_one_transition_uses_xfade(self):
|
||||
"""多片段 + 有一个非 cut 转场:使用 xfade。"""
|
||||
chains = [self._chain(0), self._chain(1)]
|
||||
transitions = [TransitionEffect.CUT, TransitionEffect.FADE]
|
||||
filter_str, duration = build_filter_complex(chains, 1280, 720, 0.5, transitions)
|
||||
|
||||
self.assertIn("xfade=transition=fade", filter_str)
|
||||
self.assertNotIn("concat=n=2:v=1:a=0", filter_str)
|
||||
self.assertAlmostEqual(duration, 5.5) # 3 + 3 - 0.5
|
||||
|
||||
def test_string_cut_value(self):
|
||||
"""字符串 'cut' 也被识别为无转场。"""
|
||||
chains = [self._chain(0), self._chain(1)]
|
||||
transitions = ["cut", "cut"]
|
||||
filter_str, _ = build_filter_complex(chains, 1280, 720, 0.5, transitions)
|
||||
self.assertIn("concat=n=2:v=1:a=0[outv]", filter_str)
|
||||
self.assertNotIn("xfade", filter_str)
|
||||
|
||||
def test_mixed_cut_and_transition(self):
|
||||
"""混合 cut 和转场:走 xfade 路径。"""
|
||||
chains = [self._chain(0), self._chain(1), self._chain(2)]
|
||||
transitions = ["cut", TransitionEffect.FADE, "cut"]
|
||||
filter_str, _ = build_filter_complex(chains, 1280, 720, 0.5, transitions)
|
||||
self.assertIn("xfade", filter_str)
|
||||
|
||||
def test_all_dissolve_transition(self):
|
||||
"""全部 dissolve 转场。"""
|
||||
chains = [self._chain(0), self._chain(1)]
|
||||
transitions = [TransitionEffect.CUT, TransitionEffect.DISSOLVE]
|
||||
filter_str, _ = build_filter_complex(chains, 1280, 720, 0.5, transitions)
|
||||
self.assertIn("xfade=transition=dissolve", filter_str)
|
||||
|
||||
|
||||
# ── 集成测试:build_clip_filter + build_filter_complex 端到端 ────────────────
|
||||
|
||||
|
||||
class TestEndToEndFilterBuilding(unittest.TestCase):
|
||||
"""端到端集成测试:从 EditPlanClip 到完整 filter_complex。"""
|
||||
|
||||
def test_two_video_clips_concat(self):
|
||||
"""两个视频片段走 concat 路径的完整流程。"""
|
||||
clip1 = _make_clip("c1", duration=5.0, clip_type="video")
|
||||
clip2 = _make_clip("c2", duration=3.0, clip_type="video")
|
||||
|
||||
chain1 = build_clip_filter(clip1, 0, 1280, 720, 25)
|
||||
chain2 = build_clip_filter(clip2, 1, 1280, 720, 25)
|
||||
|
||||
filter_str, duration = build_filter_complex(
|
||||
[chain1, chain2],
|
||||
1280,
|
||||
720,
|
||||
0.5,
|
||||
[TransitionEffect.CUT, TransitionEffect.CUT],
|
||||
)
|
||||
|
||||
# 有两个视频滤镜链
|
||||
self.assertIn("[0:v]", filter_str)
|
||||
self.assertIn("[1:v]", filter_str)
|
||||
# concat 输出
|
||||
self.assertIn("[outv]", filter_str)
|
||||
self.assertIn("[outa]", filter_str)
|
||||
# 总时长
|
||||
self.assertAlmostEqual(duration, 8.0)
|
||||
# 结构:2视频 + 2音频 + 1视频concat + 1音频concat = 6 段
|
||||
self.assertEqual(len(filter_str.split(";")), 6)
|
||||
|
||||
def test_two_clips_with_xfade(self):
|
||||
"""两个片段走 xfade 转场的完整流程。"""
|
||||
clip1 = _make_clip("c1", duration=5.0, clip_type="video")
|
||||
clip2 = _make_clip("c2", duration=4.0, clip_type="video")
|
||||
|
||||
chain1 = build_clip_filter(clip1, 0, 1920, 1080, 30)
|
||||
chain2 = build_clip_filter(clip2, 1, 1920, 1080, 30)
|
||||
|
||||
filter_str, duration = build_filter_complex(
|
||||
[chain1, chain2],
|
||||
1920,
|
||||
1080,
|
||||
0.5,
|
||||
[TransitionEffect.CUT, TransitionEffect.FADE],
|
||||
)
|
||||
|
||||
self.assertIn("xfade=transition=fade:duration=0.5", filter_str)
|
||||
self.assertIn("[outv]", filter_str)
|
||||
# 总时长减去转场
|
||||
self.assertAlmostEqual(duration, 8.5) # 5 + 4 - 0.5
|
||||
|
||||
def test_title_plus_video(self):
|
||||
"""title 片段(无音频)+ video 片段(有音频)。"""
|
||||
clip1 = _make_clip("c1", duration=2.0, clip_type="title")
|
||||
clip2 = _make_clip("c2", duration=5.0, clip_type="video")
|
||||
|
||||
chain1 = build_clip_filter(clip1, 0, 1280, 720, 25)
|
||||
chain2 = build_clip_filter(clip2, 1, 1280, 720, 25)
|
||||
|
||||
# title 无音频,video 有音频
|
||||
self.assertIsNone(chain1.audio_label)
|
||||
self.assertIsNotNone(chain2.audio_label)
|
||||
|
||||
# concat 路径
|
||||
filter_str, _ = build_filter_complex(
|
||||
[chain1, chain2],
|
||||
1280,
|
||||
720,
|
||||
0.5,
|
||||
[TransitionEffect.CUT, TransitionEffect.CUT],
|
||||
)
|
||||
# 音频 concat 只有 1 个输入(片段2)
|
||||
self.assertIn("[a1]concat=n=1:v=0:a=1[outa]", filter_str)
|
||||
self.assertNotIn("[a0]", filter_str)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user