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

"""视频调速引擎 — 基于 FFmpeg setpts + atempo 的速度调整能力。

支持：
- 0.25x ~ 4x 变速范围
- 视频调速（setpts）
- 音频调速（atempo，多级串联处理超范围值）
- 音调修正（pitch_correct，默认开启）
- 边界自动钳制，不阻断渲染
"""

from dataclasses import dataclass
from typing import Optional

# ─── 常量 ───────────────────────────────────────────────
MIN_SPEED = 0.25
MAX_SPEED = 4.0
DEFAULT_SPEED = 1.0

# atempo 单级有效范围
_ATEMPO_MIN = 0.5
_ATEMPO_MAX = 2.0


@dataclass
class SpeedConfig:
    """调速配置。

    Attributes:
        speed: 播放速度，0.25~4.0，1.0 为原速
        pitch_correct: 是否保持音调（默认 True，用 atempo 时间拉伸算法）
    """

    speed: float = DEFAULT_SPEED
    pitch_correct: bool = True

    @classmethod
    def parse(cls, data: Optional[dict]) -> "SpeedConfig":
        """从 dict 解析配置，无效值回退到默认。"""
        if not data or not isinstance(data, dict):
            return cls()

        speed = data.get("speed", DEFAULT_SPEED)
        if not isinstance(speed, (int, float)):
            speed = DEFAULT_SPEED

        pitch_correct = data.get("pitch_correct", True)
        if not isinstance(pitch_correct, bool):
            pitch_correct = True

        config = cls(speed=float(speed), pitch_correct=pitch_correct)
        config.clamp()
        return config

    def clamp(self) -> None:
        """将速度钳制到合法范围。"""
        if self.speed <= 0:
            self.speed = DEFAULT_SPEED
        elif self.speed < MIN_SPEED:
            self.speed = MIN_SPEED
        elif self.speed > MAX_SPEED:
            self.speed = MAX_SPEED

    @property
    def is_original(self) -> bool:
        """是否原速（无需调速）。"""
        return abs(self.speed - 1.0) < 1e-6


class SpeedEngine:
    """调速引擎 — 生成 FFmpeg 调速滤镜链。

    用法：
        engine = SpeedEngine()
        video_filter = engine.build_video_filter(config)
        audio_filter = engine.build_audio_filter(config)
        new_duration = engine.adjust_duration(duration, config)
    """

    def build_video_filter(self, config: SpeedConfig) -> str:
        """生成视频调速滤镜字符串。

        返回 setpts 滤镜表达式，原速时返回空字符串。
        """
        if config.is_original:
            return ""
        # setpts=PTS/speed — speed>1 加速，speed<1 减速
        return f"setpts=PTS/{config.speed:.4f}"

    def build_audio_filter(self, config: SpeedConfig) -> str:
        """生成音频调速滤镜字符串。

        atempo 单级范围 0.5~2.0，超出范围时自动多级串联：
        - 0.25x → atempo=0.5,atempo=0.5
        - 4x → atempo=2.0,atempo=2.0
        - 0.3x → atempo=0.5,atempo=0.6
        - 3x → atempo=2.0,atempo=1.5

        原速时返回空字符串。
        """
        if config.is_original:
            return ""

        speed = config.speed
        stages: list[float] = self._split_atempo_stages(speed)
        return ",".join(f"atempo={s:.4f}" for s in stages)

    @staticmethod
    def _split_atempo_stages(speed: float) -> list[float]:
        """将速度拆分为多级 atempo 串联，每级都在 [0.5, 2.0] 范围内。"""
        if _ATEMPO_MIN <= speed <= _ATEMPO_MAX:
            return [speed]

        stages: list[float] = []
        remaining = speed

        # 加速场景（speed > 2.0）
        if speed > _ATEMPO_MAX:
            while remaining > _ATEMPO_MAX:
                stages.append(_ATEMPO_MAX)
                remaining /= _ATEMPO_MAX
            stages.append(remaining)

        # 减速场景（speed < 0.5）
        else:
            while remaining < _ATEMPO_MIN:
                stages.append(_ATEMPO_MIN)
                remaining /= _ATEMPO_MIN
            stages.append(remaining)

        return stages

    def adjust_duration(self, original_duration: float, config: SpeedConfig) -> float:
        """计算调速后的时长。

        加速 → 时长变短；减速 → 时长变长。
        """
        if config.is_original or original_duration <= 0:
            return original_duration
        return original_duration / config.speed

    def build_clip_speed_filter(
        self,
        speed: float,
        pitch_correct: bool = True,
    ) -> tuple[str, str, SpeedConfig]:
        """便捷方法：从单一 speed 值生成视频+音频滤镜。

        返回 (video_filter, audio_filter, config)。
        """
        config = SpeedConfig(speed=speed, pitch_correct=pitch_correct)
        config.clamp()
        return (
            self.build_video_filter(config),
            self.build_audio_filter(config),
            config,
        )

    @staticmethod
    def resolve_clip_speed(
        clip_config: dict,
        global_speed: float = DEFAULT_SPEED,
    ) -> float:
        """从 clip config 中解析 playback_speed，0 或缺失则使用全局速度。"""
        speed = clip_config.get("playback_speed", 0) if clip_config else 0
        if not isinstance(speed, (int, float)) or speed <= 0:
            return global_speed
        return float(speed)