xiaoxia-saas/packages/domain/xfade_builder.py

"""XFade 转场滤镜构建 — 纯逻辑，无 FFmpeg 依赖.

抽离自 apps/worker/video_processing/ffmpeg_utils.py，包含：
- xfade 转场效果名称映射
- 滤镜链串联工具
- xfade 转场滤镜链构建（带时长钳制）
"""

from __future__ import annotations

import logging
from typing import Any

logger = logging.getLogger(__name__)


# ── 常量 ──────────────────────────────────────────────────────────────────────


DEFAULT_TRANSITION_DURATION = 0.5

# xfade 转场映射：transition_effect 名称 → FFmpeg xfade transition 名称
# 键同时支持 TransitionEffect 枚举值和字符串名称（向后兼容）
# "cut" 为特殊值：硬切，不使用 xfade（由调用方特殊处理）
XFADE_TRANSITION_MAP: dict[str, str] = {
    # 基础
    "fade": "fade",
    "dissolve": "dissolve",
    "crossfade": "dissolve",
    "crossdissolve": "dissolve",
    # 滑入系列
    "slideleft": "slideleft",
    "slide_left": "slideleft",
    "slideright": "slideright",
    "slide_right": "slideright",
    "slideup": "slideup",
    "slide_up": "slideup",
    "slidedown": "slidedown",
    "slide_down": "slidedown",
    "slide": "slideleft",  # 默认向左滑
    # 缩放
    "zoom": "zoomin",
    "zoomin": "zoomin",
    "zoomout": "zoomout",
    # 擦除系列
    "wipe": "wipeleft",  # 默认向左擦
    "wipeleft": "wipeleft",
    "wiperight": "wiperight",
    "wipeup": "wipeup",
    "wipedown": "wipedown",
    # 特殊效果
    "circlecrop": "circlecrop",
    "circle": "circlecrop",
    "rectcrop": "rectcrop",
    "rect": "rectcrop",
}

# 所有支持的转场效果名称（用户侧输入）
SUPPORTED_TRANSITIONS: set[str] = set(XFADE_TRANSITION_MAP.keys())

# 所有 FFmpeg xfade transition 名称（输出侧）
XFade_TRANSITION_NAMES: set[str] = set(XFADE_TRANSITION_MAP.values())


# ── 工具函数 ─────────────────────────────────────────────────────────────────


def chain_filters(filters: list[str], output_label: str, *, input_label: str = "0:v") -> str:
    """将滤镜列表串联为 FFmpeg 滤镜字符串.

    例：chain_filters(["scale=1280:720", "fps=25"], "v0")
    → "[0:v]scale=1280:720,fps=25[v0]"

    Args:
        filters: 滤镜字符串列表
        output_label: 输出标签（不带方括号）
        input_label: 输入标签（不带方括号），默认 "0:v"

    Returns:
        完整的滤镜字符串
    """
    filter_body = ",".join(filters)
    return f"[{input_label}]{filter_body}[{output_label}]"


def resolve_xfade_transition(transition_name: Any) -> str:
    """将转场效果名称映射为 FFmpeg xfade transition 名称.

    支持 TransitionEffect 枚举值和字符串名称，未知值回退到 "fade"。

    Args:
        transition_name: 转场名称（字符串或带 .value 属性的枚举）

    Returns:
        FFmpeg xfade transition 名称
    """
    # 兼容 TransitionEffect 枚举（有 .value 属性）
    if hasattr(transition_name, "value"):
        transition_name = transition_name.value
    return XFADE_TRANSITION_MAP.get(transition_name, "fade")


# ── xfade 滤镜链构建 ─────────────────────────────────────────────────────────


def build_xfade_filter_chain(
    clip_durations: list[float],
    clip_video_labels: list[str],
    transitions: list[str],
    *,
    transition_duration: float = DEFAULT_TRANSITION_DURATION,
    output_label: str = "outv",
) -> tuple[str, float]:
    """构建 xfade 转场滤镜链.

    对每步 xfade 自动钳制 transition duration，确保
    ``offset + td ≤ first_input_duration``，避免 FFmpeg exit 234。

    Args:
        clip_durations: 每个片段的时长（必须与 trim 后的实际时长一致）
        clip_video_labels: 每个片段的视频流标签（如 "v0", "v1"）
        transitions: 每个片段对应的转场效果（第一个片段的转场被忽略）
        transition_duration: 转场时长（秒）
        output_label: 最终输出标签

    Returns:
        (filter_string, estimated_total_duration)
    """
    n = len(clip_durations)
    parts: list[str] = []

    if n == 0:
        return "", 0.0

    if n == 1:
        parts.append(f"[{clip_video_labels[0]}]copy[{output_label}]")
        return ";".join(parts), clip_durations[0]

    # xfade 链 — 每步动态钳制 td，防止 offset + td > first_input_duration
    cumulative = 0.0
    prev_label = clip_video_labels[0]
    total_transition = 0.0  # 累计已使用的转场时长

    for i in range(1, n):
        cumulative += clip_durations[i - 1]

        # 当前 xfade 的第一个输入时长
        if i == 1:
            first_input_dur = clip_durations[0]
        else:
            first_input_dur = cumulative - total_transition

        # 原始 offset 计算
        offset = max(0.0, cumulative - transition_duration * i)

        # 安全钳制：offset + td 不能超过第一个输入的时长
        available = max(0.0, first_input_dur - offset)
        safe_td = min(transition_duration, available)

        # 同时不能超过剩余总时长
        remaining = max(0.0, sum(clip_durations) - cumulative)
        safe_td = min(safe_td, remaining)
        # 同时不能超过当前第二个输入（单个片段）的时长
        safe_td = min(safe_td, clip_durations[i])
        safe_td = max(0.001, safe_td)  # 至少 1ms，避免 td=0

        transition = transitions[i] if i < len(transitions) else "cut"
        xfade_transition = resolve_xfade_transition(transition)

        if i == n - 1:
            out_label = output_label
        else:
            out_label = f"xf{i}"

        parts.append(
            f"[{prev_label}][{clip_video_labels[i]}]"
            f"xfade=transition={xfade_transition}"
            f":duration={safe_td:.3f}"
            f":offset={offset:.3f}"
            f"[{out_label}]"
        )
        prev_label = out_label
        total_transition += safe_td

    # 总时长减去转场重叠部分
    total_duration = sum(clip_durations) - total_transition
    return ";".join(parts), max(0.0, total_duration)