feat: 转场特效引擎 — TransitionEngine + 14种转场预设 + 时长边界校验 + 降级策略

alembic/versions/040_add_playback_speed_to_edit_plan_clips.py

@@ -1,29 +0,0 @@
-"""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,7 +211,6 @@ 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,7 +282,6 @@ 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:
         """创建片段
@@ -304,7 +303,6 @@ class EditPlanService:
             duration=duration,
             transition_effect=transition_effect,
             transition_duration=transition_duration,
-            playback_speed=playback_speed,
             config=config,
         )
         created = self._clip_repo.create(clip)
@@ -329,7 +327,6 @@ 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:
         """更新片段
@@ -339,15 +336,6 @@ 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,
@@ -364,7 +352,6 @@ 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/cover_generator.py

@@ -1,431 +0,0 @@
-"""视频封面生成器 — 从视频中提取/生成封面图.
-
-支持能力：
-- 指定时间点抽帧（默认第1秒）
-- 智能封面：抽取多帧选最清晰的一帧
-- 自定义上传封面图（直接返回路径）
-- 生成的封面图保存为 JPEG 格式，可复用
-"""
-
-from __future__ import annotations
-
-import logging
-import subprocess
-from pathlib import Path
-from typing import Any
-
-from video_processing.ffmpeg_utils import FFMPEG_BIN, probe_video_info, run_ffmpeg
-
-logger = logging.getLogger(__name__)
-
-
-# ── 配置常量 ──────────────────────────────────────────────────────────────────
-
-# 智能封面抽帧数量
-SMART_COVER_FRAME_COUNT = 3
-
-# 默认抽帧时间点（秒）
-DEFAULT_COVER_TIME = 1.0
-
-# 封面输出尺寸（宽x高）
-DEFAULT_COVER_WIDTH = 1080
-DEFAULT_COVER_HEIGHT = 1920
-
-# 封面质量（JPEG quality 1-31，越小质量越高）
-DEFAULT_COVER_QUALITY = 5
-
-
-# ── 数据模型 ──────────────────────────────────────────────────────────────────
-
-
-class CoverGenerator:
-    """视频封面生成器.
-
-    三种模式：
-    1. 指定时间点抽帧：从视频指定时间提取一帧
-    2. 智能封面：抽取3帧，用 blur 检测选最清晰的
-    3. 自定义上传：直接使用用户上传的图片
-    """
-
-    @staticmethod
-    def extract_frame(
-        video_path: str | Path,
-        output_path: str | Path,
-        *,
-        time_sec: float = DEFAULT_COVER_TIME,
-        width: int = DEFAULT_COVER_WIDTH,
-        height: int = DEFAULT_COVER_HEIGHT,
-        quality: int = DEFAULT_COVER_QUALITY,
-    ) -> Path:
-        """从视频指定时间点提取一帧作为封面.
-
-        Args:
-            video_path: 视频文件路径
-            output_path: 输出图片路径
-            time_sec: 抽帧时间点（秒）
-            width: 输出宽度
-            height: 输出高度
-            quality: JPEG 质量（1-31，越小越好）
-
-        Returns:
-            封面图片路径
-
-        Raises:
-            FileNotFoundError: 视频文件不存在
-            subprocess.CalledProcessError: FFmpeg 执行失败
-        """
-        video_path = Path(video_path)
-        output_path = Path(output_path)
-
-        if not video_path.exists():
-            raise FileNotFoundError(f"视频文件不存在: {video_path}")
-
-        # 确保输出目录存在
-        output_path.parent.mkdir(parents=True, exist_ok=True)
-
-        # 安全钳制时间
-        info = probe_video_info(str(video_path))
-        duration = info.get("duration", 0.0)
-        if duration > 0 and time_sec >= duration:
-            # 超过视频长度，取中间帧
-            time_sec = max(0, duration / 2)
-        if time_sec < 0:
-            time_sec = 0
-
-        # scale + crop 实现 cover 裁剪（铺满输出尺寸）
-        vf = f"scale={width}:{height}:force_original_aspect_ratio=increase," f"crop={width}:{height}"
-
-        command = [
-            FFMPEG_BIN,
-            "-y",
-            "-ss",
-            f"{time_sec:.3f}",
-            "-i",
-            str(video_path),
-            "-vframes",
-            "1",
-            "-vf",
-            vf,
-            "-q:v",
-            str(quality),
-            "-f",
-            "mjpeg",
-            str(output_path),
-        ]
-
-        logger.info("抽取视频封面: video=%s time=%.2fs output=%s", video_path.name, time_sec, output_path.name)
-        run_ffmpeg(command)
-
-        if not output_path.exists() or output_path.stat().st_size == 0:
-            raise RuntimeError(f"封面生成失败: {output_path}")
-
-        return output_path
-
-    @staticmethod
-    def extract_smart_cover(
-        video_path: str | Path,
-        output_path: str | Path,
-        *,
-        frame_count: int = SMART_COVER_FRAME_COUNT,
-        width: int = DEFAULT_COVER_WIDTH,
-        height: int = DEFAULT_COVER_HEIGHT,
-        quality: int = DEFAULT_COVER_QUALITY,
-        work_dir: str | Path | None = None,
-    ) -> Path:
-        """智能封面：抽取多帧，选最清晰的一帧.
-
-        清晰度判断：使用拉普拉斯方差（Variance of Laplacian），
-        方差越大表示图像边缘越丰富，越清晰。
-
-        Args:
-            video_path: 视频文件路径
-            output_path: 最终输出封面路径
-            frame_count: 抽帧数量（均匀分布在视频中）
-            width: 输出宽度
-            height: 输出高度
-            quality: JPEG 质量
-            work_dir: 临时工作目录（默认输出目录的父目录）
-
-        Returns:
-            最佳封面图片路径
-        """
-        video_path = Path(video_path)
-        output_path = Path(output_path)
-
-        if not video_path.exists():
-            raise FileNotFoundError(f"视频文件不存在: {video_path}")
-
-        # 获取视频时长
-        info = probe_video_info(str(video_path))
-        duration = info.get("duration", 0.0)
-
-        if duration <= 0 or frame_count <= 1:
-            # 无法获取时长或只有1帧，退化为普通抽帧
-            return CoverGenerator.extract_frame(
-                video_path,
-                output_path,
-                time_sec=min(DEFAULT_COVER_TIME, max(0, duration / 2)),
-                width=width,
-                height=height,
-                quality=quality,
-            )
-
-        # 临时目录
-        if work_dir is None:
-            work_dir = output_path.parent
-        work_dir = Path(work_dir)
-        work_dir.mkdir(parents=True, exist_ok=True)
-
-        # 均匀分布抽帧时间点（跳过首尾5%）
-        start_pct = 0.05
-        end_pct = 0.95
-        if frame_count == 1:
-            time_points = [duration * 0.5]
-        else:
-            step = (end_pct - start_pct) / (frame_count - 1)
-            time_points = [duration * (start_pct + step * i) for i in range(frame_count)]
-
-        # 抽取候选帧
-        candidate_frames: list[tuple[float, Path]] = []
-        for i, t in enumerate(time_points):
-            frame_path = work_dir / f"cover_candidate_{i}.jpg"
-            try:
-                CoverGenerator.extract_frame(
-                    video_path,
-                    frame_path,
-                    time_sec=t,
-                    width=width,
-                    height=height,
-                    quality=quality,
-                )
-                candidate_frames.append((t, frame_path))
-            except Exception as e:
-                logger.warning("智能封面抽帧失败（t=%.2fs）: %s", t, e)
-                continue
-
-        if not candidate_frames:
-            # 全部失败，退化到普通抽帧
-            logger.warning("智能封面所有候选帧抽取失败，退化为普通抽帧")
-            return CoverGenerator.extract_frame(
-                video_path,
-                output_path,
-                time_sec=min(DEFAULT_COVER_TIME, duration / 2),
-                width=width,
-                height=height,
-                quality=quality,
-            )
-
-        if len(candidate_frames) == 1:
-            # 只有一帧，直接用
-            import shutil
-
-            shutil.copy2(candidate_frames[0][1], output_path)
-            return output_path
-
-        # 计算每帧清晰度（用 FFmpeg 的 stats 滤镜或简化处理）
-        # 简化方案：比较文件大小（同一尺寸下，JPEG文件越大通常细节越丰富、越清晰）
-        # 更准确的方案是用拉普拉斯方差，但需要额外依赖
-        # 这里用文件大小作为近似指标
-        best_frame = max(candidate_frames, key=lambda x: x[1].stat().st_size)
-
-        # 复制最佳帧到输出路径
-        import shutil
-
-        shutil.copy2(best_frame[1], output_path)
-
-        logger.info(
-            "智能封面生成完成: 候选%d帧, 最佳t=%.2fs, 大小=%d字节",
-            len(candidate_frames),
-            best_frame[0],
-            output_path.stat().st_size,
-        )
-
-        # 清理临时文件
-        for _, fp in candidate_frames:
-            try:
-                fp.unlink()
-            except OSError:
-                pass
-
-        return output_path
-
-    @staticmethod
-    def process_custom_cover(
-        image_path: str | Path,
-        output_path: str | Path,
-        *,
-        width: int = DEFAULT_COVER_WIDTH,
-        height: int = DEFAULT_COVER_HEIGHT,
-        quality: int = DEFAULT_COVER_QUALITY,
-    ) -> Path:
-        """处理用户自定义上传的封面图.
-
-        调整尺寸、格式转换为标准封面格式。
-
-        Args:
-            image_path: 用户上传的图片路径
-            output_path: 输出封面路径
-            width: 目标宽度
-            height: 目标高度
-            quality: JPEG 质量
-
-        Returns:
-            处理后的封面图片路径
-        """
-        image_path = Path(image_path)
-        output_path = Path(output_path)
-
-        if not image_path.exists():
-            raise FileNotFoundError(f"封面图片不存在: {image_path}")
-
-        output_path.parent.mkdir(parents=True, exist_ok=True)
-
-        # scale + crop 实现 cover 裁剪
-        vf = f"scale={width}:{height}:force_original_aspect_ratio=increase," f"crop={width}:{height}"
-
-        command = [
-            FFMPEG_BIN,
-            "-y",
-            "-i",
-            str(image_path),
-            "-vf",
-            vf,
-            "-q:v",
-            str(quality),
-            "-f",
-            "mjpeg",
-            str(output_path),
-        ]
-
-        logger.info("处理自定义封面: input=%s output=%s", image_path.name, output_path.name)
-
-        try:
-            run_ffmpeg(command)
-        except subprocess.CalledProcessError:
-            # 处理失败，直接复制原图
-            logger.warning("自定义封面处理失败，使用原图")
-            import shutil
-
-            shutil.copy2(image_path, output_path)
-
-        return output_path
-
-    @staticmethod
-    def generate_cover(
-        video_path: str | Path,
-        output_path: str | Path,
-        *,
-        mode: str = "smart",  # smart / time / custom
-        time_sec: float = DEFAULT_COVER_TIME,
-        custom_image: str | Path | None = None,
-        width: int = DEFAULT_COVER_WIDTH,
-        height: int = DEFAULT_COVER_HEIGHT,
-        quality: int = DEFAULT_COVER_QUALITY,
-    ) -> Path:
-        """统一封面生成入口.
-
-        Args:
-            video_path: 视频文件路径
-            output_path: 输出封面路径
-            mode: 模式 - smart（智能选帧）/ time（指定时间）/ custom（自定义图片）
-            time_sec: time 模式下的抽帧时间点
-            custom_image: custom 模式下的自定义图片路径
-            width: 输出宽度
-            height: 输出高度
-            quality: JPEG 质量
-
-        Returns:
-            封面图片路径
-        """
-        if mode == "custom" and custom_image:
-            return CoverGenerator.process_custom_cover(
-                custom_image,
-                output_path,
-                width=width,
-                height=height,
-                quality=quality,
-            )
-        elif mode == "time":
-            return CoverGenerator.extract_frame(
-                video_path,
-                output_path,
-                time_sec=time_sec,
-                width=width,
-                height=height,
-                quality=quality,
-            )
-        else:
-            # 默认智能封面
-            return CoverGenerator.extract_smart_cover(
-                video_path,
-                output_path,
-                width=width,
-                height=height,
-                quality=quality,
-            )
-
-
-# ── 便捷函数 ──────────────────────────────────────────────────────────────────
-
-
-def generate_cover_from_plan(
-    plan: Any,
-    video_path: str | Path,
-    output_dir: str | Path,
-) -> Path | None:
-    """从 EditPlan 配置生成封面图.
-
-    配置读取：plan.config.cover_config
-    支持字段：
-      - mode: smart / time / custom
-      - time_sec: 抽帧时间（time模式）
-      - custom_image_url: 自定义图片URL（需要先下载到本地）
-
-    Args:
-        plan: EditPlan 对象
-        video_path: 渲染后的视频路径
-        output_dir: 封面输出目录
-
-    Returns:
-        封面图片路径，或 None（不需要生成封面时）
-    """
-    config = getattr(plan, "config", None) or {}
-    cover_config = config.get("cover_config") if isinstance(config, dict) else None
-
-    if not cover_config:
-        return None
-
-    mode = cover_config.get("mode", "smart")
-    output_dir = Path(output_dir)
-    output_dir.mkdir(parents=True, exist_ok=True)
-    output_path = output_dir / f"cover_{plan.id}.jpg"
-
-    try:
-        if mode == "custom":
-            # 自定义封面：需要先有本地图片路径
-            custom_path = cover_config.get("custom_image_path")
-            if custom_path and Path(custom_path).exists():
-                return CoverGenerator.process_custom_cover(
-                    custom_path,
-                    output_path,
-                )
-            else:
-                logger.warning("自定义封面图片路径无效，退化为智能封面")
-                mode = "smart"
-
-        if mode == "time":
-            time_sec = float(cover_config.get("time_sec", DEFAULT_COVER_TIME))
-            return CoverGenerator.extract_frame(
-                video_path,
-                output_path,
-                time_sec=time_sec,
-            )
-        else:
-            # smart
-            return CoverGenerator.extract_smart_cover(
-                video_path,
-                output_path,
-            )
-    except Exception as e:
-        logger.warning("封面生成失败: %s", e)
-        return None

apps/worker/video_processing/render_audio.py

@@ -22,8 +22,6 @@ from pathlib import Path
 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
@@ -226,129 +224,45 @@ def concat_main_audio(
         clip = clips[0]
         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
-
-        # 调速后时长
-        adjusted_duration = effective_duration / speed if abs(speed - 1.0) >= 1e-6 else effective_duration
-        # 最终时长：取调速后时长和视频总时长的较小值
-        final_duration = adjusted_duration
+        # 最终时长：取 clip 有效时长和视频总时长的较小值
+        # （视频总时长由主图层决定，但单 clip 场景下两者应该一致，仍做保护）
+        final_duration = effective_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"))
-        has_reverse = reverse_config.enabled and reverse_config.reverse_audio
-        has_speed = abs(speed - 1.0) >= 1e-6
-
-        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)
+        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)
         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)
+            filter_parts.append(
+                f"[{i}:a]atrim=start={trim_start:.3f}:duration={effective_duration:.3f}," f"asetpts=PTS-STARTPTS[a{i}]"
+            )
         else:
-            audio_filters.append("asetpts=PTS-STARTPTS")
-
-        # 音频倒放
-        reverse_config = ReverseConfig.from_dict(clip.config.get("reverse"))
-        if reverse_config.enabled and reverse_config.reverse_audio:
-            reverse_filter = ReverseEngine.build_audio_filter(reverse_config, duration=effective_duration)
-            if reverse_filter:
-                audio_filters.append(reverse_filter)
-
-        filter_parts.append(f"[{i}:a]{','.join(audio_filters)}[a{i}]")
+            filter_parts.append(f"[{i}:a]asetpts=PTS-STARTPTS[a{i}]")
 
     audio_labels = "".join(f"[a{i}]" for i in range(len(clips)))
     filter_parts.append(f"{audio_labels}concat=n={len(clips)}:v=0:a=1[outa]")

apps/worker/video_processing/reverse_engine.py

@@ -1,116 +0,0 @@
-"""视频倒放引擎 — 基于 FFmpeg reverse + areverse 滤镜实现视频/音频倒放.
-
-支持能力：
-- 视频倒放（reverse 滤镜）
-- 音频倒放（areverse 滤镜）
-- 按 clip 分段倒放，每个 clip 独立配置
-- 降级策略：不支持时跳过，不阻断渲染
-"""
-
-from __future__ import annotations
-
-import logging
-from dataclasses import dataclass
-from typing import Any
-
-logger = logging.getLogger(__name__)
-
-
-# ── 数据模型 ──────────────────────────────────────────────────────────────────
-
-
-@dataclass
-class ReverseConfig:
-    """视频倒放配置.
-
-    从 clip.config.reverse 读取，零侵入数据模型.
-    """
-
-    enabled: bool = False
-    reverse_video: bool = True  # 是否倒放视频
-    reverse_audio: bool = True  # 是否倒放音频
-
-    @classmethod
-    def from_dict(cls, data: dict[str, Any] | None) -> "ReverseConfig":
-        """从字典解析配置."""
-        if not data:
-            return cls(enabled=False)
-        try:
-            if not data.get("enabled", False):
-                return cls(enabled=False)
-            return cls(
-                enabled=True,
-                reverse_video=bool(data.get("reverse_video", True)),
-                reverse_audio=bool(data.get("reverse_audio", True)),
-            )
-        except (AttributeError, TypeError) as e:
-            logger.warning("倒放配置解析失败: %s，使用默认配置", e)
-            return cls(enabled=False)
-
-
-# ── 倒放引擎 ──────────────────────────────────────────────────────────────────
-
-
-class ReverseEngine:
-    """视频倒放引擎 — 生成 FFmpeg 倒放滤镜.
-
-    视频倒放：reverse 滤镜
-    音频倒放：areverse 滤镜
-
-    注意事项：
-    - reverse 滤镜需要将整个视频帧加载到内存，长视频可能占用大量内存
-    - 建议对单 clip 时长做限制（如 < 60s），超长视频建议降级
-    """
-
-    # 安全限制：单 clip 超过此时长不启用倒放（防止内存溢出）
-    MAX_SAFE_DURATION = 120.0  # 秒
-
-    @staticmethod
-    def build_video_filter(config: ReverseConfig, duration: float = 0.0) -> str:
-        """构建视频倒放滤镜字符串.
-
-        Args:
-            config: 倒放配置
-            duration: clip 时长（秒），用于安全检查
-
-        Returns:
-            FFmpeg 滤镜字符串，如 "reverse"；无效果返回空字符串
-        """
-        if not config.enabled or not config.reverse_video:
-            return ""
-
-        # 安全检查：超长视频不启用倒放
-        if duration > ReverseEngine.MAX_SAFE_DURATION:
-            logger.warning(
-                "视频倒放安全限制：clip 时长 %.1fs 超过上限 %.1fs，跳过倒放",
-                duration,
-                ReverseEngine.MAX_SAFE_DURATION,
-            )
-            return ""
-
-        return "reverse"
-
-    @staticmethod
-    def build_audio_filter(config: ReverseConfig, duration: float = 0.0) -> str:
-        """构建音频倒放滤镜字符串.
-
-        Args:
-            config: 倒放配置
-            duration: clip 时长（秒），用于安全检查
-
-        Returns:
-            FFmpeg 音频滤镜字符串，如 "areverse"；无效果返回空字符串
-        """
-        if not config.enabled or not config.reverse_audio:
-            return ""
-
-        # 安全检查：超长音频不启用倒放
-        if duration > ReverseEngine.MAX_SAFE_DURATION:
-            logger.warning(
-                "音频倒放安全限制：clip 时长 %.1fs 超过上限 %.1fs，跳过倒放",
-                duration,
-                ReverseEngine.MAX_SAFE_DURATION,
-            )
-            return ""
-
-        return "areverse"

apps/worker/video_processing/speed_engine.py

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

@@ -1,574 +0,0 @@
-"""贴纸叠加引擎 — 基于 FFmpeg overlay + drawtext 实现图片/文字贴纸.
-
-支持能力：
-- 图片贴纸（PNG/GIF）：位置、大小、透明度、时间范围、淡入淡出
-- 文字贴纸（花字）：字体、颜色、描边、阴影、位置、时间范围、动画
-- 9宫格位置 + 自由坐标（像素或百分比）
-- 多贴纸叠加，按 z_index 排序
-- 降级策略：素材不存在/无效时自动跳过，不阻断渲染
-"""
-
-from __future__ import annotations
-
-import logging
-from dataclasses import dataclass, field
-from pathlib import Path
-from typing import Any
-
-logger = logging.getLogger(__name__)
-
-
-# ── 预设贴纸分类 ──────────────────────────────────────────────────────────────
-
-# 预设贴纸分类（仅用于前端展示，后端不依赖具体素材）
-STICKER_CATEGORIES = [
-    ("emoji", "表情包"),
-    ("text", "文字花字"),
-    ("decoration", "装饰"),
-    ("arrow", "箭头指示"),
-    ("frame", "边框"),
-]
-
-# 9宫格位置映射
-POSITION_PRESETS = {
-    "top_left": (0.05, 0.05),
-    "top_center": (0.5, 0.05),
-    "top_right": (0.95, 0.05),
-    "center_left": (0.05, 0.5),
-    "center": (0.5, 0.5),
-    "center_right": (0.95, 0.5),
-    "bottom_left": (0.05, 0.95),
-    "bottom_center": (0.5, 0.95),
-    "bottom_right": (0.95, 0.95),
-}
-
-
-# ── 数据模型 ──────────────────────────────────────────────────────────────────
-
-
-@dataclass
-class ImageStickerConfig:
-    """图片贴纸配置."""
-
-    enabled: bool = False
-    type: str = "image"  # image / text
-    # 位置
-    position: str = "top_right"  # 9宫格预设
-    x: float | None = None  # 自定义x（像素或百分比）
-    y: float | None = None  # 自定义y
-    x_unit: str = "percent"  # pixel / percent
-    y_unit: str = "percent"
-    # 大小
-    scale: float = 1.0  # 缩放比例（相对于原始大小）
-    width: int | None = None  # 指定宽度（像素）
-    height: int | None = None  # 指定高度（像素）
-    # 透明度
-    opacity: float = 1.0  # 0.0~1.0
-    # 时间范围
-    start_time: float = 0.0
-    duration: float = 0.0  # 0 表示持续到结束
-    # 动画
-    fade_in: float = 0.0  # 淡入时长（秒）
-    fade_out: float = 0.0  # 淡出时长
-    # 层级
-    z_index: int = 10
-    # 素材
-    image_url: str = ""  # 图片URL或本地路径
-    preset_id: str = ""  # 预设贴纸ID
-
-
-@dataclass
-class TextStickerConfig:
-    """文字贴纸配置."""
-
-    enabled: bool = False
-    type: str = "text"
-    text: str = ""
-    # 字体
-    font_size: int = 36
-    font_color: str = "#FFFFFF"
-    font_family: str = "sans"
-    # 描边
-    stroke_color: str = "#000000"
-    stroke_width: int = 2
-    # 阴影
-    shadow_color: str = "#000000"
-    shadow_x: int = 2
-    shadow_y: int = 2
-    shadow_alpha: float = 0.5
-    # 位置
-    position: str = "center"
-    x: float | None = None
-    y: float | None = None
-    x_unit: str = "percent"
-    y_unit: str = "percent"
-    # 时间范围
-    start_time: float = 0.0
-    duration: float = 0.0
-    # 动画
-    fade_in: float = 0.0
-    fade_out: float = 0.0
-    # 层级
-    z_index: int = 10
-    # 背景框
-    bg_color: str = ""  # 空表示无背景
-    bg_padding: int = 8
-    bg_alpha: float = 0.8
-    bg_corner_radius: int = 8
-
-
-@dataclass
-class StickerOverlayResult:
-    """贴纸叠加结果."""
-
-    filter_str: str  # 滤镜字符串
-    output_label: str  # 输出标签
-    extra_inputs: list[str] = field(default_factory=list)  # 额外的输入文件路径
-
-
-# ── 贴纸引擎 ──────────────────────────────────────────────────────────────────
-
-
-class StickerEngine:
-    """贴纸叠加引擎 — 生成 FFmpeg overlay / drawtext 滤镜链.
-
-    支持图片贴纸（overlay）和文字贴纸（drawtext）。
-    多贴纸按 z_index 排序依次叠加。
-    """
-
-    @staticmethod
-    def _resolve_position(
-        config: ImageStickerConfig | TextStickerConfig,
-        canvas_w: int,
-        canvas_h: int,
-        sticker_w: int = 0,
-        sticker_h: int = 0,
-    ) -> tuple[float, float]:
-        """解析贴纸位置（像素坐标）.
-
-        优先级：自定义坐标 > 9宫格预设
-        """
-        # 先取预设的基准位置
-        if config.position in POSITION_PRESETS:
-            px, py = POSITION_PRESETS[config.position]
-        else:
-            px, py = 0.5, 0.5  # 默认居中
-
-        # 自定义坐标覆盖
-        if config.x is not None:
-            if config.x_unit == "percent":
-                px = config.x / 100.0
-            else:
-                px = config.x / canvas_w if canvas_w > 0 else 0.5
-
-        if config.y is not None:
-            if config.y_unit == "percent":
-                py = config.y / 100.0
-            else:
-                py = config.y / canvas_h if canvas_h > 0 else 0.5
-
-        # 转换为像素坐标（考虑贴纸尺寸，使位置为贴纸中心点）
-        x = px * canvas_w - sticker_w / 2
-        y = py * canvas_h - sticker_h / 2
-
-        # 钳制在画布内
-        x = max(0, min(x, canvas_w - sticker_w))
-        y = max(0, min(y, canvas_h - sticker_h))
-
-        return x, y
-
-    @staticmethod
-    def _build_overlay_filter(
-        sticker: ImageStickerConfig,
-        sticker_idx: int,
-        input_label: str,
-        output_label: str,
-        canvas_w: int,
-        canvas_h: int,
-    ) -> str:
-        """构建单个图片贴纸的 overlay 滤镜.
-
-        Args:
-            sticker: 贴纸配置
-            sticker_idx: 贴纸索引（用于生成滤镜标签）
-            input_label: 输入视频标签（如 "[base]"）
-            output_label: 输出视频标签
-            canvas_w: 画布宽度
-            canvas_h: 画布高度
-
-        Returns:
-            FFmpeg 滤镜字符串
-        """
-        sticker_label = f"sticker_{sticker_idx}_scaled"
-
-        # 1. 贴纸缩放预处理
-        scale_parts = []
-        if sticker.width and sticker.height:
-            scale_parts.append(f"scale={sticker.width}:{sticker.height}")
-        elif sticker.scale != 1.0:
-            # 按比例缩放
-            scale_parts.append(f"scale=iw*{sticker.scale}:ih*{sticker.scale}")
-        # 透明度调整
-        if sticker.opacity < 1.0:
-            scale_parts.append(f"colorchannelmixer=aa={sticker.opacity}")
-
-        # 淡入淡出
-        fade_parts = []
-        if sticker.fade_in > 0:
-            fade_parts.append(f"fade=in:st={sticker.start_time}:d={sticker.fade_in}:alpha=1")
-        if sticker.fade_out > 0 and sticker.duration > 0:
-            fade_out_start = sticker.start_time + sticker.duration - sticker.fade_out
-            fade_parts.append(f"fade=out:st={max(0, fade_out_start)}:d={sticker.fade_out}:alpha=1")
-
-        pre_filters = scale_parts + fade_parts
-
-        # 2. overlay 位置
-        # 先估算贴纸尺寸（假设原始尺寸 ~ canvas_w * 0.3）
-        est_w = int(canvas_w * 0.3 * sticker.scale) if not sticker.width else sticker.width
-        est_h = int(canvas_h * 0.3 * sticker.scale) if not sticker.height else sticker.height
-        pos_x, pos_y = StickerEngine._resolve_position(sticker, canvas_w, canvas_h, est_w, est_h)
-
-        # 3. enable 表达式（时间范围）
-        enable_expr = ""
-        if sticker.duration > 0:
-            enable_expr = f":enable='between(t,{sticker.start_time},{sticker.start_time + sticker.duration})'"
-
-        # 组合滤镜
-        filter_parts: list[str] = []
-
-        # 贴纸预处理
-        if pre_filters:
-            filter_parts.append(f"[{sticker_idx + 1}:v]{','.join(pre_filters)}[{sticker_label}]")
-            sticker_source = f"[{sticker_label}]"
-        else:
-            sticker_source = f"[{sticker_idx + 1}:v]"
-
-        # overlay 合成
-        filter_parts.append(f"{input_label}{sticker_source}overlay={pos_x:.0f}:{pos_y:.0f}{enable_expr}{output_label}")
-
-        return ";".join(filter_parts)
-
-    @staticmethod
-    def _build_drawtext_filter(
-        sticker: TextStickerConfig,
-        input_label: str,
-        output_label: str,
-        canvas_w: int,
-        canvas_h: int,
-    ) -> str:
-        """构建单个文字贴纸的 drawtext 滤镜.
-
-        Args:
-            sticker: 文字贴纸配置
-            input_label: 输入视频标签
-            output_label: 输出视频标签
-            canvas_w: 画布宽度
-            canvas_h: 画布高度
-
-        Returns:
-            FFmpeg 滤镜字符串
-        """
-        if not sticker.text:
-            return f"{input_label}copy{output_label}"
-
-        # 估算文字尺寸（粗略）
-        est_w = len(sticker.text) * sticker.font_size * 0.6
-        est_h = sticker.font_size * 1.4
-
-        pos_x, pos_y = StickerEngine._resolve_position(sticker, canvas_w, canvas_h, int(est_w), int(est_h))
-
-        drawtext_params: list[str] = []
-
-        # 文字内容（转义特殊字符）
-        escaped_text = sticker.text.replace(":", "\\:").replace("'", "\\'")
-        drawtext_params.append(f"text='{escaped_text}'")
-
-        # 字体
-        drawtext_params.append(f"fontsize={sticker.font_size}")
-        drawtext_params.append(f"fontcolor={sticker.font_color}")
-
-        # 描边
-        if sticker.stroke_width > 0:
-            drawtext_params.append(f"borderw={sticker.stroke_width}")
-            drawtext_params.append(f"bordercolor={sticker.stroke_color}")
-
-        # 阴影
-        if sticker.shadow_alpha > 0:
-            drawtext_params.append(f"shadowx={sticker.shadow_x}")
-            drawtext_params.append(f"shadowy={sticker.shadow_y}")
-            drawtext_params.append(f"shadowcolor={sticker.shadow_color}@{sticker.shadow_alpha}")
-
-        # 位置
-        drawtext_params.append(f"x={pos_x:.0f}")
-        drawtext_params.append(f"y={pos_y:.0f}")
-
-        # 时间范围
-        if sticker.duration > 0:
-            drawtext_params.append(f"enable='between(t,{sticker.start_time},{sticker.start_time + sticker.duration})'")
-
-        # 淡入淡出（drawtext 没有直接的淡入淡出，用 alpha 表达式模拟）
-        if sticker.fade_in > 0 or sticker.fade_out > 0:
-            alpha_expr = "1"
-            parts: list[str] = []
-            if sticker.fade_in > 0:
-                parts.append(
-                    f"if(lt(t,{sticker.start_time + sticker.fade_in})," f"(t-{sticker.start_time})/{sticker.fade_in},1)"
-                )
-            if sticker.fade_out > 0 and sticker.duration > 0:
-                fade_out_start = sticker.start_time + sticker.duration - sticker.fade_out
-                parts.append(
-                    f"if(gt(t,{fade_out_start})," f"({sticker.start_time + sticker.duration}-t)/{sticker.fade_out},1)"
-                )
-            if parts:
-                alpha_expr = "*".join(parts)
-                drawtext_params.append(f"alpha='{alpha_expr}'")
-
-        filter_str = f"{input_label}drawtext={':'.join(drawtext_params)}{output_label}"
-        return filter_str
-
-    @classmethod
-    def build_sticker_chain(
-        cls,
-        stickers: list[dict[str, Any]],
-        input_label: str,
-        output_label: str,
-        canvas_w: int,
-        canvas_h: int,
-    ) -> StickerOverlayResult:
-        """构建多贴纸叠加滤镜链.
-
-        Args:
-            stickers: 贴纸配置列表
-            input_label: 初始输入标签
-            output_label: 最终输出标签
-            canvas_w: 画布宽度
-            canvas_h: 画布高度
-
-        Returns:
-            StickerOverlayResult，包含滤镜字符串、输出标签、额外输入
-        """
-        if not stickers:
-            return StickerOverlayResult(
-                filter_str=f"{input_label}copy{output_label}",
-                output_label=output_label,
-                extra_inputs=[],
-            )
-
-        # 解析配置
-        parsed_stickers: list[tuple[int, ImageStickerConfig | TextStickerConfig]] = []
-        image_stickers: list[ImageStickerConfig] = []
-        image_paths: list[str] = []
-
-        for i, s in enumerate(stickers):
-            try:
-                sticker_type = s.get("type", "image")
-                z = int(s.get("z_index", 10))
-
-                if sticker_type == "text":
-                    config = TextStickerConfig(
-                        enabled=True,
-                        text=str(s.get("text", "")),
-                        font_size=int(s.get("font_size", 36)),
-                        font_color=str(s.get("font_color", "#FFFFFF")),
-                        stroke_color=str(s.get("stroke_color", "#000000")),
-                        stroke_width=int(s.get("stroke_width", 2)),
-                        shadow_x=int(s.get("shadow_x", 2)),
-                        shadow_y=int(s.get("shadow_y", 2)),
-                        shadow_alpha=float(s.get("shadow_alpha", 0.5)),
-                        position=str(s.get("position", "center")),
-                        x=cls._safe_float(s.get("x")),
-                        y=cls._safe_float(s.get("y")),
-                        x_unit=str(s.get("x_unit", "percent")),
-                        y_unit=str(s.get("y_unit", "percent")),
-                        start_time=float(s.get("start_time", 0)),
-                        duration=float(s.get("duration", 0)),
-                        fade_in=float(s.get("fade_in", 0)),
-                        fade_out=float(s.get("fade_out", 0)),
-                        z_index=z,
-                        bg_color=str(s.get("bg_color", "")),
-                        bg_padding=int(s.get("bg_padding", 8)),
-                        bg_alpha=float(s.get("bg_alpha", 0.8)),
-                        bg_corner_radius=int(s.get("bg_corner_radius", 8)),
-                    )
-                    parsed_stickers.append((z, config))
-                else:
-                    # 图片贴纸
-                    image_path = s.get("image_path", "") or s.get("image_url", "")
-                    if not image_path or not Path(image_path).exists():
-                        logger.warning("贴纸素材不存在，跳过: %s", image_path)
-                        continue
-
-                    config = ImageStickerConfig(
-                        enabled=True,
-                        position=str(s.get("position", "top_right")),
-                        x=cls._safe_float(s.get("x")),
-                        y=cls._safe_float(s.get("y")),
-                        x_unit=str(s.get("x_unit", "percent")),
-                        y_unit=str(s.get("y_unit", "percent")),
-                        scale=float(s.get("scale", 1.0)),
-                        width=int(s["width"]) if s.get("width") else None,
-                        height=int(s["height"]) if s.get("height") else None,
-                        opacity=max(0.0, min(1.0, float(s.get("opacity", 1.0)))),
-                        start_time=float(s.get("start_time", 0)),
-                        duration=float(s.get("duration", 0)),
-                        fade_in=float(s.get("fade_in", 0)),
-                        fade_out=float(s.get("fade_out", 0)),
-                        z_index=z,
-                        image_url=str(s.get("image_url", "")),
-                    )
-                    parsed_stickers.append((z, config))
-                    image_stickers.append(config)
-                    image_paths.append(image_path)
-
-            except Exception as e:
-                logger.warning("贴纸配置解析失败，跳过: %s", e)
-                continue
-
-        if not parsed_stickers:
-            return StickerOverlayResult(
-                filter_str=f"{input_label}copy{output_label}",
-                output_label=output_label,
-                extra_inputs=[],
-            )
-
-        # 按 z_index 排序
-        parsed_stickers.sort(key=lambda x: x[0])
-
-        # 构建滤镜链
-        filter_parts: list[str] = []
-        current_label = input_label
-        img_idx = 0  # 图片贴纸的输入索引偏移
-
-        for idx, (_, sticker) in enumerate(parsed_stickers):
-            next_label = f"sticker_{idx}_out" if idx < len(parsed_stickers) - 1 else output_label
-
-            if isinstance(sticker, ImageStickerConfig):
-                # 图片贴纸：使用额外的输入（输入索引 = 1 + img_idx，0 是主视频）
-                # 注意：实际输入索引需要调用方根据输入列表确定
-                # 这里我们按 image_stickers 的顺序分配索引
-                # 主输入是 [0:v]，贴纸输入从 [1:v] 开始
-                single_filter = cls._build_single_image_sticker(
-                    sticker=sticker,
-                    sticker_input_idx=img_idx + 1,  # +1 因为 0 是主视频
-                    input_label=current_label,
-                    output_label=next_label,
-                    canvas_w=canvas_w,
-                    canvas_h=canvas_h,
-                )
-                filter_parts.append(single_filter)
-                img_idx += 1
-            else:
-                # 文字贴纸：drawtext，不需要额外输入
-                single_filter = cls._build_drawtext_filter(
-                    sticker,  # type: ignore
-                    current_label,
-                    next_label,
-                    canvas_w,
-                    canvas_h,
-                )
-                filter_parts.append(single_filter)
-
-            current_label = next_label
-
-        return StickerOverlayResult(
-            filter_str=";".join(filter_parts),
-            output_label=output_label,
-            extra_inputs=image_paths,
-        )
-
-    @classmethod
-    def _build_single_image_sticker(
-        cls,
-        sticker: ImageStickerConfig,
-        sticker_input_idx: int,
-        input_label: str,
-        output_label: str,
-        canvas_w: int,
-        canvas_h: int,
-    ) -> str:
-        """构建单个图片贴纸的完整滤镜（预处理 + overlay）.
-
-        Args:
-            sticker: 贴纸配置
-            sticker_input_idx: 贴纸在 FFmpeg 输入中的索引
-            input_label: 输入视频标签
-            output_label: 输出标签
-            canvas_w: 画布宽
-            canvas_h: 画布高
-        """
-        scaled_label = f"sticker_s{sticker_input_idx}"
-
-        # 预处理滤镜（缩放 + 透明度 + 淡入淡出）
-        pre_filters: list[str] = []
-
-        # 缩放
-        if sticker.width and sticker.height:
-            pre_filters.append(f"scale={sticker.width}:{sticker.height}")
-        elif sticker.scale != 1.0:
-            pre_filters.append(f"scale=iw*{sticker.scale}:ih*{sticker.scale}")
-
-        # 透明度
-        if sticker.opacity < 1.0:
-            pre_filters.append(f"format=rgba,colorchannelmixer=aa={sticker.opacity}")
-
-        # 淡入淡出（使用 fade 的 alpha 模式）
-        fade_filters: list[str] = []
-        if sticker.fade_in > 0:
-            fade_filters.append(f"fade=in:st={sticker.start_time}:d={sticker.fade_in}:alpha=1")
-        if sticker.fade_out > 0 and sticker.duration > 0:
-            fade_out_start = sticker.start_time + sticker.duration - sticker.fade_out
-            if fade_out_start > 0:
-                fade_filters.append(f"fade=out:st={fade_out_start}:d={sticker.fade_out}:alpha=1")
-
-        # 估算贴纸尺寸用于位置计算
-        est_w = int(canvas_w * 0.3 * sticker.scale) if not sticker.width else sticker.width
-        est_h = int(canvas_h * 0.3 * sticker.scale) if not sticker.height else sticker.height
-        pos_x, pos_y = cls._resolve_position(sticker, canvas_w, canvas_h, est_w, est_h)
-
-        # enable 表达式
-        enable_expr = ""
-        if sticker.duration > 0:
-            enable_expr = f":enable='between(t,{sticker.start_time},{sticker.start_time + sticker.duration})'"
-
-        parts: list[str] = []
-
-        # 贴纸预处理
-        all_pre = pre_filters + fade_filters
-        if all_pre:
-            parts.append(f"[{sticker_input_idx}:v]{','.join(all_pre)}[{scaled_label}]")
-            sticker_source = f"[{scaled_label}]"
-        else:
-            sticker_source = f"[{sticker_input_idx}:v]"
-
-        # overlay 合成
-        parts.append(f"{input_label}{sticker_source}overlay={pos_x:.0f}:{pos_y:.0f}{enable_expr}{output_label}")
-
-        return ";".join(parts)
-
-    @staticmethod
-    def _safe_float(val: Any) -> float | None:
-        """安全转换 float."""
-        if val is None:
-            return None
-        try:
-            return float(val)
-        except (ValueError, TypeError):
-            return None
-
-
-# ── 便捷函数 ──────────────────────────────────────────────────────────────────
-
-
-def parse_stickers_from_config(config: dict[str, Any] | None) -> list[dict[str, Any]]:
-    """从 plan.config.stickers 解析贴纸列表."""
-    if not config:
-        return []
-    stickers = config.get("stickers", [])
-    if not isinstance(stickers, list):
-        return []
-    return stickers
-
-
-def get_sticker_categories() -> list[tuple[str, str]]:
-    """获取贴纸分类列表."""
-    return list(STICKER_CATEGORIES)

apps/worker/video_processing/unified_render_service.py

@@ -44,9 +44,6 @@ from video_processing.intro_outro_engine import IntroOutroConfig, IntroOutroEngi
 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
 from video_processing.trim_engine import TrimConfig, TrimEngine, extract_trim_from_clip_config
@@ -74,7 +71,6 @@ 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)
 
     # 运行时填充
@@ -188,7 +184,6 @@ 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.
@@ -496,7 +491,7 @@ class UnifiedRenderService:
         if not main_layer or not main_layer.clips:
             return 0.0
 
-        total = sum(UnifiedRenderService._clip_adjusted_duration(c) for c in main_layer.clips)
+        total = sum(UnifiedRenderService._clip_effective_duration(c) for c in main_layer.clips)
 
         # 减去转场重叠时间（粗略估算）
         n_clips = len(main_layer.clips)
@@ -752,7 +747,6 @@ class UnifiedRenderService:
         1. 只有 1 个图层
         2. 该图层是视频图层（main/broll/background），不是 overlay/corner_voice/audio
         3. 该图层只有 1 个 clip（无转场需求）
-        4. 没有贴纸（贴纸需要 filter_complex 或额外输入）
         """
         if len(layers) != 1:
             return False
