fix(ci): Frontend Lint所有步骤共享node_modules缓存volume

feat(ci): P2债务清理 + Production部署门禁

- 升级PyJWT 2.9.0 -> 2.13.0，修复4个CVE
- 清理5处死代码（未使用import和参数）
- Production部署加健康检查门禁（API health + 登录接口 + Web前端）
- deploy-production加checkout，修复通知脚本路径问题

.ci-trigger

@@ -1 +1 @@
-# CI trigger Fri Jun 26 09:53:28 PM CST 2026
+trigger: 1784009947

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

@@ -211,6 +211,7 @@ class _PlanClipItem(BaseModel):
     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/templates.py

@@ -45,7 +45,6 @@ from packages.application.template.use_cases import (
     CreateTemplateUseCase,
     DeleteCategoryUseCase,
     DeleteTemplateUseCase,
-    GetTemplateUsageUseCase,
     GetTemplateUseCase,
     ListCategoriesUseCase,
     ListTagsUseCase,

apps/api/app/services/edit_plan_service.py

@@ -282,6 +282,7 @@ class EditPlanService:
         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:
         """创建片段
@@ -303,6 +304,7 @@ class EditPlanService:
             duration=duration,
             transition_effect=transition_effect,
             transition_duration=transition_duration,
+            playback_speed=playback_speed,
             config=config,
         )
         created = self._clip_repo.create(clip)
@@ -327,6 +329,7 @@ class EditPlanService:
         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:
         """更新片段
@@ -336,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,
@@ -352,6 +364,7 @@ class EditPlanService:
             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/web/e2e/core-generation.spec.ts

@@ -251,6 +251,9 @@ test.describe("Core generation flow", () => {
     await expect(page.locator(".xx-products-page")).toBeVisible({
       timeout: 15_000,
     });
+
+    // 清理所有路由，避免页面关闭时飞地API请求导致测试报错
+    await page.unrouteAll({ behavior: "ignoreErrors" });
   });
 
   test("generation task API creates and lists tasks", async ({ request }) => {

apps/worker/video_processing/concat_engine.py

@@ -0,0 +1,627 @@
+"""视频拼接/合并引擎 — 多段视频按顺序拼接成一个成片.
+
+基于 FFmpeg 实现两种拼接模式：
+1. **concat demuxer（stream copy）**：最快，所有视频编码参数必须一致
+2. **concat filter（重新编码）**：更灵活，支持不同分辨率/编码/帧率的视频
+
+使用场景：
+- 多段素材按顺序合并成一个视频
+- 视频分割后重新拼接
+- 片头 + 正片 + 片尾拼接
+
+降级策略：
+- 优先尝试 stream copy（速度快、无质量损失）
+- 参数不一致时自动降级到 concat filter
+- 某段视频失败时跳过，不阻断整体拼接
+"""
+
+from __future__ import annotations
+
+import logging
+import tempfile
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any
+
+from video_processing.ffmpeg_utils import FFMPEG_BIN, probe_duration, probe_video_info, run_ffmpeg
+
+logger = logging.getLogger(__name__)
+
+
+# ── 常量 ──────────────────────────────────────────────────────────────────────
+
+# concat demuxer 要求一致的参数列表
+CONCAT_DEMUXER_REQUIRED_PARAMS = [
+    "codec_name",  # 视频编码
+    "width",  # 宽度
+    "height",  # 高度
+    "r_frame_rate",  # 帧率
+    "pix_fmt",  # 像素格式
+    "sample_rate",  # 音频采样率
+    "channels",  # 音频声道数
+    "audio_codec",  # 音频编码
+]
+
+
+# ── 拼接片段配置 ──────────────────────────────────────────────────────────────
+
+
+@dataclass
+class ConcatSegment:
+    """单个拼接片段."""
+
+    video_path: str  # 视频文件路径
+    start_time: float = 0.0  # 开始时间（秒），从视频的哪个位置开始取
+    duration: float = 0.0  # 持续时长（秒），0表示取到末尾
+    has_audio: bool = True  # 是否包含音频
+
+    @classmethod
+    def from_dict(cls, seg: dict) -> "ConcatSegment":
+        """从字典创建拼接片段，带安全类型转换."""
+        try:
+            start_time = max(0.0, float(seg.get("start_time", 0.0)))
+        except (TypeError, ValueError):
+            start_time = 0.0
+
+        try:
+            duration = max(0.0, float(seg.get("duration", 0.0)))
+        except (TypeError, ValueError):
+            duration = 0.0
+
+        return cls(
+            video_path=str(seg.get("video_path", "")),
+            start_time=start_time,
+            duration=duration,
+            has_audio=bool(seg.get("has_audio", True)),
+        )
+
+
+@dataclass
+class ConcatConfig:
+    """视频拼接配置."""
+
+    segments: list[ConcatSegment] = field(default_factory=list)
+    output_width: int = 0  # 输出宽度（0=自动取第一段）
+    output_height: int = 0  # 输出高度（0=自动取第一段）
+    output_fps: float = 0.0  # 输出帧率（0=自动取第一段）
+    force_reencode: bool = False  # 强制重新编码（不用 stream copy）
+    transition: str = "none"  # 转场效果（none/crossfade）- 预留
+    transition_duration: float = 0.3  # 转场时长
+
+    @classmethod
+    def from_config_dict(cls, config: dict | None) -> "ConcatConfig":
+        """从配置字典创建 ConcatConfig."""
+        if not config or not isinstance(config, dict):
+            return cls()
+
+        segments_raw = config.get("segments", [])
+        segments: list[ConcatSegment] = []
+
+        if isinstance(segments_raw, list):
+            for s in segments_raw:
+                if isinstance(s, dict) and s.get("video_path"):
+                    try:
+                        seg = ConcatSegment.from_dict(s)
+                        if seg.video_path:
+                            segments.append(seg)
+                    except Exception:
+                        logger.warning("[concat] skip invalid segment: %s", s)
+                        continue
+
+        try:
+            output_width = max(0, int(config.get("output_width", 0)))
+        except (TypeError, ValueError):
+            output_width = 0
+
+        try:
+            output_height = max(0, int(config.get("output_height", 0)))
+        except (TypeError, ValueError):
+            output_height = 0
+
+        try:
+            output_fps = max(0.0, float(config.get("output_fps", 0.0)))
+        except (TypeError, ValueError):
+            output_fps = 0.0
+
+        return cls(
+            segments=segments,
+            output_width=output_width,
+            output_height=output_height,
+            output_fps=output_fps,
+            force_reencode=bool(config.get("force_reencode", False)),
+            transition=str(config.get("transition", "none")),
+            transition_duration=max(0.1, float(config.get("transition_duration", 0.3))),
+        )
+
+    @property
+    def has_effect(self) -> bool:
+        """是否有有效片段需要拼接."""
+        return len([s for s in self.segments if s.video_path]) >= 2
+
+    @property
+    def total_segments(self) -> int:
+        """有效片段数量."""
+        return len([s for s in self.segments if s.video_path])
+
+
+# ── 视频拼接引擎 ──────────────────────────────────────────────────────────────
+
+
+class ConcatEngine:
+    """视频拼接引擎 — 支持 stream copy 和重新编码两种模式."""
+
+    def __init__(self, work_dir: Path):
+        self.work_dir = work_dir
+        self.work_dir.mkdir(parents=True, exist_ok=True)
+
+    # ── 主入口 ────────────────────────────────────────────────────────
+
+    def concat_videos(
+        self,
+        config: ConcatConfig,
+        output_path: Path,
+    ) -> Path:
+        """拼接多段视频.
+
+        自动选择最优拼接策略：
+        1. 所有片段参数一致 → concat demuxer（stream copy，最快）
+        2. 参数不一致或有裁剪 → concat filter（重新编码）
+
+        Args:
+            config: 拼接配置
+            output_path: 输出文件路径
+
+        Returns:
+            输出文件路径
+        """
+        valid_segments = [s for s in config.segments if s.video_path]
+
+        if not valid_segments:
+            raise ValueError("No valid video segments to concat")
+
+        if len(valid_segments) == 1:
+            # 只有一段，直接复制
+            import shutil
+
+            logger.info("[concat] single segment, copy directly")
+            shutil.copy2(valid_segments[0].video_path, output_path)
+            return output_path
+
+        # 判断能否用 stream copy
+        can_stream_copy = self._can_use_stream_copy(config)
+
+        if can_stream_copy and not config.force_reencode:
+            logger.info("[concat] using concat demuxer (stream copy)")
+            try:
+                return self._concat_demuxer(config, output_path)
+            except Exception as e:
+                logger.warning("[concat] demuxer failed, fallback to filter: %s", e)
+
+        # 降级到 concat filter
+        logger.info("[concat] using concat filter (re-encode)")
+        return self._concat_filter(config, output_path)
+
+    # ── 模式判断 ──────────────────────────────────────────────────────
+
+    def _can_use_stream_copy(self, config: ConcatConfig) -> bool:
+        """判断是否可以使用 concat demuxer（stream copy）.
+
+        条件：
+        1. 所有视频编码参数一致（分辨率、帧率、编码、像素格式）
+        2. 所有音频参数一致（采样率、声道、编码）
+        3. 没有设置 start_time 裁剪（或可以通过 concat demuxer 的 inpoint/outpoint 实现）
+        4. 没有强制重新编码
+        """
+        if config.force_reencode:
+            return False
+
+        # 如果有转场效果，必须重新编码
+        if config.transition != "none":
+            return False
+
+        # 探测所有视频的参数
+        video_infos = []
+        for seg in config.segments:
+            if not seg.video_path:
+                continue
+            try:
+                info = probe_video_info(seg.video_path)
+                video_infos.append(info)
+            except Exception:
+                logger.warning("[concat] probe failed for %s", seg.video_path[-40:])
+                return False
+
+        if len(video_infos) < 2:
+            return False
+
+        # 检查参数一致性
+        base_info = video_infos[0]
+        for info in video_infos[1:]:
+            for param in CONCAT_DEMUXER_REQUIRED_PARAMS:
+                base_val = base_info.get(param)
+                curr_val = info.get(param)
+                if base_val != curr_val:
+                    logger.debug(
+                        "[concat] param mismatch: %s (%s vs %s)",
+                        param,
+                        base_val,
+                        curr_val,
+                    )
+                    return False
+
+        # 检查是否有裁剪需求
+        # concat demuxer 支持 inpoint/outpoint，所以有裁剪也可以用
+        # 但为了简单和稳定性，有裁剪时也用 filter 模式
+        # （inpoint/outpoint 不是所有格式都支持得好）
+        has_trimming = any(seg.start_time > 0 or seg.duration > 0 for seg in config.segments if seg.video_path)
+        if has_trimming:
+            return False
+
+        return True
+
+    # ── 模式1：concat demuxer（stream copy） ──────────────────────────
+
+    def _concat_demuxer(self, config: ConcatConfig, output_path: Path) -> Path:
+        """使用 concat demuxer 拼接（stream copy）.
+
+        优点：速度极快，无质量损失
+        缺点：要求所有视频参数完全一致
+        """
+        # 生成 concat 文件列表
+        list_file = self.work_dir / "concat_list.txt"
+        lines = []
+        for seg in config.segments:
+            if not seg.video_path:
+                continue
+            # 路径转义：单引号替换为 '\''
+            safe_path = str(seg.video_path).replace("'", "'\\''")
+            lines.append(f"file '{safe_path}'")
+
+        list_file.write_text("\n".join(lines), encoding="utf-8")
+
+        command = [
+            FFMPEG_BIN,
+            "-y",
+            "-f",
+            "concat",
+            "-safe",
+            "0",
+            "-i",
+            str(list_file),
+            "-c",
+            "copy",
+            "-copyts",
+            str(output_path),
+        ]
+
+        logger.info("[concat] demuxer: %d segments", config.total_segments)
+        run_ffmpeg(command)
+        return output_path
+
+    # ── 模式2：concat filter（重新编码） ──────────────────────────────
+
+    def _concat_filter(self, config: ConcatConfig, output_path: Path) -> Path:
+        """使用 concat filter 拼接（重新编码）.
+
+        优点：支持不同参数的视频，支持裁剪
+        缺点：需要重新编码，较慢
+        """
+        valid_segments = [s for s in config.segments if s.video_path]
+        num_segments = len(valid_segments)
+
+        # 构建输入参数
+        input_args: list[str] = []
+        for seg in valid_segments:
+            input_args.extend(["-i", seg.video_path])
+
+        # 确定输出参数
+        output_width, output_height, output_fps = self._get_output_params(config)
+
+        # 构建 filter_complex
+        filter_parts: list[str] = []
+        concat_inputs = ""
+
+        for i, seg in enumerate(valid_segments):
+            vid_label = f"v{i}"
+            aud_label = f"a{i}"
+
+            seg_filters: list[str] = []
+
+            # 1. 裁剪（start_time + duration）
+            if seg.start_time > 0 or seg.duration > 0:
+                start = seg.start_time
+                if seg.duration > 0:
+                    end = start + seg.duration
+                    seg_filters.append(f"trim=start={start:.3f}:end={end:.3f}")
+                else:
+                    seg_filters.append(f"trim=start={start:.3f}")
+                seg_filters.append("setpts=PTS-STARTPTS")
+
+                # 音频同步裁剪
+                if seg.has_audio:
+                    if seg.duration > 0:
+                        filter_parts.append(
+                            f"[{i}:a]atrim=start={start:.3f}:end={end:.3f}," f"asetpts=PTS-STARTPTS[{aud_label}]"
+                        )
+                    else:
+                        filter_parts.append(f"[{i}:a]atrim=start={start:.3f}," f"asetpts=PTS-STARTPTS[{aud_label}]")
+                else:
+                    # 无音频时生成静音轨
+                    filter_parts.append(
+                        f"[{i}:v]trim=start={start:.3f}," f"setpts=PTS-STARTPTS, " f"aevalsrc=0:d={0.1}[{aud_label}]"
+                    )
+            else:
+                # 无裁剪，直接用原始标签
+                if not seg.has_audio:
+                    # 无音频时需要生成静音
+                    try:
+                        dur = probe_duration(seg.video_path)
+                    except Exception:
+                        dur = 10.0
+                    filter_parts.append(f"aevalsrc=0:d={dur:.3f}:s=44100[{aud_label}]")
+
+            # 2. 缩放/帧率统一
+            vf_parts = []
+            if not seg_filters:
+                vf_parts.append(f"[{i}:v]")
+            else:
+                vf_parts.append("")
+
+            # 分辨率统一
+            if output_width and output_height:
+                vf_parts.append(
+                    f"scale={output_width}:{output_height}:force_original_aspect_ratio=decrease,"
+                    f"pad={output_width}:{output_height}:(ow-iw)/2:(oh-ih)/2:black"
+                )
+
+            # 帧率统一
+            if output_fps > 0:
+                vf_parts.append(f"fps={output_fps}")
+
+            # 像素格式统一
+            vf_parts.append("format=yuv420p")
+
+            if len(vf_parts) > 1 or (seg_filters and vf_parts):
+                if seg_filters:
+                    # 先裁剪后缩放
+                    crop_str = "".join(seg_filters)
+                    scale_str = "".join(vf_parts[1:])  # 跳过空字符串
+                    if scale_str:
+                        filter_parts.append(f"[{i}:v]{crop_str},{scale_str}[{vid_label}]")
+                    else:
+                        filter_parts.append(f"[{i}:v]{crop_str}[{vid_label}]")
+                else:
+                    filter_parts.append(f"{vf_parts[0]}{''.join(vf_parts[1:])}[{vid_label}]")
+            else:
+                if seg_filters:
+                    filter_parts.append(f"[{i}:v]{''.join(seg_filters)}[{vid_label}]")
+                else:
+                    # 什么都不需要，直接用输入
+                    pass
+
+            # 拼接 concat 的输入标签
+            if seg_filters or (output_width and output_height) or output_fps > 0:
+                concat_inputs += f"[{vid_label}]"
+            else:
+                concat_inputs += f"[{i}:v]"
+
+            # 音频标签
+            if seg.start_time > 0 or seg.duration > 0:
+                # 已经生成了 aud_label
+                pass
+            elif not seg.has_audio:
+                # 已经生成了静音 aud_label
+                pass
+            else:
+                # 使用原始音频
+                pass
+
+        # 简化处理：用更直接的方式构建 filter
+        # 重新整理一下，确保所有输入都有对应的 v_i 和 a_i 标签
+        filter_parts.clear()
+        concat_inputs = ""  # 按段交织: [v0][a0][v1][a1]...
+
+        for i, seg in enumerate(valid_segments):
+            v_label = f"v{i}_in"
+            a_label = f"a{i}_in"
+
+            # 视频处理链
+            v_steps: list[str] = [f"[{i}:v]"]
+
+            # 裁剪
+            if seg.start_time > 0 or seg.duration > 0:
+                start = seg.start_time
+                if seg.duration > 0:
+                    end = start + seg.duration
+                    v_steps.append(f"trim=start={start:.3f}:end={end:.3f},")
+                else:
+                    v_steps.append(f"trim=start={start:.3f},")
+                v_steps.append("setpts=PTS-STARTPTS,")
+
+            # 缩放
+            if output_width and output_height:
+                v_steps.append(
+                    f"scale={output_width}:{output_height}:force_original_aspect_ratio=decrease,"
+                    f"pad={output_width}:{output_height}:(ow-iw)/2:(oh-ih)/2:black,"
+                )
+
+            # 帧率
+            if output_fps > 0:
+                v_steps.append(f"fps={output_fps},")
+
+            # 像素格式
+            v_steps.append("format=yuv420p")
+
+            v_filter = "".join(v_steps) + f"[{v_label}]"
+            filter_parts.append(v_filter)
+
+            # 音频处理链
+            a_steps: list[str] = []
+            if seg.has_audio:
+                a_steps.append(f"[{i}:a]")
+
+                if seg.start_time > 0 or seg.duration > 0:
+                    start = seg.start_time
+                    if seg.duration > 0:
+                        end = start + seg.duration
+                        a_steps.append(f"atrim=start={start:.3f}:end={end:.3f},")
+                    else:
+                        a_steps.append(f"atrim=start={start:.3f},")
+                    a_steps.append("asetpts=PTS-STARTPTS,")
+
+                a_steps.append("aformat=sample_fmts=fltp:sample_rates=44100:channel_layouts=stereo")
+            else:
+                # 生成静音音频
+                try:
+                    dur = probe_duration(seg.video_path)
+                except Exception:
+                    dur = 10.0
+                # 减去裁剪
+                if seg.start_time > 0:
+                    dur = max(0.1, dur - seg.start_time)
+                if seg.duration > 0 and seg.duration < dur:
+                    dur = seg.duration
+                a_steps.append(f"aevalsrc=0:d={dur:.3f}:s=44100:c=stereo")
+
+            a_filter = "".join(a_steps) + f"[{a_label}]"
+            filter_parts.append(a_filter)
+
+            # 按段交织排列（v_i, a_i），这是 FFmpeg concat filter 要求的顺序
+            concat_inputs += f"[{v_label}][{a_label}]"
+
+        # concat filter: 输入按 [v0][a0][v1][a1]... 顺序
+        filter_parts.append(f"{concat_inputs}" f"concat=n={num_segments}:v=1:a=1[vout][aout]")
+
+        filter_complex = ";".join(filter_parts)
+
+        command = [
+            FFMPEG_BIN,
+            "-y",
+            *input_args,
+            "-filter_complex",
+            filter_complex,
+            "-map",
+            "[vout]",
+            "-map",
+            "[aout]",
+            "-c:v",
+            "libx264",
+            "-preset",
+            "fast",
+            "-crf",
+            "23",
+            "-c:a",
+            "aac",
+            "-b:a",
+            "128k",
+            "-movflags",
+            "+faststart",
+            str(output_path),
+        ]
+
+        logger.info(
+            "[concat] filter: %d segments, %dx%d, %.2f fps",
+            num_segments,
+            output_width,
+            output_height,
+            output_fps,
+        )
+        run_ffmpeg(command)
+        return output_path
+
+    # ── 辅助方法 ──────────────────────────────────────────────────────
+
+    def _get_output_params(self, config: ConcatConfig) -> tuple[int, int, float]:
+        """获取输出参数（宽、高、帧率）.
+
+        优先级：
+        1. config 中显式指定的
+        2. 第一段视频的参数
+        """
+        valid_segments = [s for s in config.segments if s.video_path]
+
+        width = config.output_width
+        height = config.output_height
+        fps = config.output_fps
+
+        # 如果没有显式指定，用第一段的参数
+        if (width == 0 or height == 0 or fps == 0) and valid_segments:
+            try:
+                info = probe_video_info(valid_segments[0].video_path)
+                if width == 0:
+                    width = int(info.get("width", 1080))
+                if height == 0:
+                    height = int(info.get("height", 1920))
+                if fps == 0:
+                    fps_str = info.get("r_frame_rate", "30/1")
+                    if "/" in str(fps_str):
+                        num, den = str(fps_str).split("/")
+                        try:
+                            fps = float(num) / float(den)
+                        except (ValueError, ZeroDivisionError):
+                            fps = 30.0
+                    else:
+                        fps = float(fps_str) if fps_str else 30.0
+            except Exception:
+                # 探测失败，用默认值
+                if width == 0:
+                    width = 1080
+                if height == 0:
+                    height = 1920
+                if fps == 0:
+                    fps = 30.0
+
+        return width, height, fps
+
+
+# ── 便捷函数 ──────────────────────────────────────────────────────────────────
+
+
+def concat_video_files(
+    video_paths: list[str],
+    output_path: Path,
+    *,
+    work_dir: Path | None = None,
+    force_reencode: bool = False,
+) -> Path:
+    """简单拼接多个视频文件.
+
+    Args:
+        video_paths: 视频文件路径列表
+        output_path: 输出路径
+        work_dir: 工作目录（默认输出文件所在目录）
+        force_reencode: 是否强制重新编码
+
+    Returns:
+        输出文件路径
+    """
+    if work_dir is None:
+        work_dir = output_path.parent
+
+    segments = [ConcatSegment(video_path=p) for p in video_paths if p]
+    config = ConcatConfig(segments=segments, force_reencode=force_reencode)
+
+    engine = ConcatEngine(work_dir)
+    return engine.concat_videos(config, output_path)
+
+
+def concat_videos_from_config(
+    config_dict: dict | None,
+    output_path: Path,
+    *,
+    work_dir: Path,
+) -> Path | None:
+    """从配置字典执行视频拼接.
+
+    降级策略：配置无效或拼接失败时返回 None.
+    """
+    config = ConcatConfig.from_config_dict(config_dict)
+    if not config.has_effect:
+        return None
+
+    try:
+        engine = ConcatEngine(work_dir)
+        return engine.concat_videos(config, output_path)
+    except Exception as e:
+        logger.error("[concat] concat failed: %s", e)
+        return None

apps/worker/video_processing/multi_track_mixer.py

@@ -0,0 +1,392 @@
+"""多轨道混音引擎 — 支持多路音频独立音量调节与混合.
+
+基于 FFmpeg amix / amerge 实现：
+- 支持任意数量音频轨道（原音、BGM、配音、音效等）
+- 每轨独立音量调节
+- 每轨独立淡入淡出
+- 每轨独立时间偏移（delay）
+- 总输出音量归一化补偿
+
+作为 render_audio.py 的增强模块，在 mix_audio 后处理阶段被调用。
+与 bgm_mixer.py 的关系：
+- bgm_mixer 专注 BGM 单轨道的复杂处理（循环、人声闪避）
+- 本模块专注多路轨道的统一音量调节与混合
+"""
+
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import TYPE_CHECKING
+
+from video_processing.ffmpeg_utils import FFMPEG_BIN, probe_duration, run_ffmpeg
+
+if TYPE_CHECKING:
+    from video_processing.render_audio import RenderContext
+
+logger = logging.getLogger(__name__)
+
+
+# ── 常量 ──────────────────────────────────────────────────────────────────────
+
+TRACK_TYPE_MAIN = "main"  # 原音（视频原声）
+TRACK_TYPE_BGM = "bgm"  # 背景音乐
+TRACK_TYPE_VOICEOVER = "voiceover"  # 配音（TTS/人声）
+TRACK_TYPE_SFX = "sfx"  # 音效
+TRACK_TYPE_AMBIENT = "ambient"  # 环境音
+
+# 各轨道默认音量（相对主音频）
+DEFAULT_VOLUMES = {
+    TRACK_TYPE_MAIN: 1.0,
+    TRACK_TYPE_BGM: 0.3,
+    TRACK_TYPE_VOICEOVER: 1.0,
+    TRACK_TYPE_SFX: 0.7,
+    TRACK_TYPE_AMBIENT: 0.2,
+}
+
+
+@dataclass
+class AudioTrack:
+    """单条音频轨道配置."""
+
+    track_id: str  # 轨道唯一标识
+    track_type: str  # 轨道类型（main/bgm/voiceover/sfx/ambient）
+    audio_path: str  # 音频文件路径
+    volume: float = 1.0  # 音量 0.0 ~ 2.0
+    fade_in: float = 0.0  # 淡入时长（秒）
+    fade_out: float = 0.0  # 淡出时长（秒）
+    start_time: float = 0.0  # 开始时间（相对于视频起点，秒）
+    duration: float = 0.0  # 持续时长（0表示到文件末尾）
+    enabled: bool = True  # 是否启用
+
+    @classmethod
+    def from_dict(cls, track: dict) -> "AudioTrack":
+        """从字典创建 AudioTrack，带安全类型转换."""
+        track_type = str(track.get("track_type", TRACK_TYPE_SFX))
+        default_vol = DEFAULT_VOLUMES.get(track_type, 1.0)
+
+        try:
+            volume = float(track.get("volume", default_vol))
+        except (TypeError, ValueError):
+            volume = default_vol
+        volume = max(0.0, min(2.0, volume))
+
+        try:
+            fade_in = max(0.0, float(track.get("fade_in", 0.0)))
+        except (TypeError, ValueError):
+            fade_in = 0.0
+
+        try:
+            fade_out = max(0.0, float(track.get("fade_out", 0.0)))
+        except (TypeError, ValueError):
+            fade_out = 0.0
+
+        try:
+            start_time = max(0.0, float(track.get("start_time", 0.0)))
+        except (TypeError, ValueError):
+            start_time = 0.0
+
+        try:
+            duration = max(0.0, float(track.get("duration", 0.0)))
+        except (TypeError, ValueError):
+            duration = 0.0
+
+        return cls(
+            track_id=str(track.get("track_id", "")),
+            track_type=track_type,
+            audio_path=str(track.get("audio_path", "")),
+            volume=volume,
+            fade_in=fade_in,
+            fade_out=fade_out,
+            start_time=start_time,
+            duration=duration,
+            enabled=bool(track.get("enabled", True)),
+        )
+
+
+@dataclass
+class MultiTrackMixConfig:
+    """多轨道混音配置."""
+
+    tracks: list[AudioTrack] = field(default_factory=list)
+    master_volume: float = 1.0  # 主输出音量
+    normalize: bool = True  # 是否自动归一化补偿
+    max_output_volume: float = 1.5  # 最大输出音量（防止爆音）
+
+    @classmethod
+    def from_config_dict(cls, config: dict | None) -> "MultiTrackMixConfig":
+        """从 plan.config.audio_tracks 字典创建配置."""
+        if not config or not isinstance(config, dict):
+            return cls()
+
+        tracks_raw = config.get("tracks", [])
+        tracks: list[AudioTrack] = []
+
+        if isinstance(tracks_raw, list):
+            for t in tracks_raw:
+                if isinstance(t, dict) and t.get("audio_path"):
+                    try:
+                        track = AudioTrack.from_dict(t)
+                        if track.enabled and track.audio_path:
+                            tracks.append(track)
+                    except Exception:
+                        logger.warning("[multi-track] skip invalid track config: %s", t)
+                        continue
+
+        try:
+            master_volume = float(config.get("master_volume", 1.0))
+            master_volume = max(0.0, min(2.0, master_volume))
+        except (TypeError, ValueError):
+            master_volume = 1.0
+
+        return cls(
+            tracks=tracks,
+            master_volume=master_volume,
+            normalize=bool(config.get("normalize", True)),
+            max_output_volume=float(config.get("max_output_volume", 1.5)),
+        )
+
+    @property
+    def has_effect(self) -> bool:
+        """是否有有效轨道需要混音."""
+        return len([t for t in self.tracks if t.enabled and t.audio_path]) > 0
+
+
+# ── 单轨道预处理 ────────────────────────────────────────────────────────────
+
+
+def _prepare_single_track(
+    ctx: "RenderContext",
+    track: AudioTrack,
+    target_duration: float,
+    output_path: Path,
+) -> bool:
+    """预处理单条轨道：音量 + 淡入淡出 + 时间偏移 + 截断.
+
+    生成一个精确对齐时间轴的音频文件，后续统一 amix 混音。
+
+    Returns:
+        True 表示处理成功，False 表示失败（跳过）
+    """
+    try:
+        audio_dur = probe_duration(track.audio_path)
+    except Exception:
+        logger.warning("[multi-track] probe failed, skip track: %s", track.track_id)
+        return False
+
+    if audio_dur <= 0:
+        return False
+
+    # 计算实际有效时长
+    effective_start = track.start_time
+    if track.duration > 0:
+        effective_dur = min(track.duration, audio_dur)
+    else:
+        effective_dur = audio_dur
+
+    # 如果轨道完全在视频时长之外，跳过
+    if effective_start >= target_duration:
+        return False
+    if effective_start + effective_dur <= 0:
+        return False
+
+    # 构建滤镜链
+    filter_parts: list[str] = []
+
+    # 1. 先截断到有效范围
+    trim_start = 0.0  # 从源文件的哪个位置开始取
+    if effective_start < 0:
+        trim_start = -effective_start
+        effective_start = 0.0
+
+    # 实际需要的源时长
+    need_dur = min(effective_dur, target_duration - effective_start)
+    if need_dur <= 0:
+        return False
+
+    filter_parts.append(f"atrim={trim_start:.3f}:{trim_start + need_dur:.3f}")
+    filter_parts.append("asetpts=N/SR/TB")
+
+    # 2. 音量调节
+    if abs(track.volume - 1.0) > 0.001:
+        filter_parts.append(f"volume={track.volume:.3f}")
+
+    # 3. 淡入
+    if track.fade_in > 0 and track.fade_in < need_dur:
+        filter_parts.append(f"afade=t=in:st=0:d={track.fade_in:.3f}")
+
+    # 4. 淡出
+    if track.fade_out > 0 and track.fade_out < need_dur:
+        fade_start = need_dur - track.fade_out
+        if fade_start > 0:
+            filter_parts.append(f"afade=t=out:st={fade_start:.3f}:d={track.fade_out:.3f}")
+
+    # 5. 时间偏移（用 adelay 实现开头静音填充）
+    if effective_start > 0.01:
+        delay_ms = int(effective_start * 1000)
+        filter_parts.append(f"adelay={delay_ms}|{delay_ms}")
+
+    # 6. 最终截断到目标总时长
+    filter_parts.append(f"atrim=0:{target_duration:.3f}")
+    filter_parts.append("asetpts=N/SR/TB")
+
+    filter_str = ",".join(filter_parts)
+
+    command = [
+        FFMPEG_BIN,
+        "-y",
+        "-i",
+        track.audio_path,
+        "-filter:a",
+        filter_str,
+        "-c:a",
+        "aac",
+        "-b:a",
+        "128k",
+        str(output_path),
+    ]
+
+    logger.info(
+        "[multi-track] prepare track: id=%s type=%s vol=%.2f start=%.2f dur=%.2f",
+        track.track_id,
+        track.track_type,
+        track.volume,
+        effective_start,
+        need_dur,
+    )
+
+    try:
+        run_ffmpeg(command)
+        return True
+    except Exception as e:
+        logger.warning("[multi-track] track prepare failed: %s, error=%s", track.track_id, e)
+        return False
+
+
+# ── 多轨道混音主入口 ─────────────────────────────────────────────────────────
+
+
+def mix_multi_track(
+    ctx: "RenderContext",
+    main_audio_path: Path,
+    config: MultiTrackMixConfig,
+    target_duration: float,
+) -> Path:
+    """多轨道混音：主音频 + 多条附加轨道.
+
+    Args:
+        ctx: 渲染上下文
+        main_audio_path: 主音频文件路径（原音）
+        config: 多轨道混音配置
+        target_duration: 目标总时长
+
+    Returns:
+        混音后的音频文件路径
+    """
+    output_path = ctx.work_dir / f"multi_track_mix_{ctx.plan_id}.aac"
+
+    if target_duration <= 0:
+        target_duration = 5.0
+
+    # 收集所有有效轨道（已预处理好的）
+    prepared_tracks: list[Path] = []
+
+    # 主音频作为第0轨
+    prepared_tracks.append(main_audio_path)
+
+    # 预处理每条附加轨道
+    for i, track in enumerate(config.tracks):
+        if not track.enabled or not track.audio_path:
+            continue
+
+        track_out = ctx.work_dir / f"track_{i}_{ctx.plan_id}.aac"
+        if _prepare_single_track(ctx, track, target_duration, track_out):
+            prepared_tracks.append(track_out)
+
+    # 如果只有主音频，直接返回（无需混音）
+    if len(prepared_tracks) <= 1:
+        import shutil
+
+        shutil.copy2(main_audio_path, output_path)
+        return output_path
+
+    # 使用 amix 混音
+    num_inputs = len(prepared_tracks)
+
+    # 构建输入参数
+    input_args: list[str] = []
+    for tp in prepared_tracks:
+        input_args.extend(["-i", str(tp)])
+
+    # amix 的 duration=first 以第一个输入（主音频）时长为准
+    # normalize 补偿：amix 会把每路音量除以 N，需要乘回来
+    # 但如果所有轨道都同时有声，可能会爆音，所以用 master_volume 控制
+    if config.normalize:
+        # 经验值：不是所有轨道都同时有声，补偿系数取 N * 0.7
+        compensate = num_inputs * 0.7
+    else:
+        compensate = 1.0
+
+    final_volume = compensate * config.master_volume
+    final_volume = min(final_volume, config.max_output_volume)
+
+    # 构建 filter_complex
+    inputs_label = "".join(f"[{i}:a]" for i in range(num_inputs))
+    filter_complex = (
+        f"{inputs_label}amix=inputs={num_inputs}:duration=first:dropout_transition=0[outa];"
+        f"[outa]volume={final_volume:.3f}[final]"
+    )
+
+    command = [
+        FFMPEG_BIN,
+        "-y",
+        *input_args,
+        "-filter_complex",
+        filter_complex,
+        "-map",
+        "[final]",
+        "-c:a",
+        "aac",
+        "-b:a",
+        "128k",
+        str(output_path),
+    ]
+
+    logger.info(
+        "[multi-track] mix %d tracks, master_vol=%.2f compensate=%.2f final_vol=%.2f",
+        num_inputs,
+        config.master_volume,
+        compensate,
+        final_volume,
+    )
+
+    try:
+        run_ffmpeg(command)
+    except Exception as e:
+        logger.error("[multi-track] mix failed, fallback to main audio only: %s", e)
+        import shutil
+
+        shutil.copy2(main_audio_path, output_path)
+
+    return output_path
+
+
+# ── 便捷函数：从 plan.config 快速混音 ───────────────────────────────────────
+
+
+def mix_audio_tracks_from_config(
+    ctx: "RenderContext",
+    main_audio_path: Path,
+    audio_tracks_config: dict | None,
+    target_duration: float,
+) -> Path:
+    """从 plan.config.audio_tracks 配置执行多轨道混音.
+
+    降级策略：配置无效或混音失败时返回主音频。
+    """
+    config = MultiTrackMixConfig.from_config_dict(audio_tracks_config)
+    if not config.has_effect:
+        return main_audio_path
+
+    return mix_multi_track(ctx, main_audio_path, config, target_duration)

apps/worker/video_processing/render_audio.py

@@ -23,6 +23,7 @@ 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
@@ -76,6 +77,7 @@ def mix_audio(
     *,
     bgm_path: str | None = None,
     bgm_config: dict | None = None,
+    audio_tracks_config: dict | None = None,
 ) -> Path | None:
     """音频后处理混音.
 
@@ -86,6 +88,8 @@ def mix_audio(
     4. 输出时长截断到 video_duration
     5. 无音频流的 clip 会被自动跳过，避免 FFmpeg 引用 [i:a] 失败
     6. 如果提供了 bgm_path，则额外混入 BGM（支持淡入淡出、循环、人声闪避）
+    7. 如果配置了 audio_tracks，则混入多轨道音频（配音、音效等）
+    8. 如果配置了降噪，最后应用降噪
 
     Args:
         ctx: 渲染上下文
@@ -93,6 +97,7 @@ def mix_audio(
         video_duration: 视频总时长（用于截断音频）
         bgm_path: BGM 音频本地路径，为 None 时不混入 BGM
         bgm_config: BGM 配置字典（volume/fade_in/fade_out/sidechain 等）
+        audio_tracks_config: 多轨道音频配置（tracks/master_volume 等）
 
     Returns:
         混音后的音频文件路径，无音频时返回 None
@@ -156,10 +161,21 @@ def mix_audio(
         try:
             # 这里 main_audio 就是 output_path，先有主音频再混 BGM
             final_path = mix_bgm_with_main(ctx, output_path, bgm_cfg, video_duration)
-            return _apply_noise_reduction_if_needed(ctx, final_path)
+            output_path = final_path
         except Exception:
             logger.exception("[bgm] BGM 混音失败，回退到无 BGM 音频: plan_id=%s", ctx.plan_id)
-            return _apply_noise_reduction_if_needed(ctx, output_path)
+
+    # ── 多轨道混音（配音/音效等） ──
+    if audio_tracks_config and audio_tracks_config.get("enabled", False):
+        from video_processing.multi_track_mixer import mix_audio_tracks_from_config
+
+        try:
+            tracks_config = audio_tracks_config.get("tracks_config") or audio_tracks_config
+            multi_output = mix_audio_tracks_from_config(ctx, output_path, tracks_config, video_duration)
+            if multi_output and multi_output != output_path:
+                output_path = multi_output
+        except Exception:
+            logger.exception("[multi-track] 多轨道混音失败，回退: plan_id=%s", ctx.plan_id)
 
     return _apply_noise_reduction_if_needed(ctx, output_path)
 
@@ -225,53 +241,118 @@ 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
 
         # 音频倒放
         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)
+        has_reverse = reverse_config.enabled and reverse_config.reverse_audio
+        has_speed = abs(speed - 1.0) >= 1e-6
 
-        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 af_filters:
-            command.extend(["-af", ",".join(af_filters)])
-        if final_duration > 0:
-            command.extend(["-t", f"{final_duration:.3f}"])
-        command.append(str(output_path))
-        run_ffmpeg(command)
+        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:
             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:
             audio_filters.append("asetpts=PTS-STARTPTS")
 

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

@@ -0,0 +1,633 @@
+"""字幕渲染引擎 — 统一管理字幕样式配置与视频烧录.
+
+与现有模块的关系：
+- render_subtitles.py：生成静态整段标题/字幕的 ASS 文件
+- subtitle_generator.py：从 ASR 时间轴生成 ASS 文件
+- 本模块：统一的字幕样式配置 + 烧录滤镜生成 + 多源字幕合并
+
+支持的字幕来源：
+1. 静态标题/字幕（title_config / subtitle_config）
+2. ASR 自动字幕（asr_subtitle_timeline）
+3. 手动字幕（manual_subtitles 时间轴）
+
+支持的样式配置：
+- 字体、字号、颜色
+- 描边（颜色、宽度）
+- 阴影（偏移、模糊、颜色）
+- 背景框（颜色、透明度、圆角、边距）
+- 位置（9宫格 + 自定义坐标）
+- 对齐方式
+- 动画（淡入淡出、滑入滑出、打字机）
+- 多行/换行规则
+"""
+
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any
+
+from video_processing.render_subtitles import generate_ass_subtitles
+from video_processing.subtitle_generator import generate_ass_from_timeline
+
+logger = logging.getLogger(__name__)
+
+
+# ── 常量 ──────────────────────────────────────────────────────────────────────
+
+# 9宫格位置映射（ASS alignment 编号）
+POSITION_ALIGNMENT = {
+    "top_left": 7,
+    "top_center": 8,
+    "top_right": 9,
+    "middle_left": 4,
+    "center": 5,
+    "middle_right": 6,
+    "bottom_left": 1,
+    "bottom_center": 2,
+    "bottom_right": 3,
+}
+
+# 位置简称兼容
+POSITION_ALIASES = {
+    "top": "top_center",
+    "bottom": "bottom_center",
+    "middle": "center",
+    "left": "middle_left",
+    "right": "middle_right",
+}
+
+DEFAULT_FONT = "思源黑体"
+DEFAULT_FONT_SIZE = 24
+DEFAULT_COLOR = "#FFFFFF"
+DEFAULT_STROKE_COLOR = "#000000"
+DEFAULT_STROKE_WIDTH = 1.5
+DEFAULT_POSITION = "bottom_center"
+DEFAULT_MAX_CHARS_PER_LINE = 20
+
+
+# ── 字幕样式配置 ────────────────────────────────────────────────────────────
+
+
+@dataclass
+class SubtitleStyle:
+    """字幕样式配置."""
+
+    font_name: str = DEFAULT_FONT
+    font_size: int = DEFAULT_FONT_SIZE
+    font_color: str = DEFAULT_COLOR
+    bold: bool = False
+    italic: bool = False
+
+    # 描边
+    stroke_enabled: bool = True
+    stroke_color: str = DEFAULT_STROKE_COLOR
+    stroke_width: float = DEFAULT_STROKE_WIDTH
+
+    # 阴影
+    shadow_enabled: bool = False
+    shadow_color: str = "#000000"
+    shadow_offset_x: int = 2
+    shadow_offset_y: int = 2
+    shadow_blur: float = 0.0
+
+    # 背景框
+    background_enabled: bool = False
+    background_color: str = "#000000"
+    background_opacity: float = 0.5  # 0.0 ~ 1.0
+    background_padding: int = 8
+    background_radius: int = 4
+
+    # 位置
+    position: str = DEFAULT_POSITION  # 9宫格位置名
+    margin_v: int = 60  # 垂直边距
+    margin_l: int = 40  # 左边距
+    margin_r: int = 40  # 右边距
+
+    # 多行
+    max_chars_per_line: int = DEFAULT_MAX_CHARS_PER_LINE
+    line_spacing: int = 0  # 行间距
+
+    # 动画
+    fade_in: float = 0.0  # 淡入时长（秒）
+    fade_out: float = 0.0  # 淡出时长（秒）
+    animation_type: str = "none"  # none/fade/slide/typewriter
+
+    @classmethod
+    def from_dict(cls, config: dict[str, Any] | None) -> "SubtitleStyle":
+        """从字典创建样式配置，带安全类型转换."""
+        if not config or not isinstance(config, dict):
+            return cls()
+
+        def safe_str(key: str, default: str) -> str:
+            val = config.get(key, default)
+            return str(val) if val is not None else default
+
+        def safe_int(key: str, default: int) -> int:
+            try:
+                return int(config.get(key, default))
+            except (TypeError, ValueError):
+                return default
+
+        def safe_float(key: str, default: float) -> float:
+            try:
+                return float(config.get(key, default))
+            except (TypeError, ValueError):
+                return default
+
+        def safe_bool(key: str, default: bool) -> bool:
+            return bool(config.get(key, default))
+
+        position = safe_str("position", DEFAULT_POSITION)
+        position = POSITION_ALIASES.get(position, position)
+        if position not in POSITION_ALIGNMENT:
+            position = DEFAULT_POSITION
+
+        return cls(
+            font_name=safe_str("font", DEFAULT_FONT),
+            font_size=safe_int("size", DEFAULT_FONT_SIZE),
+            font_color=safe_str("color", DEFAULT_COLOR),
+            bold=safe_bool("bold", False),
+            italic=safe_bool("italic", False),
+            stroke_enabled=safe_bool("stroke_enabled", True),
+            stroke_color=safe_str("stroke_color", DEFAULT_STROKE_COLOR),
+            stroke_width=safe_float("stroke_width", DEFAULT_STROKE_WIDTH),
+            shadow_enabled=safe_bool("shadow_enabled", False),
+            shadow_color=safe_str("shadow_color", "#000000"),
+            shadow_offset_x=safe_int("shadow_offset_x", 2),
+            shadow_offset_y=safe_int("shadow_offset_y", 2),
+            shadow_blur=safe_float("shadow_blur", 0.0),
+            background_enabled=safe_bool("background_enabled", False),
+            background_color=safe_str("background_color", "#000000"),
+            background_opacity=max(0.0, min(1.0, safe_float("background_opacity", 0.5))),
+            background_padding=safe_int("background_padding", 8),
+            background_radius=safe_int("background_radius", 4),
+            position=position,
+            margin_v=safe_int("margin_v", 60),
+            margin_l=safe_int("margin_l", 40),
+            margin_r=safe_int("margin_r", 40),
+            max_chars_per_line=safe_int("max_chars_per_line", DEFAULT_MAX_CHARS_PER_LINE),
+            line_spacing=safe_int("line_spacing", 0),
+            fade_in=max(0.0, safe_float("fade_in", 0.0)),
+            fade_out=max(0.0, safe_float("fade_out", 0.0)),
+            animation_type=safe_str("animation_type", "none"),
+        )
+
+    @property
+    def alignment(self) -> int:
+        """获取 ASS alignment 编号."""
+        return POSITION_ALIGNMENT.get(self.position, 2)
+
+    @property
+    def ass_font_color(self) -> str:
+        """ASS 格式颜色 &HAABBGGRR."""
+        return _hex_to_ass_color(self.font_color)
+
+    @property
+    def ass_stroke_color(self) -> str:
+        return _hex_to_ass_color(self.stroke_color)
+
+    @property
+    def ass_shadow_color(self) -> str:
+        return _hex_to_ass_color(self.shadow_color)
+
+    @property
+    def ass_background_color(self) -> str:
+        """背景框颜色（ASS BackColour），带透明度."""
+        alpha_hex = _opacity_to_ass_alpha(self.background_opacity)
+        color_bgr = _hex_to_ass_bgr(self.background_color)
+        return f"&H{alpha_hex}{color_bgr}"
+
+
+# ── 工具函数 ──────────────────────────────────────────────────────────────────
+
+
+def _hex_to_ass_color(hex_color: str) -> str:
+    """HEX → ASS 颜色 &HAABBGGRR（默认不透明）."""
+    hex_color = hex_color.lstrip("#")
+    if len(hex_color) != 6:
+        return "&H00FFFFFF"
+    r, g, b = hex_color[0:2], hex_color[2:4], hex_color[4:6]
+    return f"&H00{b.upper()}{g.upper()}{r.upper()}"
+
+
+def _hex_to_ass_bgr(hex_color: str) -> str:
+    """HEX → ASS BGR 部分（不含 alpha）."""
+    hex_color = hex_color.lstrip("#")
+    if len(hex_color) != 6:
+        return "FFFFFF"
+    r, g, b = hex_color[0:2], hex_color[2:4], hex_color[4:6]
+    return f"{b.upper()}{g.upper()}{r.upper()}"
+
+
+def _opacity_to_ass_alpha(opacity: float) -> str:
+    """不透明度 → ASS alpha（00=不透明，FF=完全透明）."""
+    alpha = 255 - int(opacity * 255)
+    return f"{alpha:02X}"
+
+
+def _escape_ass_text(text: str) -> str:
+    """转义 ASS 文本特殊字符."""
+    text = text.replace("\r\n", "\\N").replace("\n", "\\N").replace("\r", "\\N")
+    text = text.replace("{", "(").replace("}", ")")
+    return text
+
+
+def _format_ass_time(seconds: float) -> str:
+    """秒 → ASS 时间格式 H:MM:SS.cc."""
+    hours = int(seconds // 3600)
+    minutes = int((seconds % 3600) // 60)
+    secs = seconds % 60
+    return f"{hours}:{minutes:02d}:{secs:05.2f}"
+
+
+def _wrap_text(text: str, max_chars: int) -> list[str]:
+    """按字数换行，优先标点断开."""
+    if len(text) <= max_chars:
+        return [text]
+
+    lines: list[str] = []
+    remaining = text
+
+    while len(remaining) > max_chars:
+        break_point = max_chars
+        punctuations = "，。！？、；：,.;:!?"
+
+        for i in range(max_chars, max_chars // 2, -1):
+            if i < len(remaining) and remaining[i] in punctuations:
+                break_point = i + 1
+                break
+
+        lines.append(remaining[:break_point])
+        remaining = remaining[break_point:]
+
+    if remaining:
+        lines.append(remaining)
+
+    return lines
+
+
+# ── 字幕片段 ──────────────────────────────────────────────────────────────────
+
+
+@dataclass
+class SubtitleSegment:
+    """单个字幕片段."""
+
+    start: float  # 开始时间（秒）
+    end: float  # 结束时间（秒）
+    text: str  # 字幕文本
+    style_name: str = "Default"  # 使用的样式名
+
+
+# ── 字幕渲染引擎 ──────────────────────────────────────────────────────────────
+
+
+class SubtitleRenderEngine:
+    """字幕渲染引擎 — 统一管理多源字幕的 ASS 文件生成.
+
+    支持合并多个字幕来源到同一个 ASS 文件：
+    - 标题（顶部，单独样式）
+    - 字幕（底部，单独样式）
+    - ASR 时间轴字幕
+    - 手动字幕
+
+    输出一个统一的 ASS 文件，供 FFmpeg subtitles filter 烧录。
+    """
+
+    def __init__(
+        self,
+        video_width: int = 1080,
+        video_height: int = 1920,
+        video_duration: float = 0.0,
+    ):
+        self.video_width = video_width
+        self.video_height = video_height
+        self.video_duration = video_duration
+        self._styles: dict[str, SubtitleStyle] = {}
+        self._segments: list[SubtitleSegment] = []
+        self._style_counter = 0
+
+    # ── 样式管理 ──────────────────────────────────────────────────────
+
+    def add_style(self, name: str, style: SubtitleStyle) -> str:
+        """注册一个样式，返回样式名."""
+        self._styles[name] = style
+        return name
+
+    def get_or_create_style(self, base_name: str, style: SubtitleStyle) -> str:
+        """获取或创建样式（避免重复）."""
+        if base_name in self._styles:
+            return base_name
+        self._styles[base_name] = style
+        return base_name
+
+    # ── 字幕源添加 ────────────────────────────────────────────────────
+
+    def add_title(self, text: str, style: SubtitleStyle | None = None) -> None:
+        """添加整段标题（显示整个视频时长）."""
+        if not text or not text.strip():
+            return
+
+        style = style or SubtitleStyle(
+            position="top_center",
+            font_size=48,
+            bold=True,
+            stroke_enabled=True,
+            stroke_width=2.0,
+        )
+        style_name = self.get_or_create_style("TitleStyle", style)
+
+        self._segments.append(
+            SubtitleSegment(
+                start=0.0,
+                end=self.video_duration if self.video_duration > 0 else 9999.0,
+                text=text.strip(),
+                style_name=style_name,
+            )
+        )
+
+    def add_subtitle_text(self, text: str, style: SubtitleStyle | None = None) -> None:
+        """添加整段字幕（显示整个视频时长）."""
+        if not text or not text.strip():
+            return
+
+        style = style or SubtitleStyle()
+        style_name = self.get_or_create_style("SubtitleStyle", style)
+
+        self._segments.append(
+            SubtitleSegment(
+                start=0.0,
+                end=self.video_duration if self.video_duration > 0 else 9999.0,
+                text=text.strip(),
+                style_name=style_name,
+            )
+        )
+
+    def add_timeline_segments(
+        self,
+        segments: list[dict] | list[SubtitleSegment],
+        style: SubtitleStyle | None = None,
+    ) -> None:
+        """添加时间轴字幕片段（ASR 或手动字幕）.
+
+        segments 可以是：
+        - SubtitleSegment 列表
+        - dict 列表，每个 dict 含 start/end/text 字段
+        """
+        if not segments:
+            return
+
+        style = style or SubtitleStyle()
+        style_name = self.get_or_create_style("Default", style)
+
+        for seg in segments:
+            if isinstance(seg, SubtitleSegment):
+                seg.style_name = style_name
+                self._segments.append(seg)
+            elif isinstance(seg, dict):
+                try:
+                    start = float(seg.get("start", 0))
+                    end = float(seg.get("end", 0))
+                    text = str(seg.get("text", ""))
+                    if end > start and text.strip():
+                        self._segments.append(
+                            SubtitleSegment(
+                                start=start,
+                                end=end,
+                                text=text.strip(),
+                                style_name=style_name,
+                            )
+                        )
+                except (TypeError, ValueError):
+                    continue
+
+    def add_asr_timeline(self, timeline: Any, style: SubtitleStyle | None = None) -> None:
+        """从 SubtitleTimeline 对象添加 ASR 字幕."""
+        if not timeline or not hasattr(timeline, "segments") or not timeline.segments:
+            return
+
+        style = style or SubtitleStyle()
+        style_name = self.get_or_create_style("ASRStyle", style)
+
+        for seg in timeline.segments:
+            if hasattr(seg, "start") and hasattr(seg, "end") and hasattr(seg, "text"):
+                if seg.end > seg.start and seg.text.strip():
+                    self._segments.append(
+                        SubtitleSegment(
+                            start=seg.start,
+                            end=seg.end,
+                            text=seg.text.strip(),
+                            style_name=style_name,
+                        )
+                    )
+
+    # ── ASS 文件生成 ──────────────────────────────────────────────────
+
+    def generate_ass(self, output_path: Path) -> Path:
+        """生成 ASS 字幕文件.
+
+        Returns:
+            生成的文件路径；如果没有字幕内容，返回空文件。
+        """
+        if not self._segments:
+            output_path.parent.mkdir(parents=True, exist_ok=True)
+            output_path.write_text("", encoding="utf-8")
+            return output_path
+
+        # 确保至少有 Default 样式
+        if "Default" not in self._styles:
+            self._styles["Default"] = SubtitleStyle()
+
+        # 生成样式行
+        style_lines = []
+        for name, style in self._styles.items():
+            style_lines.append(self._build_ass_style_line(name, style))
+
+        # 生成事件行（按时间排序）
+        self._segments.sort(key=lambda s: s.start)
+        event_lines = []
+        for seg in self._segments:
+            event_lines.append(self._build_ass_event_line(seg))
+
+        # 组装文件
+        ass_content = f"""[Script Info]
+ScriptType: v4.00+
+PlayResX: {self.video_width}
+PlayResY: {self.video_height}
+ScaledBorderAndShadow: yes
+WrapStyle: 2
+Encoding: UTF-8
+
+[V4+ Styles]
+Format: Name, Fontname, Fontsize, PrimaryColour, SecondaryColour, OutlineColour, BackColour, Bold, Italic, Underline, StrikeOut, ScaleX, ScaleY, Spacing, Angle, BorderStyle, Outline, Shadow, Alignment, MarginL, MarginR, MarginV, Encoding
+{chr(10).join(style_lines)}
+
+[Events]
+Format: Layer, Start, End, Style, Name, MarginL, MarginR, MarginV, Effect, Text
+{chr(10).join(event_lines)}
+"""
+
+        output_path.parent.mkdir(parents=True, exist_ok=True)
+        output_path.write_text(ass_content, encoding="utf-8")
+        return output_path
+
+    def _build_ass_style_line(self, name: str, style: SubtitleStyle) -> str:
+        """构建一条 ASS Style 行."""
+        bold_val = -1 if style.bold else 0
+        italic_val = -1 if style.italic else 0
+
+        # BorderStyle: 1=outline+shadow, 3=opaque box（背景框）
+        if style.background_enabled:
+            border_style = 3
+            back_color = style.ass_background_color
+        else:
+            border_style = 1
+            back_color = style.ass_shadow_color if style.shadow_enabled else style.ass_font_color
+
+        outline_val = style.stroke_width if style.stroke_enabled else 0.0
+        shadow_val = style.shadow_offset_y if style.shadow_enabled else 0
+
+        return (
+            f"Style: {name},{style.font_name},{style.font_size},{style.ass_font_color},"
+            f"&H000000FF,{style.ass_stroke_color},{back_color},"
+            f"{bold_val},{italic_val},0,0,100,100,0,0,"
+            f"{border_style},{outline_val},{shadow_val},{style.alignment},"
+            f"{style.margin_l},{style.margin_r},{style.margin_v},1"
+        )
+
+    def _build_ass_event_line(self, seg: SubtitleSegment) -> str:
+        """构建一条 ASS Dialogue 事件行."""
+        style = self._styles.get(seg.style_name, SubtitleStyle())
+        max_chars = style.max_chars_per_line
+
+        # 自动换行
+        lines = _wrap_text(seg.text, max_chars)
+        display_text = "\\N".join(lines)
+
+        # 动画效果（淡入淡出）
+        effect_tags = ""
+        if style.fade_in > 0 or style.fade_out > 0:
+            fade_in_ms = int(style.fade_in * 1000)
+            fade_out_ms = int(style.fade_out * 1000)
+            effect_tags = f"{{\\fad({fade_in_ms},{fade_out_ms})}}"
+
+        safe_text = _escape_ass_text(display_text)
+        start_time = _format_ass_time(max(0, seg.start))
+        end_time = _format_ass_time(max(seg.start + 0.1, seg.end))
+
+        return f"Dialogue: 0,{start_time},{end_time},{seg.style_name},,0,0,0,," f"{effect_tags}{safe_text}"
+
+    @property
+    def has_subtitles(self) -> bool:
+        """是否有字幕内容."""
+        return len(self._segments) > 0
+
+
+# ── 便捷函数：从 plan.config 快速生成 ASS ────────────────────────────────────
+
+
+def build_subtitles_from_plan(
+    output_path: Path,
+    plan_config: dict,
+    *,
+    video_width: int,
+    video_height: int,
+    video_duration: float,
+    asr_timeline: Any = None,
+) -> Path | None:
+    """从 plan.config 构建字幕 ASS 文件.
+
+    支持的配置项：
+    - title_config: 标题配置（含 text/style）
+    - subtitle_config: 字幕配置（含 text/style）
+    - asr_subtitles: ASR 字幕开关 + 样式
+    - manual_subtitles: 手动字幕片段列表
+
+    Returns:
+        生成的 ASS 文件路径；如果没有任何字幕，返回 None
+    """
+    engine = SubtitleRenderEngine(
+        video_width=video_width,
+        video_height=video_height,
+        video_duration=video_duration,
+    )
+
+    has_any = False
+
+    # 1. 标题
+    title_cfg = plan_config.get("title_config") or {}
+    if isinstance(title_cfg, dict):
+        title_text = str(title_cfg.get("text", ""))
+        title_enabled = title_cfg.get("enabled", True)
+        if title_enabled and title_text.strip():
+            style_dict = title_cfg.get("style") or {}
+            style = SubtitleStyle.from_dict(style_dict)
+            # 标题默认样式：顶部、大字号、粗体
+            if style.position == DEFAULT_POSITION and style.font_size == DEFAULT_FONT_SIZE:
+                style.position = "top_center"
+                style.font_size = 48
+                style.bold = True
+            engine.add_title(title_text, style)
+            has_any = True
+
+    # 2. 静态字幕
+    sub_cfg = plan_config.get("subtitle_config") or {}
+    if isinstance(sub_cfg, dict):
+        sub_text = str(sub_cfg.get("text", ""))
+        sub_enabled = sub_cfg.get("enabled", True)
+        if sub_enabled and sub_text.strip():
+            style_dict = sub_cfg.get("style") or {}
+            style = SubtitleStyle.from_dict(style_dict)
+            engine.add_subtitle_text(sub_text, style)
+            has_any = True
+
+    # 3. ASR 自动字幕
+    asr_cfg = plan_config.get("asr_subtitles") or {}
+    if isinstance(asr_cfg, dict) and asr_cfg.get("enabled", False):
+        if asr_timeline is not None:
+            style_dict = asr_cfg.get("style") or {}
+            style = SubtitleStyle.from_dict(style_dict)
+            engine.add_asr_timeline(asr_timeline, style)
+            has_any = has_any or engine.has_subtitles
+
+    # 4. 手动字幕
+    manual_segs = plan_config.get("manual_subtitles") or []
+    if isinstance(manual_segs, list) and manual_segs:
+        style_dict = (plan_config.get("manual_subtitle_style") or {}) or {}
+        style = SubtitleStyle.from_dict(style_dict)
+        engine.add_timeline_segments(manual_segs, style)
+        has_any = has_any or engine.has_subtitles
+
+    if not has_any:
+        return None
+
+    return engine.generate_ass(output_path)
+
+
+# ── FFmpeg 烧录滤镜生成 ───────────────────────────────────────────────────────
+
+
+def build_subtitle_filter(
+    ass_path: Path | str,
+    *,
+    video_input_label: str = "0:v",
+    output_label: str = "subtitled",
+) -> str:
+    """生成 FFmpeg subtitles 滤镜字符串.
+
+    Args:
+        ass_path: ASS 字幕文件路径
+        video_input_label: 视频输入标签（如 "0:v" 或 "[v_out]"）
+        output_label: 输出标签
+
+    Returns:
+        filter_complex 片段，如 "[0:v]subtitles=xxx.ass[subtitled]"
+    """
+    # FFmpeg subtitles filter 的路径需要转义：
+    # - Windows 路径的 \ → /
+    # - 冒号 : → \:
+    # - 单引号 ' → '\''
+    safe_path = str(ass_path).replace("\\", "/").replace(":", "\\:").replace("'", "'\\''")
+    return f"{video_input_label}subtitles='{safe_path}'[{output_label}]"

apps/worker/video_processing/unified_render_service.py

@@ -28,7 +28,6 @@ from dataclasses import dataclass, field
 from pathlib import Path
 from typing import Any
 
-from video_processing.chroma_key_engine import apply_chroma_key_if_needed
 from video_processing.color_grade_engine import ColorGradeConfig, ColorGradeEngine
 from video_processing.ffmpeg_utils import (
     DEFAULT_FPS,
@@ -45,7 +44,8 @@ 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.speed_engine import SpeedConfig, SpeedEngine
+from video_processing.sticker_engine import StickerEngine
 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
@@ -73,6 +73,7 @@ class ResolvedClip:
     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)
 
     # 运行时填充
@@ -186,6 +187,7 @@ class UnifiedRenderService:
         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.
@@ -340,6 +342,7 @@ class UnifiedRenderService:
         else:
             config = self.plan.config or {}
             bgm_config = config.get("bgm", {}) or {}
+            audio_tracks_config = config.get("audio_tracks") or {}
             noise_reduction_config = config.get("audio_noise_reduction")
             ctx = RenderContext(
                 work_dir=self.work_dir,
@@ -352,6 +355,7 @@ class UnifiedRenderService:
                 video_duration,
                 bgm_path=self.bgm_path,
                 bgm_config=bgm_config,
+                audio_tracks_config=audio_tracks_config,
             )
             t_audio_end = time.time()
             audio_mix_ms = int((t_audio_end - t_audio_start) * 1000)
@@ -493,7 +497,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)
@@ -1197,6 +1201,7 @@ class UnifiedRenderService:
                 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,
@@ -1294,6 +1299,11 @@ 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:
@@ -1351,8 +1361,8 @@ 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]
 
@@ -1597,7 +1607,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
@@ -1694,3 +1704,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

apps/worker/worker_app/tasks/asset_analyzer.py

@@ -179,7 +179,7 @@ class AssetAnalyzer:
         self._video_info = info
         return info
 
-    def extract_frames(self, count: int = 10, max_frames: int = 30) -> list[np.ndarray]:
+    def extract_frames(self, count: int = 10) -> list[np.ndarray]:
         """
         从视频中均匀抽取帧
 

apps/worker/worker_app/tasks/generation.py

@@ -461,7 +461,6 @@ def _download_library_assets(
     asset_library_id: str = "",
     project_id: str = "",
     asset_ids: list[str] | None = None,
-    video_extensions: tuple = (".mp4", ".mov", ".avi", ".mkv", ".webm"),
     strict: bool = True,
     task_id: str = "",
     gen_task=None,
@@ -480,7 +479,6 @@ def _download_library_assets(
         asset_library_id: 素材库 ID（可选，与 project_id 二选一）
         project_id: 项目 ID（可选，与 asset_library_id 二选一）
         asset_ids: 指定素材 ID 列表，为空则下载全部 ready 视频素材
-        video_extensions: 支持的视频扩展名（保留兼容，当前按 file_type 过滤）
         strict: 严格模式（默认 True）。
             True — 任何素材下载失败立即抛 RuntimeError；
             False — 跳过失败素材，返回成功列表（调用方可通过日志感知失败）。

docs/schema-metadata-snapshot.json

@@ -866,6 +866,14 @@
           "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

@@ -55,6 +55,7 @@ class SQLAlchemyEditPlanClipRepository:
             duration=clip.duration,
             transition_effect=clip.transition_effect,
             transition_duration=clip.transition_duration,
+            playback_speed=clip.playback_speed,
             status=clip.status,
             config=clip.config,
         )
@@ -78,6 +79,7 @@ class SQLAlchemyEditPlanClipRepository:
         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
@@ -123,6 +125,7 @@ class SQLAlchemyEditPlanClipRepository:
             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

@@ -197,6 +197,7 @@ class EditPlanClipModel(Base):
     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

@@ -51,6 +51,7 @@ class EditPlanClip:
     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))
@@ -70,6 +71,7 @@ class EditPlanClip:
         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:
         """创建剪辑计划片段"""
@@ -81,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,
@@ -94,6 +103,7 @@ class EditPlanClip:
             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 {},
         )

requirements-base.txt

@@ -13,7 +13,7 @@ uvicorn[standard]==0.32.0
 pydantic==2.9.0
 
 # 认证核心
-pyjwt==2.9.0
+pyjwt==2.13.0
 bcrypt==4.2.0
 
 # Redis

scripts/generate_changelog.py

@@ -0,0 +1,183 @@
+#!/usr/bin/env python3
+"""
+自动生成 CHANGELOG 条目。
+
+用法:
+    python3 scripts/generate_changelog.py v0.1.128 v0.1.129
+    python3 scripts/generate_changelog.py v0.1.128 HEAD
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import re
+import sys
+import urllib.error
+import urllib.request
+from datetime import datetime
+
+GITEA_URL = os.environ.get("GITEA_URL", "https://git.xiaoxiajianji.com")
+REPO = os.environ.get("GITEA_REPO", "xiaoxia/xiaoxia-saas")
+TOKEN = os.environ.get("GITEA_TOKEN", "")
+
+
+def gitea_api(path: str) -> dict | list:
+    url = f"{GITEA_URL}/api/v1{path}"
+    req = urllib.request.Request(url)
+    if TOKEN:
+        req.add_header("Authorization", f"token {TOKEN}")
+    try:
+        with urllib.request.urlopen(req, timeout=30) as resp:
+            return json.loads(resp.read().decode())
+    except urllib.error.HTTPError as e:
+        print(f"API Error: {e.code} {e.reason}", file=sys.stderr)
+        raise
+
+
+def get_tag_date(tag: str) -> str:
+    try:
+        info = gitea_api(f"/repos/{REPO}/git/refs/tags/{tag}")
+        if isinstance(info, dict):
+            sha = info.get("object", {}).get("sha", "")
+            if sha:
+                commit = gitea_api(f"/repos/{REPO}/git/commits/{sha}")
+                if isinstance(commit, dict):
+                    return commit.get("committer", {}).get("date", "")[:10]
+    except Exception:
+        pass
+    return ""
+
+
+def get_merged_prs_between(from_tag: str, to_tag: str) -> list[dict]:
+    all_prs: list[dict] = []
+    page = 1
+    while True:
+        prs = gitea_api(f"/repos/{REPO}/pulls?state=closed&sort=merged&direction=desc" f"&per_page=50&page={page}")
+        if not isinstance(prs, list) or not prs:
+            break
+        all_prs.extend(prs)
+        if len(prs) < 50:
+            break
+        page += 1
+        if page > 10:
+            break
+
+    merged = [pr for pr in all_prs if pr.get("merged_at")]
+    from_date = get_tag_date(from_tag)
+    to_date = get_tag_date(to_tag) if not to_tag.startswith("HEAD") else datetime.now().strftime("%Y-%m-%d")
+
+    if not from_date:
+        return merged[:50]
+
+    result = []
+    for pr in merged:
+        merged_at = pr.get("merged_at", "")[:10]
+        if from_date <= merged_at <= to_date:
+            result.append(pr)
+    return result
+
+
+def categorize_pr(title: str) -> tuple[str, str]:
+    title = title.strip()
+    lower = title.lower()
+
+    m = re.match(r"^(feat|fix|chore|perf|docs|refactor|test|ci|style|build|security)\s*[:：]", title)
+    if m:
+        prefix = m.group(1)
+        clean_title = title[m.end() :].strip()
+    else:
+        prefix = ""
+        clean_title = title
+
+    if prefix in ("feat", "feature"):
+        return "✨ 功能", clean_title
+    elif prefix == "fix":
+        return "🐛 Bug 修复", clean_title
+    elif prefix in ("refactor", "chore", "style"):
+        return "🔄 重构与清理", clean_title
+    elif prefix in ("perf", "performance"):
+        return "⚡ 性能优化", clean_title
+    elif prefix == "security":
+        return "🔒 安全修复", clean_title
+    elif prefix == "docs":
+        return "📝 文档", clean_title
+    elif prefix == "test":
+        return "🧪 测试", clean_title
+    elif prefix in ("ci", "build"):
+        return "🚀 CI/CD & 基础设施", clean_title
+    else:
+        if any(k in lower for k in ["安全", "security", "cve", "漏洞"]):
+            return "🔒 安全修复", clean_title
+        elif any(k in lower for k in ["修复", "bug"]):
+            return "🐛 Bug 修复", clean_title
+        elif any(k in lower for k in ["新增", "添加", "feat", "功能"]):
+            return "✨ 功能", clean_title
+        elif any(k in lower for k in ["ci", "构建", "workflow", "pipeline"]):
+            return "🚀 CI/CD & 基础设施", clean_title
+        elif any(k in lower for k in ["测试", "test", "e2e"]):
+            return "🧪 测试", clean_title
+        else:
+            return "📌 其他", clean_title
+
+
+def generate_changelog(from_tag: str, to_tag: str, version: str = "") -> str:
+    if not version:
+        version = to_tag
+
+    prs = get_merged_prs_between(from_tag, to_tag)
+
+    categories: dict[str, list[tuple[int, str]]] = {}
+    for pr in prs:
+        cat, title = categorize_pr(pr["title"])
+        pr_num = pr["number"]
+        categories.setdefault(cat, []).append((pr_num, title))
+
+    order = [
+        "🔒 安全修复",
+        "✨ 功能",
+        "🐛 Bug 修复",
+        "⚡ 性能优化",
+        "🔄 重构与清理",
+        "📝 文档",
+        "🧪 测试",
+        "🚀 CI/CD & 基础设施",
+        "📌 其他",
+    ]
+
+    date_str = get_tag_date(to_tag) if not to_tag.startswith("HEAD") else datetime.now().strftime("%Y-%m-%d")
+    lines = [f"## [{version}] - {date_str}", ""]
+
+    for cat in order:
+        items = categories.get(cat, [])
+        if not items:
+            continue
+        lines.append(f"### {cat}")
+        lines.append("")
+        for num, title in sorted(items, key=lambda x: x[0]):
+            short_title = title.split(" — ")[0].split(" - ")[0]
+            if len(short_title) > 80:
+                short_title = short_title[:77] + "..."
+            lines.append(f"- #{num} {short_title}")
+        lines.append("")
+
+    lines.append("---")
+    lines.append("")
+    return "\n".join(lines)
+
+
+def main():
+    if len(sys.argv) < 3:
+        print(f"用法: {sys.argv[0]} <from_tag> <to_tag> [version]")
+        sys.exit(1)
+
+    from_tag = sys.argv[1]
+    to_tag = sys.argv[2]
+    version = sys.argv[3] if len(sys.argv) > 3 else ""
+
+    changelog = generate_changelog(from_tag, to_tag, version)
+    print(changelog)
+
+
+if __name__ == "__main__":
+    main()

scripts/gray_deploy.sh

@@ -0,0 +1,227 @@
+#!/bin/bash
+# 灰度发布脚本：在生产服务器上启动 canary 版本，通过 Nginx 权重切流
+# 用法: ./scripts/gray_deploy.sh <版本号> <灰度百分比>
+#
+# 前提：
+#   - 在生产服务器上执行（或通过 SSH 管道执行）
+#   - 当前已有全量运行的 production 容器
+#   - Nginx 配置在 /etc/nginx/sites-enabled/00-xiaoxia-saas
+#
+# 灰度范围：API + Web（Worker 暂时全量升级，队列消费无法按比例切流）
+
+set -euo pipefail
+
+VERSION="${1:-}"
+GRAY_PCT="${2:-10}"
+
+if [[ -z "$VERSION" ]]; then
+    echo "用法: $0 <版本号> [灰度百分比]"
+    echo "示例: $0 v0.1.130 5"
+    exit 1
+fi
+
+REGISTRY="${REGISTRY:-git.xiaoxiajianji.com/xiaoxia/xiaoxia-saas}"
+NGINX_CONF="${NGINX_CONF:-/etc/nginx/sites-enabled/00-xiaoxia-saas}"
+ENV_FILE="${ENV_FILE:-/var/lib/xiaoxia-saas-production/.env}"
+GENERATED_DIR="${GENERATED_DIR:-/var/lib/xiaoxia-saas-production/generated}"
+LEGACY_ASSETS_DIR="${LEGACY_ASSETS_DIR:-/var/lib/xiaoxia-saas-production/legacy-assets}"
+
+# Canary 端口（与 production 错开）
+CANARY_API_PORT=18001
+CANARY_WEB_PORT=13002
+
+echo "============================================"
+echo "  灰度发布"
+echo "  新版本: $VERSION"
+echo "  灰度比例: ${GRAY_PCT}%"
+echo "  Canary API 端口: $CANARY_API_PORT"
+echo "  Canary Web 端口: $CANARY_WEB_PORT"
+echo "============================================"
+
+# 1. 检查环境
+if [[ ! -f "$ENV_FILE" ]]; then
+    echo "错误: 环境文件不存在: $ENV_FILE"
+    exit 1
+fi
+
+if [[ ! -f "$NGINX_CONF" ]]; then
+    echo "错误: Nginx 配置不存在: $NGINX_CONF"
+    exit 1
+fi
+
+# 2. 拉取新版本镜像
+echo ""
+echo ">>> 拉取新版本镜像..."
+for component in api web worker; do
+    echo "  拉取 $component:$VERSION ..."
+    docker pull "${REGISTRY}-${component}:${VERSION}" 2>&1 | tail -1
+done
+echo "  ✅ 镜像拉取完成"
+
+# 3. 启动 API Canary
+echo ""
+echo ">>> 启动 API Canary 容器..."
+CANARY_API="xiaoxia-api-canary"
+if docker ps -a --format '{{.Names}}' | grep -q "^${CANARY_API}$"; then
+    echo "  停止旧 canary..."
+    docker rm -f "$CANARY_API" >/dev/null 2>&1
+fi
+
+docker run -d \
+    --name "$CANARY_API" \
+    --env-file "$ENV_FILE" \
+    --network xiaoxia-net-production \
+    -p "127.0.0.1:${CANARY_API_PORT}:8000" \
+    -e APP_ENV=production \
+    -e APP_VERSION="$VERSION-canary" \
+    -e GENERATED_FILES_DIR=/app/generated \
+    -e GENERATED_FILES_URL_PREFIX=/generated-files \
+    -e PUBLIC_API_BASE_URL=https://saas-api.xiaoxiajianji.com \
+    -v "${GENERATED_DIR}:/app/generated" \
+    --restart unless-stopped \
+    --cpus 1 \
+    --memory 1g \
+    --health-cmd "python -c \"import urllib.request; urllib.request.urlopen('http://localhost:8000/health', timeout=5)\"" \
+    --health-interval 30s \
+    --health-timeout 10s \
+    --health-retries 3 \
+    --health-start-period 40s \
+    --log-driver json-file \
+    --log-opt max-size=50m \
+    --log-opt max-file=3 \
+    "${REGISTRY}-api:${VERSION}" >/dev/null
+
+echo "  ✅ API Canary 已启动（端口 $CANARY_API_PORT）"
+
+# 4. 启动 Web Canary
+echo ""
+echo ">>> 启动 Web Canary 容器..."
+CANARY_WEB="xiaoxia-web-canary"
+if docker ps -a --format '{{.Names}}' | grep -q "^${CANARY_WEB}$"; then
+    echo "  停止旧 canary..."
+    docker rm -f "$CANARY_WEB" >/dev/null 2>&1
+fi
+
+LEGACY_VOLUME=""
+if [[ -d "$LEGACY_ASSETS_DIR" ]] && [[ -n "$(ls -A "$LEGACY_ASSETS_DIR" 2>/dev/null)" ]]; then
+    LEGACY_VOLUME="-v ${LEGACY_ASSETS_DIR}:/usr/share/nginx/html/assets-legacy/assets:ro"
+fi
+
+docker run -d \
+    --name "$CANARY_WEB" \
+    --network xiaoxia-net-production \
+    -p "127.0.0.1:${CANARY_WEB_PORT}:80" \
+    --restart unless-stopped \
+    --cpus 0.5 \
+    --memory 256m \
+    $LEGACY_VOLUME \
+    --health-cmd "wget --spider -q http://127.0.0.1:80" \
+    --health-interval 30s \
+    --health-timeout 5s \
+    --health-retries 3 \
+    --health-start-period 10s \
+    --log-driver json-file \
+    --log-opt max-size=50m \
+    --log-opt max-file=3 \
+    "${REGISTRY}-web:${VERSION}" >/dev/null
+
+echo "  ✅ Web Canary 已启动（端口 $CANARY_WEB_PORT）"
+
+# 5. 等待健康检查
+echo ""
+echo ">>> 等待 Canary 容器健康..."
+for i in $(seq 1 40); do
+    api_healthy=$(docker inspect --format='{{.State.Health.Status}}' "$CANARY_API" 2>/dev/null || echo "starting")
+    web_healthy=$(docker inspect --format='{{.State.Health.Status}}' "$CANARY_WEB" 2>/dev/null || echo "starting")
+    
+    if [[ "$api_healthy" == "healthy" && "$web_healthy" == "healthy" ]]; then
+        echo "  ✅ API + Web Canary 均健康（用时 ${i}s）"
+        break
+    fi
+    
+    if [[ "$api_healthy" == "unhealthy" ]]; then
+        echo "  ❌ API Canary 健康检查失败"
+        docker logs --tail 30 "$CANARY_API"
+        exit 1
+    fi
+    if [[ "$web_healthy" == "unhealthy" ]]; then
+        echo "  ❌ Web Canary 健康检查失败"
+        docker logs --tail 20 "$CANARY_WEB"
+        exit 1
+    fi
+    
+    sleep 3
+done
+
+# 6. 更新 Nginx 配置 - 添加 upstream 权重
+echo ""
+echo ">>> 更新 Nginx 权重（稳定: $((100-GRAY_PCT))% / 灰度: ${GRAY_PCT}%）..."
+
+# 备份
+BAK_FILE="${NGINX_CONF}.bak.gray.$(date +%Y%m%d%H%M%S)"
+cp "$NGINX_CONF" "$BAK_FILE"
+echo "  已备份: $BAK_FILE"
+
+# 生成 upstream 块
+UPSTREAM_BLOCK="
+# Gray release upstreams（自动生成 - gray_deploy.sh）
+upstream saas_api_backend {
+    server 127.0.0.1:8001 weight=$((100-GRAY_PCT));
+    server 127.0.0.1:${CANARY_API_PORT} weight=${GRAY_PCT};
+}
+
+upstream saas_web_backend {
+    server 127.0.0.1:3002 weight=$((100-GRAY_PCT));
+    server 127.0.0.1:${CANARY_WEB_PORT} weight=${GRAY_PCT};
+}
+"
+
+# 在文件最前面插入 upstream 块
+TMP_CONF=$(mktemp)
+{
+    echo "$UPSTREAM_BLOCK"
+    cat "$NGINX_CONF"
+} > "$TMP_CONF"
+
+# 替换 proxy_pass 指向 upstream
+# API: proxy_pass http://127.0.0.1:8001 -> proxy_pass http://saas_api_backend
+sed -i 's|proxy_pass http://127\.0\.0\.1:8001|proxy_pass http://saas_api_backend|g' "$TMP_CONF"
+# Web: proxy_pass http://127.0.0.1:3002/ -> proxy_pass http://saas_web_backend/
+sed -i 's|proxy_pass http://127\.0\.0\.1:3002/|proxy_pass http://saas_web_backend/|g' "$TMP_CONF"
+
+# 测试配置
+mv "$TMP_CONF" "$NGINX_CONF"
+if ! nginx -t 2>&1; then
+    echo "  ❌ Nginx 配置测试失败，回滚..."
+    cp "$BAK_FILE" "$NGINX_CONF"
+    nginx -t
+    exit 1
+fi
+
+nginx -s reload
+echo "  ✅ Nginx 已 reload，灰度生效"
+
+# 7. 验证灰度流量
+echo ""
+echo ">>> 验证灰度流量..."
+gray_hits=0
+total_hits=20
+for i in $(seq 1 $total_hits); do
+    resp=$(curl -s -o /dev/null -w "%{http_code}" -H "X-Gray-Test: 1" http://127.0.0.1:${CANARY_API_PORT}/health 2>/dev/null || echo "000")
+    if [[ "$resp" == "200" ]]; then
+        gray_hits=$((gray_hits + 1))
+    fi
+    sleep 0.1
+done
+echo "  Canary 健康验证: $gray_hits/$total_hits 请求成功"
+
+echo ""
+echo "============================================"
+echo "  ✅ 灰度发布完成"
+echo "  版本: $VERSION (${GRAY_PCT}%流量)"
+echo "  API: 127.0.0.1:$CANARY_API_PORT"
+echo "  Web: 127.0.0.1:$CANARY_WEB_PORT"
+echo "  Nginx 备份: $BAK_FILE"
+echo "  回滚: ./scripts/rollback.sh"
+echo "  Worker: 暂不灰度（队列消费无法按比例切流）"
+echo "============================================"

scripts/release.sh

@@ -0,0 +1,113 @@
+#!/bin/bash
+# 一键发布脚本：打 tag → 触发 CI 构建 → 可选灰度发布
+# 用法: ./scripts/release.sh v0.1.130 [--gray 5]
+#
+# 说明：
+#   - 打 tag 后 CI 会自动构建镜像并全量部署到生产
+#   - 加 --gray 参数则在构建完成后执行灰度切流（需 SSH 到生产服务器执行）
+#   - 加 --no-deploy 只打 tag 不触发自动部署
+
+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.130                  # 打tag + 全量发布（CI自动部署）"
+    echo "  $0 v0.1.130 --gray 5         # 打tag + 5%灰度发布"
+    echo "  $0 v0.1.130 --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.130）"
+    usage
+fi
+
+echo "============================================"
+echo "  发布版本: $VERSION"
+echo "  灰度比例: ${GRAY_PCT}%"
+echo "  自动部署: $DEPLOY"
+echo "============================================"
+
+cd "$REPO_ROOT"
+
+# 1. 确认分支
+CURRENT_BRANCH=$(git rev-parse --abbrev-ref HEAD)
+if [[ "$CURRENT_BRANCH" != "develop" ]]; then
+    echo "错误: 请在 develop 分支上打 tag"
+    exit 1
+fi
+
+# 2. 拉取最新
+echo ""
+echo ">>> 拉取最新代码..."
+git pull origin develop
+
+# 3. 检查 tag 是否已存在
+if git rev-parse "$VERSION" >/dev/null 2>&1; then
+    echo "警告: tag $VERSION 已存在，跳过打 tag"
+else
+    echo ""
+    echo ">>> 打 tag $VERSION ..."
+    git tag -a "$VERSION" -m "Release $VERSION"
+    git push origin "$VERSION"
+    echo "  ✅ Tag 已推送，CI 将自动构建生产镜像"
+fi
+
+# 4. 部署提示
+if [[ "$DEPLOY" == "true" ]]; then
+    echo ""
+    echo ">>> 构建 & 部署"
+    echo "  CI 会自动执行:"
+    echo "    1. Build Production Runtime Images（约10-15分钟）"
+    echo "    2. Deploy Production（SSH 到生产服务器部署）"
+    echo ""
+    echo "  查看进度: Gitea Actions → 对应 tag 的 run"
+    
+    if [[ "$GRAY_PCT" -gt 0 ]]; then
+        echo ""
+        echo ">>> 灰度发布"
+        echo "  构建部署完成后，在生产服务器上执行:"
+        echo "    cd /var/lib/xiaoxia-saas-production"
+        echo "    ./gray_deploy.sh $VERSION $GRAY_PCT"
+    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 "=========================================="

scripts/rollback_gray.sh

@@ -0,0 +1,77 @@
+#!/bin/bash
+# 灰度回滚脚本：切回全量稳定版本，停止 canary 容器
+# 用法: ./scripts/rollback_gray.sh
+
+set -euo pipefail
+
+NGINX_CONF="${NGINX_CONF:-/etc/nginx/sites-enabled/00-xiaoxia-saas}"
+CANARY_API="${CANARY_API:-xiaoxia-api-canary}"
+CANARY_WEB="${CANARY_WEB:-xiaoxia-web-canary}"
+
+echo "============================================"
+echo "  灰度回滚"
+echo "  目标: 全量切回稳定版本"
+echo "============================================"
+
+# 1. 找最近的灰度备份
+echo ""
+echo ">>> 查找最近的灰度备份..."
+LATEST_BAK=$(ls -t "${NGINX_CONF}".bak.gray.* 2>/dev/null | head -1 || true)
+
+if [[ -z "$LATEST_BAK" ]]; then
+    echo "  未找到灰度备份，尝试手动移除 upstream 配置..."
+    
+    # 手动回滚：移除 upstream 块，把 proxy_pass 改回 127.0.0.1
+    TMP_CONF=$(mktemp)
+    
+    # 移除 upstream 块（从 "# Gray release upstreams" 到空行结束）
+    awk '
+        /^# Gray release upstreams/ { skip=1; next }
+        skip && /^$/ && !found_first_empty { found_first_empty=1; next }
+        skip && found_first_empty && /^$/ { skip=0; found_first_empty=0; next }
+        skip { next }
+        { print }
+    ' "$NGINX_CONF" > "$TMP_CONF"
+    
+    # 把 upstream 名改回 IP
+    sed -i 's|proxy_pass http://saas_api_backend|proxy_pass http://127.0.0.1:8001|g' "$TMP_CONF"
+    sed -i 's|proxy_pass http://saas_web_backend/|proxy_pass http://127.0.0.1:3002/|g' "$TMP_CONF"
+    
+    mv "$TMP_CONF" "$NGINX_CONF"
+else
+    echo "  从备份恢复: $LATEST_BAK"
+    cp "$LATEST_BAK" "$NGINX_CONF"
+fi
+
+# 2. 测试并 reload nginx
+echo ""
+echo ">>> Nginx 测试 & reload..."
+if ! nginx -t 2>&1; then
+    echo "  ❌ Nginx 配置测试失败！请检查"
+    exit 1
+fi
+nginx -s reload
+echo "  ✅ Nginx 已回滚，全量切回稳定版本"
+
+# 3. 停止 canary 容器（延迟停止，保留30分钟便于排查）
+echo ""
+echo ">>> Canary 容器将在30分钟后停止（便于排查）"
+echo "    立即停止请执行: docker rm -f $CANARY_API $CANARY_WEB"
+
+# 30分钟后停止（后台执行，不阻塞脚本）
+(
+    sleep 1800
+    for c in "$CANARY_API" "$CANARY_WEB"; do
+        if docker ps --format '{{.Names}}' | grep -q "^${c}$"; then
+            docker stop "$c" >/dev/null 2>&1 && docker rm "$c" >/dev/null 2>&1
+            echo "[$(date)] 已停止 canary 容器: $c"
+        fi
+    done
+) &
+
+echo ""
+echo "============================================"
+echo "  ✅ 灰度回滚完成"
+echo "  流量已全部切回稳定版本"
+echo "  Canary 容器: 30分钟后自动清理"
+echo "============================================"

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