feat(ci): 新增发布与灰度部署脚本

- release.sh: 一键发布脚本（打tag + 生成changelog + 灰度部署）
- gray_deploy.sh: 灰度发布脚本（Nginx权重调整 + canary容器）
- rollback.sh: 灰度回滚脚本（切回稳定版本流量）

alembic/versions/039_add_transition_duration.py

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

alembic/versions/040_add_playback_speed_to_edit_plan_clips.py

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

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

@@ -146,6 +146,7 @@ class AIRecommendClipItem(BaseModel):
     text_content: str = Field(default="", description="文字内容")
     duration: float = Field(..., ge=0.0, description="片段时长（秒）")
     transition_effect: str = Field(default="cut", description="转场效果")
+    transition_duration: float = Field(default=0.0, ge=0.0, description="转场时长（秒），0 表示使用默认值")
     asset_id: str = Field(default="", description="关联素材 ID")
     start_time: float = Field(default=0.0, ge=0.0, description="素材截取起始时间（秒）")
     config: dict[str, Any] = Field(default_factory=dict, description="片段额外配置")
@@ -209,6 +210,8 @@ class _PlanClipItem(BaseModel):
     start_time: float
     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/api/routes/edit_plans_timeline.py

@@ -210,6 +210,7 @@ def generate_from_template(
                 start_time=c.start_time,
                 duration=c.duration,
                 transition_effect=c.transition_effect,
+                transition_duration=c.transition_duration,
                 status=c.status.value if hasattr(c.status, "value") else c.status,
                 config=c.config,
                 created_at=c.created_at,

apps/api/app/services/edit_plan_service.py

@@ -281,6 +281,8 @@ class EditPlanService:
         start_time: float = 0.0,
         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:
         """创建片段
@@ -301,6 +303,8 @@ class EditPlanService:
             start_time=start_time,
             duration=duration,
             transition_effect=transition_effect,
+            transition_duration=transition_duration,
+            playback_speed=playback_speed,
             config=config,
         )
         created = self._clip_repo.create(clip)
@@ -324,6 +328,8 @@ class EditPlanService:
         start_time: Optional[float] = None,
         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:
         """更新片段
@@ -333,6 +339,15 @@ class EditPlanService:
         """
         existing = self.get_clip_or_raise(clip_id)
 
+        # 速度边界钳制
+        if playback_speed is not None:
+            if playback_speed <= 0:
+                playback_speed = 1.0
+            elif playback_speed < 0.25:
+                playback_speed = 0.25
+            elif playback_speed > 4.0:
+                playback_speed = 4.0
+
         updated = EditPlanClip(
             id=existing.id,
             plan_id=existing.plan_id,
@@ -346,6 +361,10 @@ class EditPlanService:
             transition_effect=(
                 transition_effect.strip() if transition_effect is not None else existing.transition_effect
             ),
+            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

@@ -0,0 +1,431 @@
+"""视频封面生成器 — 从视频中提取/生成封面图.
+
+支持能力：
+- 指定时间点抽帧（默认第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/ffmpeg_utils.py

@@ -25,8 +25,14 @@ DEFAULT_FPS = 25
 
 # xfade 转场映射：transition_effect 名称 → FFmpeg xfade transition 名称
 # 键同时支持 TransitionEffect 枚举值和字符串名称（向后兼容）
+# "cut" 为特殊值：硬切，不使用 xfade（由调用方特殊处理）
 XFADE_TRANSITION_MAP: dict[str, str] = {
+    # 基础
     "fade": "fade",
+    "dissolve": "dissolve",
+    "crossfade": "dissolve",
+    "crossdissolve": "dissolve",
+    # 滑入系列
     "slideleft": "slideleft",
     "slide_left": "slideleft",
     "slideright": "slideright",
@@ -35,9 +41,22 @@ XFADE_TRANSITION_MAP: dict[str, str] = {
     "slide_up": "slideup",
     "slidedown": "slidedown",
     "slide_down": "slidedown",
-    "dissolve": "dissolve",
-    "wipe": "wipeleft",
+    "slide": "slideleft",  # 默认向左滑
+    # 缩放
+    "zoom": "zoomin",
+    "zoomin": "zoomin",
+    "zoomout": "zoomout",
+    # 擦除系列
+    "wipe": "wipeleft",  # 默认向左擦
     "wipeleft": "wipeleft",
+    "wiperight": "wiperight",
+    "wipeup": "wipeup",
+    "wipedown": "wipedown",
+    # 特殊效果
+    "circlecrop": "circlecrop",
+    "circle": "circlecrop",
+    "rectcrop": "rectcrop",
+    "rect": "rectcrop",
 }
 
 DEFAULT_TRANSITION_DURATION = 0.5

apps/worker/video_processing/render_audio.py

@@ -22,6 +22,8 @@ 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
@@ -224,45 +226,129 @@ def concat_main_audio(
         clip = clips[0]
         effective_duration = clip_effective_duration(clip)
         trim_start = getattr(clip, "start_time", 0) or 0
-        # 最终时长：取 clip 有效时长和视频总时长的较小值
-        # （视频总时长由主图层决定，但单 clip 场景下两者应该一致，仍做保护）
-        final_duration = effective_duration
+        speed = getattr(clip, "playback_speed", 1.0) or 1.0
+        if not isinstance(speed, (int, float)) or speed <= 0:
+            speed = 1.0
+
+        # 调速后时长
+        adjusted_duration = effective_duration / speed if abs(speed - 1.0) >= 1e-6 else effective_duration
+        # 最终时长：取调速后时长和视频总时长的较小值
+        final_duration = adjusted_duration
         if video_duration > 0 and (final_duration <= 0 or final_duration > video_duration):
             final_duration = video_duration
 
-        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)
+        # 音频倒放
+        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)
         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:
-            filter_parts.append(
-                f"[{i}:a]atrim=start={trim_start:.3f}:duration={effective_duration:.3f}," f"asetpts=PTS-STARTPTS[a{i}]"
-            )
+            audio_filters.append(f"atrim=start={trim_start:.3f}:duration={effective_duration:.3f}")
+            audio_filters.append("asetpts=PTS-STARTPTS")
+
+        # 音频调速 — atempo 多级串联
+        if abs(speed - 1.0) >= 1e-6:
+            from video_processing.speed_engine import SpeedConfig
+
+            config = SpeedConfig(speed=float(speed))
+            config.clamp()
+            atempo_filter = speed_engine.build_audio_filter(config)
+            if atempo_filter:
+                audio_filters.append(atempo_filter)
         else:
-            filter_parts.append(f"[{i}:a]asetpts=PTS-STARTPTS[a{i}]")
+            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}]")
 
     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

