feat: 水印 + 片头片尾引擎（视频包装能力） (#298)

apps/worker/video_processing/unified_render_service.py

@@ -40,11 +40,14 @@ from video_processing.ffmpeg_utils import (
     probe_video_info,
     run_ffmpeg,
 )
+from video_processing.intro_outro_engine import IntroOutroConfig, IntroOutroEngine
 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
+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
 
 from packages.domain.tts_config import TtsConfig
 
@@ -70,6 +73,7 @@ class ResolvedClip:
 
     # 运行时填充
     actual_duration: float = 0.0  # 素材实际时长（probe 后填充）
+    trim_config: TrimConfig | None = None  # 解析后的裁剪配置（运行时填充）
 
 
 @dataclass
@@ -359,6 +363,85 @@ class UnifiedRenderService:
         # 8. 探测输出
         duration, file_size, width, height = self._probe_output(output_path)
 
+        # 9. 片头片尾拼接（后处理）
+        intro_outro_config = IntroOutroConfig.from_dict((self.plan.config or {}).get("intro_outro"))
+        if intro_outro_config.has_intro or intro_outro_config.has_outro:
+            io_valid, io_err = intro_outro_config.validate()
+            if io_valid:
+                final_with_io = self.work_dir / f"rendered_{self.plan.id}_with_io.mp4"
+                intro_path = None
+                outro_path = None
+
+                # 生成片头
+                if intro_outro_config.has_intro:
+                    intro_path = self.work_dir / f"intro_{self.plan.id}.mp4"
+                    intro_ok = False
+                    if intro_outro_config.intro_type == "video":
+                        import shutil
+
+                        src = Path(intro_outro_config.intro_video_path)
+                        if src.exists():
+                            shutil.copy2(src, intro_path)
+                            intro_ok = True
+                        else:
+                            logger.warning("片头视频不存在，跳过片头: %s", src)
+                    elif intro_outro_config.intro_type == "text":
+                        intro_ok = IntroOutroEngine.generate_text_intro(
+                            intro_path,
+                            intro_outro_config,
+                            self.output_width,
+                            self.output_height,
+                            self.output_fps,
+                        )
+
+                    if not intro_ok:
+                        intro_path = None
+
+                # 生成片尾
+                if intro_outro_config.has_outro:
+                    outro_path = self.work_dir / f"outro_{self.plan.id}.mp4"
+                    outro_ok = False
+                    if intro_outro_config.outro_type == "video":
+                        import shutil
+
+                        src = Path(intro_outro_config.outro_video_path)
+                        if src.exists():
+                            shutil.copy2(src, outro_path)
+                            outro_ok = True
+                        else:
+                            logger.warning("片尾视频不存在，跳过片尾: %s", src)
+                    elif intro_outro_config.outro_type in ("text", "follow"):
+                        outro_ok = IntroOutroEngine.generate_text_outro(
+                            outro_path,
+                            intro_outro_config,
+                            self.output_width,
+                            self.output_height,
+                            self.output_fps,
+                        )
+
+                    if not outro_ok:
+                        outro_path = None
+
+                # 拼接
+                if intro_path or outro_path:
+                    concat_ok = IntroOutroEngine.concat_with_intro_outro(
+                        output_path,
+                        intro_path,
+                        outro_path,
+                        final_with_io,
+                        transition_duration=intro_outro_config.transition_duration,
+                        transition_effect=intro_outro_config.transition_effect,
+                    )
+                    if concat_ok and final_with_io.exists():
+                        output_path = final_with_io
+                        # 重新探测
+                        duration, file_size, width, height = self._probe_output(output_path)
+                        logger.info("[unified-render] 片头片尾拼接完成: plan_id=%s", self.plan.id)
+                    else:
+                        logger.warning("[unified-render] 片头片尾拼接失败，使用原视频: plan_id=%s", self.plan.id)
+            else:
+                logger.warning("[unified-render] 片头片尾配置无效，跳过: %s", io_err)
+
         t_total = int((time.time() - t_start) * 1000)
         logger.info(
             "[unified-render] render done: plan_id=%s total_ms=%d video_ms=%d audio_ms=%d "
@@ -968,6 +1051,7 @@ class UnifiedRenderService:
         """将 EditPlanClip 列表解析为 ResolvedClip 列表。
 
         跳过 asset_id 为空或在 asset_path_map 中找不到的片段。
+        支持多段裁剪：一个 clip 配置了 trim_segments 时会展开为多个 ResolvedClip。
         """
         resolved: list[ResolvedClip] = []
         for clip in self.clips:
@@ -987,17 +1071,75 @@ class UnifiedRenderService:
             except Exception:
                 actual_duration = clip.duration or 5.0
 
+            # 检查是否有多段裁剪配置
+            clip_config = clip.config or {}
+            trim_segments = TrimEngine.parse_segments_from_config(clip_config)
+
+            if trim_segments and len(trim_segments) > 1:
+                # 多段裁剪：展开为多个 clip
+                resolved_segments = TrimEngine.resolve_segments(trim_segments, actual_duration)
+                for i, seg in enumerate(resolved_segments):
+                    # 每个段生成一个独立的 ResolvedClip
+                    seg_clip_id = f"{clip.id}_seg_{seg.segment_id}"
+                    seg_order = clip.order + seg.order * 0.001 + i * 0.0001  # 保持排序
+                    seg_start = seg.trim.start_time
+                    seg_duration = seg.trim.duration
+
+                    rc = ResolvedClip(
+                        clip_id=seg_clip_id,
+                        asset_id=asset_id,
+                        local_path=local_path,
+                        clip_type=clip.clip_type,
+                        order=seg_order,
+                        start_time=seg_start,
+                        duration=seg_duration,
+                        transition_effect=clip.transition_effect or "cut",
+                        config={**clip_config, "_segment_id": seg.segment_id},
+                        actual_duration=actual_duration,
+                        trim_config=seg.trim,
+                    )
+                    resolved.append(rc)
+                continue
+
+            # 单段裁剪（或无裁剪）
+            # 解析裁剪配置：config 优先，否则用 clip.start_time + clip.duration
+            trim_config = extract_trim_from_clip_config(clip_config)
+            if trim_config is None and (clip.start_time > 0 or clip.duration > 0):
+                # 用旧字段构造
+                trim_config = TrimConfig(
+                    start_time=clip.start_time,
+                    duration=clip.duration,
+                )
+
+            # 钳制到实际素材时长
+            effective_trim: TrimConfig | None = None
+            final_start = clip.start_time
+            final_duration = clip.duration
+
+            if trim_config is not None and actual_duration > 0:
+                effective_trim = trim_config.validate_and_resolve(actual_duration)
+                if effective_trim.is_valid:
+                    final_start = effective_trim.start_time
+                    final_duration = effective_trim.duration
+                else:
+                    # 裁剪无效 → 使用完整素材
+                    logger.warning("裁剪配置无效，使用完整素材: clip_id=%s", clip.id)
+                    effective_trim = None
+                    final_start = 0.0
+                    final_duration = actual_duration
+
             rc = ResolvedClip(
                 clip_id=clip.id,
                 asset_id=asset_id,
                 local_path=local_path,
                 clip_type=clip.clip_type,
                 order=clip.order,
-                start_time=clip.start_time,
-                duration=clip.duration,
+                start_time=final_start,
+                duration=final_duration,
                 transition_effect=clip.transition_effect or "cut",
-                config=clip.config or {},
+                config=clip_config,
                 actual_duration=actual_duration,
+                trim_config=effective_trim,
             )
             resolved.append(rc)
 
@@ -1072,7 +1214,7 @@ class UnifiedRenderService:
 
         filter_parts: list[str] = []
 
-        # Step 1: 预处理每个 clip — scale + setpts
+        # Step 1: 预处理每个 clip — trim + scale + setpts
         # 为每个 clip 生成预处理后的标签 [v0], [v1], ...
         preprocessed_labels: list[str] = []
         for i, clip in enumerate(all_clips):
@@ -1081,11 +1223,15 @@ class UnifiedRenderService:
 
             filters: list[str] = []
 
-            # trim — 始终将输出截断到有效时长，防止 xfade offset 与实际时长不匹配
+            # trim — 裁剪到指定区间，精确到帧
             effective_duration = UnifiedRenderService._clip_effective_duration(clip)
+            trim_start = getattr(clip, "start_time", 0) or 0
 
             if effective_duration > 0:
-                filters.append(f"trim=duration={effective_duration}")
+                if trim_start > 0:
+                    filters.append(f"trim=start={trim_start:.3f}:duration={effective_duration:.3f}")
+                else:
+                    filters.append(f"trim=duration={effective_duration:.3f}")
                 filters.append("setpts=PTS-STARTPTS")
 
             # scale
@@ -1186,6 +1332,66 @@ class UnifiedRenderService:
                 filter_parts.append(f"[{final_video_label}][{overlay_label}]" f"overlay={x}:{y}[{combined_label}]")
                 final_video_label = combined_label
 
+        # 叠加水印（在字幕之前）
+        watermark_config = WatermarkConfig.from_dict((self.plan.config or {}).get("watermark"))
+        if watermark_config is not None:
+            wm_valid, wm_err = watermark_config.validate()
+            if wm_valid:
+                wm_label = "watermarked"
+                if watermark_config.mode == "image":
+                    # 图片水印：检查图片是否存在
+                    wm_path = Path(watermark_config.image_path)
+                    if wm_path.exists():
+                        # 图片水印需要额外输入，放在 filter 开头
+                        wm_idx = len(all_clips)  # 水印图是最后一个输入
+                        wm_scale = int(self.output_width * watermark_config.scale)
+
+                        # 透明度
+                        wm_filters = f"scale={wm_scale}:-1"
+                        if watermark_config.opacity < 1.0:
+                            wm_filters += f",format=rgba,colorchannelmixer=aa={watermark_config.opacity}"
+
+                        filter_parts.insert(0, f"[{wm_idx}:v]{wm_filters}[wm_scaled]")
+                        input_args.extend(["-i", str(wm_path)])
+
+                        # 位置计算（水印高度用 scale 后的宽度近似）
+                        wm_h = wm_scale  # 近似（正方形假设）
+                        x, y = WatermarkEngine.calc_position(
+                            watermark_config.position,
+                            self.output_width,
+                            self.output_height,
+                            wm_scale,
+                            wm_h,
+                            watermark_config.margin_x,
+                            watermark_config.margin_y,
+                        )
+
+                        # 滚动水印
+                        if watermark_config.scroll:
+                            x_expr = f"W-mod({watermark_config.scroll_speed}*t\\,W+w)"
+                            overlay = f"[{final_video_label}][wm_scaled]overlay=x={x_expr}:y={y}[{wm_label}]"
+                        else:
+                            overlay = f"[{final_video_label}][wm_scaled]overlay=x={x}:y={y}[{wm_label}]"
+
+                        filter_parts.append(overlay)
+                        final_video_label = wm_label
+                    else:
+                        logger.warning("水印图片不存在，跳过水印: %s", wm_path)
+                elif watermark_config.mode == "text":
+                    # 文字水印
+                    try:
+                        text_wm = WatermarkEngine.build_text_watermark_filter(
+                            f"[{final_video_label}]",
+                            f"[{wm_label}]",
+                            watermark_config,
+                            self.output_width,
+                            self.output_height,
+                        )
+                        filter_parts.append(text_wm)
+                        final_video_label = wm_label
+                    except Exception as e:
+                        logger.warning("文字水印构建失败，跳过: %s", e)
+
         # 叠加字幕（如有）+ 最终像素格式
         if ass_path is not None:
             ass_filter_path = str(ass_path).replace("\\", "/").replace(":", "\\:")