apps/worker/video_processing/multi_track_mixer_pure.py

@@ -0,0 +1,466 @@
+"""多轨混音纯逻辑模块.
+
+所有函数均为纯函数，不调用 FFmpeg、不操作文件。
+便于单元测试，也方便被其他模块复用。
+"""
+
+from __future__ import annotations
+
+from typing import Any, Optional
+
+# ── 单轨时间计算 ──────────────────────────────────────────────────────────────
+
+
+def calculate_effective_range(
+    track_start: float,
+    track_duration: float,
+    audio_duration: float,
+    target_duration: float,
+) -> tuple[float, float, float]:
+    """计算轨道的有效时间范围.
+
+    处理：
+    - 轨道时长为 0 或负时用音频完整时长
+    - 轨道开始在目标时长外时跳过
+    - 轨道开始为负时截断开头
+
+    Args:
+        track_start: 轨道开始时间（秒），可为负
+        track_duration: 轨道持续时长（秒），<=0 表示用音频全长
+        audio_duration: 音频文件实际时长（秒）
+        target_duration: 目标总时长（秒）
+
+    Returns:
+        (effective_start, need_duration, trim_start)
+        - effective_start: 在目标时间轴上的开始位置（>=0）
+        - need_duration: 需要截取的音频长度
+        - trim_start: 从源音频的哪个位置开始截取
+    """
+    if audio_duration <= 0:
+        return (0.0, 0.0, 0.0)
+
+    # 有效时长（轨道声明的时长，未被截断的）
+    if track_duration > 0:
+        effective_dur = min(track_duration, audio_duration)
+    else:
+        effective_dur = audio_duration
+
+    effective_start = track_start
+    trim_start = 0.0
+
+    # 负的开始时间：从源音频中间开始取，轨道前段被截掉
+    if effective_start < 0:
+        trim_start = -effective_start
+        # 可用时长 = 总时长 - 被截掉的前段
+        effective_dur = max(0.0, effective_dur - trim_start)
+        effective_start = 0.0
+
+    # 轨道完全在目标时长之外
+    if effective_start >= target_duration:
+        return (0.0, 0.0, 0.0)
+
+    # 轨道完全在 0 之前
+    if effective_start + effective_dur <= 0:
+        return (0.0, 0.0, 0.0)
+
+    # 实际需要的源时长
+    need_dur = min(effective_dur, target_duration - effective_start)
+    if need_dur <= 0:
+        return (0.0, 0.0, 0.0)
+
+    # 调整 trim_start 不能超过音频长度
+    if trim_start >= audio_duration:
+        return (0.0, 0.0, 0.0)
+
+    return (effective_start, need_dur, trim_start)
+
+
+def is_track_visible(
+    track_start: float,
+    track_duration: float,
+    audio_duration: float,
+    target_duration: float,
+) -> bool:
+    """判断轨道是否在目标时长范围内可见（有声音）.
+
+    Args:
+        track_start: 轨道开始时间
+        track_duration: 轨道持续时长
+        audio_duration: 音频时长
+        target_duration: 目标总时长
+
+    Returns:
+        是否可见
+    """
+    _, need_dur, _ = calculate_effective_range(track_start, track_duration, audio_duration, target_duration)
+    return need_dur > 0
+
+
+# ── 单轨滤镜链构建 ────────────────────────────────────────────────────────────
+
+
+def build_track_filter_chain(
+    volume: float,
+    fade_in: float,
+    fade_out: float,
+    effective_start: float,
+    need_duration: float,
+    trim_start: float,
+    target_duration: float,
+) -> str:
+    """构建单轨道预处理滤镜链.
+
+    处理顺序：截断 → 重置时间戳 → 音量 → 淡入 → 淡出 → 延迟 → 最终截断 → 重置时间戳
+
+    Args:
+        volume: 音量 0.0~1.0
+        fade_in: 淡入时长（秒）
+        fade_out: 淡出时长（秒）
+        effective_start: 在目标轴上的开始时间
+        need_duration: 需要截取的时长
+        trim_start: 从源音频的哪个位置开始
+        target_duration: 目标总时长
+
+    Returns:
+        逗号分隔的滤镜字符串
+    """
+    filter_parts: list[str] = []
+
+    # 1. 截断到有效范围
+    filter_parts.append(f"atrim={trim_start:.3f}:{trim_start + need_duration:.3f}")
+    filter_parts.append("asetpts=N/SR/TB")
+
+    # 2. 音量调节
+    safe_volume = max(0.0, min(2.0, volume))
+    if abs(safe_volume - 1.0) > 0.001:
+        filter_parts.append(f"volume={safe_volume:.3f}")
+
+    # 3. 淡入（必须小于总时长才有效）
+    if fade_in > 0 and fade_in < need_duration:
+        filter_parts.append(f"afade=t=in:st=0:d={fade_in:.3f}")
+
+    # 4. 淡出
+    if fade_out > 0 and fade_out < need_duration:
+        fade_start = need_duration - fade_out
+        if fade_start > 0:
+            filter_parts.append(f"afade=t=out:st={fade_start:.3f}:d={fade_out:.3f}")
+
+    # 5. 时间偏移（开头静音填充）
+    if effective_start > 0.01:
+        delay_ms = int(effective_start * 1000)
+        filter_parts.append(f"adelay={delay_ms}|{delay_ms}")
+
+    # 6. 最终截断到目标总时长
+    filter_parts.append(f"atrim=0:{target_duration:.3f}")
+    filter_parts.append("asetpts=N/SR/TB")
+
+    return ",".join(filter_parts)
+
+
+# ── amix 混音滤镜构建 ────────────────────────────────────────────────────────
+
+
+def build_amix_filter(num_inputs: int, duration_mode: str = "longest") -> str:
+    """构建 amix 混音滤镜.
+
+    Args:
+        num_inputs: 输入轨道数量
+        duration_mode: 时长模式：longest / shortest / first
+
+    Returns:
+        amix 滤镜字符串
+    """
+    if num_inputs <= 0:
+        return ""
+
+    # 校验 duration_mode
+    if duration_mode not in ("longest", "shortest", "first"):
+        duration_mode = "longest"
+
+    return f"amix=inputs={num_inputs}:duration={duration_mode}:dropout_transition=0"
+
+
+def calculate_amix_volume_compensation(num_inputs: int) -> float:
+    """计算 amix 后的音量补偿系数.
+
+    amix 会将 N 路输入每路乘以 1/N 来归一化，
+    所以需要乘以 N 来补偿（简单粗暴但有效）。
+
+    Args:
+        num_inputs: 输入轨道数量
+
+    Returns:
+        补偿系数
+    """
+    if num_inputs <= 1:
+        return 1.0
+    return float(num_inputs)
+
+
+def build_mix_filter_complex(
+    num_tracks: int,
+    has_main: bool = True,
+    duration_mode: str = "longest",
+) -> str:
+    """构建完整的混音 filter_complex.
+
+    Args:
+        num_tracks: 额外轨道数量
+        has_main: 是否有主音频
+        duration_mode: 时长模式
+
+    Returns:
+        filter_complex 字符串
+    """
+    total_inputs = num_tracks + (1 if has_main else 0)
+    if total_inputs <= 0:
+        return ""
+
+    # 输入标签
+    input_labels = "".join(f"[{i}:a]" for i in range(total_inputs))
+
+    # amix
+    amix = build_amix_filter(total_inputs, duration_mode)
+
+    # 音量补偿
+    compensation = calculate_amix_volume_compensation(total_inputs)
+    volume_filter = ""
+    if abs(compensation - 1.0) > 0.001:
+        volume_filter = f",volume={compensation}"
+
+    return f"{input_labels}{amix}{volume_filter}[mixed]"
+
+
+# ── 音量计算 ──────────────────────────────────────────────────────────────────
+
+
+def normalize_volume(volume: float) -> float:
+    """规范化音量值.
+
+    Args:
+        volume: 原始音量
+
+    Returns:
+        规范化后的音量（0.0 ~ 2.0）
+    """
+    if volume is None:
+        return 1.0
+    try:
+        v = float(volume)
+        return max(0.0, min(2.0, v))
+    except (ValueError, TypeError):
+        return 1.0
+
+
+def db_to_linear(db: float) -> float:
+    """dB 转换为线性音量.
+
+    Args:
+        db: 分贝值
+
+    Returns:
+        线性音量值
+    """
+    import math
+
+    return 10 ** (db / 20.0)
+
+
+def linear_to_db(linear: float) -> float:
+    """线性音量转换为 dB.
+
+    Args:
+        linear: 线性音量值
+
+    Returns:
+        分贝值
+    """
+    import math
+
+    if linear <= 0:
+        return -float("inf")
+    return 20.0 * math.log10(linear)
+
+
+# ── 轨道排序与过滤 ────────────────────────────────────────────────────────────
+
+
+def sort_tracks_by_priority(
+    tracks: list[dict],
+) -> list[dict]:
+    """按优先级排序轨道.
+
+    priority 数字越小优先级越高（越先播放/越底层）。
+    相同优先级保持原顺序。
+
+    Args:
+        tracks: 轨道配置列表
+
+    Returns:
+        排序后的轨道列表
+    """
+    return sorted(tracks, key=lambda t: int(t.get("priority", 100)))
+
+
+def filter_enabled_tracks(tracks: list[dict]) -> list[dict]:
+    """过滤出启用的轨道.
+
+    Args:
+        tracks: 轨道列表
+
+    Returns:
+        启用的轨道列表
+    """
+    result = []
+    for t in tracks:
+        enabled = t.get("enabled", True)
+        if bool(enabled) and enabled != "false" and enabled != 0:
+            result.append(t)
+    return result
+
+
+def count_track_types(tracks: list[dict]) -> dict[str, int]:
+    """统计各类型轨道数量.
+
+    Args:
+        tracks: 轨道列表
+
+    Returns:
+        类型计数字典
+    """
+    counts: dict[str, int] = {}
+    for t in tracks:
+        ttype = t.get("track_type", "unknown")
+        counts[ttype] = counts.get(ttype, 0) + 1
+    return counts
+
+
+# ── 配置验证 ──────────────────────────────────────────────────────────────────
+
+
+def validate_audio_track(track: dict) -> tuple[bool, list[str]]:
+    """验证单条音轨配置.
+
+    Args:
+        track: 轨道配置字典
+
+    Returns:
+        (是否合法, 错误信息列表)
+    """
+    errors: list[str] = []
+
+    # 音频路径
+    audio_path = track.get("audio_path", "")
+    if not audio_path and not track.get("asset_id"):
+        errors.append("轨道需要 audio_path 或 asset_id")
+
+    # 音量范围
+    volume = track.get("volume", 1.0)
+    try:
+        v = float(volume)
+        if v < 0:
+            errors.append("volume 不能为负数")
+        if v > 2.0:
+            errors.append("volume 建议不超过 2.0")
+    except (ValueError, TypeError):
+        errors.append("volume 必须是数字")
+
+    # 淡入淡出
+    fade_in = track.get("fade_in", 0)
+    fade_out = track.get("fade_out", 0)
+    try:
+        if float(fade_in) < 0:
+            errors.append("fade_in 不能为负数")
+    except (ValueError, TypeError):
+        errors.append("fade_in 必须是数字")
+
+    try:
+        if float(fade_out) < 0:
+            errors.append("fade_out 不能为负数")
+    except (ValueError, TypeError):
+        errors.append("fade_out 必须是数字")
+
+    # 开始时间
+    start_time = track.get("start_time", 0)
+    try:
+        float(start_time)  # 验证是否为数字
+    except (ValueError, TypeError):
+        errors.append("start_time 必须是数字")
+
+    return (len(errors) == 0, errors)
+
+
+def validate_mix_config(config: dict) -> tuple[bool, list[str]]:
+    """验证混音配置.
+
+    Args:
+        config: 混音配置
+
+    Returns:
+        (是否合法, 错误信息列表)
+    """
+    errors: list[str] = []
+
+    tracks = config.get("tracks", [])
+    if not tracks:
+        errors.append("至少需要一条轨道")
+
+    # 验证每条轨道
+    for i, track in enumerate(tracks):
+        ok, track_errors = validate_audio_track(track)
+        if not ok:
+            for err in track_errors:
+                errors.append(f"第{i+1}轨：{err}")
+
+    # 目标时长
+    target_duration = config.get("target_duration", 0)
+    try:
+        if float(target_duration) < 0:
+            errors.append("target_duration 不能为负数")
+    except (ValueError, TypeError):
+        errors.append("target_duration 必须是数字")
+
+    return (len(errors) == 0, errors)
+
+
+# ── 工具函数 ──────────────────────────────────────────────────────────────────
+
+
+def calculate_total_tracks(config: dict) -> int:
+    """计算总轨道数（含主音频）.
+
+    Args:
+        config: 混音配置
+
+    Returns:
+        总轨道数
+    """
+    tracks = config.get("tracks", [])
+    has_main = config.get("has_main_audio", True)
+    count = len(tracks)
+    if has_main:
+        count += 1
+    return count
+
+
+def estimate_mix_duration(tracks: list[dict]) -> float:
+    """估算混音总时长（所有轨道的最晚结束时间）.
+
+    Args:
+        tracks: 轨道列表，包含 start_time 和 duration
+
+    Returns:
+        估算总时长（秒）
+    """
+    max_end = 0.0
+    for t in tracks:
+        try:
+            start = float(t.get("start_time", 0))
+            dur = float(t.get("duration", 0))
+            if dur > 0:
+                end = start + dur
+                if end > max_end:
+                    max_end = end
+        except (ValueError, TypeError):
+            continue
+    return max_end

