xiaoxia-saas/packages/domain/speed_config.py

"""调速配置领域模型 — 纯逻辑，无FFmpeg依赖.

抽离自 speed_engine.py，包含：
- SpeedConfig 数据类（解析/钳制/原速判断）
- 视频/音频调速滤镜构建
- atempo 多级拆分算法
- 时长计算
"""

from __future__ import annotations

from dataclasses import dataclass
from typing import Any

# ─── 常量 ───────────────────────────────────────────────
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: dict[str, Any] | None) -> 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

    @property
    def is_fast(self) -> bool:
        """是否加速播放."""
        return self.speed > 1.0

    @property
    def is_slow(self) -> bool:
        """是否减速播放."""
        return self.speed < 1.0


# ── 滤镜构建 ────────────────────────────────────────────────────────────────


def build_video_filter(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(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] = _split_atempo_stages(speed)
    return ",".join(f"atempo={s:.4f}" for s in stages)


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(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(
    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 (
        build_video_filter(config),
        build_audio_filter(config),
        config,
    )


def resolve_clip_speed(
    clip_config: dict[str, Any] | None,
    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)