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

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/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

@@ -77,6 +77,7 @@ def mix_audio(
     *,
     bgm_path: str | None = None,
     bgm_config: dict | None = None,
+    audio_tracks_config: dict | None = None,
 ) -> Path | None:
     """音频后处理混音.
 
@@ -87,6 +88,8 @@ def mix_audio(
     4. 输出时长截断到 video_duration
     5. 无音频流的 clip 会被自动跳过，避免 FFmpeg 引用 [i:a] 失败
     6. 如果提供了 bgm_path，则额外混入 BGM（支持淡入淡出、循环、人声闪避）
+    7. 如果配置了 audio_tracks，则混入多轨道音频（配音、音效等）
+    8. 如果配置了降噪，最后应用降噪
 
     Args:
         ctx: 渲染上下文
@@ -94,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
@@ -157,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)
 

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,
@@ -46,7 +45,7 @@ from video_processing.render_audio import RenderContext, merge_audio_video, mix_
 from video_processing.render_subtitles import generate_ass_subtitles
 from video_processing.reverse_engine import ReverseConfig, ReverseEngine
 from video_processing.speed_engine import SpeedConfig, SpeedEngine
-from video_processing.sticker_engine import StickerEngine, parse_stickers_from_config
+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
@@ -343,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,
@@ -355,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)

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 — 跳过失败素材，返回成功列表（调用方可通过日志感知失败）。

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

@@ -1,9 +1,13 @@
 #!/bin/bash
-# 灰度发布脚本：通过Nginx权重调整流量比例
+# 灰度发布脚本：在生产服务器上启动 canary 版本，通过 Nginx 权重切流
 # 用法: ./scripts/gray_deploy.sh <版本号> <灰度百分比>
 #
-# 需要在目标服务器上执行，或通过SSH执行
-# 前提：服务器上运行两个版本的容器（stable + canary），Nginx做加权轮询
+# 前提：
+#   - 在生产服务器上执行（或通过 SSH 管道执行）
+#   - 当前已有全量运行的 production 容器
+#   - Nginx 配置在 /etc/nginx/sites-enabled/00-xiaoxia-saas
+#
+# 灰度范围：API + Web（Worker 暂时全量升级，队列消费无法按比例切流）
 
 set -euo pipefail
 
@@ -12,106 +16,212 @@ GRAY_PCT="${2:-10}"
 
 if [[ -z "$VERSION" ]]; then
     echo "用法: $0 <版本号> [灰度百分比]"
-    echo "示例: $0 v0.1.129 5"
+    echo "示例: $0 v0.1.130 5"
     exit 1
 fi
 
-STABLE_VERSION="${STABLE_VERSION:-current}"
 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}"
 
-echo "=========================================="
+# Canary 端口（与 production 错开）
+CANARY_API_PORT=18001
+CANARY_WEB_PORT=13002
+
+echo "============================================"
 echo "  灰度发布"
 echo "  新版本: $VERSION"
 echo "  灰度比例: ${GRAY_PCT}%"
-echo "  稳定版本: $STABLE_VERSION"
-echo "=========================================="
+echo "  Canary API 端口: $CANARY_API_PORT"
+echo "  Canary Web 端口: $CANARY_WEB_PORT"
+echo "============================================"
 
-# 1. 拉取新版本镜像
+# 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 worker web; do
+for component in api web worker; do
     echo "  拉取 $component:$VERSION ..."
     docker pull "${REGISTRY}-${component}:${VERSION}" 2>&1 | tail -1
 done
+echo "  ✅ 镜像拉取完成"
 
-# 2. 启动灰度版本容器（canary）
+# 3. 启动 API Canary
 echo ""
-echo ">>> 启动灰度版本容器..."
-
-# API canary
-CANARY_API_NAME="saas-api-canary"
-if docker ps -a --format '{{.Names}}' | grep -q "^${CANARY_API_NAME}$"; then
-    echo "  停止旧 canary 容器..."
-    docker stop "$CANARY_API_NAME" 2>/dev/null || true
-    docker rm "$CANARY_API_NAME" 2>/dev/null || true
+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
 
-echo "  启动 api canary..."
 docker run -d \
-    --name "$CANARY_API_NAME" \
-    --network saas-network \
-    -e DATABASE_URL="${DATABASE_URL}" \
-    -e REDIS_URL="${REDIS_URL}" \
-    -e FEATURE_FLAG_PROVIDER=redis \
+    --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 \
-    "${REGISTRY}-api:${VERSION}"
+    --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
 
-# Worker canary
-CANARY_WORKER_NAME="saas-worker-canary"
-if docker ps -a --format '{{.Names}}' | grep -q "^${CANARY_WORKER_NAME}$"; then
-    echo "  停止旧 worker canary..."
-    docker stop "$CANARY_WORKER_NAME" 2>/dev/null || true
-    docker rm "$CANARY_WORKER_NAME" 2>/dev/null || true
+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
 
-echo "  启动 worker canary..."
-docker run -d \
-    --name "$CANARY_WORKER_NAME" \
-    --network saas-network \
-    -e DATABASE_URL="${DATABASE_URL}" \
-    -e REDIS_URL="${REDIS_URL}" \
-    --restart unless-stopped \
-    "${REGISTRY}-worker:${VERSION}"
+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
 
-# 3. 等待容器健康
+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 ">>> 等待容器健康..."
-sleep 10
-if ! docker ps --format '{{.Names}} {{.Status}}' | grep -q "$CANARY_API_NAME"; then
-    echo "错误: API canary 容器未运行"
-    docker logs "$CANARY_API_NAME" --tail 20
+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
-echo "  ✅ API canary 运行中"
 
-# 4. 更新Nginx权重
+nginx -s reload
+echo "  ✅ Nginx 已 reload，灰度生效"
+
+# 7. 验证灰度流量
 echo ""
-echo ">>> 更新Nginx权重 (稳定: $((100-GRAY_PCT))% / 灰度: ${GRAY_PCT}%)..."
-
-NGINX_CONF="${NGINX_CONF:-/etc/nginx/conf.d/saas-api.conf}"
-if [[ -f "$NGINX_CONF" ]]; then
-    # 备份
-    cp "$NGINX_CONF" "${NGINX_CONF}.bak.$(date +%Y%m%d%H%M%S)"
-    
-    # 更新 upstream 权重（需要根据实际配置调整）
-    echo "  请手动更新 Nginx upstream 配置中的权重"
-    echo "  示例配置:"
-    cat <<EOF
-    upstream saas_api_backend {
-        server saas-api:8000 weight=$((100-GRAY_PCT));
-        server saas-api-canary:8000 weight=${GRAY_PCT};
-    }
-EOF
-    nginx -t && nginx -s reload
-    echo "  ✅ Nginx 已reload"
-else
-    echo "  警告: Nginx 配置文件不存在 ($NGINX_CONF)"
-    echo "  请手动配置灰度流量权重"
-fi
+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 "  ✅ 灰度发布完成"
-echo "  新版本: $VERSION (${GRAY_PCT}%流量)"
-echo "  监控: Grafana / 日志"
-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

@@ -1,6 +1,11 @@
 #!/bin/bash
-# 一键发布脚本：打tag → 触发生产镜像构建 → 部署到灰度
-# 用法: ./scripts/release.sh v0.1.129 [--gray 5]
+# 一键发布脚本：打 tag → 触发 CI 构建 → 可选灰度发布
+# 用法: ./scripts/release.sh v0.1.130 [--gray 5]
+#
+# 说明：
+#   - 打 tag 后 CI 会自动构建镜像并全量部署到生产
+#   - 加 --gray 参数则在构建完成后执行灰度切流（需 SSH 到生产服务器执行）
+#   - 加 --no-deploy 只打 tag 不触发自动部署
 
 set -euo pipefail
 
@@ -11,13 +16,12 @@ usage() {
     echo "用法: $0 <版本号> [--gray 百分比] [--no-deploy]"
     echo ""
     echo "示例:"
-    echo "  $0 v0.1.129                    # 打tag + 全量发布"
-    echo "  $0 v0.1.129 --gray 5           # 打tag + 5%灰度发布"
-    echo "  $0 v0.1.129 --no-deploy        # 只打tag，不部署"
+    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
@@ -47,80 +51,63 @@ while [[ $# -gt 0 ]]; do
 done
 
 if [[ -z "$VERSION" ]]; then
-    echo "错误: 请指定版本号（如 v0.1.129）"
+    echo "错误: 请指定版本号（如 v0.1.130）"
     usage
 fi
 
-echo "=========================================="
+echo "============================================"
 echo "  发布版本: $VERSION"
 echo "  灰度比例: ${GRAY_PCT}%"
 echo "  自动部署: $DEPLOY"
-echo "=========================================="
+echo "============================================"
 
 cd "$REPO_ROOT"
 
-# 1. 确认在 develop 分支
+# 1. 确认分支
 CURRENT_BRANCH=$(git rev-parse --abbrev-ref HEAD)
 if [[ "$CURRENT_BRANCH" != "develop" ]]; then
-    echo "错误: 请切换到 develop 分支后再发布"
+    echo "错误: 请在 develop 分支上打 tag"
     exit 1
 fi
 
-# 2. 拉取最新代码
+# 2. 拉取最新
 echo ""
 echo ">>> 拉取最新代码..."
 git pull origin develop
 
-# 3. 生成 CHANGELOG
-echo ""
-echo ">>> 生成 CHANGELOG..."
-if [[ -f "scripts/generate_changelog.py" ]]; then
-    PREV_TAG=$(git describe --tags --abbrev=0 HEAD^ 2>/dev/null || echo "")
-    if [[ -n "$PREV_TAG" ]]; then
-        python3 scripts/generate_changelog.py \
-            --from-tag "$PREV_TAG" \
-            --to-tag HEAD \
-            --gitea-token "${GITEA_TOKEN:-}" \
-            --output /tmp/changelog_$$.md
-        echo "CHANGELOG 已生成到 /tmp/changelog_$$.md"
-    fi
-fi
-
-# 4. 打tag
-echo ""
-echo ">>> 打 tag $VERSION ..."
+# 3. 检查 tag 是否已存在
 if git rev-parse "$VERSION" >/dev/null 2>&1; then
-    echo "警告: tag $VERSION 已存在，跳过打tag"
+    echo "警告: tag $VERSION 已存在，跳过打 tag"
 else
+    echo ""
+    echo ">>> 打 tag $VERSION ..."
     git tag -a "$VERSION" -m "Release $VERSION"
     git push origin "$VERSION"
-    echo "Tag $VERSION 已推送，触发生产镜像构建..."
+    echo "  ✅ Tag 已推送，CI 将自动构建生产镜像"
 fi
 
-# 5. 等待镜像构建
+# 4. 部署提示
 if [[ "$DEPLOY" == "true" ]]; then
     echo ""
-    echo ">>> 等待镜像构建完成（约10-15分钟）..."
-    echo "    镜像: git.xiaoxiajianji.com/xiaoxia/xiaoxia-saas-{api,worker,web}:$VERSION"
-    
-    # 这里可以加镜像存在性检查
-    echo "    （镜像构建由CI自动完成，请在Gitea Actions中确认）"
-fi
-
-# 6. 灰度部署
-if [[ "$DEPLOY" == "true" && "$GRAY_PCT" -gt 0 ]]; then
+    echo ">>> 构建 & 部署"
+    echo "  CI 会自动执行:"
+    echo "    1. Build Production Runtime Images（约10-15分钟）"
+    echo "    2. Deploy Production（SSH 到生产服务器部署）"
     echo ""
-    echo ">>> 灰度部署: ${GRAY_PCT}% 流量到 $VERSION"
-    if [[ -f "scripts/gray_deploy.sh" ]]; then
-        ./scripts/gray_deploy.sh "$VERSION" "$GRAY_PCT"
-    else
-        echo "警告: gray_deploy.sh 不存在，跳过灰度部署"
+    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 "============================================"
+echo "  ✅ 发布流程触发完成"
 echo "  版本: $VERSION"
 echo "  灰度: ${GRAY_PCT}%"
-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 "============================================"