feat: 转场特效引擎 — TransitionEngine + 14种转场预设 + 时长边界校验 + 降级策略

apps/worker/video_processing/transition_engine.py

@@ -0,0 +1,381 @@
+"""转场特效引擎 — Phase 8 智能增强.
+
+基于 FFmpeg xfade 滤镜的统一转场抽象层，提供：
+  1. 转场类型枚举与预设管理
+  2. 转场配置解析与边界校验
+  3. 降级策略（不支持的转场自动 fallback 到硬切）
+  4. xfade 滤镜链构建（封装底层 ffmpeg_utils）
+
+新增转场只需在 TransitionType 中加一项 + 在 XFADE_TRANSITION_MAP 中映射。
+"""
+
+from __future__ import annotations
+
+import logging
+import sys
+from dataclasses import dataclass
+
+if sys.version_info >= (3, 11):
+    from enum import StrEnum
+else:
+    from enum import Enum
+
+    class StrEnum(str, Enum):
+        pass
+
+
+from video_processing.ffmpeg_utils import build_xfade_filter_chain
+
+logger = logging.getLogger(__name__)
+
+
+# ── 常量 ──────────────────────────────────────────────────────────────────────
+
+# 转场时长范围（秒）
+MIN_TRANSITION_DURATION = 0.3
+MAX_TRANSITION_DURATION = 2.0
+DEFAULT_TRANSITION_DURATION = 0.5
+
+# 硬切（无转场）
+CUT_TRANSITION = "cut"
+
+
+# ── 转场类型枚举 ──────────────────────────────────────────────────────────────
+
+
+class TransitionType(StrEnum):
+    """支持的转场效果类型.
+
+    每种类型对应 FFmpeg xfade filter 的一个 transition 值。
+    新增转场只需在此添加一项，并在 _FFMPEG_XFADE_MAP 中映射。
+    """
+
+    # 硬切（无转场效果，直接拼接）
+    CUT = "cut"
+
+    # 淡入淡出（最常用，默认 fallback）
+    FADE = "fade"
+
+    # 溶解（交叉溶解）
+    DISSOLVE = "dissolve"
+
+    # 滑入系列
+    SLIDE_LEFT = "slideleft"
+    SLIDE_RIGHT = "slideright"
+    SLIDE_UP = "slideup"
+    SLIDE_DOWN = "slidedown"
+
+    # 缩放
+    ZOOM = "zoom"
+
+    # 擦除系列
+    WIPE_LEFT = "wipeleft"
+    WIPE_RIGHT = "wiperight"
+    WIPE_UP = "wipeup"
+    WIPE_DOWN = "wipedown"
+
+    # 圆形扩散
+    CIRCLE_CROP = "circlecrop"
+
+    # 矩形覆盖
+    RECT_CROP = "rectcrop"
+
+    @classmethod
+    def all_supported(cls) -> list[str]:
+        """返回所有支持的转场类型名称列表."""
+        return [t.value for t in cls if t != cls.CUT]
+
+    @classmethod
+    def is_supported(cls, name: str) -> bool:
+        """检查转场类型是否支持（不区分大小写和下划线）."""
+        normalized = _normalize_transition_name(name)
+        return normalized in _NAME_TO_ENUM_MAP
+
+
+# ── 名称 → 枚举 映射（支持多种别名）──────────────────────────────────────────
+
+
+def _normalize_transition_name(name: str) -> str:
+    """标准化转场名称：小写 + 去下划线."""
+    return name.lower().replace("_", "").replace("-", "")
+
+
+# 构建别名映射
+_NAME_TO_ENUM_MAP: dict[str, TransitionType] = {}
+for _t in TransitionType:
+    _NAME_TO_ENUM_MAP[_normalize_transition_name(_t.value)] = _t
+
+# 额外的别名
+_ALIASES: dict[str, TransitionType] = {
+    "dissolve": TransitionType.DISSOLVE,
+    "crossfade": TransitionType.DISSOLVE,
+    "crossdissolve": TransitionType.DISSOLVE,
+    "fadein": TransitionType.FADE,
+    "fadeout": TransitionType.FADE,
+    "fadeblack": TransitionType.FADE,
+    "slide": TransitionType.SLIDE_LEFT,  # 默认向左滑
+    "wipe": TransitionType.WIPE_LEFT,  # 默认向左擦
+    "zoomin": TransitionType.ZOOM,
+    "zoomout": TransitionType.ZOOM,
+    "circle": TransitionType.CIRCLE_CROP,
+    "rect": TransitionType.RECT_CROP,
+}
+for _alias, _type in _ALIASES.items():
+    _key = _normalize_transition_name(_alias)
+    if _key not in _NAME_TO_ENUM_MAP:
+        _NAME_TO_ENUM_MAP[_key] = _type
+
+
+# ── TransitionType → FFmpeg xfade transition 名称映射 ─────────────────────────
+
+
+_FFMPEG_XFADE_MAP: dict[TransitionType, str] = {
+    TransitionType.FADE: "fade",
+    TransitionType.DISSOLVE: "dissolve",
+    TransitionType.SLIDE_LEFT: "slideleft",
+    TransitionType.SLIDE_RIGHT: "slideright",
+    TransitionType.SLIDE_UP: "slideup",
+    TransitionType.SLIDE_DOWN: "slidedown",
+    TransitionType.ZOOM: "zoomin",
+    TransitionType.WIPE_LEFT: "wipeleft",
+    TransitionType.WIPE_RIGHT: "wiperight",
+    TransitionType.WIPE_UP: "wipeup",
+    TransitionType.WIPE_DOWN: "wipedown",
+    TransitionType.CIRCLE_CROP: "circlecrop",
+    TransitionType.RECT_CROP: "rectcrop",
+}
+
+
+# ── 转场配置 ──────────────────────────────────────────────────────────────────
+
+
+@dataclass(slots=True)
+class TransitionConfig:
+    """转场效果配置.
+
+    Attributes:
+        effect: 转场效果名称（见 TransitionType）
+        duration: 转场时长（秒），范围 0.3~2.0，默认 0.5
+    """
+
+    effect: str = CUT_TRANSITION
+    duration: float = DEFAULT_TRANSITION_DURATION
+
+    @classmethod
+    def parse(cls, effect: str | None = None, duration: float | None = None) -> "TransitionConfig":
+        """解析并验证转场配置，自动处理边界和降级.
+
+        Args:
+            effect: 转场效果名称（None 或空则使用默认 cut）
+            duration: 转场时长（None 则使用默认值）
+
+        Returns:
+            验证后的 TransitionConfig
+        """
+        # 处理 effect
+        final_effect = CUT_TRANSITION
+        if effect and effect.strip():
+            effect_clean = effect.strip()
+            if TransitionType.is_supported(effect_clean):
+                final_effect = _resolve_transition_enum(effect_clean).value
+            elif effect_clean.lower() == CUT_TRANSITION:
+                final_effect = CUT_TRANSITION
+            else:
+                # 降级：不支持的转场 → 硬切，不阻断渲染
+                logger.warning(
+                    "不支持的转场效果 '%s'，已降级为硬切（cut）",
+                    effect_clean,
+                )
+                final_effect = CUT_TRANSITION
+
+        # 处理 duration：边界钳制
+        final_duration = DEFAULT_TRANSITION_DURATION
+        if duration is not None:
+            try:
+                d = float(duration)
+                if d < MIN_TRANSITION_DURATION:
+                    logger.warning(
+                        "转场时长 %.3fs 小于最小值 %.1fs，已钳制到最小值",
+                        d,
+                        MIN_TRANSITION_DURATION,
+                    )
+                    final_duration = MIN_TRANSITION_DURATION
+                elif d > MAX_TRANSITION_DURATION:
+                    logger.warning(
+                        "转场时长 %.3fs 大于最大值 %.1fs，已钳制到最大值",
+                        d,
+                        MAX_TRANSITION_DURATION,
+                    )
+                    final_duration = MAX_TRANSITION_DURATION
+                else:
+                    final_duration = d
+            except (TypeError, ValueError):
+                logger.warning("无效的转场时长 '%s'，使用默认值 %.1fs", duration, DEFAULT_TRANSITION_DURATION)
+                final_duration = DEFAULT_TRANSITION_DURATION
+
+        return cls(effect=final_effect, duration=final_duration)
+
+    @property
+    def is_cut(self) -> bool:
+        """是否为硬切（无转场效果）."""
+        return self.effect == CUT_TRANSITION
+
+    @property
+    def ffmpeg_transition(self) -> str:
+        """获取对应的 FFmpeg xfade transition 名称."""
+        if self.is_cut:
+            return ""
+        enum_type = _resolve_transition_enum(self.effect)
+        return _FFMPEG_XFADE_MAP.get(enum_type, "fade")
+
+
+def _resolve_transition_enum(name: str) -> TransitionType:
+    """将名称解析为 TransitionType 枚举，必须先通过 is_supported 校验."""
+    normalized = _normalize_transition_name(name)
+    return _NAME_TO_ENUM_MAP.get(normalized, TransitionType.FADE)
+
+
+# ── 转场引擎 ──────────────────────────────────────────────────────────────────
+
+
+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"},
+        ]