feat: 视频封面生成 + 视频倒放 + 贴纸叠加三个渲染能力

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/render_audio.py

@@ -22,6 +22,7 @@ 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
 
 if TYPE_CHECKING:
     from video_processing.unified_render_service import RenderLayer, ResolvedClip
@@ -230,6 +231,14 @@ def concat_main_audio(
         if video_duration > 0 and (final_duration <= 0 or final_duration > video_duration):
             final_duration = video_duration
 
+        # 音频倒放
+        reverse_config = ReverseConfig.from_dict(clip.config.get("reverse"))
+        af_filters = []
+        if reverse_config.enabled and reverse_config.reverse_audio:
+            reverse_filter = ReverseEngine.build_audio_filter(reverse_config, duration=effective_duration)
+            if reverse_filter:
+                af_filters.append(reverse_filter)
+
         command = [
             FFMPEG_BIN,
             "-y",
@@ -243,6 +252,8 @@ def concat_main_audio(
         ]
         if trim_start > 0:
             command.extend(["-ss", f"{trim_start:.3f}"])
+        if af_filters:
+            command.extend(["-af", ",".join(af_filters)])
         if final_duration > 0:
             command.extend(["-t", f"{final_duration:.3f}"])
         command.append(str(output_path))
@@ -257,12 +268,21 @@ def concat_main_audio(
         input_args.extend(["-i", str(clip.local_path)])
         effective_duration = clip_effective_duration(clip)
         trim_start = getattr(clip, "start_time", 0) or 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")
         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/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/unified_render_service.py

@@ -45,6 +45,8 @@ 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.sticker_engine import StickerEngine, parse_stickers_from_config
 from video_processing.subtitle_generator import generate_ass_from_timeline
 from video_processing.trim_engine import TrimConfig, TrimEngine, extract_trim_from_clip_config
 from video_processing.tts_engine import TtsEngine
@@ -745,6 +747,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 +756,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 +972,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 +1067,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}"])
@@ -1270,6 +1291,13 @@ class UnifiedRenderService:
                     filters.append(f"trim=duration={effective_duration:.3f}")
                 filters.append("setpts=PTS-STARTPTS")
 
+            # 倒放滤镜（在 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)
@@ -1440,6 +1468,15 @@ 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 +1537,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]:
         """探测输出文件的时长、大小、宽高.