feat: 视频调速引擎（快进/慢放） (#294)

alembic/versions/040_add_playback_speed_to_edit_plan_clips.py

@@ -0,0 +1,29 @@
+"""add playback_speed to edit_plan_clips
+
+Revision ID: 040_playback_speed
+Revises: 039_transition_duration
+Create Date: 2026-07-14 10:00:00.000000
+"""
+
+from __future__ import annotations
+
+import sqlalchemy as sa
+
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "040_playback_speed"
+down_revision = "039_transition_duration"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "edit_plan_clips",
+        sa.Column("playback_speed", sa.Float(), nullable=False, server_default="1.0"),
+    )
+
+
+def downgrade() -> None:
+    op.drop_column("edit_plan_clips", "playback_speed")

apps/api/app/api/routes/edit_plans.py

@@ -211,6 +211,7 @@ class _PlanClipItem(BaseModel):
     duration: float
     transition_effect: str
     transition_duration: float
+    playback_speed: float = 1.0
     status: str
     config: Optional[dict[str, Any]] = None
     created_at: datetime

apps/api/app/services/edit_plan_service.py

@@ -282,6 +282,7 @@ class EditPlanService:
         duration: float = 0.0,
         transition_effect: str = "cut",
         transition_duration: float = 0.0,
+        playback_speed: float = 1.0,
         config: Optional[dict[str, Any]] = None,
     ) -> EditPlanClip:
         """创建片段
@@ -303,6 +304,7 @@ class EditPlanService:
             duration=duration,
             transition_effect=transition_effect,
             transition_duration=transition_duration,
+            playback_speed=playback_speed,
             config=config,
         )
         created = self._clip_repo.create(clip)
@@ -327,6 +329,7 @@ class EditPlanService:
         duration: Optional[float] = None,
         transition_effect: Optional[str] = None,
         transition_duration: Optional[float] = None,
+        playback_speed: Optional[float] = None,
         config: Optional[dict[str, Any]] = None,
     ) -> EditPlanClip:
         """更新片段
