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

Squash merge PR #305

apps/worker/video_processing/unified_render_service.py

@@ -44,6 +44,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.transition_engine import TransitionEngine
 from video_processing.trim_engine import TrimConfig, TrimEngine, extract_trim_from_clip_config
@@ -747,6 +749,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
@@ -755,6 +758,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(
@@ -967,6 +974,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)
@@ -1055,6 +1069,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}"])
@@ -1273,6 +1294,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)
@@ -1450,6 +1478,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:
@@ -1510,6 +1546,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]:
         """探测输出文件的时长、大小、宽高.