@@ -0,0 +1,116 @@
+"""视频倒放引擎 — 基于 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

@@ -0,0 +1,167 @@
+"""视频调速引擎 — 基于 FFmpeg setpts + atempo 的速度调整能力。
+
+支持：
+- 0.25x ~ 4x 变速范围
+- 视频调速（setpts）
+- 音频调速（atempo，多级串联处理超范围值）
+- 音调修正（pitch_correct，默认开启）
+- 边界自动钳制，不阻断渲染
+"""
+
+from dataclasses import dataclass
+from typing import Optional
+
+# ─── 常量 ───────────────────────────────────────────────
+MIN_SPEED = 0.25
+MAX_SPEED = 4.0
+DEFAULT_SPEED = 1.0
+
+# atempo 单级有效范围
+_ATEMPO_MIN = 0.5
+_ATEMPO_MAX = 2.0
+
+
+@dataclass
+class SpeedConfig:
+    """调速配置。
+
+    Attributes:
+        speed: 播放速度，0.25~4.0，1.0 为原速
+        pitch_correct: 是否保持音调（默认 True，用 atempo 时间拉伸算法）
+    """
+
+    speed: float = DEFAULT_SPEED
+    pitch_correct: bool = True
+
+    @classmethod
+    def parse(cls, data: Optional[dict]) -> "SpeedConfig":
+        """从 dict 解析配置，无效值回退到默认。"""
+        if not data or not isinstance(data, dict):
+            return cls()
+
+        speed = data.get("speed", DEFAULT_SPEED)
+        if not isinstance(speed, (int, float)):
+            speed = DEFAULT_SPEED
+
+        pitch_correct = data.get("pitch_correct", True)
+        if not isinstance(pitch_correct, bool):
+            pitch_correct = True
+
+        config = cls(speed=float(speed), pitch_correct=pitch_correct)
+        config.clamp()
+        return config
+
+    def clamp(self) -> None:
+        """将速度钳制到合法范围。"""
+        if self.speed <= 0:
+            self.speed = DEFAULT_SPEED
+        elif self.speed < MIN_SPEED:
+            self.speed = MIN_SPEED
+        elif self.speed > MAX_SPEED:
+            self.speed = MAX_SPEED
+
+    @property
+    def is_original(self) -> bool:
+        """是否原速（无需调速）。"""
+        return abs(self.speed - 1.0) < 1e-6
+
+
+class SpeedEngine:
+    """调速引擎 — 生成 FFmpeg 调速滤镜链。
+
+    用法：
+        engine = SpeedEngine()
+        video_filter = engine.build_video_filter(config)
+        audio_filter = engine.build_audio_filter(config)
+        new_duration = engine.adjust_duration(duration, config)
+    """
+
+    def build_video_filter(self, config: SpeedConfig) -> str:
+        """生成视频调速滤镜字符串。
+
+        返回 setpts 滤镜表达式，原速时返回空字符串。
+        """
+        if config.is_original:
+            return ""
+        # setpts=PTS/speed — speed>1 加速，speed<1 减速
+        return f"setpts=PTS/{config.speed:.4f}"
+
+    def build_audio_filter(self, config: SpeedConfig) -> str:
+        """生成音频调速滤镜字符串。
+
+        atempo 单级范围 0.5~2.0，超出范围时自动多级串联：
+        - 0.25x → atempo=0.5,atempo=0.5
+        - 4x → atempo=2.0,atempo=2.0
+        - 0.3x → atempo=0.5,atempo=0.6
+        - 3x → atempo=2.0,atempo=1.5
+
+        原速时返回空字符串。
+        """
+        if config.is_original:
+            return ""
+
+        speed = config.speed
+        stages: list[float] = self._split_atempo_stages(speed)
+        return ",".join(f"atempo={s:.4f}" for s in stages)
+
+    @staticmethod
+    def _split_atempo_stages(speed: float) -> list[float]:
+        """将速度拆分为多级 atempo 串联，每级都在 [0.5, 2.0] 范围内。"""
+        if _ATEMPO_MIN <= speed <= _ATEMPO_MAX:
+            return [speed]
+
+        stages: list[float] = []
+        remaining = speed
+
+        # 加速场景（speed > 2.0）
+        if speed > _ATEMPO_MAX:
+            while remaining > _ATEMPO_MAX:
+                stages.append(_ATEMPO_MAX)
+                remaining /= _ATEMPO_MAX
+            stages.append(remaining)
+
+        # 减速场景（speed < 0.5）
+        else:
+            while remaining < _ATEMPO_MIN:
+                stages.append(_ATEMPO_MIN)
+                remaining /= _ATEMPO_MIN
+            stages.append(remaining)
+
+        return stages
+
+    def adjust_duration(self, original_duration: float, config: SpeedConfig) -> float:
+        """计算调速后的时长。
+
+        加速 → 时长变短；减速 → 时长变长。
+        """
+        if config.is_original or original_duration <= 0:
+            return original_duration
+        return original_duration / config.speed
+
+    def build_clip_speed_filter(
+        self,
+        speed: float,
+        pitch_correct: bool = True,
+    ) -> tuple[str, str, SpeedConfig]:
+        """便捷方法：从单一 speed 值生成视频+音频滤镜。
+
+        返回 (video_filter, audio_filter, config)。
+        """
+        config = SpeedConfig(speed=speed, pitch_correct=pitch_correct)
+        config.clamp()
+        return (
+            self.build_video_filter(config),
+            self.build_audio_filter(config),
+            config,
+        )
+
+    @staticmethod
+    def resolve_clip_speed(
+        clip_config: dict,
+        global_speed: float = DEFAULT_SPEED,
+    ) -> float:
+        """从 clip config 中解析 playback_speed，0 或缺失则使用全局速度。"""
+        speed = clip_config.get("playback_speed", 0) if clip_config else 0
+        if not isinstance(speed, (int, float)) or speed <= 0:
+            return global_speed
+        return float(speed)

apps/worker/video_processing/sticker_engine.py

@@ -0,0 +1,574 @@
+"""贴纸叠加引擎 — 基于 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/transition_engine.py