@@ -336,6 +339,15 @@ class EditPlanService:
         """
         existing = self.get_clip_or_raise(clip_id)
 
+        # 速度边界钳制
+        if playback_speed is not None:
+            if playback_speed <= 0:
+                playback_speed = 1.0
+            elif playback_speed < 0.25:
+                playback_speed = 0.25
+            elif playback_speed > 4.0:
+                playback_speed = 4.0
+
         updated = EditPlanClip(
             id=existing.id,
             plan_id=existing.plan_id,
@@ -352,6 +364,7 @@ class EditPlanService:
             transition_duration=(
                 transition_duration if transition_duration is not None else existing.transition_duration
             ),
+            playback_speed=playback_speed if playback_speed is not None else existing.playback_speed,
             status=existing.status,
             config=config if config is not None else existing.config,
             created_at=existing.created_at,

apps/worker/video_processing/render_audio.py

@@ -23,6 +23,7 @@ from typing import TYPE_CHECKING
 
 from video_processing.ffmpeg_utils import FFMPEG_BIN, probe_has_audio, run_ffmpeg
 from video_processing.reverse_engine import ReverseConfig, ReverseEngine
+from video_processing.speed_engine import SpeedEngine
 
 if TYPE_CHECKING:
     from video_processing.unified_render_service import RenderLayer, ResolvedClip
@@ -225,53 +226,118 @@ def concat_main_audio(
         clip = clips[0]
         effective_duration = clip_effective_duration(clip)
         trim_start = getattr(clip, "start_time", 0) or 0
-        # 最终时长：取 clip 有效时长和视频总时长的较小值
-        # （视频总时长由主图层决定，但单 clip 场景下两者应该一致，仍做保护）
-        final_duration = effective_duration
+        speed = getattr(clip, "playback_speed", 1.0) or 1.0
+        if not isinstance(speed, (int, float)) or speed <= 0:
+            speed = 1.0
+
+        # 调速后时长
+        adjusted_duration = effective_duration / speed if abs(speed - 1.0) >= 1e-6 else effective_duration
+        # 最终时长：取调速后时长和视频总时长的较小值
+        final_duration = adjusted_duration
         if video_duration > 0 and (final_duration <= 0 or final_duration > video_duration):
             final_duration = video_duration
 
         # 音频倒放
         reverse_config = ReverseConfig.from_dict(clip.config.get("reverse"))
-        af_filters = []
-        if reverse_config.enabled and reverse_config.reverse_audio:
-            reverse_filter = ReverseEngine.build_audio_filter(reverse_config, duration=effective_duration)
-            if reverse_filter:
-                af_filters.append(reverse_filter)
+        has_reverse = reverse_config.enabled and reverse_config.reverse_audio
+        has_speed = abs(speed - 1.0) >= 1e-6
 
-        command = [
-            FFMPEG_BIN,
-            "-y",
-            "-i",
-            str(clip.local_path),
-            "-vn",
-            "-acodec",
-            "aac",
-            "-b:a",
-            "128k",
-        ]
-        if trim_start > 0:
-            command.extend(["-ss", f"{trim_start:.3f}"])
-        if af_filters:
-            command.extend(["-af", ",".join(af_filters)])
-        if final_duration > 0:
-            command.extend(["-t", f"{final_duration:.3f}"])
-        command.append(str(output_path))
-        run_ffmpeg(command)
+        if not has_speed and not has_reverse:
+            # 无调速无倒放：简单命令行，-ss 裁剪更高效
+            command = [
+                FFMPEG_BIN,
+                "-y",
+                "-i",
+                str(clip.local_path),
+                "-vn",
+                "-acodec",
+                "aac",
+                "-b:a",
+                "128k",
+            ]
+            if trim_start > 0:
+                command.extend(["-ss", f"{trim_start:.3f}"])
+            if final_duration > 0:
+                command.extend(["-t", f"{final_duration:.3f}"])
+            command.append(str(output_path))
+            run_ffmpeg(command)
+        else:
+            # 有调速或倒放：用 filter_complex
+            speed_engine = SpeedEngine()
+            audio_filters = []
+            if effective_duration > 0:
+                audio_filters.append(f"atrim=start={trim_start:.3f}:duration={effective_duration:.3f}")
+                audio_filters.append("asetpts=PTS-STARTPTS")
+
+            # 音频调速
+            if has_speed:
+                from video_processing.speed_engine import SpeedConfig
+
+                config = SpeedConfig(speed=float(speed))
+                config.clamp()
+                atempo_filter = speed_engine.build_audio_filter(config)
+                if atempo_filter:
+                    audio_filters.append(atempo_filter)
+
+            # 音频倒放
+            if has_reverse:
+                reverse_filter = ReverseEngine.build_audio_filter(reverse_config, duration=effective_duration)
+                if reverse_filter:
+                    audio_filters.append(reverse_filter)
+
+            filter_parts: list[str] = [f"[0:a]{','.join(audio_filters)}[outa]"]
+            if video_duration > 0 and final_duration < adjusted_duration:
+                filter_parts.append(f"[outa]atrim=0:{final_duration:.3f}[final_audio]")
+                final_label = "final_audio"
+            else:
+                final_label = "outa"
+
+            filter_complex = ";".join(filter_parts)
+            command = [
+                FFMPEG_BIN,
+                "-y",
+                "-i",
+                str(clip.local_path),
+                "-filter_complex",
+                filter_complex,
+                "-map",
+                f"[{final_label}]",
+                "-acodec",
+                "aac",
+                "-b:a",
+                "128k",
+                str(output_path),
+            ]
+            run_ffmpeg(command)
         return
 
     # 多 clip，用 filter_complex concat
     input_args: list[str] = []
     filter_parts: list[str] = []
+    speed_engine = SpeedEngine()
 
     for i, clip in enumerate(clips):
         input_args.extend(["-i", str(clip.local_path)])
         effective_duration = clip_effective_duration(clip)
         trim_start = getattr(clip, "start_time", 0) or 0
+        speed = getattr(clip, "playback_speed", 1.0) or 1.0
+        if not isinstance(speed, (int, float)) or speed <= 0:
+            speed = 1.0
+
         audio_filters: list[str] = []
         if effective_duration > 0:
             audio_filters.append(f"atrim=start={trim_start:.3f}:duration={effective_duration:.3f}")
             audio_filters.append("asetpts=PTS-STARTPTS")
+
+        # 音频调速 — atempo 多级串联
+        if abs(speed - 1.0) >= 1e-6:
+            from video_processing.speed_engine import SpeedConfig
+
+            config = SpeedConfig(speed=float(speed))
+            config.clamp()
+            atempo_filter = speed_engine.build_audio_filter(config)
+            if atempo_filter:
+                audio_filters.append(atempo_filter)
         else:
             audio_filters.append("asetpts=PTS-STARTPTS")
 

apps/worker/video_processing/speed_engine.py

@@ -0,0 +1,167 @@
+"""视频调速引擎 — 基于 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)

