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

"""视频倒放引擎 — 基于 FFmpeg reverse + areverse 滤镜实现视频/音频倒放.

支持能力：
- 视频倒放（reverse 滤镜）
- 音频倒放（areverse 滤镜）
- 按 clip 分段倒放，每个 clip 独立配置
- 降级策略：不支持时跳过，不阻断渲染
"""

from __future__ import annotations

import logging
from dataclasses import dataclass
from typing import Any

logger = logging.getLogger(__name__)


# ── 数据模型 ──────────────────────────────────────────────────────────────────


@dataclass
class ReverseConfig:
    """视频倒放配置.

    从 clip.config.reverse 读取，零侵入数据模型.
    """

    enabled: bool = False
    reverse_video: bool = True  # 是否倒放视频
    reverse_audio: bool = True  # 是否倒放音频

    @classmethod
    def from_dict(cls, data: dict[str, Any] | None) -> "ReverseConfig":
        """从字典解析配置."""
        if not isinstance(data, dict):
            return cls(enabled=False)
        try:
            if not data.get("enabled", False):
                return cls(enabled=False)
            return cls(
                enabled=True,
                reverse_video=bool(data.get("reverse_video", True)),
                reverse_audio=bool(data.get("reverse_audio", True)),
            )
        except (AttributeError, TypeError) as e:
            logger.warning("倒放配置解析失败: %s，使用默认配置", e)
            return cls(enabled=False)


# ── 倒放引擎 ──────────────────────────────────────────────────────────────────


class ReverseEngine:
    """视频倒放引擎 — 生成 FFmpeg 倒放滤镜.

    视频倒放：reverse 滤镜
    音频倒放：areverse 滤镜

    注意事项：
    - reverse 滤镜需要将整个视频帧加载到内存，长视频可能占用大量内存
    - 建议对单 clip 时长做限制（如 < 60s），超长视频建议降级
    """

    # 安全限制：单 clip 超过此时长不启用倒放（防止内存溢出）
    MAX_SAFE_DURATION = 120.0  # 秒

    @staticmethod
    def build_video_filter(config: ReverseConfig, duration: float = 0.0) -> str:
        """构建视频倒放滤镜字符串.

        Args:
            config: 倒放配置
            duration: clip 时长（秒），用于安全检查

        Returns:
            FFmpeg 滤镜字符串，如 "reverse"；无效果返回空字符串
        """
        if not config.enabled or not config.reverse_video:
            return ""

        # 安全检查：超长视频不启用倒放
        if duration > ReverseEngine.MAX_SAFE_DURATION:
            logger.warning(
                "视频倒放安全限制：clip 时长 %.1fs 超过上限 %.1fs，跳过倒放",
                duration,
                ReverseEngine.MAX_SAFE_DURATION,
            )
            return ""

        return "reverse"

    @staticmethod
    def build_audio_filter(config: ReverseConfig, duration: float = 0.0) -> str:
        """构建音频倒放滤镜字符串.

        Args:
            config: 倒放配置
            duration: clip 时长（秒），用于安全检查

        Returns:
            FFmpeg 音频滤镜字符串，如 "areverse"；无效果返回空字符串
        """
        if not config.enabled or not config.reverse_audio:
            return ""

        # 安全检查：超长音频不启用倒放
        if duration > ReverseEngine.MAX_SAFE_DURATION:
            logger.warning(
                "音频倒放安全限制：clip 时长 %.1fs 超过上限 %.1fs，跳过倒放",
                duration,
                ReverseEngine.MAX_SAFE_DURATION,
            )
            return ""

        return "areverse"