@@ -0,0 +1,381 @@
+"""转场特效引擎 — Phase 8 智能增强.
+
+基于 FFmpeg xfade 滤镜的统一转场抽象层，提供：
+  1. 转场类型枚举与预设管理
+  2. 转场配置解析与边界校验
+  3. 降级策略（不支持的转场自动 fallback 到硬切）
+  4. xfade 滤镜链构建（封装底层 ffmpeg_utils）
+
+新增转场只需在 TransitionType 中加一项 + 在 XFADE_TRANSITION_MAP 中映射。
+"""
+
+from __future__ import annotations
+
+import logging
+import sys
+from dataclasses import dataclass
+
+if sys.version_info >= (3, 11):
+    from enum import StrEnum
+else:
+    from enum import Enum
+
+    class StrEnum(str, Enum):
+        pass
+
+
+from video_processing.ffmpeg_utils import build_xfade_filter_chain
+
+logger = logging.getLogger(__name__)
+
+
+# ── 常量 ──────────────────────────────────────────────────────────────────────
+
+# 转场时长范围（秒）
+MIN_TRANSITION_DURATION = 0.3
+MAX_TRANSITION_DURATION = 2.0
+DEFAULT_TRANSITION_DURATION = 0.5
+
+# 硬切（无转场）
+CUT_TRANSITION = "cut"
+
+
+# ── 转场类型枚举 ──────────────────────────────────────────────────────────────
+
+
+class TransitionType(StrEnum):
+    """支持的转场效果类型.
+
+    每种类型对应 FFmpeg xfade filter 的一个 transition 值。
+    新增转场只需在此添加一项，并在 _FFMPEG_XFADE_MAP 中映射。
+    """
+
+    # 硬切（无转场效果，直接拼接）
+    CUT = "cut"
+
+    # 淡入淡出（最常用，默认 fallback）
+    FADE = "fade"
+
+    # 溶解（交叉溶解）
+    DISSOLVE = "dissolve"
+
+    # 滑入系列
+    SLIDE_LEFT = "slideleft"
+    SLIDE_RIGHT = "slideright"
+    SLIDE_UP = "slideup"
+    SLIDE_DOWN = "slidedown"
+
+    # 缩放
+    ZOOM = "zoom"
+
+    # 擦除系列
+    WIPE_LEFT = "wipeleft"
+    WIPE_RIGHT = "wiperight"
+    WIPE_UP = "wipeup"
+    WIPE_DOWN = "wipedown"
+
+    # 圆形扩散
+    CIRCLE_CROP = "circlecrop"
+
+    # 矩形覆盖
+    RECT_CROP = "rectcrop"
+
+    @classmethod
+    def all_supported(cls) -> list[str]:
+        """返回所有支持的转场类型名称列表."""
+        return [t.value for t in cls if t != cls.CUT]
+
+    @classmethod
+    def is_supported(cls, name: str) -> bool:
+        """检查转场类型是否支持（不区分大小写和下划线）."""
+        normalized = _normalize_transition_name(name)
+        return normalized in _NAME_TO_ENUM_MAP
+
+
+# ── 名称 → 枚举 映射（支持多种别名）──────────────────────────────────────────
+
+
+def _normalize_transition_name(name: str) -> str:
+    """标准化转场名称：小写 + 去下划线."""
+    return name.lower().replace("_", "").replace("-", "")
+
+
+# 构建别名映射
+_NAME_TO_ENUM_MAP: dict[str, TransitionType] = {}
+for _t in TransitionType:
+    _NAME_TO_ENUM_MAP[_normalize_transition_name(_t.value)] = _t
+
+# 额外的别名
+_ALIASES: dict[str, TransitionType] = {
+    "dissolve": TransitionType.DISSOLVE,
+    "crossfade": TransitionType.DISSOLVE,
+    "crossdissolve": TransitionType.DISSOLVE,
+    "fadein": TransitionType.FADE,
+    "fadeout": TransitionType.FADE,
+    "fadeblack": TransitionType.FADE,
+    "slide": TransitionType.SLIDE_LEFT,  # 默认向左滑
+    "wipe": TransitionType.WIPE_LEFT,  # 默认向左擦
+    "zoomin": TransitionType.ZOOM,
+    "zoomout": TransitionType.ZOOM,
+    "circle": TransitionType.CIRCLE_CROP,
+    "rect": TransitionType.RECT_CROP,
+}
+for _alias, _type in _ALIASES.items():
+    _key = _normalize_transition_name(_alias)
+    if _key not in _NAME_TO_ENUM_MAP:
+        _NAME_TO_ENUM_MAP[_key] = _type
+
+
+# ── TransitionType → FFmpeg xfade transition 名称映射 ─────────────────────────
+
+
+_FFMPEG_XFADE_MAP: dict[TransitionType, str] = {
+    TransitionType.FADE: "fade",
+    TransitionType.DISSOLVE: "dissolve",
+    TransitionType.SLIDE_LEFT: "slideleft",
+    TransitionType.SLIDE_RIGHT: "slideright",
+    TransitionType.SLIDE_UP: "slideup",
+    TransitionType.SLIDE_DOWN: "slidedown",
+    TransitionType.ZOOM: "zoomin",
+    TransitionType.WIPE_LEFT: "wipeleft",
+    TransitionType.WIPE_RIGHT: "wiperight",
+    TransitionType.WIPE_UP: "wipeup",
+    TransitionType.WIPE_DOWN: "wipedown",
+    TransitionType.CIRCLE_CROP: "circlecrop",
+    TransitionType.RECT_CROP: "rectcrop",
+}
+
+
+# ── 转场配置 ──────────────────────────────────────────────────────────────────
+
+
+@dataclass(slots=True)
+class TransitionConfig:
+    """转场效果配置.
+
+    Attributes:
+        effect: 转场效果名称（见 TransitionType）
+        duration: 转场时长（秒），范围 0.3~2.0，默认 0.5
+    """
+
+    effect: str = CUT_TRANSITION
+    duration: float = DEFAULT_TRANSITION_DURATION
+
+    @classmethod
+    def parse(cls, effect: str | None = None, duration: float | None = None) -> "TransitionConfig":
+        """解析并验证转场配置，自动处理边界和降级.
+
+        Args:
+            effect: 转场效果名称（None 或空则使用默认 cut）
+            duration: 转场时长（None 则使用默认值）
+
+        Returns:
+            验证后的 TransitionConfig
+        """
+        # 处理 effect
+        final_effect = CUT_TRANSITION
+        if effect and effect.strip():
+            effect_clean = effect.strip()
+            if TransitionType.is_supported(effect_clean):
+                final_effect = _resolve_transition_enum(effect_clean).value
+            elif effect_clean.lower() == CUT_TRANSITION:
+                final_effect = CUT_TRANSITION
+            else:
+                # 降级：不支持的转场 → 硬切，不阻断渲染
+                logger.warning(
+                    "不支持的转场效果 '%s'，已降级为硬切（cut）",
+                    effect_clean,
+                )
+                final_effect = CUT_TRANSITION
+
+        # 处理 duration：边界钳制
+        final_duration = DEFAULT_TRANSITION_DURATION
+        if duration is not None:
+            try:
+                d = float(duration)
+                if d < MIN_TRANSITION_DURATION:
+                    logger.warning(
+                        "转场时长 %.3fs 小于最小值 %.1fs，已钳制到最小值",
+                        d,
+                        MIN_TRANSITION_DURATION,
+                    )
+                    final_duration = MIN_TRANSITION_DURATION
+                elif d > MAX_TRANSITION_DURATION:
+                    logger.warning(
+                        "转场时长 %.3fs 大于最大值 %.1fs，已钳制到最大值",
+                        d,
+                        MAX_TRANSITION_DURATION,
+                    )
+                    final_duration = MAX_TRANSITION_DURATION
+                else:
+                    final_duration = d
+            except (TypeError, ValueError):
+                logger.warning("无效的转场时长 '%s'，使用默认值 %.1fs", duration, DEFAULT_TRANSITION_DURATION)
+                final_duration = DEFAULT_TRANSITION_DURATION
+
+        return cls(effect=final_effect, duration=final_duration)
+
+    @property
+    def is_cut(self) -> bool:
+        """是否为硬切（无转场效果）."""
+        return self.effect == CUT_TRANSITION
+
+    @property
+    def ffmpeg_transition(self) -> str:
+        """获取对应的 FFmpeg xfade transition 名称."""
+        if self.is_cut:
+            return ""
+        enum_type = _resolve_transition_enum(self.effect)
+        return _FFMPEG_XFADE_MAP.get(enum_type, "fade")
+
+
+def _resolve_transition_enum(name: str) -> TransitionType:
+    """将名称解析为 TransitionType 枚举，必须先通过 is_supported 校验."""
+    normalized = _normalize_transition_name(name)
+    return _NAME_TO_ENUM_MAP.get(normalized, TransitionType.FADE)
+
+
+# ── 转场引擎 ──────────────────────────────────────────────────────────────────
+
+
+class TransitionEngine:
+    """转场特效引擎.
+
+    封装转场配置验证、降级策略和 xfade 滤镜链构建，
+    供 UnifiedRenderService 等上层调用。
+
+    用法::
+
+        engine = TransitionEngine(default_duration=0.5)
+        config = engine.resolve_config("fade", 0.8)
+        filter_str, total_dur = engine.build_xfade_chain(
+            clip_durations=[3.0, 4.0, 5.0],
+            clip_video_labels=["v0", "v1", "v2"],
+            transitions=["cut", "fade", "dissolve"],
+        )
+    """
+
+    def __init__(self, default_duration: float = DEFAULT_TRANSITION_DURATION) -> None:
+        """初始化转场引擎.
+
+        Args:
+            default_duration: 默认转场时长（秒），用于未指定时长的 clip
+        """
+        self._default_duration = default_duration
+
+    def resolve_config(
+        self,
+        effect: str | None = None,
+        duration: float | None = None,
+    ) -> TransitionConfig:
+        """解析单个转场配置，应用验证和降级.
+
+        Args:
+            effect: 转场效果名称
+            duration: 转场时长
+
+        Returns:
+            验证后的 TransitionConfig
+        """
+        # 若未指定 duration，使用引擎默认值
+        dur = duration if duration is not None else self._default_duration
+        return TransitionConfig.parse(effect=effect, duration=dur)
+
+    def resolve_clip_transitions(
+        self,
+        clip_transitions: list[str],
+        clip_durations: list[float] | None = None,
+    ) -> list[TransitionConfig]:
+        """批量解析 clip 级别的转场配置.
+
+        Args:
+            clip_transitions: 每个 clip 的转场效果名称列表
+            clip_durations: 每个 clip 的时长列表（用于验证转场时长不超过片段时长）
+
+        Returns:
+            TransitionConfig 列表
+        """
+        configs: list[TransitionConfig] = []
+        for i, effect in enumerate(clip_transitions):
+            cfg = self.resolve_config(effect=effect)
+            # 额外校验：转场时长不能超过对应 clip 时长的一半（保守限制）
+            if clip_durations and i < len(clip_durations) and not cfg.is_cut:
+                max_safe_duration = max(MIN_TRANSITION_DURATION, clip_durations[i] * 0.5)
+                if cfg.duration > max_safe_duration:
+                    cfg = TransitionConfig(effect=cfg.effect, duration=max_safe_duration)
+            configs.append(cfg)
+        return configs
+
+    def build_xfade_chain(
+        self,
+        clip_durations: list[float],
+        clip_video_labels: list[str],
+        transitions: list[str],
+        *,
+        transition_duration: float | None = None,
+        output_label: str = "outv",
+    ) -> tuple[str, float]:
+        """构建 xfade 转场滤镜链.
+
+        对每步转场应用验证和降级，然后调用底层 ffmpeg_utils 构建。
+
+        Args:
+            clip_durations: 每个片段的时长
+            clip_video_labels: 每个片段的视频流标签
+            transitions: 每个片段对应的转场效果
+            transition_duration: 统一转场时长，None 则使用引擎默认值
+            output_label: 最终输出标签
+
+        Returns:
+            (filter_string, estimated_total_duration)
+        """
+        if len(clip_durations) <= 1:
+            return build_xfade_filter_chain(
+                clip_durations=clip_durations,
+                clip_video_labels=clip_video_labels,
+                transitions=transitions,
+                transition_duration=transition_duration or self._default_duration,
+                output_label=output_label,
+            )
+
+        # 解析所有转场配置
+        resolved = self.resolve_clip_transitions(transitions, clip_durations)
+        resolved_effects = [c.effect for c in resolved]
+
+        # 使用统一的时长（取各转场中最大的时长作为基准，底层会做每步钳制）
+        dur = transition_duration or self._default_duration
+        if not dur:
+            dur = max(c.duration for c in resolved) if resolved else DEFAULT_TRANSITION_DURATION
+
+        # 调用底层构建
+        return build_xfade_filter_chain(
+            clip_durations=clip_durations,
+            clip_video_labels=clip_video_labels,
+            transitions=resolved_effects,
+            transition_duration=dur,
+            output_label=output_label,
+        )
+
+    @staticmethod
+    def supported_transitions() -> list[dict[str, str]]:
+        """获取所有支持的转场效果列表（用于 API 返回给前端）.
+
+        Returns:
+            [{name, display_name, category}, ...]
+        """
+        return [
+            {"name": "cut", "display_name": "硬切", "category": "basic"},
+            {"name": "fade", "display_name": "淡入淡出", "category": "basic"},
+            {"name": "dissolve", "display_name": "溶解", "category": "basic"},
+            {"name": "slideleft", "display_name": "左滑入", "category": "slide"},
+            {"name": "slideright", "display_name": "右滑入", "category": "slide"},
+            {"name": "slideup", "display_name": "上滑入", "category": "slide"},
+            {"name": "slidedown", "display_name": "下滑入", "category": "slide"},
+            {"name": "zoom", "display_name": "缩放", "category": "zoom"},
+            {"name": "wipeleft", "display_name": "左擦除", "category": "wipe"},
+            {"name": "wiperight", "display_name": "右擦除", "category": "wipe"},
+            {"name": "wipeup", "display_name": "上擦除", "category": "wipe"},
+            {"name": "wipedown", "display_name": "下擦除", "category": "wipe"},
+            {"name": "circlecrop", "display_name": "圆形扩散", "category": "special"},
+            {"name": "rectcrop", "display_name": "矩形扩散", "category": "special"},
+        ]