tests/unit/test_multi_track_mixer_pure.py

@@ -0,0 +1,638 @@
+"""多轨混音纯逻辑单元测试."""
+
+from __future__ import annotations
+
+import math
+
+import pytest
+from video_processing.multi_track_mixer_pure import (
+    build_amix_filter,
+    build_mix_filter_complex,
+    build_track_filter_chain,
+    calculate_amix_volume_compensation,
+    calculate_effective_range,
+    calculate_total_tracks,
+    count_track_types,
+    db_to_linear,
+    estimate_mix_duration,
+    filter_enabled_tracks,
+    is_track_visible,
+    linear_to_db,
+    normalize_volume,
+    sort_tracks_by_priority,
+    validate_audio_track,
+    validate_mix_config,
+)
+
+# ─────────────────────────────────────────────────────────────────────────────
+# 时间计算测试
+# ─────────────────────────────────────────────────────────────────────────────
+
+
+class TestCalculateEffectiveRange:
+    """有效时间范围计算测试."""
+
+    def test_normal_track(self):
+        """正常轨道."""
+        start, dur, trim = calculate_effective_range(5, 10, 30, 60)
+        assert start == 5.0
+        assert dur == 10.0
+        assert trim == 0.0
+
+    def test_track_longer_than_audio(self):
+        """轨道时长超过音频长度."""
+        start, dur, trim = calculate_effective_range(0, 100, 30, 60)
+        assert start == 0.0
+        assert dur == 30.0  # 用音频全长
+
+    def test_zero_track_duration(self):
+        """轨道时长为 0（用音频全长）."""
+        start, dur, trim = calculate_effective_range(0, 0, 30, 60)
+        assert start == 0.0
+        assert dur == 30.0
+
+    def test_negative_start_time(self):
+        """负开始时间（从音频中间取）."""
+        start, dur, trim = calculate_effective_range(-5, 20, 30, 60)
+        assert start == 0.0
+        assert dur == 15.0  # 20 - 5 = 15
+        assert trim == 5.0
+
+    def test_track_after_target(self):
+        """轨道完全在目标之后."""
+        start, dur, trim = calculate_effective_range(100, 10, 30, 60)
+        assert dur == 0.0
+
+    def test_track_before_zero(self):
+        """轨道完全在 0 之前."""
+        start, dur, trim = calculate_effective_range(-50, 10, 30, 60)
+        assert dur == 0.0
+
+    def test_zero_audio_duration(self):
+        """音频时长为 0."""
+        start, dur, trim = calculate_effective_range(0, 10, 0, 60)
+        assert dur == 0.0
+
+    def test_track_extends_beyond_target(self):
+        """轨道超出目标时长."""
+        start, dur, trim = calculate_effective_range(50, 20, 30, 60)
+        assert start == 50.0
+        assert dur == 10.0  # 60 - 50 = 10
+
+    def test_full_target_duration(self):
+        """轨道覆盖整个目标时长."""
+        start, dur, trim = calculate_effective_range(0, 0, 100, 60)
+        assert start == 0.0
+        assert dur == 60.0
+
+
+class TestIsTrackVisible:
+    """轨道可见性测试."""
+
+    def test_visible_track(self):
+        """可见轨道."""
+        assert is_track_visible(5, 10, 30, 60) is True
+
+    def test_invisible_after_target(self):
+        """目标之后不可见."""
+        assert is_track_visible(100, 10, 30, 60) is False
+
+    def test_invisible_zero_duration(self):
+        """零时长不可见."""
+        assert is_track_visible(0, 0, 0, 60) is False
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# 滤镜链构建测试
+# ─────────────────────────────────────────────────────────────────────────────
+
+
+class TestBuildTrackFilterChain:
+    """单轨滤镜链构建测试."""
+
+    def test_basic_structure(self):
+        """基本结构：截断+音量+淡入淡出+延迟+截断."""
+        result = build_track_filter_chain(
+            volume=0.5,
+            fade_in=1.0,
+            fade_out=1.0,
+            effective_start=5.0,
+            need_duration=10.0,
+            trim_start=0.0,
+            target_duration=60.0,
+        )
+        assert "atrim=0.000:10.000" in result
+        assert "volume=0.500" in result
+        assert "afade=t=in:st=0:d=1.000" in result
+        assert "afade=t=out" in result
+        assert "adelay=5000|5000" in result
+        assert "atrim=0:60.000" in result
+
+    def test_volume_1_0_skipped(self):
+        """音量为 1.0 不添加 volume 滤镜."""
+        result = build_track_filter_chain(
+            volume=1.0,
+            fade_in=0,
+            fade_out=0,
+            effective_start=0,
+            need_duration=10,
+            trim_start=0,
+            target_duration=60,
+        )
+        assert "volume=" not in result
+
+    def test_no_fade_in(self):
+        """无淡入."""
+        result = build_track_filter_chain(
+            volume=1.0,
+            fade_in=0,
+            fade_out=2.0,
+            effective_start=0,
+            need_duration=10,
+            trim_start=0,
+            target_duration=60,
+        )
+        assert "afade=t=in" not in result
+        assert "afade=t=out" in result
+
+    def test_no_delay(self):
+        """无延迟（effective_start 很小）."""
+        result = build_track_filter_chain(
+            volume=1.0,
+            fade_in=0,
+            fade_out=0,
+            effective_start=0.001,
+            need_duration=10,
+            trim_start=0,
+            target_duration=60,
+        )
+        assert "adelay" not in result
+
+    def test_with_delay(self):
+        """有延迟."""
+        result = build_track_filter_chain(
+            volume=1.0,
+            fade_in=0,
+            fade_out=0,
+            effective_start=2.5,
+            need_duration=10,
+            trim_start=0,
+            target_duration=60,
+        )
+        assert "adelay=2500|2500" in result
+
+    def test_fade_in_longer_than_duration(self):
+        """淡入超过总时长，不添加淡入."""
+        result = build_track_filter_chain(
+            volume=1.0,
+            fade_in=20,
+            fade_out=0,
+            effective_start=0,
+            need_duration=10,
+            trim_start=0,
+            target_duration=60,
+        )
+        assert "afade=t=in" not in result
+
+    def test_fade_out_at_start(self):
+        """淡出从 0 开始（很短的音频）."""
+        result = build_track_filter_chain(
+            volume=1.0,
+            fade_in=0,
+            fade_out=15,
+            effective_start=0,
+            need_duration=10,
+            trim_start=0,
+            target_duration=60,
+        )
+        # fade_out > need_duration，不添加
+        assert "afade=t=out" not in result
+
+    def test_trim_start_nonzero(self):
+        """从音频中间开始截取."""
+        result = build_track_filter_chain(
+            volume=1.0,
+            fade_in=0,
+            fade_out=0,
+            effective_start=0,
+            need_duration=5,
+            trim_start=3.0,
+            target_duration=60,
+        )
+        assert "atrim=3.000:8.000" in result  # 3.0 to 3.0+5.0
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# amix 滤镜测试
+# ─────────────────────────────────────────────────────────────────────────────
+
+
+class TestBuildAmixFilter:
+    """amix 滤镜构建测试."""
+
+    def test_two_inputs(self):
+        """两路输入."""
+        result = build_amix_filter(2)
+        assert "amix=inputs=2" in result
+        assert "duration=longest" in result
+
+    def test_five_inputs(self):
+        """五路输入."""
+        result = build_amix_filter(5)
+        assert "amix=inputs=5" in result
+
+    def test_zero_inputs(self):
+        """零输入."""
+        assert build_amix_filter(0) == ""
+
+    def test_duration_shortest(self):
+        """shortest 模式."""
+        result = build_amix_filter(3, "shortest")
+        assert "duration=shortest" in result
+
+    def test_invalid_duration_mode(self):
+        """无效模式，默认 longest."""
+        result = build_amix_filter(3, "invalid")
+        assert "duration=longest" in result
+
+
+class TestCalculateAmixVolumeCompensation:
+    """音量补偿计算测试."""
+
+    def test_single_track(self):
+        """单轨，无需补偿."""
+        assert calculate_amix_volume_compensation(1) == 1.0
+
+    def test_two_tracks(self):
+        """两轨，补偿 2x."""
+        assert calculate_amix_volume_compensation(2) == 2.0
+
+    def test_five_tracks(self):
+        """五轨，补偿 5x."""
+        assert calculate_amix_volume_compensation(5) == 5.0
+
+    def test_zero_tracks(self):
+        """零轨，返回 1."""
+        assert calculate_amix_volume_compensation(0) == 1.0
+
+
+class TestBuildMixFilterComplex:
+    """完整混音滤镜测试."""
+
+    def test_with_main_and_two_tracks(self):
+        """主音频 + 2 条轨道."""
+        result = build_mix_filter_complex(2, has_main=True)
+        assert "[0:a][1:a][2:a]" in result  # 3 路输入
+        assert "amix=inputs=3" in result
+        assert "volume=3" in result  # 3x 补偿
+        assert "[mixed]" in result
+
+    def test_no_main_three_tracks(self):
+        """无主音频，3 条轨道."""
+        result = build_mix_filter_complex(3, has_main=False)
+        assert "[0:a][1:a][2:a]" in result
+        assert "amix=inputs=3" in result
+        assert "[mixed]" in result
+
+    def test_zero_tracks_no_main(self):
+        """无轨道无主音频."""
+        assert build_mix_filter_complex(0, has_main=False) == ""
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# 音量计算测试
+# ─────────────────────────────────────────────────────────────────────────────
+
+
+class TestNormalizeVolume:
+    """音量规范化测试."""
+
+    def test_normal_volume(self):
+        """正常音量."""
+        assert normalize_volume(0.5) == 0.5
+
+    def test_none_default(self):
+        """None 默认 1.0."""
+        assert normalize_volume(None) == 1.0
+
+    def test_below_zero_clamped(self):
+        """负值钳制到 0."""
+        assert normalize_volume(-5) == 0.0
+
+    def test_above_max_clamped(self):
+        """超过上限钳制."""
+        assert normalize_volume(3.0) == 2.0
+
+    def test_string_input(self):
+        """字符串输入."""
+        assert normalize_volume("0.5") == 0.5
+
+    def test_invalid_string(self):
+        """无效字符串默认 1.0."""
+        assert normalize_volume("abc") == 1.0
+
+
+class TestDbConversion:
+    """dB 转换测试."""
+
+    def test_0_db_is_unity(self):
+        """0 dB = 1.0."""
+        assert db_to_linear(0) == pytest.approx(1.0)
+
+    def test_negative_db(self):
+        """负 dB < 1."""
+        assert db_to_linear(-6) == pytest.approx(0.5, rel=0.01)
+
+    def test_positive_db(self):
+        """正 dB > 1."""
+        assert db_to_linear(6) == pytest.approx(2.0, rel=0.01)
+
+    def test_round_trip(self):
+        """往返转换."""
+        original = 0.5
+        db = linear_to_db(original)
+        result = db_to_linear(db)
+        assert result == pytest.approx(original)
+
+    def test_zero_linear_is_negative_inf(self):
+        """零线性值 = -inf dB."""
+        assert math.isinf(linear_to_db(0))
+        assert linear_to_db(0) < 0
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# 轨道排序与过滤测试
+# ─────────────────────────────────────────────────────────────────────────────
+
+
+class TestSortTracksByPriority:
+    """轨道优先级排序测试."""
+
+    def test_sorted_by_priority(self):
+        """按优先级排序."""
+        tracks = [
+            {"priority": 10, "name": "high"},
+            {"priority": 1, "name": "highest"},
+            {"priority": 100, "name": "low"},
+        ]
+        result = sort_tracks_by_priority(tracks)
+        assert result[0]["name"] == "highest"
+        assert result[1]["name"] == "high"
+        assert result[2]["name"] == "low"
+
+    def test_default_priority_100(self):
+        """默认优先级 100."""
+        tracks = [
+            {"priority": 50, "name": "mid"},
+            {"name": "default"},
+        ]
+        result = sort_tracks_by_priority(tracks)
+        assert result[0]["name"] == "mid"
+        assert result[1]["name"] == "default"
+
+    def test_same_preserves_order(self):
+        """同优先级保持顺序."""
+        tracks = [
+            {"priority": 10, "name": "first"},
+            {"priority": 10, "name": "second"},
+        ]
+        result = sort_tracks_by_priority(tracks)
+        assert result[0]["name"] == "first"
+        assert result[1]["name"] == "second"
+
+    def test_empty_list(self):
+        """空列表."""
+        assert sort_tracks_by_priority([]) == []
+
+
+class TestFilterEnabledTracks:
+    """启用轨道过滤测试."""
+
+    def test_all_enabled(self):
+        """全部启用."""
+        tracks = [{"enabled": True}, {"enabled": True}]
+        assert len(filter_enabled_tracks(tracks)) == 2
+
+    def test_mixed(self):
+        """混合."""
+        tracks = [
+            {"enabled": True, "name": "a"},
+            {"enabled": False, "name": "b"},
+        ]
+        result = filter_enabled_tracks(tracks)
+        assert len(result) == 1
+        assert result[0]["name"] == "a"
+
+    def test_default_enabled(self):
+        """默认启用."""
+        tracks = [{"name": "a"}]
+        assert len(filter_enabled_tracks(tracks)) == 1
+
+    def test_empty_list(self):
+        """空列表."""
+        assert filter_enabled_tracks([]) == []
+
+
+class TestCountTrackTypes:
+    """轨道类型统计测试."""
+
+    def test_mixed_types(self):
+        """混合类型."""
+        tracks = [
+            {"track_type": "bgm"},
+            {"track_type": "voiceover"},
+            {"track_type": "bgm"},
+            {"track_type": "sfx"},
+        ]
+        counts = count_track_types(tracks)
+        assert counts["bgm"] == 2
+        assert counts["voiceover"] == 1
+        assert counts["sfx"] == 1
+
+    def test_default_type(self):
+        """默认类型."""
+        tracks = [{}]
+        counts = count_track_types(tracks)
+        assert counts["unknown"] == 1
+
+    def test_empty_list(self):
+        """空列表."""
+        assert count_track_types([]) == {}
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# 配置验证测试
+# ─────────────────────────────────────────────────────────────────────────────
+
+
+class TestValidateAudioTrack:
+    """单轨验证测试."""
+
+    def test_valid_track(self):
+        """合法轨道."""
+        ok, errors = validate_audio_track(
+            {
+                "audio_path": "/audio.mp3",
+                "volume": 0.8,
+                "fade_in": 1.0,
+                "fade_out": 2.0,
+            }
+        )
+        assert ok is True
+        assert len(errors) == 0
+
+    def test_missing_path(self):
+        """缺路径."""
+        ok, errors = validate_audio_track({})
+        assert ok is False
+        assert any("audio_path" in e or "asset_id" in e for e in errors)
+
+    def test_negative_volume(self):
+        """负音量."""
+        ok, errors = validate_audio_track(
+            {
+                "audio_path": "/a.mp3",
+                "volume": -1,
+            }
+        )
+        assert ok is False
+        assert any("volume" in e for e in errors)
+
+    def test_negative_fade_in(self):
+        """负淡入."""
+        ok, errors = validate_audio_track(
+            {
+                "audio_path": "/a.mp3",
+                "fade_in": -1,
+            }
+        )
+        assert ok is False
+        assert any("fade_in" in e for e in errors)
+
+    def test_negative_fade_out(self):
+        """负淡出."""
+        ok, errors = validate_audio_track(
+            {
+                "audio_path": "/a.mp3",
+                "fade_out": -1,
+            }
+        )
+        assert ok is False
+        assert any("fade_out" in e for e in errors)
+
+    def test_invalid_volume_type(self):
+        """无效音量类型."""
+        ok, errors = validate_audio_track(
+            {
+                "audio_path": "/a.mp3",
+                "volume": "loud",
+            }
+        )
+        assert ok is False
+        assert any("volume" in e for e in errors)
+
+    def test_with_asset_id(self):
+        """有 asset_id 无 audio_path 也合法."""
+        ok, errors = validate_audio_track({"asset_id": "123"})
+        assert ok is True
+
+
+class TestValidateMixConfig:
+    """混音配置验证测试."""
+
+    def test_valid_config(self):
+        """合法配置."""
+        ok, errors = validate_mix_config(
+            {
+                "tracks": [
+                    {"audio_path": "/a.mp3", "volume": 0.5},
+                    {"audio_path": "/b.mp3", "volume": 0.8},
+                ],
+                "target_duration": 60,
+            }
+        )
+        assert ok is True
+
+    def test_empty_tracks(self):
+        """空轨道列表."""
+        ok, errors = validate_mix_config({"tracks": []})
+        assert ok is False
+        assert any("至少需要" in e for e in errors)
+
+    def test_invalid_track(self):
+        """无效轨道."""
+        ok, errors = validate_mix_config(
+            {
+                "tracks": [
+                    {"audio_path": "/a.mp3"},
+                    {},  # 无效
+                ],
+            }
+        )
+        assert ok is False
+        assert len(errors) >= 1
+
+    def test_negative_target_duration(self):
+        """负目标时长."""
+        ok, errors = validate_mix_config(
+            {
+                "tracks": [{"audio_path": "/a.mp3"}],
+                "target_duration": -10,
+            }
+        )
+        assert ok is False
+        assert any("target_duration" in e for e in errors)
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# 工具函数测试
+# ─────────────────────────────────────────────────────────────────────────────
+
+
+class TestCalculateTotalTracks:
+    """总轨道数计算测试."""
+
+    def test_with_main(self):
+        """含主音频."""
+        assert calculate_total_tracks({"tracks": [1, 2, 3]}) == 4
+
+    def test_without_main(self):
+        """不含主音频."""
+        assert (
+            calculate_total_tracks(
+                {
+                    "tracks": [1, 2],
+                    "has_main_audio": False,
+                }
+            )
+            == 2
+        )
+
+    def test_empty_tracks_with_main(self):
+        """无轨道，只有主音频."""
+        assert calculate_total_tracks({"tracks": []}) == 1
+
+
+class TestEstimateMixDuration:
+    """混音时长估算测试."""
+
+    def test_multiple_tracks(self):
+        """多轨道取最长结束时间."""
+        tracks = [
+            {"start_time": 0, "duration": 10},
+            {"start_time": 5, "duration": 20},  # 结束 25
+            {"start_time": 2, "duration": 5},
+        ]
+        assert estimate_mix_duration(tracks) == pytest.approx(25.0)
+
+    def test_empty_list(self):
+        """空列表."""
+        assert estimate_mix_duration([]) == 0.0
+
+    def test_zero_duration_tracks_ignored(self):
+        """零时长轨道忽略."""
+        tracks = [
+            {"start_time": 0, "duration": 0},
+            {"start_time": 5, "duration": 10},
+        ]
+        assert estimate_mix_duration(tracks) == pytest.approx(15.0)