apps/worker/video_processing/concat_engine_pure.py

@@ -0,0 +1,493 @@
+"""视频拼接引擎纯逻辑模块.
+
+所有函数均为纯函数，不调用 FFmpeg、不操作文件。
+便于单元测试，也方便被其他模块复用。
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Any, Optional
+
+# ── 帧率解析 ──────────────────────────────────────────────────────────────────
+
+
+def parse_fps(fps_value: Any) -> float:
+    """解析帧率字符串/数值.
+
+    支持格式：
+    - 数字: 30 → 30.0
+    - 分数: "30/1" → 30.0, "24000/1001" → 23.976...
+    - 字符串数字: "30" → 30.0
+
+    Args:
+        fps_value: 帧率值（字符串、数字等）
+
+    Returns:
+        帧率（fps），失败返回 30.0
+    """
+    if fps_value is None:
+        return 30.0
+
+    try:
+        fps_str = str(fps_value).strip()
+        if not fps_str:
+            return 30.0
+
+        if "/" in fps_str:
+            num_str, den_str = fps_str.split("/", 1)
+            num = float(num_str)
+            den = float(den_str)
+            if den == 0:
+                return 30.0
+            return num / den
+
+        return float(fps_str)
+    except (ValueError, TypeError, ZeroDivisionError):
+        return 30.0
+
+
+def format_fps_filter(fps: float) -> str:
+    """格式化 fps 滤镜参数.
+
+    Args:
+        fps: 帧率
+
+    Returns:
+        fps 滤镜字符串
+    """
+    # 接近整数时用整数形式
+    if abs(fps - round(fps)) < 0.001:
+        return f"fps={int(fps)}"
+    return f"fps={fps:.3f}"
+
+
+# ── 输出参数计算 ──────────────────────────────────────────────────────────────
+
+
+def resolve_output_params(
+    config_width: int,
+    config_height: int,
+    config_fps: float,
+    first_video_info: Optional[dict] = None,
+    default_width: int = 1080,
+    default_height: int = 1920,
+    default_fps: float = 30.0,
+) -> tuple[int, int, float]:
+    """计算输出视频参数.
+
+    优先级：
+    1. config 中显式指定的（非 0 值）
+    2. 第一段视频的探测参数
+    3. 默认值
+
+    Args:
+        config_width: 配置的宽度（0 表示未指定）
+        config_height: 配置的高度（0 表示未指定）
+        config_fps: 配置的帧率（0 表示未指定）
+        first_video_info: 第一段视频的探测信息字典
+        default_width: 默认宽度
+        default_height: 默认高度
+        default_fps: 默认帧率
+
+    Returns:
+        (宽度, 高度, 帧率)
+    """
+    width = config_width
+    height = config_height
+    fps = config_fps
+
+    info = first_video_info or {}
+
+    # 宽度：用配置 → 探测 → 默认
+    if width == 0:
+        width = int(info.get("width", default_width))
+
+    # 高度
+    if height == 0:
+        height = int(info.get("height", default_height))
+
+    # 帧率
+    if fps == 0:
+        fps_str = info.get("r_frame_rate", f"{int(default_fps)}/1")
+        fps = parse_fps(fps_str)
+
+    # 确保都是有效值
+    width = max(1, width)
+    height = max(1, height)
+    fps = max(1.0, fps)
+
+    return width, height, fps
+
+
+def calculate_scaled_size(
+    src_w: int,
+    src_h: int,
+    target_w: int,
+    target_h: int,
+) -> tuple[int, int, int, int]:
+    """计算等比缩放后的尺寸和填充偏移.
+
+    保持宽高比，不足的部分用黑边填充。
+
+    Args:
+        src_w: 原始宽度
+        src_h: 原始高度
+        target_w: 目标宽度
+        target_h: 目标高度
+
+    Returns:
+        (缩放后宽度, 缩放后高度, X偏移, Y偏移)
+    """
+    if src_w <= 0 or src_h <= 0:
+        return (target_w, target_h, 0, 0)
+
+    src_ratio = src_w / src_h
+    target_ratio = target_w / target_h
+
+    if abs(src_ratio - target_ratio) < 0.001:
+        # 比例相同，直接缩放
+        return (target_w, target_h, 0, 0)
+    elif src_ratio > target_ratio:
+        # 源更宽，以宽度为准，上下填充
+        scaled_w = target_w
+        scaled_h = int(target_w / src_ratio)
+        offset_x = 0
+        offset_y = (target_h - scaled_h) // 2
+        return (scaled_w, scaled_h, offset_x, offset_y)
+    else:
+        # 源更高，以高度为准，左右填充
+        scaled_h = target_h
+        scaled_w = int(target_h * src_ratio)
+        offset_x = (target_w - scaled_w) // 2
+        offset_y = 0
+        return (scaled_w, scaled_h, offset_x, offset_y)
+
+
+# ── stream copy 判断 ─────────────────────────────────────────────────────────
+
+
+def can_use_stream_copy(
+    segments: list[dict],
+    target_width: int,
+    target_height: int,
+    target_fps: float,
+    force_reencode: bool = False,
+) -> bool:
+    """判断是否可以使用 stream copy（无损拼接）.
+
+    stream copy 条件：
+    1. force_reencode 为 False
+    2. 所有视频段的编码格式、分辨率、帧率均相同
+    3. 目标参数与源参数一致（不需要转码）
+
+    Args:
+        segments: 视频段列表，每个元素包含 codec_name/width/height/fps
+        target_width: 目标宽度
+        target_height: 目标高度
+        target_fps: 目标帧率
+        force_reencode: 是否强制重编码
+
+    Returns:
+        是否可以用 stream copy
+    """
+    if force_reencode:
+        return False
+
+    if not segments:
+        return False
+
+    # 用第一段作为基准
+    first = segments[0]
+    base_codec = first.get("codec_name", "")
+    base_width = int(first.get("width", 0))
+    base_height = int(first.get("height", 0))
+    base_fps = parse_fps(first.get("r_frame_rate", "30/1"))
+
+    # 目标参数必须与基准一致
+    if target_width != base_width or target_height != base_height:
+        return False
+
+    if abs(target_fps - base_fps) > 0.01:
+        return False
+
+    # 所有段必须参数一致
+    for seg in segments[1:]:
+        if seg.get("codec_name", "") != base_codec:
+            return False
+        if int(seg.get("width", 0)) != base_width:
+            return False
+        if int(seg.get("height", 0)) != base_height:
+            return False
+        seg_fps = parse_fps(seg.get("r_frame_rate", "30/1"))
+        if abs(seg_fps - base_fps) > 0.01:
+            return False
+
+    return True
+
+
+# ── 文件列表生成（demuxer 模式） ───────────────────────────────────────────────
+
+
+def generate_concat_file_list(
+    video_paths: list[str],
+) -> str:
+    """生成 concat demuxer 模式的文件列表内容.
+
+    格式：
+        file '/path/to/video1.mp4'
+        file '/path/to/video2.mp4'
+
+    Args:
+        video_paths: 视频文件路径列表
+
+    Returns:
+        文件列表文本内容
+    """
+    lines = []
+    for path in video_paths:
+        # 转义单引号
+        escaped = path.replace("'", "'\\''")
+        lines.append(f"file '{escaped}'")
+    return "\n".join(lines) + "\n"
+
+
+# ── 滤镜链构建 ────────────────────────────────────────────────────────────────
+
+
+def build_scale_pad_filter(
+    target_w: int,
+    target_h: int,
+    src_w: int = 0,
+    src_h: int = 0,
+) -> str:
+    """构建 scale + pad 滤镜（等比缩放+黑边填充）.
+
+    Args:
+        target_w: 目标宽度
+        target_h: 目标高度
+        src_w: 源宽度（0 表示未知，用 iw/ih）
+        src_h: 源高度（0 表示未知）
+
+    Returns:
+        滤镜字符串
+    """
+    # 使用 FFmpeg 表达式，动态计算
+    return (
+        f"scale={target_w}:{target_h}:force_original_aspect_ratio=decrease,"
+        f"pad={target_w}:{target_h}:(ow-iw)/2:(oh-ih)/2:black"
+    )
+
+
+def build_fps_filter(fps: float) -> str:
+    """构建 fps 滤镜.
+
+    Args:
+        fps: 目标帧率
+
+    Returns:
+        fps 滤镜字符串
+    """
+    return format_fps_filter(fps)
+
+
+def build_setpts_filter() -> str:
+    """构建 setpts 滤镜（重置时间戳）.
+
+    Returns:
+        setpts 滤镜字符串
+    """
+    return "setpts=PTS-STARTPTS"
+
+
+def build_concat_filter(
+    num_inputs: int,
+    has_audio: bool = True,
+) -> str:
+    """构建 concat 滤镜.
+
+    Args:
+        num_inputs: 输入数量
+        has_audio: 是否包含音频轨
+
+    Returns:
+        concat 滤镜字符串（包含输入标签）
+    """
+    if num_inputs <= 0:
+        return ""
+
+    n = num_inputs
+    v = 1  # 视频轨数
+    a = 1 if has_audio else 0  # 音频轨数
+
+    # 构建输入标签
+    input_labels = "".join(f"[{i}:v][{i}:a]" if has_audio else f"[{i}:v]" for i in range(n))
+
+    output_label = "[concat_v]" + ("[concat_a]" if has_audio else "")
+
+    return f"{input_labels}concat=n={n}:v={v}:a={a}{output_label}"
+
+
+def build_single_segment_filter_chain(
+    target_width: int,
+    target_height: int,
+    target_fps: float,
+    segment_index: int,
+    has_audio: bool = True,
+) -> str:
+    """构建单段视频的预处理滤镜链.
+
+    每段视频需要：缩放填充 → 帧率统一 → 重置时间戳
+
+    Args:
+        target_width: 目标宽度
+        target_height: 目标高度
+        target_fps: 目标帧率
+        segment_index: 段索引（用于生成标签）
+        has_audio: 是否包含音频
+
+    Returns:
+        滤镜字符串
+    """
+    scale_pad = build_scale_pad_filter(target_width, target_height)
+    fps = build_fps_filter(target_fps)
+    setpts = build_setpts_filter()
+
+    input_v = f"[{segment_index}:v]"
+    output_v = f"[v{segment_index}]"
+
+    video_chain = f"{input_v}{scale_pad},{fps},{setpts}{output_v}"
+
+    if has_audio:
+        input_a = f"[{segment_index}:a]"
+        output_a = f"[a{segment_index}]"
+        # 音频也需要重置时间戳
+        audio_chain = f"{input_a}asetpts=PTS-STARTPTS{output_a}"
+        return f"{video_chain};{audio_chain}"
+
+    return video_chain
+
+
+# ── 配置验证 ──────────────────────────────────────────────────────────────────
+
+
+def validate_concat_config(config: dict) -> tuple[bool, list[str]]:
+    """验证拼接配置.
+
+    Args:
+        config: 配置字典
+
+    Returns:
+        (是否合法, 错误信息列表)
+    """
+    errors: list[str] = []
+
+    segments = config.get("segments", [])
+    if not segments:
+        errors.append("至少需要一个视频段")
+        return (False, errors)
+
+    if len(segments) < 1:
+        errors.append("视频段数量不能少于 1")
+
+    # 检查每个段
+    for i, seg in enumerate(segments):
+        video_path = seg.get("video_path", "")
+        if not video_path:
+            errors.append(f"第 {i+1} 段缺少 video_path")
+
+    # 输出参数
+    output_width = config.get("output_width", 0)
+    output_height = config.get("output_height", 0)
+    if output_width < 0:
+        errors.append("output_width 不能为负数")
+    if output_height < 0:
+        errors.append("output_height 不能为负数")
+
+    output_fps = config.get("output_fps", 0)
+    if output_fps < 0:
+        errors.append("output_fps 不能为负数")
+
+    return (len(errors) == 0, errors)
+
+
+# ── 路径验证 ──────────────────────────────────────────────────────────────────
+
+
+def validate_video_path(video_path: str, work_dir: str | Path) -> tuple[bool, str]:
+    """验证视频路径是否安全.
+
+    检查：
+    1. 路径不为空
+    2. 路径不包含 .. 回溯
+    3. 路径在 work_dir 内（安全边界）
+
+    Args:
+        video_path: 视频文件路径
+        work_dir: 工作目录
+
+    Returns:
+        (是否合法, 错误信息)
+    """
+    if not video_path:
+        return (False, "视频路径不能为空")
+
+    path_str = str(video_path)
+    work_str = str(work_dir)
+
+    # 检查路径遍历
+    if ".." in Path(path_str).parts:
+        return (False, "视频路径不能包含 .. 回溯")
+
+    # 绝对路径才做边界检查；相对路径默认相对于 work_dir
+    if not Path(path_str).is_absolute():
+        return (True, "")
+
+    # 绝对路径检查是否在工作目录内
+    try:
+        video_abs = Path(path_str).resolve()
+        work_abs = Path(work_str).resolve()
+        if work_abs.is_absolute() and not str(video_abs).startswith(str(work_abs)):
+            return (False, "视频路径必须在工作目录内")
+    except (OSError, ValueError):
+        pass  # 解析失败时跳过边界检查
+
+    return (True, "")
+
+
+# ── 工具函数 ──────────────────────────────────────────────────────────────────
+
+
+def estimate_total_duration(segments: list[dict]) -> float:
+    """估算总时长.
+
+    Args:
+        segments: 视频段列表，每个元素包含 duration 字段
+
+    Returns:
+        总时长（秒）
+    """
+    total = 0.0
+    for seg in segments:
+        dur = seg.get("duration", 0)
+        try:
+            total += float(dur)
+        except (ValueError, TypeError):
+            pass
+    return total
+
+
+def count_valid_segments(segments: list[dict]) -> int:
+    """统计有效视频段数量（有 video_path 的）.
+
+    Args:
+        segments: 视频段列表
+
+    Returns:
+        有效段数量
+    """
+    count = 0
+    for seg in segments:
+        if seg.get("video_path"):
+            count += 1
+    return count

tests/unit/test_concat_engine_pure.py

@@ -0,0 +1,534 @@
+"""视频拼接引擎纯逻辑单元测试."""
+
+from __future__ import annotations
+
+import pytest
+from video_processing.concat_engine_pure import (
+    build_concat_filter,
+    build_fps_filter,
+    build_scale_pad_filter,
+    build_single_segment_filter_chain,
+    calculate_scaled_size,
+    can_use_stream_copy,
+    count_valid_segments,
+    estimate_total_duration,
+    format_fps_filter,
+    generate_concat_file_list,
+    parse_fps,
+    resolve_output_params,
+    validate_concat_config,
+    validate_video_path,
+)
+
+# ─────────────────────────────────────────────────────────────────────────────
+# 帧率解析测试
+# ─────────────────────────────────────────────────────────────────────────────
+
+
+class TestParseFps:
+    """parse_fps 测试."""
+
+    def test_integer_fps(self):
+        """整数帧率."""
+        assert parse_fps(30) == 30.0
+
+    def test_float_fps(self):
+        """浮点帧率."""
+        assert parse_fps(29.97) == pytest.approx(29.97)
+
+    def test_string_integer(self):
+        """字符串整数."""
+        assert parse_fps("30") == 30.0
+
+    def test_string_fraction(self):
+        """分数字符串（30/1）."""
+        assert parse_fps("30/1") == 30.0
+
+    def test_fraction_24000_1001(self):
+        """23.976 帧率."""
+        result = parse_fps("24000/1001")
+        assert result == pytest.approx(23.976, rel=0.01)
+
+    def test_none_input(self):
+        """None 输入返回默认值."""
+        assert parse_fps(None) == 30.0
+
+    def test_empty_string(self):
+        """空字符串返回默认值."""
+        assert parse_fps("") == 30.0
+
+    def test_invalid_string(self):
+        """无效字符串."""
+        assert parse_fps("abc") == 30.0
+
+    def test_zero_denominator(self):
+        """分母为 0."""
+        assert parse_fps("30/0") == 30.0
+
+    def test_negative_fps(self):
+        """负帧率."""
+        assert parse_fps(-30) == -30.0
+
+
+class TestFormatFpsFilter:
+    """format_fps_filter 测试."""
+
+    def test_integer_fps(self):
+        """整数帧率."""
+        assert format_fps_filter(30.0) == "fps=30"
+
+    def test_float_fps(self):
+        """浮点帧率."""
+        result = format_fps_filter(29.97)
+        assert result.startswith("fps=")
+        assert "29.97" in result
+
+    def test_near_integer(self):
+        """接近整数."""
+        assert format_fps_filter(30.0001) == "fps=30"
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# 输出参数计算测试
+# ─────────────────────────────────────────────────────────────────────────────
+
+
+class TestResolveOutputParams:
+    """resolve_output_params 测试."""
+
+    def test_all_specified(self):
+        """全部显式指定."""
+        w, h, fps = resolve_output_params(1920, 1080, 60.0)
+        assert w == 1920
+        assert h == 1080
+        assert fps == 60.0
+
+    def test_no_specified_use_defaults(self):
+        """全部未指定，用默认值."""
+        w, h, fps = resolve_output_params(0, 0, 0)
+        assert w == 1080
+        assert h == 1920
+        assert fps == 30.0
+
+    def test_use_first_video_info(self):
+        """用第一段视频信息."""
+        info = {"width": 1280, "height": 720, "r_frame_rate": "24/1"}
+        w, h, fps = resolve_output_params(0, 0, 0, info)
+        assert w == 1280
+        assert h == 720
+        assert fps == 24.0
+
+    def test_partial_specified(self):
+        """部分指定，未指定的用探测值."""
+        info = {"width": 1280, "height": 720, "r_frame_rate": "24/1"}
+        w, h, fps = resolve_output_params(1920, 0, 0, info)
+        assert w == 1920  # 指定的
+        assert h == 720  # 探测的
+        assert fps == 24.0
+
+    def test_zero_size_clamped(self):
+        """零尺寸被钳制."""
+        w, h, fps = resolve_output_params(0, 0, 0, {})
+        assert w >= 1
+        assert h >= 1
+        assert fps >= 1.0
+
+    def test_custom_defaults(self):
+        """自定义默认值."""
+        w, h, fps = resolve_output_params(0, 0, 0, None, 640, 480, 25.0)
+        assert w == 640
+        assert h == 480
+        assert fps == 25.0
+
+
+class TestCalculateScaledSize:
+    """calculate_scaled_size 测试."""
+
+    def test_same_ratio(self):
+        """比例相同."""
+        sw, sh, ox, oy = calculate_scaled_size(1920, 1080, 1920, 1080)
+        assert sw == 1920
+        assert sh == 1080
+        assert ox == 0
+        assert oy == 0
+
+    def test_wider_source(self):
+        """源更宽，上下填黑边."""
+        sw, sh, ox, oy = calculate_scaled_size(1920, 1080, 1080, 1920)
+        assert sw == 1080  # 以宽度为准
+        assert sh < 1920  # 高度按比例
+        assert ox == 0
+        assert oy > 0  # 垂直居中
+
+    def test_taller_source(self):
+        """源更高，左右填黑边."""
+        sw, sh, ox, oy = calculate_scaled_size(1080, 1920, 1920, 1080)
+        assert sh == 1080  # 以高度为准
+        assert sw < 1920  # 宽度按比例
+        assert ox > 0  # 水平居中
+        assert oy == 0
+
+    def test_zero_source(self):
+        """零尺寸源."""
+        sw, sh, ox, oy = calculate_scaled_size(0, 0, 100, 100)
+        assert sw == 100
+        assert sh == 100
+
+    def test_scale_down(self):
+        """缩小."""
+        sw, sh, ox, oy = calculate_scaled_size(1920, 1080, 640, 360)
+        assert sw == 640
+        assert sh == 360
+        assert ox == 0
+        assert oy == 0
+
+    def test_scale_up(self):
+        """放大."""
+        sw, sh, ox, oy = calculate_scaled_size(640, 360, 1920, 1080)
+        assert sw == 1920
+        assert sh == 1080
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# stream copy 判断测试
+# ─────────────────────────────────────────────────────────────────────────────
+
+
+class TestCanUseStreamCopy:
+    """can_use_stream_copy 测试."""
+
+    def test_identical_segments(self):
+        """所有段参数相同，可以 stream copy."""
+        segs = [
+            {"codec_name": "h264", "width": 1920, "height": 1080, "r_frame_rate": "30/1"},
+            {"codec_name": "h264", "width": 1920, "height": 1080, "r_frame_rate": "30/1"},
+        ]
+        assert can_use_stream_copy(segs, 1920, 1080, 30.0) is True
+
+    def test_force_reencode(self):
+        """强制重编码."""
+        segs = [
+            {"codec_name": "h264", "width": 1920, "height": 1080, "r_frame_rate": "30/1"},
+        ]
+        assert can_use_stream_copy(segs, 1920, 1080, 30.0, force_reencode=True) is False
+
+    def test_different_codec(self):
+        """编码不同."""
+        segs = [
+            {"codec_name": "h264", "width": 1920, "height": 1080, "r_frame_rate": "30/1"},
+            {"codec_name": "hevc", "width": 1920, "height": 1080, "r_frame_rate": "30/1"},
+        ]
+        assert can_use_stream_copy(segs, 1920, 1080, 30.0) is False
+
+    def test_different_resolution(self):
+        """分辨率不同."""
+        segs = [
+            {"codec_name": "h264", "width": 1920, "height": 1080, "r_frame_rate": "30/1"},
+            {"codec_name": "h264", "width": 1280, "height": 720, "r_frame_rate": "30/1"},
+        ]
+        assert can_use_stream_copy(segs, 1920, 1080, 30.0) is False
+
+    def test_different_fps(self):
+        """帧率不同."""
+        segs = [
+            {"codec_name": "h264", "width": 1920, "height": 1080, "r_frame_rate": "30/1"},
+            {"codec_name": "h264", "width": 1920, "height": 1080, "r_frame_rate": "60/1"},
+        ]
+        assert can_use_stream_copy(segs, 1920, 1080, 30.0) is False
+
+    def test_target_differs(self):
+        """目标参数与源不同."""
+        segs = [
+            {"codec_name": "h264", "width": 1920, "height": 1080, "r_frame_rate": "30/1"},
+        ]
+        assert can_use_stream_copy(segs, 1280, 720, 30.0) is False
+
+    def test_empty_segments(self):
+        """空列表."""
+        assert can_use_stream_copy([], 1920, 1080, 30.0) is False
+
+    def test_single_segment(self):
+        """单段."""
+        segs = [{"codec_name": "h264", "width": 1920, "height": 1080, "r_frame_rate": "30/1"}]
+        assert can_use_stream_copy(segs, 1920, 1080, 30.0) is True
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# 文件列表生成测试
+# ─────────────────────────────────────────────────────────────────────────────
+
+
+class TestGenerateConcatFileList:
+    """generate_concat_file_list 测试."""
+
+    def test_single_file(self):
+        """单个文件."""
+        result = generate_concat_file_list(["/a.mp4"])
+        assert "file '/a.mp4'" in result
+        assert result.endswith("\n")
+
+    def test_multiple_files(self):
+        """多个文件."""
+        result = generate_concat_file_list(["/a.mp4", "/b.mp4", "/c.mp4"])
+        lines = result.strip().split("\n")
+        assert len(lines) == 3
+        assert lines[0] == "file '/a.mp4'"
+        assert lines[1] == "file '/b.mp4'"
+        assert lines[2] == "file '/c.mp4'"
+
+    def test_empty_list(self):
+        """空列表."""
+        result = generate_concat_file_list([])
+        assert result == "\n"
+
+    def test_path_with_single_quote(self):
+        """路径包含单引号（转义）."""
+        result = generate_concat_file_list(["/path/to/file's.mp4"])
+        # 单引号应该被转义
+        assert "'\\''" in result or file
+        assert "file '" in result
+
+    def test_path_with_spaces(self):
+        """路径包含空格."""
+        result = generate_concat_file_list(["/path/to/my video.mp4"])
+        assert "my video" in result
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# 滤镜构建测试
+# ─────────────────────────────────────────────────────────────────────────────
+
+
+class TestBuildScalePadFilter:
+    """scale+pad 滤镜测试."""
+
+    def test_contains_scale(self):
+        """包含 scale."""
+        result = build_scale_pad_filter(1920, 1080)
+        assert "scale=" in result
+
+    def test_contains_pad(self):
+        """包含 pad."""
+        result = build_scale_pad_filter(1920, 1080)
+        assert "pad=" in result
+        assert "1920:1080" in result
+
+    def test_force_original_aspect_ratio(self):
+        """保持宽高比."""
+        result = build_scale_pad_filter(1920, 1080)
+        assert "force_original_aspect_ratio=decrease" in result
+
+    def test_black_padding(self):
+        """黑边填充."""
+        result = build_scale_pad_filter(1920, 1080)
+        assert ":black" in result
+
+
+class TestBuildFpsFilter:
+    """fps 滤镜测试."""
+
+    def test_integer_fps(self):
+        """整数帧率."""
+        assert build_fps_filter(30.0) == "fps=30"
+
+    def test_float_fps(self):
+        """浮点帧率."""
+        result = build_fps_filter(29.97)
+        assert result.startswith("fps=")
+
+
+class TestBuildConcatFilter:
+    """concat 滤镜测试."""
+
+    def test_two_inputs_with_audio(self):
+        """两路输入，有音频."""
+        result = build_concat_filter(2, has_audio=True)
+        assert "[0:v][0:a][1:v][1:a]concat=n=2:v=1:a=1" in result
+        assert "[concat_v][concat_a]" in result
+
+    def test_three_inputs_video_only(self):
+        """三路输入，无音频."""
+        result = build_concat_filter(3, has_audio=False)
+        assert "[0:v][1:v][2:v]concat=n=3:v=1:a=0" in result
+        assert "[concat_v]" in result
+
+    def test_single_input(self):
+        """单路输入."""
+        result = build_concat_filter(1, has_audio=True)
+        assert "[0:v][0:a]concat=n=1:v=1:a=1" in result
+
+    def test_zero_inputs(self):
+        """零输入."""
+        assert build_concat_filter(0) == ""
+
+
+class TestBuildSingleSegmentFilterChain:
+    """单段滤镜链测试."""
+
+    def test_with_audio(self):
+        """有音频."""
+        result = build_single_segment_filter_chain(1920, 1080, 30.0, 0)
+        assert "scale=" in result
+        assert "fps=" in result
+        assert "setpts=PTS-STARTPTS" in result
+        assert "asetpts=PTS-STARTPTS" in result
+        assert "[v0]" in result
+        assert "[a0]" in result
+
+    def test_video_only(self):
+        """无音频."""
+        result = build_single_segment_filter_chain(1920, 1080, 30.0, 1, has_audio=False)
+        assert "scale=" in result
+        assert "setpts=" in result
+        assert "asetpts" not in result
+        assert "[v1]" in result
+
+    def test_segment_index_in_labels(self):
+        """段索引在标签中."""
+        result = build_single_segment_filter_chain(1920, 1080, 30.0, 5)
+        assert "[5:v]" in result
+        assert "[v5]" in result
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# 配置验证测试
+# ─────────────────────────────────────────────────────────────────────────────
+
+
+class TestValidateConcatConfig:
+    """配置验证测试."""
+
+    def test_valid_config(self):
+        """合法配置."""
+        config = {
+            "segments": [{"video_path": "/a.mp4"}, {"video_path": "/b.mp4"}],
+            "output_width": 1920,
+            "output_height": 1080,
+            "output_fps": 30,
+        }
+        ok, errors = validate_concat_config(config)
+        assert ok is True
+        assert len(errors) == 0
+
+    def test_empty_segments(self):
+        """空段列表."""
+        ok, errors = validate_concat_config({"segments": []})
+        assert ok is False
+        assert any("至少需要" in e or "视频段" in e for e in errors)
+
+    def test_missing_video_path(self):
+        """缺少 video_path."""
+        config = {"segments": [{"video_path": "/a.mp4"}, {}]}
+        ok, errors = validate_concat_config(config)
+        assert ok is False
+        assert any("video_path" in e for e in errors)
+
+    def test_negative_width(self):
+        """负宽度."""
+        config = {"segments": [{"video_path": "/a.mp4"}], "output_width": -100}
+        ok, errors = validate_concat_config(config)
+        assert ok is False
+        assert any("output_width" in e for e in errors)
+
+    def test_negative_height(self):
+        """负高度."""
+        config = {"segments": [{"video_path": "/a.mp4"}], "output_height": -100}
+        ok, errors = validate_concat_config(config)
+        assert ok is False
+        assert any("output_height" in e for e in errors)
+
+    def test_negative_fps(self):
+        """负帧率."""
+        config = {"segments": [{"video_path": "/a.mp4"}], "output_fps": -30}
+        ok, errors = validate_concat_config(config)
+        assert ok is False
+        assert any("output_fps" in e for e in errors)
+
+    def test_zero_output_params_ok(self):
+        """零输出参数合法（表示自动探测）."""
+        config = {"segments": [{"video_path": "/a.mp4"}]}
+        ok, errors = validate_concat_config(config)
+        assert ok is True
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# 路径验证测试
+# ─────────────────────────────────────────────────────────────────────────────
+
+
+class TestValidateVideoPath:
+    """视频路径验证测试."""
+
+    def test_empty_path(self):
+        """空路径."""
+        ok, msg = validate_video_path("", "/work")
+        assert ok is False
+        assert "不能为空" in msg
+
+    def test_path_traversal(self):
+        """路径遍历."""
+        ok, msg = validate_video_path("../etc/passwd", "/work")
+        assert ok is False
+        assert "回溯" in msg or ".." in msg
+
+    def test_valid_relative_path(self):
+        """相对路径（不检查边界）."""
+        ok, msg = validate_video_path("video.mp4", "/work")
+        assert ok is True
+
+    def test_valid_absolute_path(self):
+        """绝对路径在工作目录内."""
+        ok, msg = validate_video_path("/work/sub/video.mp4", "/work")
+        assert ok is True
+
+    def test_path_outside_work_dir(self):
+        """路径在工作目录外."""
+        ok, msg = validate_video_path("/etc/passwd", "/work")
+        assert ok is False
+        assert "工作目录" in msg
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# 工具函数测试
+# ─────────────────────────────────────────────────────────────────────────────
+
+
+class TestEstimateTotalDuration:
+    """总时长估算测试."""
+
+    def test_multiple_segments(self):
+        """多段视频."""
+        segs = [{"duration": 10}, {"duration": 20.5}, {"duration": 5}]
+        assert estimate_total_duration(segs) == pytest.approx(35.5)
+
+    def test_empty_list(self):
+        """空列表."""
+        assert estimate_total_duration([]) == 0.0
+
+    def test_invalid_duration_skipped(self):
+        """无效时长跳过."""
+        segs = [{"duration": 10}, {"duration": "abc"}, {"duration": 20}]
+        assert estimate_total_duration(segs) == pytest.approx(30.0)
+
+    def test_missing_duration(self):
+        """缺 duration 字段."""
+        segs = [{}, {"duration": 10}]
+        assert estimate_total_duration(segs) == pytest.approx(10.0)
+
+
+class TestCountValidSegments:
+    """有效段统计测试."""
+
+    def test_all_valid(self):
+        """全部有效."""
+        segs = [{"video_path": "/a.mp4"}, {"video_path": "/b.mp4"}]
+        assert count_valid_segments(segs) == 2
+
+    def test_some_invalid(self):
+        """部分无效."""
+        segs = [{"video_path": "/a.mp4"}, {}, {"video_path": ""}]
+        assert count_valid_segments(segs) == 1
+
+    def test_empty_list(self):
+        """空列表."""
+        assert count_valid_segments([]) == 0