apps/worker/video_processing/unified_render_service.py

@@ -36,7 +36,6 @@ from video_processing.ffmpeg_utils import (
     DEFAULT_OUTPUT_WIDTH,
     DEFAULT_TRANSITION_DURATION,
     FFMPEG_BIN,
-    build_xfade_filter_chain,
     probe_duration,
     probe_video_info,
     run_ffmpeg,
@@ -45,7 +44,11 @@ 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
 from video_processing.tts_engine import TtsEngine
 from video_processing.watermark_engine import WatermarkConfig, WatermarkEngine
@@ -70,6 +73,8 @@ class ResolvedClip:
     start_time: float = 0.0
     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)
 
     # 运行时填充
@@ -182,6 +187,8 @@ class UnifiedRenderService:
         self.transition_duration = transition_duration
         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.
@@ -489,7 +496,7 @@ class UnifiedRenderService:
         if not main_layer or not main_layer.clips:
             return 0.0
 
-        total = sum(UnifiedRenderService._clip_effective_duration(c) for c in main_layer.clips)
+        total = sum(UnifiedRenderService._clip_adjusted_duration(c) for c in main_layer.clips)
 
         # 减去转场重叠时间（粗略估算）
         n_clips = len(main_layer.clips)
@@ -745,6 +752,7 @@ class UnifiedRenderService:
         1. 只有 1 个图层
         2. 该图层是视频图层（main/broll/background），不是 overlay/corner_voice/audio
         3. 该图层只有 1 个 clip（无转场需求）
