feat: 画中画（PiP）能力 - 多图层叠加引擎 (#299)

apps/worker/video_processing/unified_render_service.py

@@ -40,6 +40,7 @@ from video_processing.ffmpeg_utils import (
     probe_video_info,
     run_ffmpeg,
 )
+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.subtitle_generator import generate_ass_from_timeline
@@ -207,15 +208,21 @@ class UnifiedRenderService:
         # 4. 生成 ASS 字幕文件（如果有 title/subtitle 配置）
         ass_path = self._maybe_generate_ass(video_duration)
 
+        # 4.5 解析画中画配置
+        pip_config = PiPConfig.from_dict((self.plan.config or {}).get("pip_config"))
+        pip_sources = self._resolve_pip_sources(pip_config) if pip_config.enabled else []
+        has_pip = len(pip_sources) > 0
+
         # 灰度埋点：开始渲染
         layer_roles = [layer.role for layer in layers]
         clip_counts = {layer.role: len(layer.clips) for layer in layers}
         logger.info(
-            "[unified-render] start render: plan_id=%s clip_count=%d layers=%s clip_counts=%s",
+            "[unified-render] start render: plan_id=%s clip_count=%d layers=%s clip_counts=%s pip_layers=%d",
             self.plan.id,
             len(resolved),
             layer_roles,
             clip_counts,
+            len(pip_sources),
         )
 
         # 5. 视频主渲染
@@ -223,7 +230,8 @@ class UnifiedRenderService:
         video_only_path = self.work_dir / f"rendered_{self.plan.id}_video.mp4"
         output_path = self.work_dir / f"rendered_{self.plan.id}.mp4"
 
-        is_pass_through = self._can_use_pass_through(layers)
+        # 有画中画时不走直通（需要额外图层叠加）
+        is_pass_through = self._can_use_pass_through(layers) and not has_pip
         pass_through_has_audio = False
         used_stream_copy = False
 
@@ -249,6 +257,11 @@ class UnifiedRenderService:
                 )
         else:
             filter_complex, input_args = self._build_filter_complex(layers, ass_path=ass_path)
+
+            # 追加画中画滤镜
+            if has_pip:
+                filter_complex, input_args = self._append_pip_filters(filter_complex, input_args, pip_sources)
+
             self._execute_ffmpeg(filter_complex, input_args, video_only_path)
 
         t_video_end = time.time()
@@ -1121,3 +1134,96 @@ class UnifiedRenderService:
         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
+
+    # ── 画中画（PiP）相关方法 ──────────────────────────────────────────────────
+
+    def _resolve_pip_sources(self, pip_config: PiPConfig) -> list[tuple[str, PiPLayerConfig, Path]]:
+        """解析画中画图层的素材源，返回可用的图层列表.
+
+        降级策略：素材不存在或无效的图层自动跳过，不阻断渲染。
+
+        Returns:
+            [(input_label_placeholder, layer_config, local_path), ...]
+            input_label 在 build_pip_filters 中会用实际的输入索引替换
+        """
+        if not pip_config.enabled:
+            return []
+
+        engine = PiPEngine(
+            output_width=self.output_width,
+            output_height=self.output_height,
+            output_fps=self.output_fps,
+        )
+
+        result = []
+        for i, layer in enumerate(pip_config.layers):
+            path = engine.validate_layer_source(layer, self.asset_path_map)
+            if path is None:
+                logger.warning("PiP图层素材不可用，跳过: layer_index=%d source=%s", i, layer.source)
+                continue
+            # 标签占位，实际输入索引由 build_pip_filters 内部管理
+            result.append((f"pip_src_{i}", layer, path))
+
+        return result
+
+    def _append_pip_filters(
+        self,
+        filter_complex: str,
+        input_args: list[str],
+        pip_sources: list[tuple[str, Any, Path]],
+    ) -> tuple[str, list[str]]:
+        """将画中画滤镜追加到 filter_complex 末尾.
+
+        处理逻辑：
+        1. 将原 final_video 标签重命名为 pip_base（作为PiP的底层视频）
+        2. 追加 PiP 预处理和 overlay 滤镜
+        3. PiP 最终输出命名为 final_video
+
+        Args:
+            filter_complex: 原 filter_complex 字符串
+            input_args: 原输入参数列表
+            pip_sources: PiP 素材列表 [(label, layer_config, path), ...]
+
+        Returns:
+            (new_filter_complex, new_input_args)
+        """
+        if not pip_sources:
+            return filter_complex, input_args
+
+        pip_engine = PiPEngine(
+            output_width=self.output_width,
+            output_height=self.output_height,
+            output_fps=self.output_fps,
+        )
+
+        # 1. 将原 final_video 改为 pip_base
+        new_filter = filter_complex.replace("[final_video]", "[pip_base]")
+
+        # 2. 构建 PiP 滤镜链
+        # 主输入数量 = len(input_args) // 2（每个输入占 "-i path" 两个参数）
+        base_input_idx = len(input_args) // 2
+        pip_filter_parts, pip_input_args, final_label = pip_engine.build_pip_filters(
+            base_label="pip_base",
+            pip_sources=pip_sources,
+            base_input_idx=base_input_idx,
+        )
+
+        if not pip_filter_parts:
+            # 没有有效PiP滤镜，恢复原标签
+            return filter_complex, input_args
+
+        # 3. 追加 PiP 滤镜 + 最终格式转换（输出为 final_video）
+        pip_filter_str = ";".join(pip_filter_parts)
+        final_format = f"[{final_label}]format=yuv420p[final_video]"
+        new_filter = f"{new_filter};{pip_filter_str};{final_format}"
+
+        # 4. 追加输入参数
+        new_input_args = list(input_args) + pip_input_args
+
+        logger.info(
+            "[unified-render] appended PiP filters: layers=%d new_inputs=%d",
+            len(pip_sources),
+            len(pip_input_args) // 2,
+        )
+
+        return new_filter, new_input_args