test(wave121): 抽离xfade_builder领域模型 + 77单测
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (push) Successful in 1m55s
CI/CD Pipeline / Validate - Code Quality (push) Failing after 4m21s
CI/CD Pipeline / Validate - Type Check (mypy) (push) Successful in 2m35s
CI/CD Pipeline / Validate - Migration (alembic) (push) Successful in 2m25s
CI/CD Pipeline / Unit Tests (push) Successful in 9m9s
CI/CD Pipeline / Integration Tests (push) Successful in 3m0s
CI/CD Pipeline / Frontend Lint (push) Successful in 2m44s
CI/CD Pipeline / Frontend Unit Tests (push) Successful in 2m13s
CI/CD Pipeline / Build Staging API Image (push) Successful in 12m53s
CI/CD Pipeline / Build Staging Web Image (push) Successful in 2m34s
CI/CD Pipeline / Build Staging Worker Image (push) Successful in 5m7s
CI/CD Pipeline / ACR Image Cleanup (push) Successful in 54s
CI/CD Pipeline / Staging E2E Tests (push) Successful in 2m23s
CI/CD Pipeline / Staging API Integration Tests (push) Successful in 4m44s
CI/CD Pipeline / Canary Release to Production (push) Failing after 1228h3m34s
CI/CD Pipeline / Production Browser E2E (push) Failing after 1228h20m56s
CI/CD Pipeline / Build Production Worker Image (push) Failing after 1228h21m2s
CI/CD Pipeline / Build Production Web Image (push) Failing after 1228h21m3s
CI/CD Pipeline / Build Production API Image (push) Failing after 1228h21m4s
CI/CD Pipeline / PR Build Worker Image (push) Failing after 1228h23m39s
CI/CD Pipeline / PR Build Web Image (push) Failing after 1228h23m39s
CI/CD Pipeline / PR Build API Image (push) Failing after 1228h23m39s
CI/CD Pipeline / Check if frontend-only change (push) Failing after 1228h23m40s
CI/CD Pipeline / Deploy Production (push) Failing after 1228h53m43s

This commit is contained in:
2026-07-27 11:04:17 +08:00
parent 925b365d6a
commit d2c067e9bd
3 changed files with 596 additions and 144 deletions
+38 -144
View File
@@ -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)
+187
View File
@@ -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)
+371
View File
@@ -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