+        4. 没有贴纸（贴纸需要 filter_complex 或额外输入）
         """
         if len(layers) != 1:
             return False
@@ -753,6 +761,10 @@ 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(
@@ -965,6 +977,13 @@ 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)
@@ -1053,6 +1072,13 @@ 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}"])
@@ -1173,6 +1199,8 @@ class UnifiedRenderService:
                 start_time=final_start,
                 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,
@@ -1270,6 +1298,18 @@ 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)
@@ -1320,21 +1360,28 @@ class UnifiedRenderService:
         for layer in layers:
             layer_clip_indices = [all_clips.index(c) for c in layer.clips]
             layer_labels = [preprocessed_labels[i] for i in layer_clip_indices]
-            # 使用 trim 后的有效时长，与 Step 1 的 trim=duration 保持一致
-            layer_durations = [UnifiedRenderService._clip_effective_duration(all_clips[i]) for i in layer_clip_indices]
+            # 使用调速后的实际时长，与 Step 1 的调速处理保持一致
+            layer_durations = [UnifiedRenderService._clip_adjusted_duration(all_clips[i]) for i in layer_clip_indices]
             layer_transitions = [all_clips[i].transition_effect for i in layer_clip_indices]
+            layer_transition_durations = [all_clips[i].transition_duration for i in layer_clip_indices]
 
             if len(layer_labels) == 1:
                 # 单 clip 层，直接使用预处理标签
                 layer_output_labels[layer.role] = layer_labels[0]
             else:
-                # 多 clip 层，用 xfade 串联
+                # 多 clip 层，用 TransitionEngine 构建转场链
                 out_label = f"{layer.role}_merged"
-                xfade_filter, _ = build_xfade_filter_chain(
+                # 计算该层使用的转场时长（取首个非零值，否则用默认）
+                layer_dur = 0.0
+                for d in layer_transition_durations:
+                    if d > 0:
+                        layer_dur = d
+                        break
+                xfade_filter, _ = self._transition_engine.build_xfade_chain(
                     clip_durations=layer_durations,
                     clip_video_labels=layer_labels,
                     transitions=layer_transitions,
-                    transition_duration=self.transition_duration,
+                    transition_duration=layer_dur if layer_dur > 0 else None,
                     output_label=out_label,
                 )
                 if xfade_filter:
@@ -1440,6 +1487,14 @@ 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:
@@ -1500,6 +1555,40 @@ 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]:
         """探测输出文件的时长、大小、宽高.
 
@@ -1517,7 +1606,7 @@ class UnifiedRenderService:
 
     @staticmethod
     def _clip_effective_duration(clip: ResolvedClip) -> float:
-        """计算 clip 的有效时长."""
+        """计算 clip 的有效时长（原速 trim 后时长）."""
         if clip.duration > 0:
             return min(clip.duration, clip.actual_duration) if clip.actual_duration > 0 else clip.duration
         return clip.actual_duration if clip.actual_duration > 0 else 0.0
@@ -1614,3 +1703,20 @@ class UnifiedRenderService:
         )
 
         return new_filter, new_input_args
+
+    @staticmethod
+    def _clip_speed(clip: ResolvedClip) -> float:
+        """获取 clip 的播放速度，无效值回退到 1.0."""
+        speed = getattr(clip, "playback_speed", 1.0)
+        if not isinstance(speed, (int, float)) or speed <= 0:
+            return 1.0
+        return float(speed)
+
+    @staticmethod
+    def _clip_adjusted_duration(clip: ResolvedClip) -> float:
+        """计算调速后的 clip 实际时长（用于拼接计算）."""
+        base = UnifiedRenderService._clip_effective_duration(clip)
+        speed = UnifiedRenderService._clip_speed(clip)
+        if abs(speed - 1.0) < 1e-6:
+            return base
+        return base / speed

docs/schema-metadata-snapshot.json

@@ -858,6 +858,22 @@
           "type": "VARCHAR(20)",
           "unique": false
         },
+        {
+          "index": false,
+          "name": "transition_duration",
+          "nullable": false,
+          "primary_key": false,
+          "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

@@ -54,6 +54,8 @@ class SQLAlchemyEditPlanClipRepository:
             start_time=clip.start_time,
             duration=clip.duration,
             transition_effect=clip.transition_effect,
+            transition_duration=clip.transition_duration,
+            playback_speed=clip.playback_speed,
             status=clip.status,
             config=clip.config,
         )
@@ -76,6 +78,8 @@ class SQLAlchemyEditPlanClipRepository:
         model.start_time = clip.start_time
         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
@@ -120,6 +124,8 @@ class SQLAlchemyEditPlanClipRepository:
             start_time=model.start_time or 0.0,
             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

@@ -196,6 +196,8 @@ class EditPlanClipModel(Base):
     start_time = Column(Float, nullable=False, default=0.0)
     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

@@ -50,6 +50,8 @@ class EditPlanClip:
     start_time: float = 0.0
     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))
@@ -68,6 +70,8 @@ class EditPlanClip:
         start_time: float = 0.0,
         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:
         """创建剪辑计划片段"""
@@ -79,6 +83,13 @@ class EditPlanClip:
             raise ValueError("start_time 不能为负数")
         if duration < 0:
             raise ValueError("duration 不能为负数")
