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

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

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

注：核心领域模型已抽离到 packages/domain/speed_config.py，
本模块保留薄包装层，确保向后兼容。
"""

from __future__ import annotations

from packages.domain.speed_config import MAX_SPEED  # noqa: F401 — 向后兼容
from packages.domain.speed_config import MIN_SPEED  # noqa: F401 — 向后兼容
from packages.domain.speed_config import (
    DEFAULT_SPEED,
    SpeedConfig,
    _split_atempo_stages,
)
from packages.domain.speed_config import adjust_duration as _adjust_duration_base  # noqa: F401 — 向后兼容
from packages.domain.speed_config import build_audio_filter as _build_audio_filter_base
from packages.domain.speed_config import build_video_filter as _build_video_filter_base
from packages.domain.speed_config import resolve_clip_speed as _resolve_clip_speed_base


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

    薄包装层，实际逻辑委托给 packages.domain.speed_config。
    """

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

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

    @staticmethod
    def _split_atempo_stages(speed: float) -> list[float]:
        """将速度拆分为多级 atempo 串联（内部方法，向后兼容）."""
        return _split_atempo_stages(speed)

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

    def build_clip_speed_filter(
        self,
        speed: float,
        pitch_correct: bool = True,
    ) -> tuple[str, str, SpeedConfig]:
        """便捷方法：从单一 speed 值生成视频+音频滤镜."""
        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 或缺失则使用全局速度."""
        return _resolve_clip_speed_base(clip_config, global_speed)


# ── 向后兼容：模块级函数（重构前的 API） ─────────────────────────
def build_video_filter(config):
    """向后兼容：模块级 build_video_filter."""
    return _build_video_filter_base(config)


def build_audio_filter(config):
    """向后兼容：模块级 build_audio_filter."""
    return _build_audio_filter_base(config)


def adjust_duration(original_duration, config):
    """向后兼容：模块级 adjust_duration."""
    return _adjust_duration_base(original_duration, config)


def resolve_clip_speed(clip_config, global_speed=DEFAULT_SPEED):
    """向后兼容：模块级 resolve_clip_speed."""
    return _resolve_clip_speed_base(clip_config, global_speed)


def build_clip_speed_filter(speed, pitch_correct=True):
    """向后兼容：模块级 build_clip_speed_filter."""
    config = SpeedConfig(speed=speed, pitch_correct=pitch_correct)
    config.clamp()
    return (
        build_video_filter(config),
        build_audio_filter(config),
        config,
    )