diff --git a/apps/worker/video_processing/ffmpeg_utils.py b/apps/worker/video_processing/ffmpeg_utils.py index 8ac5f493a..8f9ee5d8e 100755 --- a/apps/worker/video_processing/ffmpeg_utils.py +++ b/apps/worker/video_processing/ffmpeg_utils.py @@ -20,6 +20,17 @@ from shared.ffmpeg_utils import ( # noqa: F401 run_ffmpeg, ) +# xfade 转场纯逻辑已抽离到 domain 层，这里 re-export 保持向后兼容 +from packages.domain.xfade_builder import ( # noqa: F401 + DEFAULT_TRANSITION_DURATION as _default_transition_duration_base, + SUPPORTED_TRANSITIONS, + XFADE_TRANSITION_MAP, + XFade_TRANSITION_NAMES, + build_xfade_filter_chain as _build_xfade_filter_chain_base, + chain_filters as _chain_filters_base, + resolve_xfade_transition as _resolve_xfade_transition_base, +) + logger = logging.getLogger(__name__) # ── 常量（Worker 层业务相关） ──────────────────────────────────────────────── @@ -28,43 +39,34 @@ DEFAULT_OUTPUT_WIDTH = 1280 DEFAULT_OUTPUT_HEIGHT = 720 DEFAULT_FPS = 25 -# xfade 转场映射：transition_effect 名称 → FFmpeg xfade transition 名称 -# 键同时支持 TransitionEffect 枚举值和字符串名称（向后兼容） -# "cut" 为特殊值：硬切，不使用 xfade（由调用方特殊处理） -XFADE_TRANSITION_MAP: dict[str, str] = { - # 基础 - "fade": "fade", - "dissolve": "dissolve", - "crossfade": "dissolve", - "crossdissolve": "dissolve", - # 滑入系列 - "slideleft": "slideleft", - "slide_left": "slideleft", - "slideright": "slideright", - "slide_right": "slideright", - "slideup": "slideup", - "slide_up": "slideup", - "slidedown": "slidedown", - "slide_down": "slidedown", - "slide": "slideleft", # 默认向左滑 - # 缩放 - "zoom": "zoomin", - "zoomin": "zoomin", - "zoomout": "zoomout", - # 擦除系列 - "wipe": "wipeleft", # 默认向左擦 - "wipeleft": "wipeleft", - "wiperight": "wiperight", - "wipeup": "wipeup", - "wipedown": "wipedown", - # 特殊效果 - "circlecrop": "circlecrop", - "circle": "circlecrop", - "rectcrop": "rectcrop", - "rect": "rectcrop", -} +# 向后兼容：DEFAULT_TRANSITION_DURATION 从 domain 层导出 +DEFAULT_TRANSITION_DURATION = _default_transition_duration_base -DEFAULT_TRANSITION_DURATION = 0.5 + +# 向后兼容：薄包装函数 +def chain_filters(filters: list[str], output_label: str, *, input_label: str = "0:v") -> str: + return _chain_filters_base(filters, output_label, input_label=input_label) + + +def resolve_xfade_transition(transition_name: Any) -> str: + return _resolve_xfade_transition_base(transition_name) + + +def build_xfade_filter_chain( + clip_durations: list[float], + clip_video_labels: list[str], + transitions: list[str], + *, + transition_duration: float = DEFAULT_TRANSITION_DURATION, + output_label: str = "outv", +) -> tuple[str, float]: + return _build_xfade_filter_chain_base( + clip_durations, + clip_video_labels, + transitions, + transition_duration=transition_duration, + output_label=output_label, + ) # ── FFprobe 探测 ────────────────────────────────────────────────────────────── @@ -304,111 +306,3 @@ def normalize_video( ] run_ffmpeg(command) return {"width": width, "height": height, "path": output_path} - - -# ── xfade / concat 滤镜构建 ────────────────────────────────────────────────── - - -def chain_filters(filters: list[str], output_label: str, *, input_label: str = "0:v") -> str: - """将滤镜列表串联为 FFmpeg 滤镜字符串。 - - 例：chain_filters(["scale=1280:720", "fps=25"], "v0") - → "[0:v]scale=1280:720,fps=25[v0]" - """ - filter_body = ",".join(filters) - return f"[{input_label}]{filter_body}[{output_label}]" - - -def resolve_xfade_transition(transition_name: str) -> str: - """将转场效果名称映射为 FFmpeg xfade transition 名称。 - - 支持 TransitionEffect 枚举值和字符串名称，未知值回退到 "fade"。 - """ - # 兼容 TransitionEffect 枚举（有 .value 属性） - if hasattr(transition_name, "value"): - transition_name = transition_name.value - return XFADE_TRANSITION_MAP.get(transition_name, "fade") - - -def build_xfade_filter_chain( - clip_durations: list[float], - clip_video_labels: list[str], - transitions: list[str], - *, - transition_duration: float = DEFAULT_TRANSITION_DURATION, - output_label: str = "outv", -) -> tuple[str, float]: - """构建 xfade 转场滤镜链。 - - 对每步 xfade 自动钳制 transition duration，确保 - ``offset + td ≤ first_input_duration``，避免 FFmpeg exit 234。 - - Args: - clip_durations: 每个片段的时长（必须与 trim 后的实际时长一致） - clip_video_labels: 每个片段的视频流标签（如 "v0", "v1"） - transitions: 每个片段对应的转场效果（第一个片段的转场被忽略） - transition_duration: 转场时长（秒） - output_label: 最终输出标签 - - Returns: - (filter_string, estimated_total_duration) - """ - n = len(clip_durations) - parts: list[str] = [] - - if n == 0: - return "", 0.0 - - if n == 1: - parts.append(f"[{clip_video_labels[0]}]copy[{output_label}]") - return ";".join(parts), clip_durations[0] - - # xfade 链 — 每步动态钳制 td，防止 offset + td > first_input_duration - cumulative = 0.0 - prev_label = clip_video_labels[0] - total_transition = 0.0 # 累计已使用的转场时长 - - for i in range(1, n): - cumulative += clip_durations[i - 1] - - # 当前 xfade 的第一个输入时长 - if i == 1: - first_input_dur = clip_durations[0] - else: - first_input_dur = cumulative - total_transition - - # 原始 offset 计算 - offset = max(0.0, cumulative - transition_duration * i) - - # 安全钳制：offset + td 不能超过第一个输入的时长 - available = max(0.0, first_input_dur - offset) - safe_td = min(transition_duration, available) - - # 同时不能超过剩余总时长 - remaining = max(0.0, sum(clip_durations) - cumulative) - safe_td = min(safe_td, remaining) - # 同时不能超过当前第二个输入（单个片段）的时长 - safe_td = min(safe_td, clip_durations[i]) - safe_td = max(0.001, safe_td) # 至少 1ms，避免 td=0 - - transition = transitions[i] if i < len(transitions) else "cut" - xfade_transition = resolve_xfade_transition(transition) - - if i == n - 1: - out_label = output_label - else: - out_label = f"xf{i}" - - parts.append( - f"[{prev_label}][{clip_video_labels[i]}]" - f"xfade=transition={xfade_transition}" - f":duration={safe_td:.3f}" - f":offset={offset:.3f}" - f"[{out_label}]" - ) - prev_label = out_label - total_transition += safe_td - - # 总时长减去转场重叠部分 - total_duration = sum(clip_durations) - total_transition - return ";".join(parts), max(0.0, total_duration) diff --git a/packages/domain/xfade_builder.py b/packages/domain/xfade_builder.py new file mode 100755 index 000000000..c5ec2a505 --- /dev/null +++ b/packages/domain/xfade_builder.py @@ -0,0 +1,187 @@ +"""XFade 转场滤镜构建 — 纯逻辑，无 FFmpeg 依赖. + +抽离自 apps/worker/video_processing/ffmpeg_utils.py，包含： +- xfade 转场效果名称映射 +- 滤镜链串联工具 +- xfade 转场滤镜链构建（带时长钳制） +""" + +from __future__ import annotations + +import logging +from typing import Any + +logger = logging.getLogger(__name__) + + +# ── 常量 ────────────────────────────────────────────────────────────────────── + + +DEFAULT_TRANSITION_DURATION = 0.5 + +# xfade 转场映射：transition_effect 名称 → FFmpeg xfade transition 名称 +# 键同时支持 TransitionEffect 枚举值和字符串名称（向后兼容） +# "cut" 为特殊值：硬切，不使用 xfade（由调用方特殊处理） +XFADE_TRANSITION_MAP: dict[str, str] = { + # 基础 + "fade": "fade", + "dissolve": "dissolve", + "crossfade": "dissolve", + "crossdissolve": "dissolve", + # 滑入系列 + "slideleft": "slideleft", + "slide_left": "slideleft", + "slideright": "slideright", + "slide_right": "slideright", + "slideup": "slideup", + "slide_up": "slideup", + "slidedown": "slidedown", + "slide_down": "slidedown", + "slide": "slideleft", # 默认向左滑 + # 缩放 + "zoom": "zoomin", + "zoomin": "zoomin", + "zoomout": "zoomout", + # 擦除系列 + "wipe": "wipeleft", # 默认向左擦 + "wipeleft": "wipeleft", + "wiperight": "wiperight", + "wipeup": "wipeup", + "wipedown": "wipedown", + # 特殊效果 + "circlecrop": "circlecrop", + "circle": "circlecrop", + "rectcrop": "rectcrop", + "rect": "rectcrop", +} + +# 所有支持的转场效果名称（用户侧输入） +SUPPORTED_TRANSITIONS: set[str] = set(XFADE_TRANSITION_MAP.keys()) + +# 所有 FFmpeg xfade transition 名称（输出侧） +XFade_TRANSITION_NAMES: set[str] = set(XFADE_TRANSITION_MAP.values()) + + +# ── 工具函数 ───────────────────────────────────────────────────────────────── + + +def chain_filters(filters: list[str], output_label: str, *, input_label: str = "0:v") -> str: + """将滤镜列表串联为 FFmpeg 滤镜字符串. + + 例：chain_filters(["scale=1280:720", "fps=25"], "v0") + → "[0:v]scale=1280:720,fps=25[v0]" + + Args: + filters: 滤镜字符串列表 + output_label: 输出标签（不带方括号） + input_label: 输入标签（不带方括号），默认 "0:v" + + Returns: + 完整的滤镜字符串 + """ + filter_body = ",".join(filters) + return f"[{input_label}]{filter_body}[{output_label}]" + + +def resolve_xfade_transition(transition_name: Any) -> str: + """将转场效果名称映射为 FFmpeg xfade transition 名称. + + 支持 TransitionEffect 枚举值和字符串名称，未知值回退到 "fade"。 + + Args: + transition_name: 转场名称（字符串或带 .value 属性的枚举） + + Returns: + FFmpeg xfade transition 名称 + """ + # 兼容 TransitionEffect 枚举（有 .value 属性） + if hasattr(transition_name, "value"): + transition_name = transition_name.value + return XFADE_TRANSITION_MAP.get(transition_name, "fade") + + +# ── xfade 滤镜链构建 ───────────────────────────────────────────────────────── + + +def build_xfade_filter_chain( + clip_durations: list[float], + clip_video_labels: list[str], + transitions: list[str], + *, + transition_duration: float = DEFAULT_TRANSITION_DURATION, + output_label: str = "outv", +) -> tuple[str, float]: + """构建 xfade 转场滤镜链. + + 对每步 xfade 自动钳制 transition duration，确保 + ``offset + td ≤ first_input_duration``，避免 FFmpeg exit 234。 + + Args: + clip_durations: 每个片段的时长（必须与 trim 后的实际时长一致） + clip_video_labels: 每个片段的视频流标签（如 "v0", "v1"） + transitions: 每个片段对应的转场效果（第一个片段的转场被忽略） + transition_duration: 转场时长（秒） + output_label: 最终输出标签 + + Returns: + (filter_string, estimated_total_duration) + """ + n = len(clip_durations) + parts: list[str] = [] + + if n == 0: + return "", 0.0 + + if n == 1: + parts.append(f"[{clip_video_labels[0]}]copy[{output_label}]") + return ";".join(parts), clip_durations[0] + + # xfade 链 — 每步动态钳制 td，防止 offset + td > first_input_duration + cumulative = 0.0 + prev_label = clip_video_labels[0] + total_transition = 0.0 # 累计已使用的转场时长 + + for i in range(1, n): + cumulative += clip_durations[i - 1] + + # 当前 xfade 的第一个输入时长 + if i == 1: + first_input_dur = clip_durations[0] + else: + first_input_dur = cumulative - total_transition + + # 原始 offset 计算 + offset = max(0.0, cumulative - transition_duration * i) + + # 安全钳制：offset + td 不能超过第一个输入的时长 + available = max(0.0, first_input_dur - offset) + safe_td = min(transition_duration, available) + + # 同时不能超过剩余总时长 + remaining = max(0.0, sum(clip_durations) - cumulative) + safe_td = min(safe_td, remaining) + # 同时不能超过当前第二个输入（单个片段）的时长 + safe_td = min(safe_td, clip_durations[i]) + safe_td = max(0.001, safe_td) # 至少 1ms，避免 td=0 + + transition = transitions[i] if i < len(transitions) else "cut" + xfade_transition = resolve_xfade_transition(transition) + + if i == n - 1: + out_label = output_label + else: + out_label = f"xf{i}" + + parts.append( + f"[{prev_label}][{clip_video_labels[i]}]" + f"xfade=transition={xfade_transition}" + f":duration={safe_td:.3f}" + f":offset={offset:.3f}" + f"[{out_label}]" + ) + prev_label = out_label + total_transition += safe_td + + # 总时长减去转场重叠部分 + total_duration = sum(clip_durations) - total_transition + return ";".join(parts), max(0.0, total_duration) diff --git a/tests/unit/test_xfade_builder.py b/tests/unit/test_xfade_builder.py new file mode 100755 index 000000000..b69b75d4d --- /dev/null +++ b/tests/unit/test_xfade_builder.py @@ -0,0 +1,371 @@ +"""XFade 转场滤镜构建领域模型单元测试.""" + +from __future__ import annotations + +import pytest + +from packages.domain.xfade_builder import ( + DEFAULT_TRANSITION_DURATION, + SUPPORTED_TRANSITIONS, + XFADE_TRANSITION_MAP, + XFade_TRANSITION_NAMES, + build_xfade_filter_chain, + chain_filters, + resolve_xfade_transition, +) + +# ── 常量测试 ───────────────────────────────────────────────────────────────── + + +class TestConstants: + def test_default_transition_duration(self): + assert DEFAULT_TRANSITION_DURATION == 0.5 + + def test_xfade_transition_map_not_empty(self): + assert len(XFADE_TRANSITION_MAP) > 0 + + def test_supported_transitions(self): + assert "fade" in SUPPORTED_TRANSITIONS + assert "dissolve" in SUPPORTED_TRANSITIONS + + def test_xfade_transition_names(self): + assert "fade" in XFade_TRANSITION_NAMES + assert "dissolve" in XFade_TRANSITION_NAMES + + +# ── chain_filters 测试 ────────────────────────────────────────────────────── + + +class TestChainFilters: + def test_single_filter(self): + result = chain_filters(["scale=1280:720"], "v0") + assert result == "[0:v]scale=1280:720[v0]" + + def test_multiple_filters(self): + result = chain_filters(["scale=1280:720", "fps=25"], "v0") + assert result == "[0:v]scale=1280:720,fps=25[v0]" + + def test_empty_filters(self): + result = chain_filters([], "out") + assert result == "[0:v][out]" + + def test_custom_input_label(self): + result = chain_filters(["scale=640:480"], "out", input_label="1:v") + assert result == "[1:v]scale=640:480[out]" + + def test_three_filters(self): + result = chain_filters(["trim=0:5", "setpts=PTS-STARTPTS", "fps=30"], "v1") + assert result == "[0:v]trim=0:5,setpts=PTS-STARTPTS,fps=30[v1]" + + +# ── resolve_xfade_transition 测试 ─────────────────────────────────────────── + + +class TestResolveXfadeTransition: + def test_fade(self): + assert resolve_xfade_transition("fade") == "fade" + + def test_dissolve(self): + assert resolve_xfade_transition("dissolve") == "dissolve" + + def test_crossfade_maps_to_dissolve(self): + assert resolve_xfade_transition("crossfade") == "dissolve" + + def test_slideleft(self): + assert resolve_xfade_transition("slideleft") == "slideleft" + + def test_slide_left_maps_to_slideleft(self): + assert resolve_xfade_transition("slide_left") == "slideleft" + + def test_slide_default_left(self): + assert resolve_xfade_transition("slide") == "slideleft" + + def test_slideup(self): + assert resolve_xfade_transition("slideup") == "slideup" + + def test_zoom_maps_to_zoomin(self): + assert resolve_xfade_transition("zoom") == "zoomin" + + def test_zoomin(self): + assert resolve_xfade_transition("zoomin") == "zoomin" + + def test_wipe_default_left(self): + assert resolve_xfade_transition("wipe") == "wipeleft" + + def test_wipeup(self): + assert resolve_xfade_transition("wipeup") == "wipeup" + + def test_circle_maps_to_circlecrop(self): + assert resolve_xfade_transition("circle") == "circlecrop" + + def test_rect_maps_to_rectcrop(self): + assert resolve_xfade_transition("rect") == "rectcrop" + + def test_unknown_falls_back_to_fade(self): + assert resolve_xfade_transition("nonexistent_effect") == "fade" + + def test_empty_string_falls_back_to_fade(self): + assert resolve_xfade_transition("") == "fade" + + def test_enum_with_value_attribute(self): + """测试带 .value 属性的枚举对象.""" + + class FakeEnum: + def __init__(self, val): + self.value = val + + assert resolve_xfade_transition(FakeEnum("fade")) == "fade" + assert resolve_xfade_transition(FakeEnum("slideleft")) == "slideleft" + assert resolve_xfade_transition(FakeEnum("unknown")) == "fade" + + +# ── build_xfade_filter_chain 测试 ─────────────────────────────────────────── + + +class TestBuildXfadeFilterChain: + # ── 边界情况 ────────────────────────────────────────────────────── + + def test_empty_clips(self): + result, duration = build_xfade_filter_chain([], [], []) + assert result == "" + assert duration == 0.0 + + def test_single_clip(self): + result, duration = build_xfade_filter_chain([10.0], ["v0"], ["none"]) + assert "copy" in result + assert "[v0]copy[outv]" in result + assert duration == 10.0 + + def test_single_clip_custom_output_label(self): + result, duration = build_xfade_filter_chain([5.0], ["a0"], ["none"], output_label="final") + assert "[a0]copy[final]" in result + assert duration == 5.0 + + # ── 两片段基础测试 ──────────────────────────────────────────────── + + def test_two_clips_basic(self): + result, duration = build_xfade_filter_chain([10.0, 10.0], ["v0", "v1"], ["none", "fade"]) + assert "xfade=transition=fade" in result + assert "[v0][v1]" in result + assert "[outv]" in result + # 总时长 = 10 + 10 - 0.5 = 19.5 + assert abs(duration - 19.5) < 0.01 + + def test_two_clips_custom_duration(self): + result, duration = build_xfade_filter_chain( + [10.0, 10.0], + ["v0", "v1"], + ["none", "fade"], + transition_duration=1.0, + ) + assert "duration=1.000" in result + # 总时长 = 10 + 10 - 1.0 = 19.0 + assert abs(duration - 19.0) < 0.01 + + def test_two_clips_offset(self): + """两片段时 offset 应该为 0（cumulative - td * 1 = 10 - 0.5 = 9.5？不对）。 + + 对于两个片段： + - cumulative = clip_durations[0] = 10.0 + - offset = max(0, cumulative - td * i) = max(0, 10.0 - 0.5 * 1) = 9.5 + - duration=0.5, offset=9.5 + """ + result, _ = build_xfade_filter_chain([10.0, 10.0], ["v0", "v1"], ["none", "fade"]) + assert "offset=9.500" in result + + # ── 多片段测试 ──────────────────────────────────────────────────── + + def test_three_clips(self): + result, duration = build_xfade_filter_chain( + [5.0, 5.0, 5.0], + ["v0", "v1", "v2"], + ["none", "fade", "dissolve"], + ) + # 应该有两个 xfade + assert result.count("xfade=") == 2 + # 第一个 xfade 输出标签 xf1，第二个 xfade 输出 outv + assert "xf1" in result + assert "[outv]" in result + # 总时长 ≈ 5 + 5 + 5 - 0.5 - 0.5 = 14.0 + assert abs(duration - 14.0) < 0.1 + + def test_five_clips(self): + result, duration = build_xfade_filter_chain( + [3.0, 3.0, 3.0, 3.0, 3.0], + ["v0", "v1", "v2", "v3", "v4"], + ["none", "fade", "fade", "fade", "fade"], + ) + assert result.count("xfade=") == 4 + # 总时长 ≈ 15 - 4 * 0.5 = 13.0 + assert abs(duration - 13.0) < 0.2 + + # ── 转场效果测试 ────────────────────────────────────────────────── + + def test_dissolve_transition(self): + result, _ = build_xfade_filter_chain([10.0, 10.0], ["v0", "v1"], ["none", "dissolve"]) + assert "transition=dissolve" in result + + def test_slideleft_transition(self): + result, _ = build_xfade_filter_chain([10.0, 10.0], ["v0", "v1"], ["none", "slideleft"]) + assert "transition=slideleft" in result + + def test_cut_uses_fade(self): + """cut 转场效果应该回退到 fade.""" + result, _ = build_xfade_filter_chain([10.0, 10.0], ["v0", "v1"], ["none", "cut"]) + # cut 不是 XFADE_TRANSITION_MAP 的键，所以会回退到 fade + assert "transition=fade" in result + + def test_transitions_shorter_than_clips(self): + """如果 transitions 列表比 clips 短，剩余的用 'cut'（→ fade）.""" + result, _ = build_xfade_filter_chain( + [5.0, 5.0, 5.0], + ["v0", "v1", "v2"], + ["none"], # 只有一个 + ) + # 第二个转场（index 2）会回退到 cut → fade + assert result.count("transition=fade") == 2 + + # ── 时长钳制测试 ────────────────────────────────────────────────── + + def test_short_first_clip_truncates_transition(self): + """第一个片段非常短，转场时长应该被钳制.""" + result, duration = build_xfade_filter_chain( + [0.3, 10.0], + ["v0", "v1"], + ["none", "fade"], + transition_duration=1.0, + ) + # offset = max(0, 0.3 - 1.0 * 1) = 0.0 + # available = max(0, 0.3 - 0.0) = 0.3 + # safe_td = min(1.0, 0.3, 剩余 10.0, clip_durations[1] 10.0) = 0.3 + assert "duration=0.300" in result + assert abs(duration - 10.0) < 0.01 # 0.3 + 10.0 - 0.3 = 10.0 + + def test_zero_duration_clips(self): + """零时长片段的边界情况.""" + result, duration = build_xfade_filter_chain([0.0, 5.0], ["v0", "v1"], ["none", "fade"]) + # 第一个片段 0 时长，转场时长应该被钳制到最小值 0.001 + # offset = max(0, 0 - 0.5) = 0 + # available = max(0, 0 - 0) = 0 + # safe_td = min(0.5, 0, ...) = min(0.5, 0, 5.0, 5.0) = 0 → max(0.001, 0) = 0.001 + assert "duration=0.001" in result + + def test_very_long_transition_duration(self): + """转场时长超过所有片段时长.""" + result, duration = build_xfade_filter_chain( + [2.0, 2.0], + ["v0", "v1"], + ["none", "fade"], + transition_duration=5.0, + ) + # offset = max(0, 2.0 - 5.0) = 0 + # available = max(0, 2.0 - 0) = 2.0 + # safe_td = min(5.0, 2.0, 剩余 2.0, 2.0) = 2.0 + assert "duration=2.000" in result + assert abs(duration - 2.0) < 0.01 # 2 + 2 - 2 = 2 + + # ── 标签测试 ────────────────────────────────────────────────────── + + def test_custom_labels(self): + result, _ = build_xfade_filter_chain( + [10.0, 10.0], + ["clip_a", "clip_b"], + ["none", "fade"], + output_label="final_v", + ) + assert "[clip_a][clip_b]" in result + assert "[final_v]" in result + + def test_intermediate_labels_three_clips(self): + result, _ = build_xfade_filter_chain([5.0, 5.0, 5.0], ["v0", "v1", "v2"], ["none", "fade", "fade"]) + # 第一个 xfade 输出 xf1 + assert "[xf1][v2]" in result or result.count("[xf1]") >= 1 + + # ── 总时长计算验证 ──────────────────────────────────────────────── + + def test_total_duration_two_equal_clips(self): + _, duration = build_xfade_filter_chain([8.0, 8.0], ["v0", "v1"], ["none", "fade"]) + # 8 + 8 - 0.5 = 15.5 + assert abs(duration - 15.5) < 0.01 + + def test_total_duration_no_transition_impossible(self): + """即使 transition_duration=0，也有最小 0.001 的钳制.""" + _, duration = build_xfade_filter_chain( + [10.0, 10.0], + ["v0", "v1"], + ["none", "fade"], + transition_duration=0.0, + ) + # transition_duration=0，但 safe_td 有下限 0.001 + assert duration < 20.0 # 应该小于 20（有重叠） + assert duration > 19.9 # 但接近 20 + + # ── 滤镜字符串格式验证 ──────────────────────────────────────────── + + def test_filter_format_contains_xfade_keyword(self): + result, _ = build_xfade_filter_chain([10.0, 10.0], ["v0", "v1"], ["none", "fade"]) + assert "xfade=" in result + + def test_filter_uses_semicolon_separator(self): + """多步 xfade 之间用分号分隔.""" + result, _ = build_xfade_filter_chain([5.0, 5.0, 5.0], ["v0", "v1", "v2"], ["none", "fade", "fade"]) + assert ";" in result + # 3个片段 → 2个xfade → 1个分号 + assert result.count("xfade=") == 2 + assert result.count(";") == 1 + + def test_filter_has_transition_param(self): + result, _ = build_xfade_filter_chain([10.0, 10.0], ["v0", "v1"], ["none", "fade"]) + assert "transition=fade" in result + + def test_filter_has_duration_param(self): + result, _ = build_xfade_filter_chain([10.0, 10.0], ["v0", "v1"], ["none", "fade"]) + assert "duration=" in result + + def test_filter_has_offset_param(self): + result, _ = build_xfade_filter_chain([10.0, 10.0], ["v0", "v1"], ["none", "fade"]) + assert "offset=" in result + + # ── 各种转场效果遍历测试 ────────────────────────────────────────── + + @pytest.mark.parametrize( + "transition_name", + list(XFADE_TRANSITION_MAP.keys()), + ) + def test_all_supported_transitions(self, transition_name): + """所有支持的转场效果都应该能正确生成滤镜.""" + result, duration = build_xfade_filter_chain([10.0, 10.0], ["v0", "v1"], ["none", transition_name]) + expected = XFADE_TRANSITION_MAP[transition_name] + assert f"transition={expected}" in result + assert duration > 0 + + # ── 四片段复杂场景 ──────────────────────────────────────────────── + + def test_four_clips_different_durations(self): + durations = [3.0, 5.0, 2.0, 7.0] + result, duration = build_xfade_filter_chain( + durations, + ["v0", "v1", "v2", "v3"], + ["none", "fade", "dissolve", "slideleft"], + ) + assert result.count("xfade=") == 3 + # 总时长 = sum(durations) - 3 * 0.5 ≈ 17 - 1.5 = 15.5 + assert abs(duration - 15.5) < 0.2 + + # ── transition_duration = 0 的边界 ─────────────────────────────── + + def test_zero_transition_duration_minimum_clamped(self): + result, _ = build_xfade_filter_chain( + [10.0, 10.0], + ["v0", "v1"], + ["none", "fade"], + transition_duration=0.0, + ) + # 至少 0.001 + assert "duration=0.001" in result + + # ── 单片段自定义输出标签 ───────────────────────────────────────── + + def test_single_clip_output_label(self): + result, _ = build_xfade_filter_chain([5.0], ["v0"], ["none"], output_label="result") + assert "[v0]copy[result]" in result