xiaoxia-saas/apps/worker/video_processing/multi_track_mixer_pure.py

"""多轨混音纯逻辑模块.

所有函数均为纯函数，不调用 FFmpeg、不操作文件。
便于单元测试，也方便被其他模块复用。
"""

from __future__ import annotations

# ── 单轨时间计算 ──────────────────────────────────────────────────────────────


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:
        线性音量值
    """

    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