test(wave117): 抽离speed_config领域模型 + 57单测

apps/worker/video_processing/speed_engine.py

@@ -1,4 +1,4 @@
-"""视频调速引擎 — 基于 FFmpeg setpts + atempo 的速度调整能力。
+"""视频调速引擎 — 基于 FFmpeg setpts + atempo 的速度调整能力.
 
 支持：
 - 0.25x ~ 4x 变速范围
@@ -6,147 +6,57 @@
 - 音频调速（atempo，多级串联处理超范围值）
 - 音调修正（pitch_correct，默认开启）
 - 边界自动钳制，不阻断渲染
+
+注：核心领域模型已抽离到 packages/domain/speed_config.py，
+本模块保留薄包装层，确保向后兼容。
 """
 
-from dataclasses import dataclass
+from __future__ import annotations
+
 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
+from packages.domain.speed_config import (  # noqa: F401 — 向后兼容
+    DEFAULT_SPEED,
+    MAX_SPEED,
+    MIN_SPEED,
+    SpeedConfig,
+    _split_atempo_stages,
+    adjust_duration as _adjust_duration_base,
+    build_audio_filter as _build_audio_filter_base,
+    build_video_filter as _build_video_filter_base,
+    resolve_clip_speed as _resolve_clip_speed_base,
+)
 
 
 class SpeedEngine:
-    """调速引擎 — 生成 FFmpeg 调速滤镜链。
+    """调速引擎 — 生成 FFmpeg 调速滤镜链.
 
-    用法：
-        engine = SpeedEngine()
-        video_filter = engine.build_video_filter(config)
-        audio_filter = engine.build_audio_filter(config)
-        new_duration = engine.adjust_duration(duration, config)
+    薄包装层，实际逻辑委托给 packages.domain.speed_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}"
+        """生成视频调速滤镜字符串."""
+        return _build_video_filter_base(config)
 
     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)
+        """生成音频调速滤镜字符串."""
+        return _build_audio_filter_base(config)
 
     @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
+        """将速度拆分为多级 atempo 串联（内部方法，向后兼容）."""
+        return _split_atempo_stages(speed)
 
     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
+        """计算调速后的时长."""
+        return _adjust_duration_base(original_duration, config)
 
     def build_clip_speed_filter(
         self,
         speed: float,
         pitch_correct: bool = True,
     ) -> tuple[str, str, SpeedConfig]:
-        """便捷方法：从单一 speed 值生成视频+音频滤镜。
-
-        返回 (video_filter, audio_filter, config)。
-        """
+        """便捷方法：从单一 speed 值生成视频+音频滤镜."""
         config = SpeedConfig(speed=speed, pitch_correct=pitch_correct)
         config.clamp()
         return (
@@ -160,8 +70,5 @@ class SpeedEngine:
         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)
+        """从 clip config 中解析 playback_speed，0 或缺失则使用全局速度."""
+        return _resolve_clip_speed_base(clip_config, global_speed)