+        # 速度边界钳制
+        if playback_speed <= 0:
+            playback_speed = 1.0
+        elif playback_speed < 0.25:
+            playback_speed = 0.25
+        elif playback_speed > 4.0:
+            playback_speed = 4.0
 
         return cls(
             id=uuid4().hex,
@@ -91,6 +102,8 @@ class EditPlanClip:
             start_time=start_time,
             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

@@ -0,0 +1,117 @@
+#!/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

@@ -0,0 +1,126 @@
+#!/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

@@ -0,0 +1,42 @@
+#!/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,6 +138,70 @@ 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,6 +103,70 @@ 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):
@@ -470,8 +534,8 @@ class TestRetryProjectTask:
         assert data["task_type"] == "generation"
         assert data["status"] == "pending"
         assert "current_step" in data
-        # 验证新任务的 ID 不同于原任务
-        assert data["source_id"] != "gen-failed-1"
+        # 原地重试：source_id 保持不变（复用同一个任务）
+        assert data["source_id"] == "gen-failed-1"
         # 验证 Celery 任务被发送
         assert mock_celery.send_task.called
 
@@ -630,11 +694,13 @@ class TestTaskCenterCrossEndpoint:
         retry_resp = tc.post("/tasks/gen-fail-cross/retry")
         assert retry_resp.status_code == 200
 
-        # 3. 再次列出，应有2个任务（旧的failed + 新的pending）
+        # 3. 再次列出：原地重试，任务数不变（仍是1个），但状态变为 pending
         list_resp2 = tc.get("/tasks")
         assert list_resp2.status_code == 200
         items2 = list_resp2.json()["items"]
-        assert len(items2) == 2
+        assert len(items2) == 1
+        assert items2[0]["status"] == "pending"
+        assert items2[0]["source_id"] == "gen-fail-cross"
 
         test_app.dependency_overrides.clear()
 

tests/unit/test_cover_reverse_sticker.py

@@ -0,0 +1,860 @@
+"""封面生成 + 视频倒放 + 贴纸叠加 单元测试.
+
+覆盖三个新渲染能力的核心场景和降级逻辑。
+"""
+
+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,6 +193,70 @@ 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,6 +200,70 @@ 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,6 +64,38 @@ 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

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

tests/unit/test_transition_engine.py

@@ -0,0 +1,484 @@
+"""转场特效引擎单测 — Phase 8 智能增强."""
+
+from __future__ import annotations
+
+import pytest
+from video_processing.transition_engine import (
+    CUT_TRANSITION,
+    DEFAULT_TRANSITION_DURATION,
+    MAX_TRANSITION_DURATION,
+    MIN_TRANSITION_DURATION,
+    TransitionConfig,
+    TransitionEngine,
+    TransitionType,
+    _normalize_transition_name,
+)
+
+# ── TransitionType 枚举测试 ──────────────────────────────────────────────────
+
+
+class TestTransitionType:
+    """TransitionType 枚举测试."""
+
+    def test_all_supported_count(self):
+        """支持的转场类型数量（不含cut）."""
+        supported = TransitionType.all_supported()
+        # 至少 8 种：fade, dissolve, slide*4, zoom, wipe*4, circlecrop, rectcrop
+        assert len(supported) >= 8
+        assert "fade" in supported
+        assert "dissolve" in supported
+        assert "zoom" in supported
+        assert "circlecrop" in supported
+        assert "rectcrop" in supported
+
+    def test_slide_directions(self):
+        """四个方向的滑入转场都支持."""
+        assert TransitionType.is_supported("slideleft")
+        assert TransitionType.is_supported("slideright")
+        assert TransitionType.is_supported("slideup")
+        assert TransitionType.is_supported("slidedown")
+
+    def test_wipe_directions(self):
+        """四个方向的擦除转场都支持."""
+        assert TransitionType.is_supported("wipeleft")
+        assert TransitionType.is_supported("wiperight")
+        assert TransitionType.is_supported("wipeup")
+        assert TransitionType.is_supported("wipedown")
+
+    def test_is_supported_case_insensitive(self):
+        """大小写不敏感."""
+        assert TransitionType.is_supported("FADE")
+        assert TransitionType.is_supported("Fade")
+        assert TransitionType.is_supported("fade")
+
+    def test_is_supported_with_underscores(self):
+        """下划线不影响判断."""
+        assert TransitionType.is_supported("slide_left")
+        assert TransitionType.is_supported("slide-left")
+
+    def test_is_supported_aliases(self):
+        """别名支持."""
+        assert TransitionType.is_supported("crossfade")
+        assert TransitionType.is_supported("dissolve")
+        assert TransitionType.is_supported("zoomin")
+        assert TransitionType.is_supported("wipe")
+
+    def test_unsupported_transition(self):
+        """不支持的转场返回 False."""
+        assert not TransitionType.is_supported("nonexistent_effect")
+        assert not TransitionType.is_supported("random_stuff")
+        assert not TransitionType.is_supported("")
+
+    def test_cut_not_in_supported(self):
+        """硬切不在"支持的转场效果"列表中（它不是特效）."""
+        supported = TransitionType.all_supported()
+        assert "cut" not in supported
+
+
+# ── 名称标准化测试 ────────────────────────────────────────────────────────────
+
+
+class TestNormalizeTransitionName:
+    """名称标准化函数测试."""
+
+    def test_lowercase(self):
+        """大写转小写."""
+        assert _normalize_transition_name("FADE") == "fade"
+        assert _normalize_transition_name("Fade") == "fade"
+
+    def test_remove_underscores(self):
+        """移除下划线."""
+        assert _normalize_transition_name("slide_left") == "slideleft"
+        assert _normalize_transition_name("slide_up") == "slideup"
+
+    def test_remove_hyphens(self):
+        """移除连字符."""
+        assert _normalize_transition_name("slide-left") == "slideleft"
+
+    def test_mixed(self):
+        """混合情况."""
+        assert _normalize_transition_name("Slide_Left") == "slideleft"
+        assert _normalize_transition_name("FADE-IN") == "fadein"
+
+
+# ── TransitionConfig 测试 ────────────────────────────────────────────────────
+
+
+class TestTransitionConfig:
+    """TransitionConfig 配置解析测试."""
+
+    # ── 默认值 ──
+
+    def test_default_config(self):
+        """默认配置是硬切."""
+        cfg = TransitionConfig.parse()
+        assert cfg.effect == CUT_TRANSITION
+        assert cfg.duration == DEFAULT_TRANSITION_DURATION
+        assert cfg.is_cut is True
+
+    def test_none_effect(self):
+        """None effect 降级为 cut."""
+        cfg = TransitionConfig.parse(effect=None)
+        assert cfg.effect == CUT_TRANSITION
+        assert cfg.is_cut is True
+
+    def test_empty_effect(self):
+        """空字符串 effect 降级为 cut."""
+        cfg = TransitionConfig.parse(effect="")
+        assert cfg.effect == CUT_TRANSITION
+        assert cfg.is_cut is True
+
+    # ── 有效转场类型 ──
+
+    def test_fade_effect(self):
+        """fade 转场."""
+        cfg = TransitionConfig.parse(effect="fade")
+        assert cfg.effect == "fade"
+        assert cfg.is_cut is False
+        assert cfg.ffmpeg_transition == "fade"
+
+    def test_dissolve_effect(self):
+        """dissolve 转场."""
+        cfg = TransitionConfig.parse(effect="dissolve")
+        assert cfg.effect == "dissolve"
+        assert cfg.ffmpeg_transition == "dissolve"
+
+    def test_zoom_effect(self):
+        """zoom 转场 → FFmpeg zoomin."""
+        cfg = TransitionConfig.parse(effect="zoom")
+        assert cfg.effect == "zoom"
+        assert cfg.ffmpeg_transition == "zoomin"
+
+    def test_slide_left_alias(self):
+        """slide_left 别名."""
+        cfg = TransitionConfig.parse(effect="slide_left")
+        assert cfg.effect == "slideleft"
+        assert cfg.ffmpeg_transition == "slideleft"
+
+    def test_wipe_alias(self):
+        """wipe 别名 → 默认向左擦."""
+        cfg = TransitionConfig.parse(effect="wipe")
+        assert cfg.effect == "wipeleft"
+        assert cfg.ffmpeg_transition == "wipeleft"
+
+    def test_circlecrop_effect(self):
+        """圆形扩散转场."""
+        cfg = TransitionConfig.parse(effect="circlecrop")
+        assert cfg.effect == "circlecrop"
+        assert cfg.ffmpeg_transition == "circlecrop"
+
+    def test_rectcrop_effect(self):
+        """矩形扩散转场."""
+        cfg = TransitionConfig.parse(effect="rectcrop")
+        assert cfg.effect == "rectcrop"
+        assert cfg.ffmpeg_transition == "rectcrop"
+
+    # ── 降级策略 ──
+
+    def test_unsupported_fallback_to_cut(self):
+        """不支持的转场自动降级为硬切，不阻断渲染."""
+        cfg = TransitionConfig.parse(effect="nonexistent_effect")
+        assert cfg.effect == CUT_TRANSITION
+        assert cfg.is_cut is True
+
+    def test_unsupported_whitespace_fallback(self):
+        """带空格的不支持转场也降级."""
+        cfg = TransitionConfig.parse(effect="  bad effect  ")
+        assert cfg.effect == CUT_TRANSITION
+
+    # ── 时长边界校验 ──
+
+    def test_default_duration(self):
+        """默认时长 0.5s."""
+        cfg = TransitionConfig.parse(effect="fade")
+        assert cfg.duration == 0.5
+
+    def test_duration_within_range(self):
+        """正常范围内的时长."""
+        cfg = TransitionConfig.parse(effect="fade", duration=1.0)
+        assert cfg.duration == 1.0
+
+    def test_duration_min_boundary(self):
+        """最小值边界."""
+        cfg = TransitionConfig.parse(effect="fade", duration=MIN_TRANSITION_DURATION)
+        assert cfg.duration == MIN_TRANSITION_DURATION
+
+    def test_duration_max_boundary(self):
+        """最大值边界."""
+        cfg = TransitionConfig.parse(effect="fade", duration=MAX_TRANSITION_DURATION)
+        assert cfg.duration == MAX_TRANSITION_DURATION
+
+    def test_duration_below_min_clamped(self):
+        """低于最小值的时长被钳制."""
+        cfg = TransitionConfig.parse(effect="fade", duration=0.1)
+        assert cfg.duration == MIN_TRANSITION_DURATION
+        assert cfg.duration >= MIN_TRANSITION_DURATION
+
+    def test_duration_above_max_clamped(self):
+        """高于最大值的时长被钳制."""
+        cfg = TransitionConfig.parse(effect="fade", duration=5.0)
+        assert cfg.duration == MAX_TRANSITION_DURATION
+        assert cfg.duration <= MAX_TRANSITION_DURATION
+
+    def test_duration_zero_default_for_effect(self):
+        """有转场效果但 duration=0 时使用默认值."""
+        # 0.0 会被当作小于最小值钳制到 0.3
+        cfg = TransitionConfig.parse(effect="fade", duration=0.0)
+        assert cfg.duration == MIN_TRANSITION_DURATION
+
+    def test_duration_negative_clamped(self):
+        """负时长被钳制到最小值."""
+        cfg = TransitionConfig.parse(effect="fade", duration=-1.0)
+        assert cfg.duration == MIN_TRANSITION_DURATION
+
+    def test_duration_none_uses_default(self):
+        """None duration 使用默认值."""
+        cfg = TransitionConfig.parse(effect="fade", duration=None)
+        assert cfg.duration == DEFAULT_TRANSITION_DURATION
+
+    def test_duration_invalid_type(self):
+        """无效类型的时长使用默认值."""
+        cfg = TransitionConfig.parse(effect="fade", duration="abc")  # type: ignore
+        assert cfg.duration == DEFAULT_TRANSITION_DURATION
+
+    # ── cut 的 ffmpeg_transition ──
+
+    def test_cut_ffmpeg_transition_empty(self):
+        """硬切没有对应的 FFmpeg xfade transition."""
+        cfg = TransitionConfig.parse(effect="cut")
+        assert cfg.ffmpeg_transition == ""
+
+
+# ── TransitionEngine 测试 ────────────────────────────────────────────────────
+
+
+class TestTransitionEngine:
+    """TransitionEngine 转场引擎测试."""
+
+    def test_default_engine(self):
+        """默认引擎初始化."""
+        engine = TransitionEngine()
+        assert engine is not None
+
+    def test_custom_default_duration(self):
+        """自定义默认时长."""
+        engine = TransitionEngine(default_duration=1.0)
+        cfg = engine.resolve_config(effect="fade")
+        assert cfg.duration == 1.0
+
+    def test_resolve_config_fade(self):
+        """解析 fade 配置."""
+        engine = TransitionEngine()
+        cfg = engine.resolve_config(effect="fade", duration=0.8)
+        assert cfg.effect == "fade"
+        assert cfg.duration == 0.8
+
+    def test_resolve_config_fallback(self):
+        """不支持的转场降级."""
+        engine = TransitionEngine()
+        cfg = engine.resolve_config(effect="unknown_effect")
+        assert cfg.effect == CUT_TRANSITION
+        assert cfg.is_cut is True
+
+    def test_resolve_config_duration_clamp(self):
+        """时长边界钳制."""
+        engine = TransitionEngine()
+        cfg = engine.resolve_config(effect="fade", duration=3.0)
+        assert cfg.duration == MAX_TRANSITION_DURATION
+
+    # ── 批量解析 ──
+
+    def test_resolve_clip_transitions_all_valid(self):
+        """批量解析全部有效转场."""
+        engine = TransitionEngine()
+        configs = engine.resolve_clip_transitions(["cut", "fade", "dissolve", "slideleft"])
+        assert len(configs) == 4
+        assert configs[0].effect == "cut"
+        assert configs[0].is_cut is True
+        assert configs[1].effect == "fade"
+        assert configs[2].effect == "dissolve"
+        assert configs[3].effect == "slideleft"
+
+    def test_resolve_clip_transitions_with_fallback(self):
+        """批量解析包含不支持的转场，自动降级."""
+        engine = TransitionEngine()
+        configs = engine.resolve_clip_transitions(["fade", "bad_effect", "dissolve", "worse_effect"])
+        assert len(configs) == 4
+        assert configs[0].effect == "fade"
+        assert configs[1].effect == "cut"  # 降级
+        assert configs[2].effect == "dissolve"
+        assert configs[3].effect == "cut"  # 降级
+
+    def test_resolve_clip_transitions_with_durations(self):
+        """带时长校验的批量解析（转场时长不超过片段时长的一半）."""
+        engine = TransitionEngine(default_duration=1.0)
+        # 片段只有 1.0s，转场时长被限制在 0.5s
+        configs = engine.resolve_clip_transitions(
+            ["fade", "dissolve"],
+            clip_durations=[1.0, 1.0],
+        )
+        assert len(configs) == 2
+        # 1.0s 默认值超过了片段时长的一半 (0.5s)，所以被钳制
+        assert configs[0].duration <= 0.5
+        assert configs[1].duration <= 0.5
+
+    def test_resolve_clip_transitions_short_clip_min_bound(self):
+        """超短片段的转场时长至少为最小值."""
+        engine = TransitionEngine()
+        configs = engine.resolve_clip_transitions(
+            ["fade"],
+            clip_durations=[0.1],  # 极短片段
+        )
+        assert len(configs) == 1
+        # 0.1 * 0.5 = 0.05 < MIN_TRANSITION_DURATION，所以用最小值
+        assert configs[0].duration == MIN_TRANSITION_DURATION
+
+    # ── xfade 滤镜链构建 ──
+
+    def test_build_xfade_single_clip(self):
+        """单 clip 直接 copy."""
+        engine = TransitionEngine()
+        filter_str, total_dur = engine.build_xfade_chain(
+            clip_durations=[5.0],
+            clip_video_labels=["v0"],
+            transitions=["cut"],
+            output_label="outv",
+        )
+        assert "copy" in filter_str
+        assert "[outv]" in filter_str
+        assert total_dur == pytest.approx(5.0, abs=0.01)
+
+    def test_build_xfade_two_clips_fade(self):
+        """两个 clip 之间 fade 转场."""
+        engine = TransitionEngine()
+        filter_str, total_dur = engine.build_xfade_chain(
+            clip_durations=[3.0, 4.0],
+            clip_video_labels=["v0", "v1"],
+            transitions=["cut", "fade"],
+            output_label="outv",
+        )
+        assert "xfade" in filter_str
+        assert "transition=fade" in filter_str
+        # 总时长 = 3 + 4 - transition_duration (0.5) = 6.5
+        assert total_dur == pytest.approx(6.5, abs=0.1)
+
+    def test_build_xfade_three_clips_mixed(self):
+        """三个 clip 混合转场."""
+        engine = TransitionEngine()
+        filter_str, total_dur = engine.build_xfade_chain(
+            clip_durations=[3.0, 4.0, 5.0],
+            clip_video_labels=["v0", "v1", "v2"],
+            transitions=["cut", "fade", "dissolve"],
+            output_label="outv",
+        )
+        assert "xfade" in filter_str
+        assert "transition=fade" in filter_str
+        assert "transition=dissolve" in filter_str
+        # 总时长 ≈ 3 + 4 + 5 - 2 * 0.5 = 11.0
+        assert total_dur == pytest.approx(11.0, abs=0.2)
+
+    def test_build_xfade_with_custom_duration(self):
+        """自定义转场时长."""
+        engine = TransitionEngine(default_duration=0.5)
+        filter_str, total_dur = engine.build_xfade_chain(
+            clip_durations=[3.0, 4.0],
+            clip_video_labels=["v0", "v1"],
+            transitions=["cut", "fade"],
+            transition_duration=1.0,
+            output_label="outv",
+        )
+        assert "xfade" in filter_str
+        # 总时长 = 3 + 4 - 1.0 = 6.0
+        assert total_dur == pytest.approx(6.0, abs=0.1)
+
+    def test_build_xfade_zoom_transition(self):
+        """zoom 转场滤镜构建."""
+        engine = TransitionEngine()
+        filter_str, _ = engine.build_xfade_chain(
+            clip_durations=[3.0, 4.0],
+            clip_video_labels=["v0", "v1"],
+            transitions=["cut", "zoom"],
+        )
+        assert "xfade" in filter_str
+        assert "transition=zoomin" in filter_str  # zoom → zoomin
+
+    def test_build_xfade_slide_directions(self):
+        """四个方向的滑入转场."""
+        engine = TransitionEngine()
+        for direction in ["slideleft", "slideright", "slideup", "slidedown"]:
+            filter_str, _ = engine.build_xfade_chain(
+                clip_durations=[3.0, 4.0],
+                clip_video_labels=["v0", "v1"],
+                transitions=["cut", direction],
+            )
+            assert f"transition={direction}" in filter_str
+
+    def test_build_xfade_fallback_transition(self):
+        """不支持的转场降级后构建（降级为cut，等效于极短fade）."""
+        engine = TransitionEngine()
+        # bad_effect 降级为 cut，cut 使用极短转场
+        filter_str, _ = engine.build_xfade_chain(
+            clip_durations=[3.0, 4.0],
+            clip_video_labels=["v0", "v1"],
+            transitions=["cut", "bad_effect"],
+        )
+        # 降级后是 cut，cut 会被 xfade 层映射为 fade（因为 cut 不在 map 里）
+        # 但时长会很短，所以仍然有 xfade
+        assert "xfade" in filter_str
+
+    # ── 支持的转场列表 ──
+
+    def test_supported_transitions_list(self):
+        """获取支持的转场列表（给 API 用）."""
+        transitions = TransitionEngine.supported_transitions()
+        assert len(transitions) >= 10  # cut + 至少 9 种特效
+        # 检查结构
+        for t in transitions:
+            assert "name" in t
+            assert "display_name" in t
+            assert "category" in t
+        # 检查分类
+        names = [t["name"] for t in transitions]
+        assert "cut" in names
+        assert "fade" in names
+        assert "zoom" in names
+        assert "circlecrop" in names
+
+
+# ── 集成测试：与 UnifiedRenderService 协作 ────────────────────────────────────
+
+
+class TestTransitionIntegration:
+    """转场引擎与统一渲染服务的集成测试."""
+
+    def test_unified_render_service_has_transition_engine(self):
+        """UnifiedRenderService 内部有 TransitionEngine 实例."""
+        from pathlib import Path
+
+        from video_processing.unified_render_service import UnifiedRenderService
+
+        # 构造最小化的服务实例
+        service = UnifiedRenderService(
+            plan=None,
+            clips=[],
+            asset_path_map={},
+            work_dir=Path("/tmp"),
+        )
+        assert hasattr(service, "_transition_engine")
+        assert isinstance(service._transition_engine, TransitionEngine)
+
+    def test_resolved_clip_has_transition_duration(self):
+        """ResolvedClip 有 transition_duration 字段."""
+        from video_processing.unified_render_service import ResolvedClip
+
+        rc = ResolvedClip(
+            clip_id="test",
+            asset_id="asset1",
+            local_path=__file__,  # 随便一个存在的路径
+            clip_type="main",
+            order=0,
+            transition_effect="fade",
+            transition_duration=0.8,
+        )
+        assert rc.transition_duration == 0.8
+        assert rc.transition_effect == "fade"