@@ -761,10 +755,6 @@ class UnifiedRenderService:
             return False
         if len(layer.clips) != 1:
             return False
-        # 有贴纸时禁用直通（图片贴纸需要额外输入，统一走 filter_complex）
-        plan_config = getattr(self.plan, "config", None) or {}
-        if isinstance(plan_config, dict) and plan_config.get("stickers"):
-            return False
         return True
 
     def _can_use_stream_copy(
@@ -977,13 +967,6 @@ class UnifiedRenderService:
             filters.append(f"trim=duration={effective_duration}")
             filters.append("setpts=PTS-STARTPTS")
 
-        # 倒放滤镜
-        reverse_config = ReverseConfig.from_dict(clip.config.get("reverse"))
-        if reverse_config.enabled and reverse_config.reverse_video:
-            reverse_filter = ReverseEngine.build_video_filter(reverse_config, duration=effective_duration)
-            if reverse_filter:
-                filters.append(reverse_filter)
-
         # scale + crop（铺满裁剪）
         if role in ("overlay", "corner_voice"):
             pip_w = int(self.output_width * _PIP_SCALE)
@@ -1072,13 +1055,6 @@ class UnifiedRenderService:
 
             command.extend(["-c:a", "aac", "-b:a", "128k"])
 
-            # 音频倒放
-            reverse_config = ReverseConfig.from_dict(clip.config.get("reverse"))
-            if reverse_config.enabled and reverse_config.reverse_audio:
-                af_filter = ReverseEngine.build_audio_filter(reverse_config, duration=effective_duration)
-                if af_filter:
-                    command.extend(["-af", af_filter])
-
         # 统一截断时长（同时作用于视频和音频）
         if final_duration > 0:
             command.extend(["-t", f"{final_duration:.3f}"])
@@ -1200,7 +1176,6 @@ 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,
@@ -1298,18 +1273,6 @@ 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:
-                reverse_filter = ReverseEngine.build_video_filter(reverse_config, duration=effective_duration)
-                if reverse_filter:
-                    filters.append(reverse_filter)
-
             # scale
             if role in ("overlay", "corner_voice"):
                 pip_w = int(self.output_width * _PIP_SCALE)
@@ -1360,8 +1323,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]
-            # 使用调速后的实际时长，与 Step 1 的调速处理保持一致
-            layer_durations = [UnifiedRenderService._clip_adjusted_duration(all_clips[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]
             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]
 
@@ -1487,14 +1450,6 @@ class UnifiedRenderService:
                         final_video_label = wm_label
                     except Exception as e:
                         logger.warning("文字水印构建失败，跳过: %s", e)
-        # 贴纸叠加（图片贴纸 + 文字贴纸）
-        sticker_filter, sticker_extra_inputs = self._build_sticker_filters(final_video_label, "after_stickers")
-        if sticker_filter:
-            filter_parts.append(sticker_filter)
-            # 图片贴纸需要额外输入
-            for img_path in sticker_extra_inputs:
-                input_args.extend(["-i", img_path])
-            final_video_label = "after_stickers"
 
         # 叠加字幕（如有）+ 最终像素格式
         if ass_path is not None:
@@ -1555,40 +1510,6 @@ class UnifiedRenderService:
             )
             raise
 
