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

"""转场特效引擎 — Phase 8 智能增强.

基于 FFmpeg xfade 滤镜的统一转场抽象层，提供：
  1. 转场类型枚举与预设管理
  2. 转场配置解析与边界校验
  3. 降级策略（不支持的转场自动 fallback 到硬切）
  4. xfade 滤镜链构建（封装底层 ffmpeg_utils）

新增转场只需在 TransitionType 中加一项 + 在 XFADE_TRANSITION_MAP 中映射。
"""

from __future__ import annotations

import logging

from video_processing.ffmpeg_utils import build_xfade_filter_chain

from packages.domain.transition_config import (  # noqa: F401 — 向后兼容
    CUT_TRANSITION,
    DEFAULT_TRANSITION_DURATION,
    MAX_TRANSITION_DURATION,
    MIN_TRANSITION_DURATION,
    TransitionConfig,
    TransitionType,
)

logger = logging.getLogger(__name__)


# ── 转场引擎 ──────────────────────────────────────────────────────────────────


class TransitionEngine:
    """转场特效引擎.

    封装转场配置验证、降级策略和 xfade 滤镜链构建，
    供 UnifiedRenderService 等上层调用。

    用法::

        engine = TransitionEngine(default_duration=0.5)
        config = engine.resolve_config("fade", 0.8)
        filter_str, total_dur = engine.build_xfade_chain(
            clip_durations=[3.0, 4.0, 5.0],
            clip_video_labels=["v0", "v1", "v2"],
            transitions=["cut", "fade", "dissolve"],
        )
    """

    def __init__(self, default_duration: float = DEFAULT_TRANSITION_DURATION) -> None:
        """初始化转场引擎.

        Args:
            default_duration: 默认转场时长（秒），用于未指定时长的 clip
        """
        self._default_duration = default_duration

    def resolve_config(
        self,
        effect: str | None = None,
        duration: float | None = None,
    ) -> TransitionConfig:
        """解析单个转场配置，应用验证和降级.

        Args:
            effect: 转场效果名称
            duration: 转场时长

        Returns:
            验证后的 TransitionConfig
        """
        # 若未指定 duration，使用引擎默认值
        dur = duration if duration is not None else self._default_duration
        return TransitionConfig.parse(effect=effect, duration=dur)

    def resolve_clip_transitions(
        self,
        clip_transitions: list[str],
        clip_durations: list[float] | None = None,
    ) -> list[TransitionConfig]:
        """批量解析 clip 级别的转场配置.

        Args:
            clip_transitions: 每个 clip 的转场效果名称列表
            clip_durations: 每个 clip 的时长列表（用于验证转场时长不超过片段时长）

        Returns:
            TransitionConfig 列表
        """
        configs: list[TransitionConfig] = []
        for i, effect in enumerate(clip_transitions):
            cfg = self.resolve_config(effect=effect)
            # 额外校验：转场时长不能超过对应 clip 时长的一半（保守限制）
            if clip_durations and i < len(clip_durations) and not cfg.is_cut:
                max_safe_duration = max(MIN_TRANSITION_DURATION, clip_durations[i] * 0.5)
                if cfg.duration > max_safe_duration:
                    cfg = TransitionConfig(effect=cfg.effect, duration=max_safe_duration)
            configs.append(cfg)
        return configs

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

        对每步转场应用验证和降级，然后调用底层 ffmpeg_utils 构建。

        Args:
            clip_durations: 每个片段的时长
            clip_video_labels: 每个片段的视频流标签
            transitions: 每个片段对应的转场效果
            transition_duration: 统一转场时长，None 则使用引擎默认值
            output_label: 最终输出标签

        Returns:
            (filter_string, estimated_total_duration)
        """
        if len(clip_durations) <= 1:
            return build_xfade_filter_chain(
                clip_durations=clip_durations,
                clip_video_labels=clip_video_labels,
                transitions=transitions,
                transition_duration=transition_duration or self._default_duration,
                output_label=output_label,
            )

        # 解析所有转场配置
        resolved = self.resolve_clip_transitions(transitions, clip_durations)
        resolved_effects = [c.effect for c in resolved]

        # 使用统一的时长（取各转场中最大的时长作为基准，底层会做每步钳制）
        dur = transition_duration or self._default_duration
        if not dur:
            dur = max(c.duration for c in resolved) if resolved else DEFAULT_TRANSITION_DURATION

        # 调用底层构建
        return build_xfade_filter_chain(
            clip_durations=clip_durations,
            clip_video_labels=clip_video_labels,
            transitions=resolved_effects,
            transition_duration=dur,
            output_label=output_label,
        )

    @staticmethod
    def supported_transitions() -> list[dict[str, str]]:
        """获取所有支持的转场效果列表（用于 API 返回给前端）.

        Returns:
            [{name, display_name, category}, ...]
        """
        return [
            {"name": "cut", "display_name": "硬切", "category": "basic"},
            {"name": "fade", "display_name": "淡入淡出", "category": "basic"},
            {"name": "dissolve", "display_name": "溶解", "category": "basic"},
            {"name": "slideleft", "display_name": "左滑入", "category": "slide"},
            {"name": "slideright", "display_name": "右滑入", "category": "slide"},
            {"name": "slideup", "display_name": "上滑入", "category": "slide"},
            {"name": "slidedown", "display_name": "下滑入", "category": "slide"},
            {"name": "zoom", "display_name": "缩放", "category": "zoom"},
            {"name": "wipeleft", "display_name": "左擦除", "category": "wipe"},
            {"name": "wiperight", "display_name": "右擦除", "category": "wipe"},
            {"name": "wipeup", "display_name": "上擦除", "category": "wipe"},
            {"name": "wipedown", "display_name": "下擦除", "category": "wipe"},
            {"name": "circlecrop", "display_name": "圆形扩散", "category": "special"},
            {"name": "rectcrop", "display_name": "矩形扩散", "category": "special"},
        ]