From 4db9353b18a9ba2c2197113dd2ec762c82ef6129 Mon Sep 17 00:00:00 2001 From: CI Bot Date: Sun, 26 Jul 2026 18:13:41 +0800 Subject: [PATCH] =?UTF-8?q?test(wave101):=20=E6=8A=BD=E5=8F=96video=5Ffilt?= =?UTF-8?q?er=5Fbuilder=E7=BA=AF=E9=80=BB=E8=BE=91=E6=A8=A1=E5=9D=97=20+?= =?UTF-8?q?=2073=E4=B8=AA=E5=8D=95=E6=B5=8B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 从 video_compose_service.py 抽离 FFmpeg 滤镜构建纯逻辑到 packages/domain/video_filter_builder.py： - ClipFilterChain 数据类 - build_clip_filter: 单片段滤镜链（scale/pad/format/fps/setpts/trim） - build_concat_filter: concat 拼接（含音频归一化） - build_xfade_filter: xfade 转场链（fade/slide/dissolve/wipe） - build_filter_complex: 策略选择（单片段/concat/xfade） - chain_filters / has_audio 工具函数 - 常量与 XFADE_TRANSITION_MAP video_compose_service.py 保留向后兼容别名（私有方法+模块级函数均委托到新模块） 测试：+73 个单测，覆盖所有边界 - 原有 42 个测试全部通过（向后兼容验证） - 新增 73 个测试，覆盖数据类/常量/所有函数的正反向路径 - 648行 → 436行（-33%），新模块 379 行 --- .../api/app/services/video_compose_service.py | 275 +----- packages/domain/video_filter_builder.py | 375 ++++++++ tests/unit/test_video_filter_builder.py | 855 ++++++++++++++++++ 3 files changed, 1260 insertions(+), 245 deletions(-) create mode 100755 packages/domain/video_filter_builder.py create mode 100755 tests/unit/test_video_filter_builder.py diff --git a/apps/api/app/services/video_compose_service.py b/apps/api/app/services/video_compose_service.py index 9809663ed..b196f9901 100755 --- a/apps/api/app/services/video_compose_service.py +++ b/apps/api/app/services/video_compose_service.py @@ -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) diff --git a/packages/domain/video_filter_builder.py b/packages/domain/video_filter_builder.py new file mode 100755 index 000000000..7cbfc36b6 --- /dev/null +++ b/packages/domain/video_filter_builder.py @@ -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]") diff --git a/tests/unit/test_video_filter_builder.py b/tests/unit/test_video_filter_builder.py new file mode 100755 index 000000000..5f7c749e7 --- /dev/null +++ b/tests/unit/test_video_filter_builder.py @@ -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() -- 2.54.0