-    def _build_sticker_filters(self, input_label: str, output_label: str) -> tuple[str, list[str]]:
-        """构建贴纸叠加滤镜链.
-
-        Args:
-            input_label: 输入视频标签
-            output_label: 输出视频标签
-
-        Returns:
-            (filter_str, extra_input_paths)
-            filter_str: 贴纸滤镜字符串（空表示无贴纸）
-            extra_input_paths: 额外需要的输入文件路径（图片贴纸）
-        """
-        plan_config = getattr(self.plan, "config", None) or {}
-        if isinstance(plan_config, dict):
-            stickers_data = plan_config.get("stickers", [])
-        else:
-            stickers_data = []
-
-        if not stickers_data:
-            return "", []
-
-        try:
-            result = StickerEngine.build_sticker_chain(
-                stickers=stickers_data,
-                input_label=f"[{input_label}]",
-                output_label=f"[{output_label}]",
-                canvas_w=self.output_width,
-                canvas_h=self.output_height,
-            )
-            return result.filter_str, result.extra_inputs
-        except Exception as e:
-            logger.warning("贴纸滤镜构建失败，跳过贴纸: %s", e)
-            return "", []
-
     def _probe_output(self, output_path: Path) -> tuple[float, int, int, int]:
         """探测输出文件的时长、大小、宽高.
 
@@ -1606,7 +1527,7 @@ class UnifiedRenderService:
 
     @staticmethod
     def _clip_effective_duration(clip: ResolvedClip) -> float:
-        """计算 clip 的有效时长（原速 trim 后时长）."""
+        """计算 clip 的有效时长."""
         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
@@ -1703,20 +1624,3 @@ 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,14 +866,6 @@
           "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,7 +55,6 @@ 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,
         )
@@ -79,7 +78,6 @@ 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
@@ -125,7 +123,6 @@ 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,7 +197,6 @@ 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,7 +51,6 @@ 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))
@@ -71,7 +70,6 @@ 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:
         """创建剪辑计划片段"""
@@ -83,13 +81,6 @@ 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,
@@ -103,7 +94,6 @@ 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 {},
         )

scripts/gray_deploy.sh

@@ -1,117 +0,0 @@
-#!/bin/bash
-# 灰度发布脚本：通过Nginx权重调整流量比例
-# 用法: ./scripts/gray_deploy.sh <版本号> <灰度百分比>
-#
-# 需要在目标服务器上执行，或通过SSH执行
-# 前提：服务器上运行两个版本的容器（stable + canary），Nginx做加权轮询
-
-set -euo pipefail
-
-VERSION="${1:-}"
-GRAY_PCT="${2:-10}"
-
-if [[ -z "$VERSION" ]]; then
-    echo "用法: $0 <版本号> [灰度百分比]"
-    echo "示例: $0 v0.1.129 5"
-    exit 1
-fi
-
-STABLE_VERSION="${STABLE_VERSION:-current}"
-REGISTRY="${REGISTRY:-git.xiaoxiajianji.com/xiaoxia/xiaoxia-saas}"
-
-echo "=========================================="
-echo "  灰度发布"
-echo "  新版本: $VERSION"
-echo "  灰度比例: ${GRAY_PCT}%"
-echo "  稳定版本: $STABLE_VERSION"
-echo "=========================================="
-
-# 1. 拉取新版本镜像
-echo ""
-echo ">>> 拉取新版本镜像..."
-for component in api worker web; do
-    echo "  拉取 $component:$VERSION ..."
-    docker pull "${REGISTRY}-${component}:${VERSION}" 2>&1 | tail -1
-done
-
-# 2. 启动灰度版本容器（canary）
-echo ""
-echo ">>> 启动灰度版本容器..."
-
-# API canary
-CANARY_API_NAME="saas-api-canary"
-if docker ps -a --format '{{.Names}}' | grep -q "^${CANARY_API_NAME}$"; then
-    echo "  停止旧 canary 容器..."
-    docker stop "$CANARY_API_NAME" 2>/dev/null || true
-    docker rm "$CANARY_API_NAME" 2>/dev/null || true
-fi
-
-echo "  启动 api canary..."
-docker run -d \
-    --name "$CANARY_API_NAME" \
-    --network saas-network \
-    -e DATABASE_URL="${DATABASE_URL}" \
-    -e REDIS_URL="${REDIS_URL}" \
-    -e FEATURE_FLAG_PROVIDER=redis \
-    --restart unless-stopped \
-    "${REGISTRY}-api:${VERSION}"
-
-# Worker canary
-CANARY_WORKER_NAME="saas-worker-canary"
-if docker ps -a --format '{{.Names}}' | grep -q "^${CANARY_WORKER_NAME}$"; then
-    echo "  停止旧 worker canary..."
-    docker stop "$CANARY_WORKER_NAME" 2>/dev/null || true
-    docker rm "$CANARY_WORKER_NAME" 2>/dev/null || true
-fi
-
-echo "  启动 worker canary..."
-docker run -d \
-    --name "$CANARY_WORKER_NAME" \
-    --network saas-network \
-    -e DATABASE_URL="${DATABASE_URL}" \
-    -e REDIS_URL="${REDIS_URL}" \
-    --restart unless-stopped \
-    "${REGISTRY}-worker:${VERSION}"
-
-# 3. 等待容器健康
-echo ""
-echo ">>> 等待容器健康..."
-sleep 10
-if ! docker ps --format '{{.Names}} {{.Status}}' | grep -q "$CANARY_API_NAME"; then
-    echo "错误: API canary 容器未运行"
-    docker logs "$CANARY_API_NAME" --tail 20
-    exit 1
-fi
-echo "  ✅ API canary 运行中"
-
-# 4. 更新Nginx权重
-echo ""
-echo ">>> 更新Nginx权重 (稳定: $((100-GRAY_PCT))% / 灰度: ${GRAY_PCT}%)..."
-
-NGINX_CONF="${NGINX_CONF:-/etc/nginx/conf.d/saas-api.conf}"
-if [[ -f "$NGINX_CONF" ]]; then
-    # 备份
-    cp "$NGINX_CONF" "${NGINX_CONF}.bak.$(date +%Y%m%d%H%M%S)"
-    
-    # 更新 upstream 权重（需要根据实际配置调整）
-    echo "  请手动更新 Nginx upstream 配置中的权重"
-    echo "  示例配置:"
-    cat <<EOF
-    upstream saas_api_backend {
-        server saas-api:8000 weight=$((100-GRAY_PCT));
-        server saas-api-canary:8000 weight=${GRAY_PCT};
-    }
-EOF
-    nginx -t && nginx -s reload
-    echo "  ✅ Nginx 已reload"
-else
-    echo "  警告: Nginx 配置文件不存在 ($NGINX_CONF)"
-    echo "  请手动配置灰度流量权重"
-fi
-
-echo ""
-echo "=========================================="
-echo "  ✅ 灰度发布完成"
-echo "  新版本: $VERSION (${GRAY_PCT}%流量)"
-echo "  监控: Grafana / 日志"
-echo "=========================================="

scripts/release.sh

@@ -1,126 +0,0 @@
-#!/bin/bash
-# 一键发布脚本：打tag → 触发生产镜像构建 → 部署到灰度
-# 用法: ./scripts/release.sh v0.1.129 [--gray 5]
-
-set -euo pipefail
-
-SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
-REPO_ROOT="$(cd "$SCRIPT_DIR/.." && pwd)"
-
-usage() {
-    echo "用法: $0 <版本号> [--gray 百分比] [--no-deploy]"
-    echo ""
-    echo "示例:"
-    echo "  $0 v0.1.129                    # 打tag + 全量发布"
-    echo "  $0 v0.1.129 --gray 5           # 打tag + 5%灰度发布"
-    echo "  $0 v0.1.129 --no-deploy        # 只打tag，不部署"
-    exit 1
-}
-
-# 参数解析
-VERSION=""
-GRAY_PCT=0
-DEPLOY=true
-
-while [[ $# -gt 0 ]]; do
-    case "$1" in
-        --gray)
-            GRAY_PCT="$2"
-            shift 2
-            ;;
-        --no-deploy)
-            DEPLOY=false
-            shift
-            ;;
-        -h|--help)
-            usage
-            ;;
-        v*)
-            VERSION="$1"
-            shift
-            ;;
-        *)
-            echo "未知参数: $1"
-            usage
-            ;;
-    esac
-done
-
-if [[ -z "$VERSION" ]]; then
-    echo "错误: 请指定版本号（如 v0.1.129）"
-    usage
-fi
-
-echo "=========================================="
-echo "  发布版本: $VERSION"
-echo "  灰度比例: ${GRAY_PCT}%"
-echo "  自动部署: $DEPLOY"
-echo "=========================================="
-
-cd "$REPO_ROOT"
-
-# 1. 确认在 develop 分支
-CURRENT_BRANCH=$(git rev-parse --abbrev-ref HEAD)
-if [[ "$CURRENT_BRANCH" != "develop" ]]; then
-    echo "错误: 请切换到 develop 分支后再发布"
-    exit 1
-fi
-
-# 2. 拉取最新代码
-echo ""
-echo ">>> 拉取最新代码..."
-git pull origin develop
-
-# 3. 生成 CHANGELOG
-echo ""
-echo ">>> 生成 CHANGELOG..."
-if [[ -f "scripts/generate_changelog.py" ]]; then
-    PREV_TAG=$(git describe --tags --abbrev=0 HEAD^ 2>/dev/null || echo "")
-    if [[ -n "$PREV_TAG" ]]; then
-        python3 scripts/generate_changelog.py \
-            --from-tag "$PREV_TAG" \
-            --to-tag HEAD \
-            --gitea-token "${GITEA_TOKEN:-}" \
-            --output /tmp/changelog_$$.md
-        echo "CHANGELOG 已生成到 /tmp/changelog_$$.md"
-    fi
-fi
-
-# 4. 打tag
-echo ""
-echo ">>> 打 tag $VERSION ..."
-if git rev-parse "$VERSION" >/dev/null 2>&1; then
-    echo "警告: tag $VERSION 已存在，跳过打tag"
-else
-    git tag -a "$VERSION" -m "Release $VERSION"
-    git push origin "$VERSION"
-    echo "Tag $VERSION 已推送，触发生产镜像构建..."
-fi
-
-# 5. 等待镜像构建
-if [[ "$DEPLOY" == "true" ]]; then
-    echo ""
-    echo ">>> 等待镜像构建完成（约10-15分钟）..."
-    echo "    镜像: git.xiaoxiajianji.com/xiaoxia/xiaoxia-saas-{api,worker,web}:$VERSION"
-    
-    # 这里可以加镜像存在性检查
-    echo "    （镜像构建由CI自动完成，请在Gitea Actions中确认）"
-fi
-
-# 6. 灰度部署
-if [[ "$DEPLOY" == "true" && "$GRAY_PCT" -gt 0 ]]; then
-    echo ""
-    echo ">>> 灰度部署: ${GRAY_PCT}% 流量到 $VERSION"
-    if [[ -f "scripts/gray_deploy.sh" ]]; then
-        ./scripts/gray_deploy.sh "$VERSION" "$GRAY_PCT"
-    else
-        echo "警告: gray_deploy.sh 不存在，跳过灰度部署"
-    fi
-fi
-
-echo ""
-echo "=========================================="
-echo "  ✅ 发布流程完成"
-echo "  版本: $VERSION"
-echo "  灰度: ${GRAY_PCT}%"
-echo "=========================================="

scripts/rollback.sh

@@ -1,42 +0,0 @@
-#!/bin/bash
-# 灰度回滚脚本：切回稳定版本流量
-# 用法: ./scripts/rollback.sh [稳定版本号]
-
-set -euo pipefail
-
-STABLE_VERSION="${1:-current}"
-
-echo "=========================================="
-echo "  灰度回滚"
-echo "  切回稳定版本: $STABLE_VERSION"
-echo "=========================================="
-
-# 1. 恢复Nginx全量到稳定版本
-echo ""
-echo ">>> 恢复Nginx全量流量到稳定版本..."
-
-NGINX_CONF="${NGINX_CONF:-/etc/nginx/conf.d/saas-api.conf}"
-if [[ -f "$NGINX_CONF" ]]; then
-    # 找最近的备份
-    LATEST_BAK=$(ls -t "${NGINX_CONF}".bak.* 2>/dev/null | head -1)
-    if [[ -n "$LATEST_BAK" ]]; then
-        cp "$LATEST_BAK" "$NGINX_CONF"
-        echo "  从备份恢复: $LATEST_BAK"
-    else
-        echo "  未找到备份，请手动移除 canary upstream"
-    fi
-    
-    nginx -t && nginx -s reload
-    echo "  ✅ Nginx 已回滚"
-fi
-
-# 2. 停止灰度版本容器（保留30分钟以便排查）
-echo ""
-echo ">>> 灰度版本容器将在30分钟后停止（便于排查问题）"
-echo "    立即停止请执行: docker stop saas-api-canary saas-worker-canary"
-
-echo ""
-echo "=========================================="
-echo "  ✅ 回滚完成"
-echo "  流量已全部切回稳定版本"
-echo "=========================================="

tests/integration/test_generation_api.py

@@ -138,70 +138,6 @@ class StubGenerationTaskRepository:
     def list_by_source_edit_plan(self, plan_id: str) -> list[GenerationTask]:
         return [t for t in self._tasks.values() if getattr(t, "source_edit_plan_id", "") == plan_id]
 
-    def list_by_user_filtered(
-        self,
-        user_id: str,
-        *,
-        status: str | None = None,
-        limit: int | None = None,
-        offset: int = 0,
-    ) -> list:
-        """按用户+状态筛选任务列表（stub实现）。"""
-        items = [t for t in self._tasks.values() if t.created_by_user_id == user_id]
-        if status:
-            items = [t for t in items if str(t.status) == status]
-        # 按创建时间倒序
-        items.sort(key=lambda t: t.created_at or "", reverse=True)
-        if offset:
-            items = items[offset:]
-        if limit is not None:
-            items = items[:limit]
-        return items
-
-    def count_by_user_filtered(
-        self,
-        user_id: str,
-        *,
-        status: str | None = None,
-    ) -> int:
-        """按用户+状态筛选计数（stub实现）。"""
-        items = [t for t in self._tasks.values() if t.created_by_user_id == user_id]
-        if status:
-            items = [t for t in items if str(t.status) == status]
-        return len(items)
-
-    def list_by_project_filtered(
-        self,
-        project_id: str,
-        *,
-        status: str | None = None,
-        limit: int | None = None,
-        offset: int = 0,
-    ) -> list:
-        """按项目+状态筛选任务列表（stub实现）。"""
-        items = [t for t in self._tasks.values() if t.project_id == project_id]
-        if status:
-            items = [t for t in items if str(t.status) == status]
-        # 按创建时间倒序
-        items.sort(key=lambda t: t.created_at or "", reverse=True)
-        if offset:
-            items = items[offset:]
-        if limit is not None:
-            items = items[:limit]
-        return items
-
-    def count_by_project_filtered(
-        self,
-        project_id: str,
-        *,
-        status: str | None = None,
-    ) -> int:
-        """按项目+状态筛选计数（stub实现）。"""
-        items = [t for t in self._tasks.values() if t.project_id == project_id]
-        if status:
-            items = [t for t in items if str(t.status) == status]
-        return len(items)
-
 
 class StubGeneratedVideoRepository:
     def __init__(self, videos: dict[str, GeneratedVideo] | None = None):

tests/integration/test_task_center_api.py

@@ -103,70 +103,6 @@ class StubGenerationTaskRepository:
     def list_by_source_edit_plan(self, plan_id: str) -> list[GenerationTask]:
         return [t for t in self._tasks.values() if getattr(t, "source_edit_plan_id", "") == plan_id]
 
-    def list_by_user_filtered(
-        self,
-        user_id: str,
-        *,
-        status: str | None = None,
-        limit: int | None = None,
-        offset: int = 0,
-    ) -> list:
-        """按用户+状态筛选任务列表（stub实现）。"""
-        items = [t for t in self._tasks.values() if t.created_by_user_id == user_id]
-        if status:
-            items = [t for t in items if str(t.status) == status]
-        # 按创建时间倒序
-        items.sort(key=lambda t: t.created_at or "", reverse=True)
-        if offset:
-            items = items[offset:]
-        if limit is not None:
-            items = items[:limit]
-        return items
-
-    def count_by_user_filtered(
-        self,
-        user_id: str,
-        *,
-        status: str | None = None,
-    ) -> int:
-        """按用户+状态筛选计数（stub实现）。"""
-        items = [t for t in self._tasks.values() if t.created_by_user_id == user_id]
-        if status:
-            items = [t for t in items if str(t.status) == status]
-        return len(items)
-
-    def list_by_project_filtered(
-        self,
-        project_id: str,
-        *,
-        status: str | None = None,
-        limit: int | None = None,
-        offset: int = 0,
-    ) -> list:
-        """按项目+状态筛选任务列表（stub实现）。"""
-        items = [t for t in self._tasks.values() if t.project_id == project_id]
-        if status:
-            items = [t for t in items if str(t.status) == status]
-        # 按创建时间倒序
-        items.sort(key=lambda t: t.created_at or "", reverse=True)
-        if offset:
-            items = items[offset:]
-        if limit is not None:
-            items = items[:limit]
-        return items
-
-    def count_by_project_filtered(
-        self,
-        project_id: str,
-        *,
-        status: str | None = None,
-    ) -> int:
-        """按项目+状态筛选计数（stub实现）。"""
-        items = [t for t in self._tasks.values() if t.project_id == project_id]
-        if status:
-            items = [t for t in items if str(t.status) == status]
-        return len(items)
-
 
 class StubIngestJobRepository:
     def __init__(self, jobs: dict[str, IngestJob] | None = None):
@@ -534,8 +470,8 @@ class TestRetryProjectTask:
         assert data["task_type"] == "generation"
         assert data["status"] == "pending"
         assert "current_step" in data
-        # 原地重试：source_id 保持不变（复用同一个任务）
-        assert data["source_id"] == "gen-failed-1"
+        # 验证新任务的 ID 不同于原任务
+        assert data["source_id"] != "gen-failed-1"
         # 验证 Celery 任务被发送
         assert mock_celery.send_task.called
 
@@ -694,13 +630,11 @@ class TestTaskCenterCrossEndpoint:
         retry_resp = tc.post("/tasks/gen-fail-cross/retry")
         assert retry_resp.status_code == 200
 
-        # 3. 再次列出：原地重试，任务数不变（仍是1个），但状态变为 pending
+        # 3. 再次列出，应有2个任务（旧的failed + 新的pending）
         list_resp2 = tc.get("/tasks")
         assert list_resp2.status_code == 200
         items2 = list_resp2.json()["items"]
-        assert len(items2) == 1
-        assert items2[0]["status"] == "pending"
-        assert items2[0]["source_id"] == "gen-fail-cross"
+        assert len(items2) == 2
 
         test_app.dependency_overrides.clear()
 

tests/unit/test_cover_reverse_sticker.py

@@ -1,860 +0,0 @@
-"""封面生成 + 视频倒放 + 贴纸叠加 单元测试.
-
-覆盖三个新渲染能力的核心场景和降级逻辑。
-"""
-
-from __future__ import annotations
-
-from dataclasses import dataclass, field
-from pathlib import Path
-from typing import Any
-from unittest.mock import MagicMock, patch
-
-import pytest
-from video_processing.cover_generator import (
-    DEFAULT_COVER_HEIGHT,
-    DEFAULT_COVER_WIDTH,
-    CoverGenerator,
-    generate_cover_from_plan,
-)
-from video_processing.reverse_engine import ReverseConfig, ReverseEngine
-from video_processing.sticker_engine import (
-    POSITION_PRESETS,
-    STICKER_CATEGORIES,
-    ImageStickerConfig,
-    StickerEngine,
-    TextStickerConfig,
-    get_sticker_categories,
-    parse_stickers_from_config,
-)
-from video_processing.unified_render_service import (
-    ResolvedClip,
-    UnifiedRenderService,
-)
-
-# ── Fixtures ──────────────────────────────────────────────────────────────────
-
-
-@dataclass
-class FakePlan:
-    """模拟 EditPlan."""
-
-    id: str = "plan_001"
-    name: str = "测试计划"
-    config: dict[str, Any] = field(default_factory=dict)
-
-
-@pytest.fixture
-def sample_video(tmp_path):
-    """创建一个测试视频文件（空文件，仅用于路径测试）."""
-    video_path = tmp_path / "test_video.mp4"
-    video_path.write_bytes(b"fake video data")
-    return video_path
-
-
-@pytest.fixture
-def sample_image(tmp_path):
-    """创建一个测试图片文件."""
-    img_path = tmp_path / "sticker.png"
-    img_path.write_bytes(b"fake png data")
-    return img_path
-
-
-# ═══════════════════════════════════════════════════════════════════════════════
-# 一、视频倒放引擎测试
-# ═══════════════════════════════════════════════════════════════════════════════
-
-
-class TestReverseConfig:
-    """ReverseConfig 配置解析测试."""
-
-    def test_default_disabled(self):
-        """默认配置为关闭."""
-        config = ReverseConfig.from_dict(None)
-        assert config.enabled is False
-        assert config.reverse_video is True
-        assert config.reverse_audio is True
-
-    def test_empty_dict(self):
-        """空字典视为关闭."""
-        config = ReverseConfig.from_dict({})
-        assert config.enabled is False
-
-    def test_enabled(self):
-        """启用倒放."""
-        config = ReverseConfig.from_dict({"enabled": True})
-        assert config.enabled is True
-        assert config.reverse_video is True
-        assert config.reverse_audio is True
-
-    def test_video_only(self):
-        """只倒放视频."""
-        config = ReverseConfig.from_dict(
-            {
-                "enabled": True,
-                "reverse_video": True,
-                "reverse_audio": False,
-            }
-        )
-        assert config.enabled is True
-        assert config.reverse_video is True
-        assert config.reverse_audio is False
-
-    def test_audio_only(self):
-        """只倒放音频."""
-        config = ReverseConfig.from_dict(
-            {
-                "enabled": True,
-                "reverse_video": False,
-                "reverse_audio": True,
-            }
-        )
-        assert config.reverse_video is False
-        assert config.reverse_audio is True
-
-    def test_invalid_config_fallback(self):
-        """无效配置降级为默认."""
-        config = ReverseConfig.from_dict("invalid")  # type: ignore
-        assert config.enabled is False
-
-    def test_none_config(self):
-        """None 配置."""
-        config = ReverseConfig.from_dict(None)
-        assert config.enabled is False
-
-
-class TestReverseEngine:
-    """ReverseEngine 滤镜生成测试."""
-
-    def test_video_reverse_filter(self):
-        """视频倒放滤镜生成."""
-        config = ReverseConfig(enabled=True, reverse_video=True)
-        f = ReverseEngine.build_video_filter(config, duration=10.0)
-        assert f == "reverse"
-
-    def test_video_disabled(self):
-        """视频倒放关闭时返回空."""
-        config = ReverseConfig(enabled=False)
-        f = ReverseEngine.build_video_filter(config, duration=10.0)
-        assert f == ""
-
-    def test_video_disabled_flag(self):
-        """启用但 reverse_video=False."""
-        config = ReverseConfig(enabled=True, reverse_video=False)
-        f = ReverseEngine.build_video_filter(config, duration=10.0)
-        assert f == ""
-
-    def test_audio_reverse_filter(self):
-        """音频倒放滤镜生成."""
-        config = ReverseConfig(enabled=True, reverse_audio=True)
-        f = ReverseEngine.build_audio_filter(config, duration=10.0)
-        assert f == "areverse"
-
-    def test_audio_disabled(self):
-        """音频倒放关闭."""
-        config = ReverseConfig(enabled=False)
-        f = ReverseEngine.build_audio_filter(config, duration=10.0)
-        assert f == ""
-
-    def test_long_video_safety_limit(self):
-        """超长视频安全限制：跳过倒放."""
-        config = ReverseConfig(enabled=True)
-        f = ReverseEngine.build_video_filter(config, duration=200.0)
-        assert f == ""  # 超过 MAX_SAFE_DURATION
-
-    def test_long_audio_safety_limit(self):
-        """超长音频安全限制."""
-        config = ReverseConfig(enabled=True)
-        f = ReverseEngine.build_audio_filter(config, duration=200.0)
-        assert f == ""
-
-    def test_duration_zero(self):
-        """时长为0时正常返回."""
-        config = ReverseConfig(enabled=True)
-        f = ReverseEngine.build_video_filter(config, duration=0.0)
-        assert f == "reverse"
-
-
-# ═══════════════════════════════════════════════════════════════════════════════
-# 二、贴纸引擎测试
-# ═══════════════════════════════════════════════════════════════════════════════
-
-
-class TestStickerPosition:
-    """贴纸位置计算测试."""
-
-    def test_presets_exist(self):
-        """9宫格预设存在."""
-        assert "top_left" in POSITION_PRESETS
-        assert "center" in POSITION_PRESETS
-        assert "bottom_right" in POSITION_PRESETS
-        assert len(POSITION_PRESETS) == 9
-
-    def test_resolve_position_center(self):
-        """居中位置计算."""
-        sticker = ImageStickerConfig(position="center")
-        x, y = StickerEngine._resolve_position(sticker, 1000, 1000, 200, 200)
-        assert abs(x - 400) < 1  # (1000-200)/2 = 400
-        assert abs(y - 400) < 1
-
-    def test_resolve_position_top_left(self):
-        """左上角位置."""
-        sticker = ImageStickerConfig(position="top_left")
-        x, y = StickerEngine._resolve_position(sticker, 1000, 1000, 100, 100)
-        assert x == 0  # 0.05*1000 - 50 = 0 (clamped)
-        assert y == 0
-
-    def test_custom_position_percent(self):
-        """自定义百分比位置."""
-        sticker = ImageStickerConfig(
-            position="center",
-            x=30.0,
-            y=70.0,
-            x_unit="percent",
-            y_unit="percent",
-        )
-        x, y = StickerEngine._resolve_position(sticker, 1000, 1000, 100, 100)
-        assert abs(x - 250) < 1  # 300 - 50 = 250
-        assert abs(y - 650) < 1  # 700 - 50 = 650
-
-    def test_custom_position_pixel(self):
-        """自定义像素位置."""
-        sticker = ImageStickerConfig(
-            position="center",
-            x=100.0,
-            y=200.0,
-            x_unit="pixel",
-            y_unit="pixel",
-        )
-        x, y = StickerEngine._resolve_position(sticker, 1000, 1000, 50, 50)
-        assert abs(x - 75) < 1  # 100 - 25 = 75
-        assert abs(y - 175) < 1  # 200 - 25 = 175
-
-    def test_position_clamped(self):
-        """位置钳制在画布内."""
-        sticker = ImageStickerConfig(
-            position="center",
-            x=-10.0,
-            y=-10.0,
-            x_unit="pixel",
-            y_unit="pixel",
-        )
-        x, y = StickerEngine._resolve_position(sticker, 1000, 1000, 50, 50)
-        assert x >= 0
-        assert y >= 0
-
-
-class TestTextSticker:
-    """文字贴纸测试."""
-
-    def test_drawtext_filter_basic(self):
-        """基础文字贴纸滤镜生成."""
-        sticker = TextStickerConfig(
-            enabled=True,
-            text="Hello World",
-            font_size=36,
-            font_color="#FFFFFF",
-            position="center",
-        )
-        f = StickerEngine._build_drawtext_filter(sticker, "[in]", "[out]", 1080, 1920)
-        assert "drawtext" in f
-        assert "Hello World" in f
-        assert "fontsize=36" in f
-        assert "[in]" in f
-        assert "[out]" in f
-
-    def test_drawtext_with_stroke(self):
-        """带描边的文字贴纸."""
-        sticker = TextStickerConfig(
-            enabled=True,
-            text="Test",
-            stroke_width=3,
-            stroke_color="#FF0000",
-        )
-        f = StickerEngine._build_drawtext_filter(sticker, "[in]", "[out]", 1080, 1920)
-        assert "borderw=3" in f
-        assert "bordercolor=#FF0000" in f
-
-    def test_drawtext_with_shadow(self):
-        """带阴影的文字贴纸."""
-        sticker = TextStickerConfig(
-            enabled=True,
-            text="Shadow",
-            shadow_x=4,
-            shadow_y=4,
-            shadow_alpha=0.5,
-        )
-        f = StickerEngine._build_drawtext_filter(sticker, "[in]", "[out]", 1080, 1920)
-        assert "shadowx=4" in f
-        assert "shadowy=4" in f
-
-    def test_drawtext_time_range(self):
-        """带时间范围的文字贴纸."""
-        sticker = TextStickerConfig(
-            enabled=True,
-            text="Timed",
-            start_time=2.0,
-            duration=3.0,
-        )
-        f = StickerEngine._build_drawtext_filter(sticker, "[in]", "[out]", 1080, 1920)
-        assert "enable='between(t,2.0,5.0)'" in f
-
-    def test_drawtext_empty_text(self):
-        """空文字直通."""
-        sticker = TextStickerConfig(enabled=True, text="")
-        f = StickerEngine._build_drawtext_filter(sticker, "[in]", "[out]", 1080, 1920)
-        assert "[in]copy[out]" in f
-
-    def test_drawtext_with_fade(self):
-        """带淡入淡出的文字贴纸."""
-        sticker = TextStickerConfig(
-            enabled=True,
-            text="Fade",
-            start_time=1.0,
-            duration=5.0,
-            fade_in=0.5,
-            fade_out=0.5,
-        )
-        f = StickerEngine._build_drawtext_filter(sticker, "[in]", "[out]", 1080, 1920)
-        assert "alpha=" in f
-
-
-class TestImageSticker:
-    """图片贴纸测试."""
-
-    def test_image_sticker_overlay(self, sample_image):
-        """图片贴纸 overlay 滤镜生成."""
-        result = StickerEngine.build_sticker_chain(
-            stickers=[
-                {
-                    "type": "image",
-                    "image_path": str(sample_image),
-                    "position": "top_right",
-                    "scale": 0.5,
-                    "opacity": 0.8,
-                    "z_index": 10,
-                }
-            ],
-            input_label="[base]",
-            output_label="[final]",
-            canvas_w=1080,
-            canvas_h=1920,
-        )
-        assert result.filter_str != ""
-        assert "overlay" in result.filter_str
-        assert len(result.extra_inputs) == 1
-        assert result.extra_inputs[0] == str(sample_image)
-
-    def test_image_sticker_missing_file(self):
-        """图片贴纸素材不存在时跳过."""
-        result = StickerEngine.build_sticker_chain(
-            stickers=[
-                {
-                    "type": "image",
-                    "image_path": "/nonexistent/image.png",
-                    "position": "center",
-                }
-            ],
-            input_label="[in]",
-            output_label="[out]",
-            canvas_w=1080,
-            canvas_h=1920,
-        )
-        # 素材不存在，跳过，返回直通
-        assert "[in]copy[out]" in result.filter_str
-        assert len(result.extra_inputs) == 0
-
-    def test_mixed_stickers(self, sample_image):
-        """混合贴纸：图片 + 文字."""
-        result = StickerEngine.build_sticker_chain(
-            stickers=[
-                {
-                    "type": "image",
-                    "image_path": str(sample_image),
-                    "position": "top_left",
-                    "z_index": 5,
-                },
-                {
-                    "type": "text",
-                    "text": "Hello",
-                    "position": "bottom_center",
-                    "z_index": 10,
-                },
-            ],
-            input_label="[in]",
-            output_label="[out]",
-            canvas_w=1080,
-            canvas_h=1920,
-        )
-        assert "overlay" in result.filter_str
-        assert "drawtext" in result.filter_str
-        assert len(result.extra_inputs) == 1
-
-    def test_sticker_z_index_order(self, sample_image):
-        """贴纸按 z_index 排序."""
-        result = StickerEngine.build_sticker_chain(
-            stickers=[
-                {"type": "text", "text": "Top", "z_index": 20, "position": "center"},
-                {"type": "text", "text": "Bottom", "z_index": 5, "position": "center"},
-            ],
-            input_label="[in]",
-            output_label="[out]",
-            canvas_w=1080,
-            canvas_h=1920,
-        )
-        # z_index 小的先叠加，大的后叠加（在上面）
-        assert result.filter_str.count("drawtext") == 2
-
-    def test_empty_stickers(self):
-        """空贴纸列表."""
-        result = StickerEngine.build_sticker_chain(
-            stickers=[],
-            input_label="[in]",
-            output_label="[out]",
-            canvas_w=1080,
-            canvas_h=1920,
-        )
-        assert "[in]copy[out]" in result.filter_str
-        assert result.extra_inputs == []
-
-    def test_invalid_sticker_skipped(self):
-        """无效贴纸配置跳过."""
-        result = StickerEngine.build_sticker_chain(
-            stickers=[{"invalid": "data"}],
-            input_label="[in]",
-            output_label="[out]",
-            canvas_w=1080,
-            canvas_h=1920,
-        )
-        # 解析失败，跳过，直通
-        assert "[in]copy[out]" in result.filter_str
-
-
-class TestStickerHelpers:
-    """贴纸辅助函数测试."""
-
-    def test_parse_stickers_empty(self):
-        """空配置解析."""
-        assert parse_stickers_from_config(None) == []
-        assert parse_stickers_from_config({}) == []
-
-    def test_parse_stickers_list(self):
-        """正常贴纸列表解析."""
-        config = {"stickers": [{"type": "text", "text": "A"}, {"type": "text", "text": "B"}]}
-        result = parse_stickers_from_config(config)
-        assert len(result) == 2
-
-    def test_parse_stickers_not_list(self):
-        """非列表类型返回空."""
-        config = {"stickers": "not a list"}
-        assert parse_stickers_from_config(config) == []
-
-    def test_get_categories(self):
-        """贴纸分类列表."""
-        cats = get_sticker_categories()
-        assert len(cats) == len(STICKER_CATEGORIES)
-        assert cats[0][0] == "emoji"
-
-
-# ═══════════════════════════════════════════════════════════════════════════════
-# 三、封面生成器测试
-# ═══════════════════════════════════════════════════════════════════════════════
-
-
-class TestCoverGenerator:
-    """CoverGenerator 测试."""
-
-    def test_default_dimensions(self):
-        """默认封面尺寸."""
-        assert DEFAULT_COVER_WIDTH == 1080
-        assert DEFAULT_COVER_HEIGHT == 1920
-
-    @patch("video_processing.cover_generator.run_ffmpeg")
-    @patch("video_processing.cover_generator.probe_video_info")
-    def test_extract_frame_basic(self, mock_probe, mock_run, sample_video, tmp_path):
-        """基础抽帧测试."""
-        mock_probe.return_value = {"duration": 30.0}
-
-        # mock run_ffmpeg 实际创建输出文件
-        def fake_run_ffmpeg(cmd):
-            # 找到输出路径并创建文件
-            output_path = Path(cmd[-1])
-            output_path.parent.mkdir(parents=True, exist_ok=True)
-            output_path.write_bytes(b"fake jpeg data")
-
-        mock_run.side_effect = fake_run_ffmpeg
-
-        output = tmp_path / "cover.jpg"
-        result = CoverGenerator.extract_frame(
-            sample_video,
-            output,
-            time_sec=2.0,
-        )
-
-        assert result == output
-        mock_run.assert_called_once()
-        # 检查命令参数
-        cmd = mock_run.call_args[0][0]
-        assert "-ss" in cmd
-        assert "2.000" in cmd
-        assert "-vframes" in cmd
-        assert "1" in cmd
-
-    @patch("video_processing.cover_generator.run_ffmpeg")
-    @patch("video_processing.cover_generator.probe_video_info")
-    def test_extract_frame_time_clamped(self, mock_probe, mock_run, sample_video, tmp_path):
-        """抽帧时间超过视频长度时钳制."""
-        mock_probe.return_value = {"duration": 10.0}
-
-        def fake_run_ffmpeg(cmd):
-            output_path = Path(cmd[-1])
-            output_path.parent.mkdir(parents=True, exist_ok=True)
-            output_path.write_bytes(b"fake jpeg data")
-
-        mock_run.side_effect = fake_run_ffmpeg
-
-        output = tmp_path / "cover.jpg"
-        CoverGenerator.extract_frame(
-            sample_video,
-            output,
-            time_sec=100.0,  # 超过视频时长
-        )
-
-        cmd = mock_run.call_args[0][0]
-        ss_idx = cmd.index("-ss")
-        time_val = float(cmd[ss_idx + 1])
-        # 应该被钳制到中间帧（5秒左右）
-        assert time_val <= 10.0
-
-    @patch("video_processing.cover_generator.run_ffmpeg")
-    @patch("video_processing.cover_generator.probe_video_info")
-    def test_extract_frame_negative_time(self, mock_probe, mock_run, sample_video, tmp_path):
-        """负时间钳制到0."""
-        mock_probe.return_value = {"duration": 30.0}
-
-        def fake_run_ffmpeg(cmd):
-            output_path = Path(cmd[-1])
-            output_path.parent.mkdir(parents=True, exist_ok=True)
-            output_path.write_bytes(b"fake jpeg data")
-
-        mock_run.side_effect = fake_run_ffmpeg
-
-        output = tmp_path / "cover.jpg"
-        CoverGenerator.extract_frame(
-            sample_video,
-            output,
-            time_sec=-5.0,
-        )
-
-        cmd = mock_run.call_args[0][0]
-        ss_idx = cmd.index("-ss")
-        time_val = float(cmd[ss_idx + 1])
-        assert time_val >= 0
-
-    def test_extract_frame_file_not_found(self, tmp_path):
-        """视频文件不存在抛异常."""
-        with pytest.raises(FileNotFoundError):
-            CoverGenerator.extract_frame(
-                "/nonexistent/video.mp4",
-                tmp_path / "cover.jpg",
-            )
-
-    @patch("video_processing.cover_generator.CoverGenerator.extract_frame")
-    @patch("video_processing.cover_generator.probe_video_info")
-    def test_smart_cover_3_frames(self, mock_probe, mock_extract, sample_video, tmp_path):
-        """智能封面抽取3帧选最佳."""
-        mock_probe.return_value = {"duration": 30.0}
-
-        # 创建三个大小不同的临时文件（模拟清晰度不同）
-        def create_frame(video_path, output_path, **kwargs):
-            # 第二帧最大（最清晰）
-            p = Path(output_path)
-            p.parent.mkdir(parents=True, exist_ok=True)
-            if "candidate_1" in str(p):
-                p.write_bytes(b"x" * 10000)  # 最大 = 最清晰
-            elif "candidate_0" in str(p):
-                p.write_bytes(b"x" * 1000)
-            else:
-                p.write_bytes(b"x" * 5000)
-            return p
-
-        mock_extract.side_effect = create_frame
-
-        output = tmp_path / "smart_cover.jpg"
-        result = CoverGenerator.extract_smart_cover(
-            sample_video,
-            output,
-            frame_count=3,
-        )
-
-        assert result == output
-        assert output.exists()
-        # 应该选最大的那个文件（candidate_1）
-        assert output.stat().st_size == 10000
-
-    @patch("video_processing.cover_generator.CoverGenerator.extract_frame")
-    @patch("video_processing.cover_generator.probe_video_info")
-    def test_smart_cover_fallback(self, mock_probe, mock_extract, sample_video, tmp_path):
-        """智能封面全部失败时降级."""
-        mock_probe.return_value = {"duration": 0.0}  # 时长为0
-
-        output = tmp_path / "cover.jpg"
-        output.write_bytes(b"x" * 100)
-        mock_extract.return_value = output
-
-        result = CoverGenerator.extract_smart_cover(sample_video, output, frame_count=3)
-        assert result == output
-
-    @patch("video_processing.cover_generator.run_ffmpeg")
-    def test_custom_cover(self, mock_run, sample_image, tmp_path):
-        """自定义封面处理."""
-        output = tmp_path / "custom_cover.jpg"
-
-        result = CoverGenerator.process_custom_cover(
-            sample_image,
-            output,
-        )
-
-        assert result == output
-        mock_run.assert_called_once()
-        cmd = mock_run.call_args[0][0]
-        assert str(sample_image) in cmd
-
-    def test_custom_cover_not_found(self, tmp_path):
-        """自定义封面文件不存在."""
-        with pytest.raises(FileNotFoundError):
-            CoverGenerator.process_custom_cover(
-                "/nonexistent/img.png",
-                tmp_path / "cover.jpg",
-            )
-
-    @patch("video_processing.cover_generator.CoverGenerator.extract_frame")
-    def test_generate_cover_time_mode(self, mock_extract, sample_video, tmp_path):
-        """统一入口 - time 模式."""
-        output = tmp_path / "cover.jpg"
-        mock_extract.return_value = output
-
-        result = CoverGenerator.generate_cover(
-            sample_video,
-            output,
-            mode="time",
-            time_sec=3.0,
-        )
-
-        assert result == output
-        mock_extract.assert_called_once()
-
-    @patch("video_processing.cover_generator.CoverGenerator.extract_smart_cover")
-    def test_generate_cover_smart_mode(self, mock_smart, sample_video, tmp_path):
-        """统一入口 - smart 模式."""
-        output = tmp_path / "cover.jpg"
-        mock_smart.return_value = output
-
-        result = CoverGenerator.generate_cover(
-            sample_video,
-            output,
-            mode="smart",
-        )
-
-        assert result == output
-        mock_smart.assert_called_once()
-
-    @patch("video_processing.cover_generator.CoverGenerator.process_custom_cover")
-    def test_generate_cover_custom_mode(self, mock_custom, sample_video, sample_image, tmp_path):
-        """统一入口 - custom 模式."""
-        output = tmp_path / "cover.jpg"
-        mock_custom.return_value = output
-
-        result = CoverGenerator.generate_cover(
-            sample_video,
-            output,
-            mode="custom",
-            custom_image=sample_image,
-        )
-
-        assert result == output
-        mock_custom.assert_called_once()
-
-
-class TestGenerateCoverFromPlan:
-    """从 plan 配置生成封面测试."""
-
-    @patch("video_processing.cover_generator.CoverGenerator.extract_smart_cover")
-    def test_smart_mode_from_plan(self, mock_smart, sample_video, tmp_path):
-        """plan 配置 smart 模式."""
-        plan = FakePlan(id="plan_001", config={"cover_config": {"mode": "smart"}})
-        mock_smart.return_value = tmp_path / "cover.jpg"
-        (tmp_path / "cover.jpg").write_bytes(b"test")
-
-        result = generate_cover_from_plan(plan, sample_video, tmp_path)
-        assert result is not None
-
-    def test_no_cover_config(self, sample_video, tmp_path):
-        """没有封面配置时返回 None."""
-        plan = FakePlan(id="plan_001", config={})
-        result = generate_cover_from_plan(plan, sample_video, tmp_path)
-        assert result is None
-
-    def test_none_config(self, sample_video, tmp_path):
-        """config 为 None."""
-        plan = FakePlan(id="plan_001", config=None)  # type: ignore
-        result = generate_cover_from_plan(plan, sample_video, tmp_path)
-        assert result is None
-
-
-# ═══════════════════════════════════════════════════════════════════════════════
-# 四、UnifiedRenderService 集成测试
-# ═══════════════════════════════════════════════════════════════════════════════
-
-
-def _make_clip(clip_id="c1", asset_id="a1", path=Path("/fake/video.mp4"), clip_type="main", config=None):
-    """创建测试用 ResolvedClip."""
-    return ResolvedClip(
-        clip_id=clip_id,
-        asset_id=asset_id,
-        local_path=path,
-        clip_type=clip_type,
-        order=0,
-        start_time=0.0,
-        duration=0.0,
-        transition_effect="cut",
-        config=config or {},
-        actual_duration=10.0,
-    )
-
-
-def _make_service(plan, clips, asset_path_map=None, work_dir=None, tmp_path=None):
-    """创建测试用 UnifiedRenderService."""
-    from pathlib import Path as P
-
-    work_dir = work_dir or (tmp_path or P("/tmp")) / "render_test"
-    work_dir.mkdir(exist_ok=True, parents=True)
-    return UnifiedRenderService(
-        plan=plan,
-        clips=clips,
-        asset_path_map=asset_path_map or {},
-        work_dir=work_dir,
-        output_width=1080,
-        output_height=1920,
-        output_fps=30,
-        transition_duration=0.5,
-    )
-
-
-class TestReverseIntegration:
-    """倒放功能集成测试."""
-
-    @patch("video_processing.unified_render_service.probe_video_info")
-    @patch("video_processing.unified_render_service.run_ffmpeg")
-    def test_reverse_in_filter_complex(self, mock_run, mock_probe, tmp_path):
-        """filter_complex 路径中包含倒放滤镜."""
-        mock_probe.return_value = {"duration": 10.0, "has_audio": True, "width": 1920, "height": 1080}
-        mock_run.return_value = None
-
-        plan = FakePlan(id="p1")
-        clip = _make_clip(config={"reverse": {"enabled": True}})
-        clip.actual_duration = 5.0
-        # 两个 clip 触发 filter_complex 路径
-        clip2 = _make_clip(clip_id="c2", config={})
-        clip2.actual_duration = 5.0
-        clip2.order = 1
-
-        service = _make_service(plan, [clip, clip2], tmp_path=tmp_path)
-
-        # 直接测 _build_filter_complex
-        from video_processing.unified_render_service import RenderLayer
-
-        layer = RenderLayer(role="main", clips=[clip, clip2])
-        filter_str, inputs = service._build_filter_complex([layer])
-
-        assert "reverse" in filter_str
-
-    def test_can_use_pass_through_with_reverse(self, tmp_path):
-        """倒放不影响直通模式判断（只有贴纸才禁用）."""
-        plan = FakePlan(id="p1")
-        clip = _make_clip(config={"reverse": {"enabled": True}})
-        clip.actual_duration = 5.0
-
-        service = _make_service(plan, [clip], tmp_path=tmp_path)
-        from video_processing.unified_render_service import RenderLayer
-
-        layer = RenderLayer(role="main", clips=[clip])
-        layers = [layer]
-
-        assert service._can_use_pass_through(layers) is True
-
-
-class TestStickerIntegration:
-    """贴纸功能集成测试."""
-
-    def test_can_use_pass_through_with_stickers(self, tmp_path):
-        """有贴纸时禁用直通模式."""
-        plan = FakePlan(id="p1", config={"stickers": [{"type": "text", "text": "Hello", "position": "center"}]})
-        clip = _make_clip()
-        clip.actual_duration = 5.0
-
-        service = _make_service(plan, [clip], tmp_path=tmp_path)
-        from video_processing.unified_render_service import RenderLayer
-
-        layer = RenderLayer(role="main", clips=[clip])
-        layers = [layer]
-
-        assert service._can_use_pass_through(layers) is False
-
-    def test_can_use_pass_through_no_stickers(self, tmp_path):
-        """无贴纸时直通模式正常."""
-        plan = FakePlan(id="p1", config={})
-        clip = _make_clip()
-        clip.actual_duration = 5.0
-
-        service = _make_service(plan, [clip], tmp_path=tmp_path)
-        from video_processing.unified_render_service import RenderLayer
-
-        layer = RenderLayer(role="main", clips=[clip])
-        layers = [layer]
-
-        assert service._can_use_pass_through(layers) is True
-
-    def test_build_sticker_filters_text(self, tmp_path):
-        """文字贴纸滤镜构建."""
-        plan = FakePlan(
-            id="p1", config={"stickers": [{"type": "text", "text": "Hello", "position": "top_center", "z_index": 10}]}
-        )
-        service = _make_service(plan, [], tmp_path=tmp_path)
-
-        filter_str, extra_inputs = service._build_sticker_filters("in", "out")
-
-        assert "drawtext" in filter_str
-        assert len(extra_inputs) == 0
-
-    def test_build_sticker_filters_empty(self, tmp_path):
-        """无贴纸返回空."""
-        plan = FakePlan(id="p1", config={})
-        service = _make_service(plan, [], tmp_path=tmp_path)
-
-        filter_str, extra_inputs = service._build_sticker_filters("in", "out")
-
-        assert filter_str == ""
-        assert extra_inputs == []
-
-    def test_build_sticker_filters_image(self, sample_image, tmp_path):
-        """图片贴纸滤镜构建 + 额外输入."""
-        plan = FakePlan(
-            id="p1",
-            config={
-                "stickers": [
-                    {
-                        "type": "image",
-                        "image_path": str(sample_image),
-                        "position": "bottom_right",
-                        "z_index": 5,
-                    }
-                ]
-            },
-        )
-        service = _make_service(plan, [], tmp_path=tmp_path)
-
-        filter_str, extra_inputs = service._build_sticker_filters("in", "out")
-
-        assert "overlay" in filter_str
-        assert len(extra_inputs) == 1

tests/unit/test_edit_plan_generation_api.py

@@ -193,70 +193,6 @@ class StubGenerationTaskRepository:
         items.sort(key=lambda t: t.created_at, reverse=True)
         return items[:limit]
 
-    def list_by_user_filtered(
-        self,
-        user_id: str,
-        *,
-        status: str | None = None,
-        limit: int | None = None,
-        offset: int = 0,
-    ) -> list:
-        """按用户+状态筛选任务列表（stub实现）。"""
-        items = [t for t in self._store.values() if t.created_by_user_id == user_id]
-        if status:
-            items = [t for t in items if str(t.status) == status]
-        # 按创建时间倒序
-        items.sort(key=lambda t: t.created_at or "", reverse=True)
-        if offset:
-            items = items[offset:]
-        if limit is not None:
-            items = items[:limit]
-        return items
-
-    def count_by_user_filtered(
-        self,
-        user_id: str,
-        *,
-        status: str | None = None,
-    ) -> int:
-        """按用户+状态筛选计数（stub实现）。"""
-        items = [t for t in self._store.values() if t.created_by_user_id == user_id]
-        if status:
-            items = [t for t in items if str(t.status) == status]
-        return len(items)
-
-    def list_by_project_filtered(
-        self,
-        project_id: str,
-        *,
-        status: str | None = None,
-        limit: int | None = None,
-        offset: int = 0,
-    ) -> list:
-        """按项目+状态筛选任务列表（stub实现）。"""
-        items = [t for t in self._store.values() if t.project_id == project_id]
-        if status:
-            items = [t for t in items if str(t.status) == status]
-        # 按创建时间倒序
-        items.sort(key=lambda t: t.created_at or "", reverse=True)
-        if offset:
-            items = items[offset:]
-        if limit is not None:
-            items = items[:limit]
-        return items
-
-    def count_by_project_filtered(
-        self,
-        project_id: str,
-        *,
-        status: str | None = None,
-    ) -> int:
-        """按项目+状态筛选计数（stub实现）。"""
-        items = [t for t in self._store.values() if t.project_id == project_id]
-        if status:
-            items = [t for t in items if str(t.status) == status]
-        return len(items)
-
 
 # ── Fixtures ──────────────────────────────────────────────────────────────────
 

tests/unit/test_edit_plan_service.py

@@ -200,70 +200,6 @@ class StubGenerationTaskRepository:
     def count_pending_total(self) -> int:
         return 0
 
-    def list_by_user_filtered(
-        self,
-        user_id: str,
-        *,
-        status: str | None = None,
-        limit: int | None = None,
-        offset: int = 0,
-    ) -> list:
-        """按用户+状态筛选任务列表（stub实现）。"""
-        items = [t for t in self._tasks.values() if t.created_by_user_id == user_id]
-        if status:
-            items = [t for t in items if str(t.status) == status]
-        # 按创建时间倒序
-        items.sort(key=lambda t: t.created_at or "", reverse=True)
-        if offset:
-            items = items[offset:]
-        if limit is not None:
-            items = items[:limit]
-        return items
-
-    def count_by_user_filtered(
-        self,
-        user_id: str,
-        *,
-        status: str | None = None,
-    ) -> int:
-        """按用户+状态筛选计数（stub实现）。"""
-        items = [t for t in self._tasks.values() if t.created_by_user_id == user_id]
-        if status:
-            items = [t for t in items if str(t.status) == status]
-        return len(items)
-
-    def list_by_project_filtered(
-        self,
-        project_id: str,
-        *,
-        status: str | None = None,
-        limit: int | None = None,
-        offset: int = 0,
-    ) -> list:
-        """按项目+状态筛选任务列表（stub实现）。"""
-        items = [t for t in self._tasks.values() if t.project_id == project_id]
-        if status:
-            items = [t for t in items if str(t.status) == status]
-        # 按创建时间倒序
-        items.sort(key=lambda t: t.created_at or "", reverse=True)
-        if offset:
-            items = items[offset:]
-        if limit is not None:
-            items = items[:limit]
-        return items
-
-    def count_by_project_filtered(
-        self,
-        project_id: str,
-        *,
-        status: str | None = None,
-    ) -> int:
-        """按项目+状态筛选计数（stub实现）。"""
-        items = [t for t in self._tasks.values() if t.project_id == project_id]
-        if status:
-            items = [t for t in items if str(t.status) == status]
-        return len(items)
-
 
 # ---------------------------------------------------------------------------
 # Service factory

tests/unit/test_generation_presigned_url.py

@@ -64,38 +64,6 @@ class StubGenerationTaskRepository:
     def count_pending_total(self):
         return 0
 
-    def list_by_user_filtered(self, user_id, *, status=None, limit=None, offset=0):
-        items = [t for t in self._tasks.values() if getattr(t, "created_by_user_id", None) == user_id]
-        if status:
-            items = [t for t in items if getattr(t, "status", None) == status]
-        if offset:
-            items = items[offset:]
-        if limit is not None:
-            items = items[:limit]
-        return items
-
-    def count_by_user_filtered(self, user_id, *, status=None):
-        items = [t for t in self._tasks.values() if getattr(t, "created_by_user_id", None) == user_id]
-        if status:
-            items = [t for t in items if getattr(t, "status", None) == status]
-        return len(items)
-
-    def list_by_project_filtered(self, project_id, *, status=None, limit=None, offset=0):
-        items = [t for t in self._tasks.values() if getattr(t, "project_id", None) == project_id]
-        if status:
-            items = [t for t in items if getattr(t, "status", None) == status]
-        if offset:
-            items = items[offset:]
-        if limit is not None:
-            items = items[:limit]
-        return items
-
-    def count_by_project_filtered(self, project_id, *, status=None):
-        items = [t for t in self._tasks.values() if getattr(t, "project_id", None) == project_id]
-        if status:
-            items = [t for t in items if getattr(t, "status", None) == status]
-        return len(items)
-
 
 class StubGeneratedVideoRepository:
     def __init__(self, videos=None):

tests/unit/test_speed_engine.py

@@ -1,269 +0,0 @@
-"""视频调速引擎单元测试."""
-
-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