Compare commits

...

1 Commits

Author SHA1 Message Date
CI Bot 4db9353b18 test(wave101): 抽取video_filter_builder纯逻辑模块 + 73个单测
CI/CD Pipeline / Check if frontend-only change (pull_request) Successful in 11s
CI/CD Pipeline / Validate - Type Check (mypy) (pull_request) Successful in 58s
CI/CD Pipeline / Validate - Migration (alembic) (pull_request) Successful in 58s
CI/CD Pipeline / Frontend Lint (pull_request) Failing after 53s
CI/CD Pipeline / Build Staging API Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging Web Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging Worker Image (pull_request) Has been skipped
CI/CD Pipeline / PR Build Worker Image (pull_request) Successful in 41s
CI/CD Pipeline / Validate - Code Quality (pull_request) Failing after 2m4s
CI/CD Pipeline / PR Build Web Image (pull_request) Successful in 1m2s
PR Automation / Auto Merge on CI Green + Approved (pull_request) Successful in 51s
Preview Deploy / Deploy Preview Environment (pull_request) Successful in 57s
CI/CD Pipeline / PR Build API Image (pull_request) Successful in 4m3s
PR Automation / Auto Approve on CI Green (pull_request) Successful in 3m17s
AI Code Review / AI Code Review (pull_request) Successful in 6m38s
CI/CD Pipeline / Frontend Unit Tests (pull_request) Has been skipped
CI/CD Pipeline / Build Production API Image (pull_request) Has been skipped
CI/CD Pipeline / Build Production Web Image (pull_request) Has been skipped
CI/CD Pipeline / Build Production Worker Image (pull_request) Has been skipped
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (pull_request) Has been skipped
CI/CD Pipeline / Integration Tests (pull_request) Successful in 2m30s
CI/CD Pipeline / Unit Tests (pull_request) Successful in 3m38s
CI/CD Pipeline / Deploy Production (pull_request) Has been skipped
CI/CD Pipeline / Staging E2E Tests (pull_request) Has been skipped
CI/CD Pipeline / Staging API Integration Tests (pull_request) Has been skipped
CI/CD Pipeline / ACR Image Cleanup (pull_request) Has been skipped
CI/CD Pipeline / Production Browser E2E (pull_request) Has been skipped
CI/CD Pipeline / Canary Release to Production (pull_request) Has been skipped
ACR Cleanup / ACR Image Cleanup (pull_request_target) Has been cancelled
Preview Cleanup / Cleanup Preview Environment (pull_request) Successful in 20s
从 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 行
2026-07-26 18:43:08 +08:00
3 changed files with 1260 additions and 245 deletions
+30 -245
View File
@@ -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: 统一像素格式为 yuv420pH.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)
+375
View File
@@ -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 滤镜链到 partsconcat 路径)。
与原实现保持一致:归一化输出标签复用 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]")
+855
View File
@@ -0,0 +1,855 @@
"""video_filter_builder 单元测试 — FFmpeg 滤镜构建纯逻辑层。
覆盖:
- ClipFilterChain 数据类
- 常量与映射表
- build_clip_filter:单片段滤镜链
- chain_filters:滤镜串联工具
- has_audio:音频流判断
- build_concat_filterconcat 滤镜
- build_xfade_filterxfade 转场滤镜
- 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)
# 音频 concatn=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()