apps/worker/video_processing/unified_render_service.py

@@ -45,6 +45,7 @@ from video_processing.pip_engine import PiPConfig, PiPEngine, PiPLayerConfig
 from video_processing.render_audio import RenderContext, merge_audio_video, mix_audio
 from video_processing.render_subtitles import generate_ass_subtitles
 from video_processing.reverse_engine import ReverseConfig, ReverseEngine
+from video_processing.speed_engine import SpeedConfig, SpeedEngine
 from video_processing.sticker_engine import StickerEngine, parse_stickers_from_config
 from video_processing.subtitle_generator import generate_ass_from_timeline
 from video_processing.transition_engine import TransitionEngine
@@ -73,6 +74,7 @@ class ResolvedClip:
     duration: float = 0.0  # 0 表示使用素材完整时长
     transition_effect: str = "cut"
     transition_duration: float = 0.0  # 0 表示使用全局默认值
+    playback_speed: float = 1.0  # 0 或 1.0 表示原速
     config: dict[str, Any] = field(default_factory=dict)
 
     # 运行时填充
@@ -186,6 +188,7 @@ class UnifiedRenderService:
         self.asr_service = asr_service
         self.bgm_path = bgm_path
         self._transition_engine = TransitionEngine(default_duration=transition_duration)
+        self._speed_engine = SpeedEngine()
 
     def render(self) -> RenderResult:
         """执行渲染，返回 RenderResult.
@@ -493,7 +496,7 @@ class UnifiedRenderService:
         if not main_layer or not main_layer.clips:
             return 0.0
 
-        total = sum(UnifiedRenderService._clip_effective_duration(c) for c in main_layer.clips)
+        total = sum(UnifiedRenderService._clip_adjusted_duration(c) for c in main_layer.clips)
 
         # 减去转场重叠时间（粗略估算）
         n_clips = len(main_layer.clips)
@@ -1197,6 +1200,7 @@ class UnifiedRenderService:
                 duration=final_duration,
                 transition_effect=clip.transition_effect or "cut",
                 transition_duration=getattr(clip, "transition_duration", 0.0) or 0.0,
+                playback_speed=getattr(clip, "playback_speed", 1.0) or 1.0,
                 config=clip_config,
                 actual_duration=actual_duration,
                 trim_config=effective_trim,
@@ -1294,6 +1298,11 @@ class UnifiedRenderService:
                     filters.append(f"trim=duration={effective_duration:.3f}")
                 filters.append("setpts=PTS-STARTPTS")
 
+            # 调速 — 基于 setpts 改变播放速度
+            speed = UnifiedRenderService._clip_speed(clip)
+            if abs(speed - 1.0) >= 1e-6:
+                filters.append(f"setpts=PTS/{speed:.4f}")
+
             # 倒放滤镜（在 trim 之后、scale 之前应用）
             reverse_config = ReverseConfig.from_dict(clip.config.get("reverse"))
             if reverse_config.enabled and reverse_config.reverse_video:
@@ -1351,8 +1360,8 @@ class UnifiedRenderService:
         for layer in layers:
             layer_clip_indices = [all_clips.index(c) for c in layer.clips]
             layer_labels = [preprocessed_labels[i] for i in layer_clip_indices]
-            # 使用 trim 后的有效时长，与 Step 1 的 trim=duration 保持一致
-            layer_durations = [UnifiedRenderService._clip_effective_duration(all_clips[i]) for i in layer_clip_indices]
+            # 使用调速后的实际时长，与 Step 1 的调速处理保持一致
+            layer_durations = [UnifiedRenderService._clip_adjusted_duration(all_clips[i]) for i in layer_clip_indices]
             layer_transitions = [all_clips[i].transition_effect for i in layer_clip_indices]
             layer_transition_durations = [all_clips[i].transition_duration for i in layer_clip_indices]
 
@@ -1597,7 +1606,7 @@ class UnifiedRenderService:
 
     @staticmethod
     def _clip_effective_duration(clip: ResolvedClip) -> float:
-        """计算 clip 的有效时长."""
+        """计算 clip 的有效时长（原速 trim 后时长）."""
         if clip.duration > 0:
             return min(clip.duration, clip.actual_duration) if clip.actual_duration > 0 else clip.duration
         return clip.actual_duration if clip.actual_duration > 0 else 0.0
@@ -1694,3 +1703,20 @@ class UnifiedRenderService:
         )
 
         return new_filter, new_input_args
+
+    @staticmethod
+    def _clip_speed(clip: ResolvedClip) -> float:
+        """获取 clip 的播放速度，无效值回退到 1.0."""
+        speed = getattr(clip, "playback_speed", 1.0)
+        if not isinstance(speed, (int, float)) or speed <= 0:
+            return 1.0
+        return float(speed)
+
+    @staticmethod
+    def _clip_adjusted_duration(clip: ResolvedClip) -> float:
+        """计算调速后的 clip 实际时长（用于拼接计算）."""
+        base = UnifiedRenderService._clip_effective_duration(clip)
+        speed = UnifiedRenderService._clip_speed(clip)
+        if abs(speed - 1.0) < 1e-6:
+            return base
+        return base / speed

docs/schema-metadata-snapshot.json

@@ -866,6 +866,14 @@
           "type": "FLOAT",
           "unique": false
         },
+        {
+          "index": false,
+          "name": "playback_speed",
+          "nullable": false,
+          "primary_key": false,
+          "type": "FLOAT",
+          "unique": false
+        },
         {
           "index": true,
           "name": "status",

packages/adapters/sqlalchemy_impl/edit_plan_clip_repository.py

@@ -55,6 +55,7 @@ class SQLAlchemyEditPlanClipRepository:
             duration=clip.duration,
             transition_effect=clip.transition_effect,
             transition_duration=clip.transition_duration,
+            playback_speed=clip.playback_speed,
             status=clip.status,
             config=clip.config,
         )
@@ -78,6 +79,7 @@ class SQLAlchemyEditPlanClipRepository:
         model.duration = clip.duration
         model.transition_effect = clip.transition_effect
         model.transition_duration = clip.transition_duration
+        model.playback_speed = clip.playback_speed
         model.status = clip.status
         model.config = clip.config
         model.updated_at = clip.updated_at
@@ -123,6 +125,7 @@ class SQLAlchemyEditPlanClipRepository:
             duration=model.duration or 0.0,
             transition_effect=model.transition_effect or "cut",
             transition_duration=getattr(model, "transition_duration", 0.0) or 0.0,
+            playback_speed=model.playback_speed or 1.0,
             status=EditPlanClipStatus(model.status) if model.status else EditPlanClipStatus.PENDING,
             config=model.config or {},
             created_at=model.created_at,

packages/adapters/sqlalchemy_impl/models.py

@@ -197,6 +197,7 @@ class EditPlanClipModel(Base):
     duration = Column(Float, nullable=False, default=0.0)
     transition_effect = Column(String(20), nullable=False, default="cut")
     transition_duration = Column(Float, nullable=False, default=0.0)
+    playback_speed = Column(Float, nullable=False, default=1.0)
     status = Column(String(20), nullable=False, default="pending", index=True)
     config = Column(JSON, nullable=False, default=dict)
     created_at = Column(DateTime, nullable=False, default=lambda: datetime.now(timezone.utc))

packages/domain/edit_plan_clip.py

@@ -51,6 +51,7 @@ class EditPlanClip:
     duration: float = 0.0
     transition_effect: str = "cut"
     transition_duration: float = 0.0  # 0 表示使用全局默认值
+    playback_speed: float = 1.0  # 0 或 1.0 表示原速，范围 0.25~4.0
     status: EditPlanClipStatus = EditPlanClipStatus.PENDING
     config: dict[str, Any] = field(default_factory=dict)
     created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
@@ -70,6 +71,7 @@ class EditPlanClip:
         duration: float = 0.0,
         transition_effect: str = "cut",
         transition_duration: float = 0.0,
+        playback_speed: float = 1.0,
         config: dict[str, Any] | None = None,
     ) -> EditPlanClip:
         """创建剪辑计划片段"""
@@ -81,6 +83,13 @@ class EditPlanClip:
             raise ValueError("start_time 不能为负数")
         if duration < 0:
             raise ValueError("duration 不能为负数")
+        # 速度边界钳制
+        if playback_speed <= 0:
+            playback_speed = 1.0
+        elif playback_speed < 0.25:
+            playback_speed = 0.25
+        elif playback_speed > 4.0:
+            playback_speed = 4.0
 
         return cls(
             id=uuid4().hex,
@@ -94,6 +103,7 @@ class EditPlanClip:
             duration=duration,
             transition_effect=transition_effect.strip() or "cut",
             transition_duration=max(0.0, transition_duration),
+            playback_speed=playback_speed,
             status=EditPlanClipStatus.PENDING,
             config=config or {},
         )

tests/unit/test_speed_engine.py

@@ -0,0 +1,269 @@
+"""视频调速引擎单元测试."""
+
+import pytest
+from video_processing.speed_engine import (
+    MAX_SPEED,
+    MIN_SPEED,
+    SpeedConfig,
+    SpeedEngine,
+)
+
+# ─── SpeedConfig 解析与校验 ──────────────────────────────────
+
+
+class TestSpeedConfig:
+    def test_default_values(self):
+        config = SpeedConfig()
+        assert config.speed == 1.0
+        assert config.pitch_correct is True
+
+    def test_parse_none(self):
+        config = SpeedConfig.parse(None)
+        assert config.speed == 1.0
+        assert config.pitch_correct is True
+
+    def test_parse_empty_dict(self):
+        config = SpeedConfig.parse({})
+        assert config.speed == 1.0
+
+    def test_parse_valid_speed(self):
+        config = SpeedConfig.parse({"speed": 2.0})
+        assert config.speed == 2.0
+
+    def test_parse_pitch_correct_false(self):
+        config = SpeedConfig.parse({"pitch_correct": False})
+        assert config.pitch_correct is False
+
+    def test_parse_invalid_speed_type(self):
+        config = SpeedConfig.parse({"speed": "fast"})
+        assert config.speed == 1.0
+
+    def test_parse_invalid_pitch_type(self):
+        config = SpeedConfig.parse({"pitch_correct": "yes"})
+        assert config.pitch_correct is True
+
+    def test_clamp_below_min(self):
+        config = SpeedConfig(speed=0.1)
+        config.clamp()
+        assert config.speed == MIN_SPEED
+
+    def test_clamp_zero(self):
+        config = SpeedConfig(speed=0)
+        config.clamp()
+        assert config.speed == 1.0
+
+    def test_clamp_negative(self):
+        config = SpeedConfig(speed=-1.0)
+        config.clamp()
+        assert config.speed == 1.0
+
+    def test_clamp_above_max(self):
+        config = SpeedConfig(speed=10.0)
+        config.clamp()
+        assert config.speed == MAX_SPEED
+
+    def test_clamp_within_range(self):
+        config = SpeedConfig(speed=1.5)
+        config.clamp()
+        assert config.speed == 1.5
+
+    def test_is_original_true(self):
+        config = SpeedConfig(speed=1.0)
+        assert config.is_original is True
+
+    def test_is_original_false(self):
+        config = SpeedConfig(speed=1.5)
+        assert config.is_original is False
+
+    def test_parse_clamps_automatically(self):
+        """parse 方法应该自动调用 clamp."""
+        config = SpeedConfig.parse({"speed": 100.0})
+        assert config.speed == MAX_SPEED
+
+
+# ─── SpeedEngine 视频滤镜 ────────────────────────────────────
+
+
+class TestSpeedEngineVideoFilter:
+    def setup_method(self):
+        self.engine = SpeedEngine()
+
+    def test_original_speed_returns_empty(self):
+        config = SpeedConfig(speed=1.0)
+        assert self.engine.build_video_filter(config) == ""
+
+    def test_double_speed(self):
+        config = SpeedConfig(speed=2.0)
+        result = self.engine.build_video_filter(config)
+        assert "setpts=PTS/2.0" in result
+
+    def test_half_speed(self):
+        config = SpeedConfig(speed=0.5)
+        result = self.engine.build_video_filter(config)
+        assert "setpts=PTS/0.5" in result
+
+    def test_quarter_speed(self):
+        config = SpeedConfig(speed=0.25)
+        result = self.engine.build_video_filter(config)
+        assert "setpts=PTS/0.25" in result
+
+    def test_quad_speed(self):
+        config = SpeedConfig(speed=4.0)
+        result = self.engine.build_video_filter(config)
+        assert "setpts=PTS/4.0" in result
+
+
+# ─── SpeedEngine 音频滤镜（atempo 多级串联） ─────────────────
+
+
+class TestSpeedEngineAudioFilter:
+    def setup_method(self):
+        self.engine = SpeedEngine()
+
+    def test_original_speed_returns_empty(self):
+        config = SpeedConfig(speed=1.0)
+        assert self.engine.build_audio_filter(config) == ""
+
+    def test_double_speed_single_stage(self):
+        """2x 在 atempo 单级范围内，只需一个 atempo."""
+        config = SpeedConfig(speed=2.0)
+        result = self.engine.build_audio_filter(config)
+        assert result == "atempo=2.0000"
+
+    def test_half_speed_single_stage(self):
+        config = SpeedConfig(speed=0.5)
+        result = self.engine.build_audio_filter(config)
+        assert result == "atempo=0.5000"
+
+    def test_quad_speed_two_stages(self):
+        """4x 需要两级 atempo: 2.0 * 2.0."""
+        config = SpeedConfig(speed=4.0)
+        result = self.engine.build_audio_filter(config)
+        assert result == "atempo=2.0000,atempo=2.0000"
+
+    def test_quarter_speed_two_stages(self):
+        """0.25x 需要两级 atempo: 0.5 * 0.5."""
+        config = SpeedConfig(speed=0.25)
+        result = self.engine.build_audio_filter(config)
+        assert result == "atempo=0.5000,atempo=0.5000"
+
+    def test_triple_speed_two_stages(self):
+        """3x: 2.0 * 1.5."""
+        config = SpeedConfig(speed=3.0)
+        result = self.engine.build_audio_filter(config)
+        parts = result.split(",")
+        assert len(parts) == 2
+        assert "atempo=2.0000" in parts
+        assert "atempo=1.5000" in parts
+
+    def test_03_speed_two_stages(self):
+        """0.3x: 0.5 * 0.6."""
+        config = SpeedConfig(speed=0.3)
+        result = self.engine.build_audio_filter(config)
+        parts = result.split(",")
+        assert len(parts) == 2
+        assert "atempo=0.5000" in parts
+        assert "atempo=0.6000" in parts
+
+    def test_split_atempo_inside_range(self):
+        """0.5~2.0 范围内只返回一级."""
+        stages = SpeedEngine._split_atempo_stages(1.5)
+        assert len(stages) == 1
+        assert stages[0] == 1.5
+
+    def test_split_atempo_boundary_min(self):
+        stages = SpeedEngine._split_atempo_stages(0.5)
+        assert len(stages) == 1
+        assert stages[0] == 0.5
+
+    def test_split_atempo_boundary_max(self):
+        stages = SpeedEngine._split_atempo_stages(2.0)
+        assert len(stages) == 1
+        assert stages[0] == 2.0
+
+    def test_split_atempo_product_equals_speed(self):
+        """所有级联的乘积应该等于原速度."""
+        test_cases = [0.25, 0.3, 0.5, 0.75, 1.0, 1.5, 2.0, 3.0, 4.0]
+        for speed in test_cases:
+            stages = SpeedEngine._split_atempo_stages(speed)
+            product = 1.0
+            for s in stages:
+                product *= s
+            assert abs(product - speed) < 1e-6, f"speed={speed}, stages={stages}, product={product}"
+
+    def test_split_atempo_all_in_range(self):
+        """所有级都应该在 0.5~2.0 范围内."""
+        test_cases = [0.25, 0.3, 0.5, 0.75, 1.0, 1.5, 2.0, 3.0, 4.0]
+        for speed in test_cases:
+            stages = SpeedEngine._split_atempo_stages(speed)
+            for s in stages:
+                assert 0.5 <= s <= 2.0, f"speed={speed}, stage={s} out of range"
+
+
+# ─── SpeedEngine 时长计算 ────────────────────────────────────
+
+
+class TestSpeedEngineDuration:
+    def setup_method(self):
+        self.engine = SpeedEngine()
+
+    def test_original_speed_same_duration(self):
+        config = SpeedConfig(speed=1.0)
+        assert self.engine.adjust_duration(10.0, config) == 10.0
+
+    def test_double_speed_half_duration(self):
+        config = SpeedConfig(speed=2.0)
+        assert self.engine.adjust_duration(10.0, config) == 5.0
+
+    def test_half_speed_double_duration(self):
+        config = SpeedConfig(speed=0.5)
+        assert self.engine.adjust_duration(10.0, config) == 20.0
+
+    def test_quad_speed_quarter_duration(self):
+        config = SpeedConfig(speed=4.0)
+        assert self.engine.adjust_duration(10.0, config) == 2.5
+
+    def test_zero_duration(self):
+        config = SpeedConfig(speed=2.0)
+        assert self.engine.adjust_duration(0.0, config) == 0.0
+
+    def test_negative_duration(self):
+        config = SpeedConfig(speed=2.0)
+        assert self.engine.adjust_duration(-1.0, config) == -1.0
+
+
+# ─── SpeedEngine 便捷方法 ────────────────────────────────────
+
+
+class TestSpeedEngineHelper:
+    def setup_method(self):
+        self.engine = SpeedEngine()
+
+    def test_build_clip_speed_filter_original(self):
+        v_f, a_f, cfg = self.engine.build_clip_speed_filter(1.0)
+        assert v_f == ""
+        assert a_f == ""
+        assert cfg.speed == 1.0
+
+    def test_build_clip_speed_filter_2x(self):
+        v_f, a_f, cfg = self.engine.build_clip_speed_filter(2.0)
+        assert "setpts=PTS/2.0" in v_f
+        assert "atempo=2.0" in a_f
+        assert cfg.speed == 2.0
+
+    def test_build_clip_speed_clamped(self):
+        _, _, cfg = self.engine.build_clip_speed_filter(100.0)
+        assert cfg.speed == MAX_SPEED
+
+    def test_resolve_clip_speed_default(self):
+        assert SpeedEngine.resolve_clip_speed({}) == 1.0
+        assert SpeedEngine.resolve_clip_speed(None) == 1.0
+
+    def test_resolve_clip_speed_zero_uses_global(self):
+        assert SpeedEngine.resolve_clip_speed({"playback_speed": 0}, 1.5) == 1.5
+
+    def test_resolve_clip_speed_custom(self):
+        assert SpeedEngine.resolve_clip_speed({"playback_speed": 2.0}) == 2.0
+
+    def test_resolve_clip_speed_invalid_type(self):
+        assert SpeedEngine.resolve_clip_speed({"playback_speed": "fast"}) == 1.0