fix: flake8 F811(重复import) + F541(空f-string)

apps/worker/video_processing/bgm_mixer.py

@@ -0,0 +1,313 @@
+"""BGM 混音模块 — 背景音乐与主音频混合.
+
+基于 FFmpeg 实现：
+- BGM 音量调节
+- 淡入淡出（afade）
+- 循环播放（aloop，短 BGM 铺长视频）
+- 人声闪避（sidechaincompress，有人声时BGM自动降低音量）
+- amix 混音
+
+作为 render_audio.py 的增强模块，在 mix_audio 后处理阶段被调用。
+"""
+
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass
+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__)
+
+
+@dataclass
+class BGMConfig:
+    """BGM 混音配置（内部使用，从 plan.config.bgm 转换而来）"""
+
+    bgm_path: str  # BGM 本地文件路径
+    volume: float = 0.3  # 0.0 ~ 1.0
+    fade_in: float = 0.0  # 淡入时长（秒）
+    fade_out: float = 0.0  # 淡出时长（秒）
+    loop_enabled: bool = True  # 是否循环铺满
+    sidechain_enabled: bool = False  # 人声闪避
+    sidechain_ratio: float = 0.3  # 闪避时音量降低比例
+    sidechain_attack: float = 0.02  # 攻击时间
+    sidechain_release: float = 0.5  # 释放时间
+    sidechain_threshold: float = -25.0  # 触发阈值（dB）
+
+    @classmethod
+    def from_config_dict(cls, bgm_path: str, config: dict) -> "BGMConfig":
+        """从 plan.config.bgm 字典创建 BGMConfig。"""
+        return cls(
+            bgm_path=bgm_path,
+            volume=float(config.get("volume", 0.3)),
+            fade_in=float(config.get("fade_in", 0.0)),
+            fade_out=float(config.get("fade_out", 0.0)),
+            loop_enabled=bool(config.get("loop_enabled", True)),
+            sidechain_enabled=bool(config.get("sidechain_enabled", False)),
+            sidechain_ratio=float(config.get("sidechain_ratio", 0.3)),
+            sidechain_attack=float(config.get("sidechain_attack", 0.02)),
+            sidechain_release=float(config.get("sidechain_release", 0.5)),
+            sidechain_threshold=float(config.get("sidechain_threshold", -25.0)),
+        )
+
+
+# ── BGM 预处理 ────────────────────────────────────────────────────────────────
+
+
+def prepare_bgm_track(
+    ctx: "RenderContext",
+    bgm: BGMConfig,
+    target_duration: float,
+) -> Path:
+    """预处理 BGM 轨道：循环/截断 + 音量 + 淡入淡出.
+
+    生成一个时长精确等于 target_duration 的 BGM 音频文件。
+    后续再与主音频混音。
+
+    Args:
+        ctx: 渲染上下文
+        bgm: BGM 配置
+        target_duration: 目标时长（秒），通常等于视频总时长
+
+    Returns:
+        处理后的 BGM 音频文件路径
+    """
+    output_path = ctx.work_dir / f"bgm_processed_{ctx.plan_id}.aac"
+
+    if target_duration <= 0:
+        target_duration = 5.0  # 兜底
+
+    bgm_dur = probe_duration(bgm.bgm_path)
+    needs_loop = bgm.loop_enabled and bgm_dur > 0 and bgm_dur < target_duration * 0.9
+
+    # 构建滤镜链
+    filter_parts: list[str] = []
+    input_looped: bool = False
+
+    if needs_loop:
+        # 计算需要循环多少次才能铺满
+        loop_count = max(1, int(target_duration / bgm_dur) + 2)
+        # aloop 滤镜：循环指定次数
+        filter_parts.append(f"aloop=loop={loop_count}:size=0")
+        input_looped = True
+
+    # 音量调节
+    volume = max(0.0, min(1.0, bgm.volume))
+    if abs(volume - 1.0) > 0.001:
+        filter_parts.append(f"volume={volume:.3f}")
+
+    # 淡入
+    if bgm.fade_in > 0:
+        filter_parts.append(f"afade=t=in:st=0:d={bgm.fade_in:.3f}")
+
+    # 淡出（从 target_duration - fade_out 开始）
+    if bgm.fade_out > 0 and target_duration > bgm.fade_out:
+        fade_start = target_duration - bgm.fade_out
+        filter_parts.append(f"afade=t=out:st={fade_start:.3f}:d={bgm.fade_out:.3f}")
+
+    # 最终截断到目标时长
+    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",
+        bgm.bgm_path,
+        "-filter:a",
+        filter_str,
+        "-c:a",
+        "aac",
+        "-b:a",
+        "128k",
+        str(output_path),
+    ]
+
+    logger.info(
+        "[bgm] prepare BGM track: path=%s dur=%.2f target=%.2f loop=%s fade_in=%.2f fade_out=%.2f",
+        bgm.bgm_path[-40:],
+        bgm_dur,
+        target_duration,
+        needs_loop,
+        bgm.fade_in,
+        bgm.fade_out,
+    )
+
+    run_ffmpeg(command)
+    return output_path
+
+
+# ── BGM + 主音频混音 ──────────────────────────────────────────────────────────
+
+
+def mix_bgm_with_main(
+    ctx: "RenderContext",
+    main_audio_path: Path,
+    bgm: BGMConfig,
+    target_duration: float,
+) -> Path:
+    """将 BGM 与主音频混合.
+
+    两种模式：
+    1. 普通混音（sidechain 关闭）：amix 两路音频
+    2. 人声闪避（sidechain 开启）：用 sidechaincompress 让 BGM 跟随主音频音量自动调整
+
+    Args:
+        ctx: 渲染上下文
+        main_audio_path: 主音频文件路径（人声/原始音频）
+        bgm: BGM 配置
+        target_duration: 目标时长
+
+    Returns:
+        混音后的音频文件路径
+    """
+    output_path = ctx.work_dir / f"audio_with_bgm_{ctx.plan_id}.aac"
+
+    # 先预处理 BGM 轨道（循环/音量/淡入淡出/截断）
+    bgm_processed = prepare_bgm_track(ctx, bgm, target_duration)
+
+    if not bgm.sidechain_enabled:
+        # 普通 amix 混音
+        _mix_simple(main_audio_path, bgm_processed, output_path)
+    else:
+        # sidechain 人声闪避混音
+        _mix_sidechain(main_audio_path, bgm_processed, output_path, bgm)
+
+    return output_path
+
+
+def _mix_simple(main_path: Path, bgm_path: Path, output_path: Path) -> None:
+    """简单 amix 混音：主音频 + BGM = 输出.
+
+    主音频权重 1.0，BGM 已经在预处理阶段调好了音量。
+    amix 会自动归一化，需要用 volume 补偿。
+    """
+    # 使用 amix：inputs=2，duration=first（以主音频时长为准）
+    # 然后用 volume=2 补偿 amix 的衰减（2路输入每路平均乘0.5）
+    filter_complex = "[0:a][1:a]amix=inputs=2:duration=first:dropout_transition=0[outa];" "[outa]volume=2[final]"
+
+    command = [
+        FFMPEG_BIN,
+        "-y",
+        "-i",
+        str(main_path),
+        "-i",
+        str(bgm_path),
+        "-filter_complex",
+        filter_complex,
+        "-map",
+        "[final]",
+        "-c:a",
+        "aac",
+        "-b:a",
+        "128k",
+        str(output_path),
+    ]
+
+    logger.info("[bgm] simple amix mix")
+    run_ffmpeg(command)
+
+
+def _mix_sidechain(
+    main_path: Path,
+    bgm_path: Path,
+    output_path: Path,
+    bgm: BGMConfig,
+) -> None:
+    """sidechain 人声闪避混音.
+
+    原理：
+    - 主音频作为 sidechain 信号源
+    - BGM 轨道经过 sidechaincompress，根据主音频音量动态调整 BGM 音量
+    - 最后 amix 混音
+
+    FFmpeg sidechaincompress 参数：
+    - threshold: 触发阈值（dB），主音频超过此值时开始压缩
+    - ratio: 压缩比，越高压缩越狠
+    - attack: 攻击时间（秒）
+    - release: 释放时间（秒）
+    """
+    # sidechain_ratio 表示闪避时 BGM 音量降低比例
+    # ratio = 1 / (1 - sidechain_ratio)，但实际压缩比需要更精细调整
+    # 简化处理：把 ratio 映射到 2:1 ~ 10:1 范围
+    ratio = max(2.0, min(10.0, 1.0 / (1.0 - bgm.sidechain_ratio)))
+
+    filter_complex = (
+        # BGM 经过 sidechain 压缩，用主音频做触发
+        f"[1:a][0:a]sidechaincompress="
+        f"threshold={bgm.sidechain_threshold}dB:"
+        f"ratio={ratio:.1f}:"
+        f"attack={bgm.sidechain_attack:.3f}:"
+        f"release={bgm.sidechain_release:.3f}:"
+        f"knee=6[bgm_comp];"
+        # 主音频 + 压缩后的 BGM 混音
+        f"[0:a][bgm_comp]amix=inputs=2:duration=first:dropout_transition=0[outa];"
+        f"[outa]volume=1.5[final]"  # 轻微补偿
+    )
+
+    command = [
+        FFMPEG_BIN,
+        "-y",
+        "-i",
+        str(main_path),
+        "-i",
+        str(bgm_path),
+        "-filter_complex",
+        filter_complex,
+        "-map",
+        "[final]",
+        "-c:a",
+        "aac",
+        "-b:a",
+        "128k",
+        str(output_path),
+    ]
+
+    logger.info(
+        "[bgm] sidechain mix: threshold=%.1fdB ratio=%.1f attack=%.3f release=%.3f",
+        bgm.sidechain_threshold,
+        ratio,
+        bgm.sidechain_attack,
+        bgm.sidechain_release,
+    )
+    run_ffmpeg(command)
+
+
+# ── 纯 BGM 模式（无主音频） ──────────────────────────────────────────────────
+
+
+def build_bgm_only(
+    ctx: "RenderContext",
+    bgm: BGMConfig,
+    target_duration: float,
+) -> Path:
+    """只有 BGM、没有主音频时，直接生成 BGM 音频.
+
+    Args:
+        ctx: 渲染上下文
+        bgm: BGM 配置
+        target_duration: 目标时长
+
+    Returns:
+        BGM 音频文件路径
+    """
+    output_path = ctx.work_dir / f"bgm_only_{ctx.plan_id}.aac"
+
+    if target_duration <= 0:
+        target_duration = 5.0
+
+    bgm_processed = prepare_bgm_track(ctx, bgm, target_duration)
+
+    # 直接复制
+    import shutil
+
+    shutil.copy2(bgm_processed, output_path)
+    return output_path

apps/worker/video_processing/render_audio.py

@@ -70,6 +70,9 @@ def mix_audio(
     ctx: RenderContext,
     layers: list[RenderLayer],
     video_duration: float,
+    *,
+    bgm_path: str | None = None,
+    bgm_config: dict | None = None,
 ) -> Path | None:
     """音频后处理混音.
 
@@ -79,11 +82,14 @@ def mix_audio(
     3. 独立音频轨（audio role）用 amix 混入
     4. 输出时长截断到 video_duration
     5. 无音频流的 clip 会被自动跳过，避免 FFmpeg 引用 [i:a] 失败
+    6. 如果提供了 bgm_path，则额外混入 BGM（支持淡入淡出、循环、人声闪避）
 
     Args:
         ctx: 渲染上下文
         layers: 图层列表
         video_duration: 视频总时长（用于截断音频）
+        bgm_path: BGM 音频本地路径，为 None 时不混入 BGM
+        bgm_config: BGM 配置字典（volume/fade_in/fade_out/sidechain 等）
 
     Returns:
         混音后的音频文件路径，无音频时返回 None
@@ -116,6 +122,15 @@ def mix_audio(
     audio_clips = [c for c in audio_clips if clip_has_audio(ctx, c)]
 
     if not main_clips and not audio_clips:
+        # 没有主音频也没有独立音频 → 检查是否有 BGM
+        if bgm_path and bgm_config and bgm_config.get("enabled", False):
+            from video_processing.bgm_mixer import BGMConfig, build_bgm_only
+
+            bgm_cfg = BGMConfig.from_config_dict(bgm_path, bgm_config)
+            try:
+                return build_bgm_only(ctx, bgm_cfg, video_duration)
+            except Exception:
+                logger.exception("[bgm] 纯BGM生成失败: plan_id=%s", ctx.plan_id)
         return None
 
     # 构建音频处理命令
@@ -124,10 +139,25 @@ def mix_audio(
     # 简单场景：只有主图层 + 无独立音频 → 直接从视频提取音频并拼接
     if main_clips and not audio_clips:
         concat_main_audio(ctx, main_clips, output_path, video_duration)
-        return output_path
+    else:
+        # 有独立音频轨 → amix 混音
+        mix_with_independent_audio(ctx, main_clips, audio_clips, output_path, video_duration)
+
+    # ── BGM 混音 ──
+    if bgm_path and bgm_config and bgm_config.get("enabled", False):
+        from video_processing.bgm_mixer import BGMConfig, mix_bgm_with_main
+
+        bgm_cfg = BGMConfig.from_config_dict(bgm_path, bgm_config)
+        bgm_output = ctx.work_dir / f"audio_with_bgm_{ctx.plan_id}.aac"
+
+        try:
+            # 这里 main_audio 就是 output_path，先有主音频再混 BGM
+            final_path = mix_bgm_with_main(ctx, output_path, bgm_cfg, video_duration)
+            return final_path
+        except Exception:
+            logger.exception("[bgm] BGM 混音失败，回退到无 BGM 音频: plan_id=%s", ctx.plan_id)
+            return output_path
 
-    # 有独立音频轨 → amix 混音
-    mix_with_independent_audio(ctx, main_clips, audio_clips, output_path, video_duration)
     return output_path
 
 

apps/worker/video_processing/unified_render_service.py

@@ -165,6 +165,7 @@ class UnifiedRenderService:
         output_fps: int = DEFAULT_FPS,
         transition_duration: float = DEFAULT_TRANSITION_DURATION,
         asr_service: Any = None,  # ASRService 实例，用于自动生成字幕
+        bgm_path: str | None = None,  # BGM 本地文件路径
     ):
         self.plan = plan
         self.clips = clips
@@ -175,6 +176,7 @@ class UnifiedRenderService:
         self.output_fps = output_fps
         self.transition_duration = transition_duration
         self.asr_service = asr_service
+        self.bgm_path = bgm_path
 
     def render(self) -> RenderResult:
         """执行渲染，返回 RenderResult.
@@ -288,9 +290,55 @@ class UnifiedRenderService:
         if is_pass_through:
             # 直通场景已在一次调用中完成视频+音频
             has_audio = pass_through_has_audio
+            # 直通模式下也支持 BGM 混音：提取音频 → 混 BGM → 合并回视频
+            if self.bgm_path and pass_through_has_audio:
+                config = self.plan.config or {}
+                bgm_config = config.get("bgm", {}) or {}
+                if bgm_config.get("enabled", False):
+                    ctx = RenderContext(work_dir=self.work_dir, plan_id=self.plan.id)
+                    from video_processing.bgm_mixer import BGMConfig, mix_bgm_with_main
+
+                    bgm_cfg = BGMConfig.from_config_dict(self.bgm_path, bgm_config)
+                    # 从直通输出中提取音频
+                    main_audio_path = self.work_dir / f"pass_through_audio_{self.plan.id}.aac"
+                    extract_cmd = [
+                        FFMPEG_BIN,
+                        "-y",
+                        "-i",
+                        str(output_path),
+                        "-vn",
+                        "-acodec",
+                        "aac",
+                        "-b:a",
+                        "128k",
+                        str(main_audio_path),
+                    ]
+                    try:
+                        from video_processing.ffmpeg_utils import run_ffmpeg
+
+                        run_ffmpeg(extract_cmd)
+                        final_audio = mix_bgm_with_main(ctx, main_audio_path, bgm_cfg, video_duration)
+                        # 合并回视频
+
+                        bgm_output = self.work_dir / f"rendered_{self.plan.id}_bgm.mp4"
+                        merge_audio_video(ctx, output_path, final_audio, bgm_output)
+                        output_path = bgm_output
+                        logger.info("[unified-render] pass-through BGM mix done: plan_id=%s", self.plan.id)
+                    except Exception:
+                        logger.exception(
+                            "[unified-render] pass-through BGM mix failed, skipping: plan_id=%s", self.plan.id
+                        )
         else:
             ctx = RenderContext(work_dir=self.work_dir, plan_id=self.plan.id)
-            audio_path = mix_audio(ctx, layers, video_duration)
+            config = self.plan.config or {}
+            bgm_config = config.get("bgm", {}) or {}
+            audio_path = mix_audio(
+                ctx,
+                layers,
+                video_duration,
+                bgm_path=self.bgm_path,
+                bgm_config=bgm_config,
+            )
             t_audio_end = time.time()
             audio_mix_ms = int((t_audio_end - t_audio_start) * 1000)
             has_audio = audio_path is not None

apps/worker/worker_app/tasks/generation.py

@@ -335,6 +335,86 @@ def _download_voice_asset(voice_library_id: str, local_path: Path) -> bool:
     return download_asset(storage_key, local_path)
 
 
+def _prepare_bgm_track(
+    *,
+    bgm_config: dict,
+    temp_path: Path,
+    task_id: str = "",
+) -> str | None:
+    """准备 BGM 音频文件（下载到本地）.
+
+    支持 3 种来源（按优先级）：
+    1. audio_url — 外部直链 URL（最高优先级）
+    2. asset_id — 素材库中的音频素材
+    3. preset_id — 预设 BGM 库
+
+    Returns:
+        BGM 本地文件路径，准备失败返回 None
+    """
+    from urllib.parse import urlparse
+
+    audio_url = bgm_config.get("audio_url", "") or ""
+    asset_id = bgm_config.get("asset_id", "") or ""
+    preset_id = bgm_config.get("preset_id", "") or ""
+
+    bgm_file = temp_path / f"bgm_{task_id or 'track'}.mp3"
+
+    # 优先级1：外部直链 URL
+    if audio_url:
+        try:
+            parsed = urlparse(audio_url)
+            if parsed.scheme in ("http", "https"):
+                import urllib.request
+
+                logger.info("[task_id=%s] [BGM] 从URL下载: %s", task_id, audio_url[:80])
+                urllib.request.urlretrieve(audio_url, bgm_file)  # nosec B310
+                if bgm_file.exists() and bgm_file.stat().st_size > 0:
+                    return str(bgm_file)
+        except Exception as e:
+            logger.warning("[task_id=%s] [BGM] URL下载失败: %s", task_id, e)
+
+    # 优先级2：素材库素材
+    if asset_id:
+        try:
+            from app.core.db import SessionLocal
+
+            from packages.adapters.sqlalchemy_impl.models import AssetModel
+
+            session = SessionLocal()
+            try:
+                model = session.query(AssetModel).filter(AssetModel.id == asset_id).first()
+                if model and model.file_url:
+                    storage_key = model.file_url
+                    logger.info("[task_id=%s] [BGM] 从素材库下载: asset_id=%s", task_id, asset_id)
+                    ok = download_asset(storage_key, bgm_file)
+                    if ok and bgm_file.exists() and bgm_file.stat().st_size > 0:
+                        return str(bgm_file)
+            finally:
+                session.close()
+        except Exception as e:
+            logger.warning("[task_id=%s] [BGM] 素材库下载失败: %s", task_id, e)
+
+    # 优先级3：预设 BGM 库
+    if preset_id:
+        try:
+            from packages.domain.preset_bgm import get_preset_bgm
+
+            preset = get_preset_bgm(preset_id)
+            if preset and preset.audio_url:
+                import urllib.request
+
+                logger.info("[task_id=%s] [BGM] 从预设库下载: preset_id=%s", task_id, preset_id)
+                urllib.request.urlretrieve(preset.audio_url, bgm_file)  # nosec B310
+                if bgm_file.exists() and bgm_file.stat().st_size > 0:
+                    return str(bgm_file)
+        except Exception as e:
+            logger.warning("[task_id=%s] [BGM] 预设库下载失败: %s", task_id, e)
+
+    # 所有来源都失败
+    logger.warning("[task_id=%s] [BGM] 所有来源都无法获取BGM，跳过", task_id)
+    return None
+
+
 def _verify_url_accessible(url: str, timeout: float = 10.0, retries: int = 2) -> bool:
     """HEAD 请求校验 URL 可访问（含重试，防止 OSS 抖动误报）。
 
@@ -884,6 +964,22 @@ def _render_video(
         )
     else:
         logger.info("[task_id=%s] [渲染] unified 引擎 FFmpeg 渲染开始", task_id)
+
+        # ── 准备 BGM 音频 ──
+        bgm_path: str | None = None
+        plan_config = virtual_plan.config or {}
+        bgm_config = plan_config.get("bgm", {}) or {}
+        if bgm_config.get("enabled", False):
+            try:
+                bgm_path = _prepare_bgm_track(
+                    bgm_config=bgm_config,
+                    temp_path=temp_path,
+                    task_id=task_id,
+                )
+            except Exception as bgm_err:
+                logger.warning("[task_id=%s] [BGM] 准备失败，跳过BGM: %s", task_id, bgm_err)
+                bgm_path = None
+
         render_service = UnifiedRenderService(
             plan=virtual_plan,
             clips=virtual_clips,
@@ -893,6 +989,7 @@ def _render_video(
             output_height=OUTPUT_HEIGHT,
             output_fps=int(OUTPUT_FPS),
             asr_service=get_asr_service(),
+            bgm_path=bgm_path,
         )
         render_result = render_service.render()
         render_output_path = render_result.output_path

packages/domain/config_schemas.py

@@ -122,9 +122,22 @@ class SubtitleConfig(BaseModel):
 class BGMConfig(BaseModel):
     """BGM 配置"""
 
+    enabled: bool = Field(default=False, description="是否启用 BGM")
     source: BGMSource = Field(default=BGMSource.LIBRARY, description="BGM 来源")
-    asset_id: str = Field(default="", description="BGM 素材 ID")
-    volume: float = Field(default=0.3, ge=0.0, le=1.0, description="音量 (0.0 ~ 1.0)")
+    asset_id: str = Field(default="", description="BGM 素材 ID（来源为 library/upload 时使用）")
+    preset_id: str = Field(default="", description="预设 BGM ID（来源为 ai_recommend 或使用内置库时使用）")
+    audio_url: str = Field(default="", description="BGM 音频 URL（外部直链，优先级最高）")
+    volume: float = Field(default=0.3, ge=0.0, le=1.0, description="BGM 音量 (0.0 ~ 1.0)")
+    fade_in: float = Field(default=0.0, ge=0.0, le=30.0, description="淡入时长（秒）")
+    fade_out: float = Field(default=0.0, ge=0.0, le=30.0, description="淡出时长（秒）")
+    loop_enabled: bool = Field(default=True, description="BGM 是否循环播放以铺满整个视频时长")
+    sidechain_enabled: bool = Field(default=False, description="是否启用人声闪避（有人声时 BGM 自动降低音量）")
+    sidechain_ratio: float = Field(
+        default=0.3, ge=0.0, le=1.0, description="人声闪避时 BGM 音量降低比例（0.3 = 降低30%）"
+    )
+    sidechain_attack: float = Field(default=0.02, ge=0.001, le=1.0, description="人声闪避攻击时间（秒）")
+    sidechain_release: float = Field(default=0.5, ge=0.01, le=5.0, description="人声闪避释放时间（秒）")
+    sidechain_threshold: float = Field(default=-25.0, ge=-60.0, le=0.0, description="人声闪避触发阈值（dB）")
 
 
 # ── 完整 config 模型 ─────────────────────────────────────────────────────────
@@ -190,9 +203,20 @@ DEFAULT_EDIT_PLAN_CONFIG: dict = {
         "animation": "fade_in",
     },
     "bgm": {
+        "enabled": False,
         "source": "library",
         "asset_id": "",
+        "preset_id": "",
+        "audio_url": "",
         "volume": 0.3,
+        "fade_in": 0.0,
+        "fade_out": 0.0,
+        "loop_enabled": True,
+        "sidechain_enabled": False,
+        "sidechain_ratio": 0.3,
+        "sidechain_attack": 0.02,
+        "sidechain_release": 0.5,
+        "sidechain_threshold": -25.0,
     },
     "editing_mode": "one_take",
 }

packages/domain/preset_bgm.py

@@ -0,0 +1,160 @@
+"""预设 BGM 库 — 免费可商用背景音乐清单.
+
+按风格分类，存储在 OSS 或 CDN 上。
+实际音频文件由运维统一上传，这里只维护元数据清单。
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+
+
+@dataclass(frozen=True)
+class PresetBGM:
+    """预设 BGM 条目"""
+
+    id: str
+    name: str
+    style: str  # 风格分类：upbeat/relax/tech/commerce/emotional/cinematic
+    duration: float  # 时长（秒）
+    artist: str = ""
+    description: str = ""
+    tags: list[str] = field(default_factory=list)
+    audio_url: str = ""  # CDN/OSS 地址，空字符串表示待部署
+
+
+# ── 预设库清单 ────────────────────────────────────────────────────────────────
+
+PRESET_BGM_LIBRARY: list[PresetBGM] = [
+    # 轻快 upbeat
+    PresetBGM(
+        id="bgm_upbeat_001",
+        name="阳光清晨",
+        style="upbeat",
+        duration=120.0,
+        artist="免费商用音乐库",
+        description="轻快明亮的吉他+钢琴，适合vlog、生活记录",
+        tags=["轻快", "阳光", "吉他", "vlog"],
+    ),
+    PresetBGM(
+        id="bgm_upbeat_002",
+        name="活力节拍",
+        style="upbeat",
+        duration=95.0,
+        artist="免费商用音乐库",
+        description="电子鼓点+合成器，节奏明快，适合运动、产品展示",
+        tags=["轻快", "电子", "活力", "运动"],
+    ),
+    PresetBGM(
+        id="bgm_upbeat_003",
+        name="夏日漫步",
+        style="upbeat",
+        duration=110.0,
+        artist="免费商用音乐库",
+        description="Ukulele+口哨，轻松愉悦，适合旅行、美食",
+        tags=["轻快", "夏日", "ukulele", "旅行"],
+    ),
+    # 治愈 relax
+    PresetBGM(
+        id="bgm_relax_001",
+        name="静谧时光",
+        style="relax",
+        duration=180.0,
+        artist="免费商用音乐库",
+        description="温柔钢琴独奏，治愈系，适合读书、冥想",
+        tags=["治愈", "钢琴", "安静", "冥想"],
+    ),
+    PresetBGM(
+        id="bgm_relax_002",
+        name="雨后森林",
+        style="relax",
+        duration=150.0,
+        artist="免费商用音乐库",
+        description="自然白噪音+轻柔吉他，放松减压",
+        tags=["治愈", "自然", "放松", "环境音"],
+    ),
+    PresetBGM(
+        id="bgm_relax_003",
+        name="月光奏鸣曲",
+        style="relax",
+        duration=200.0,
+        artist="古典音乐(公版)",
+        description="贝多芬经典钢琴作品，公版免费",
+        tags=["治愈", "古典", "钢琴", "优雅"],
+    ),
+    # 科技 tech
+    PresetBGM(
+        id="bgm_tech_001",
+        name="未来科技",
+        style="tech",
+        duration=85.0,
+        artist="免费商用音乐库",
+        description="电子合成器+科技鼓点，适合数码产品、科技解说",
+        tags=["科技", "电子", "未来感", "数码"],
+    ),
+    PresetBGM(
+        id="bgm_tech_002",
+        name="数据脉冲",
+        style="tech",
+        duration=100.0,
+        artist="免费商用音乐库",
+        description="极简电子节奏，适合数据分析、AI类视频",
+        tags=["科技", "极简", "数据", "AI"],
+    ),
+    # 电商 commerce
+    PresetBGM(
+        id="bgm_commerce_001",
+        name="心动时刻",
+        style="commerce",
+        duration=75.0,
+        artist="免费商用音乐库",
+        description="时尚动感节奏，适合商品展示、带货视频",
+        tags=["电商", "时尚", "动感", "带货"],
+    ),
+    PresetBGM(
+        id="bgm_commerce_002",
+        name="品质生活",
+        style="commerce",
+        duration=90.0,
+        artist="免费商用音乐库",
+        description="高级感轻音乐，适合品牌宣传、高端产品",
+        tags=["电商", "高端", "品牌", "品质"],
+    ),
+]
+
+# ── 风格分类字典 ──────────────────────────────────────────────────────────────
+
+BGM_STYLES: dict[str, str] = {
+    "upbeat": "轻快",
+    "relax": "治愈",
+    "tech": "科技",
+    "commerce": "电商",
+    "emotional": "情感",
+    "cinematic": "电影",
+}
+
+
+# ── 工具函数 ──────────────────────────────────────────────────────────────────
+
+
+def get_preset_bgm(bgm_id: str) -> PresetBGM | None:
+    """按 ID 获取预设 BGM。"""
+    for bgm in PRESET_BGM_LIBRARY:
+        if bgm.id == bgm_id:
+            return bgm
+    return None
+
+
+def list_preset_bgm_by_style(style: str) -> list[PresetBGM]:
+    """按风格筛选预设 BGM。"""
+    return [bgm for bgm in PRESET_BGM_LIBRARY if bgm.style == style]
+
+
+def search_preset_bgm(keyword: str) -> list[PresetBGM]:
+    """按关键词搜索预设 BGM（名称+标签+描述）。"""
+    kw = keyword.lower()
+    results = []
+    for bgm in PRESET_BGM_LIBRARY:
+        if kw in bgm.name.lower() or kw in bgm.description.lower() or any(kw in tag.lower() for tag in bgm.tags):
+            results.append(bgm)
+    return results

tests/unit/test_bgm_mixer.py

@@ -0,0 +1,359 @@
+"""BGM 混音单元测试.
+
+测试：
+- BGMConfig 配置解析与边界值
+- 预设 BGM 库查询
+- 纯 BGM 音频生成（端到端 ffmpeg）
+- BGM + 主音频混音（端到端 ffmpeg）
+- 淡入淡出效果
+- 音量边界（0 和 1）
+- sidechain 人声闪避
+"""
+
+import sys
+import tempfile
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "worker"))
+sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "api"))
+
+import pytest
+from video_processing.bgm_mixer import BGMConfig, build_bgm_only, mix_bgm_with_main, prepare_bgm_track
+from video_processing.render_audio import RenderContext
+
+# ── Fixtures ──────────────────────────────────────────────────────────────────
+
+
+@pytest.fixture
+def work_dir(tmp_path):
+    return tmp_path
+
+
+@pytest.fixture
+def ctx(work_dir):
+    return RenderContext(work_dir=work_dir, plan_id="test_plan")
+
+
+@pytest.fixture
+def main_audio_path(work_dir):
+    """生成 10 秒测试主音频（正弦波模拟人声）。"""
+    import subprocess
+
+    path = work_dir / "main.aac"
+    # 生成 10 秒 440Hz 正弦波模拟主音频
+    subprocess.run(
+        [
+            "ffmpeg",
+            "-y",
+            "-f",
+            "lavfi",
+            "-i",
+            "sine=frequency=440:duration=10:sample_rate=44100",
+            "-c:a",
+            "aac",
+            "-b:a",
+            "128k",
+            str(path),
+        ],
+        capture_output=True,
+        check=True,
+        timeout=30,
+    )
+    return str(path)
+
+
+@pytest.fixture
+def bgm_audio_path(work_dir):
+    """生成 5 秒测试 BGM（更低频率模拟背景音乐）。"""
+    import subprocess
+
+    path = work_dir / "bgm.aac"
+    # 生成 5 秒 220Hz 正弦波模拟 BGM
+    subprocess.run(
+        [
+            "ffmpeg",
+            "-y",
+            "-f",
+            "lavfi",
+            "-i",
+            "sine=frequency=220:duration=5:sample_rate=44100",
+            "-c:a",
+            "aac",
+            "-b:a",
+            "128k",
+            str(path),
+        ],
+        capture_output=True,
+        check=True,
+        timeout=30,
+    )
+    return str(path)
+
+
+# ── BGMConfig 测试 ───────────────────────────────────────────────────────────
+
+
+class TestBGMConfig:
+    """BGMConfig 配置解析测试。"""
+
+    def test_default_values(self):
+        cfg = BGMConfig(bgm_path="/tmp/bgm.mp3")
+        assert cfg.volume == 0.3
+        assert cfg.fade_in == 0.0
+        assert cfg.fade_out == 0.0
+        assert cfg.loop_enabled is True
+        assert cfg.sidechain_enabled is False
+        assert cfg.sidechain_ratio == 0.3
+
+    def test_from_config_dict(self):
+        config_dict = {
+            "enabled": True,
+            "volume": 0.5,
+            "fade_in": 2.0,
+            "fade_out": 3.0,
+            "loop_enabled": False,
+            "sidechain_enabled": True,
+            "sidechain_ratio": 0.5,
+        }
+        cfg = BGMConfig.from_config_dict("/bgm.mp3", config_dict)
+        assert cfg.bgm_path == "/bgm.mp3"
+        assert cfg.volume == 0.5
+        assert cfg.fade_in == 2.0
+        assert cfg.fade_out == 3.0
+        assert cfg.loop_enabled is False
+        assert cfg.sidechain_enabled is True
+        assert cfg.sidechain_ratio == 0.5
+
+    def test_volume_clamped_by_config_schema(self):
+        """音量边界由 Pydantic Schema 在入口层保证，内部直接使用。"""
+        from packages.domain.config_schemas import BGMConfig as BGMConfigSchema
+
+        # 边界值测试
+        cfg = BGMConfigSchema(enabled=True, volume=0.0)
+        assert cfg.volume == 0.0
+
+        cfg = BGMConfigSchema(enabled=True, volume=1.0)
+        assert cfg.volume == 1.0
+
+    def test_fade_boundaries(self):
+        from packages.domain.config_schemas import BGMConfig as BGMConfigSchema
+
+        # 0 是合法值
+        cfg = BGMConfigSchema(fade_in=0, fade_out=0)
+        assert cfg.fade_in == 0.0
+        assert cfg.fade_out == 0.0
+
+
+# ── 预设 BGM 库测试 ─────────────────────────────────────────────────────────
+
+
+class TestPresetBGM:
+    """预设 BGM 库查询测试。"""
+
+    def test_total_count(self):
+        from packages.domain.preset_bgm import PRESET_BGM_LIBRARY
+
+        assert len(PRESET_BGM_LIBRARY) >= 10
+
+    def test_get_preset_by_id(self):
+        from packages.domain.preset_bgm import get_preset_bgm
+
+        bgm = get_preset_bgm("bgm_upbeat_001")
+        assert bgm is not None
+        assert bgm.name == "阳光清晨"
+        assert bgm.style == "upbeat"
+
+    def test_get_preset_not_found(self):
+        from packages.domain.preset_bgm import get_preset_bgm
+
+        assert get_preset_bgm("nonexistent") is None
+
+    def test_list_by_style(self):
+        from packages.domain.preset_bgm import list_preset_bgm_by_style
+
+        upbeat = list_preset_bgm_by_style("upbeat")
+        assert len(upbeat) >= 3
+        assert all(b.style == "upbeat" for b in upbeat)
+
+    def test_search_by_keyword(self):
+        from packages.domain.preset_bgm import search_preset_bgm
+
+        results = search_preset_bgm("钢琴")
+        assert len(results) >= 2
+        assert any("钢琴" in b.tags for b in results)
+
+    def test_all_presets_have_basic_fields(self):
+        from packages.domain.preset_bgm import PRESET_BGM_LIBRARY
+
+        for bgm in PRESET_BGM_LIBRARY:
+            assert bgm.id, f"{bgm.name} 缺少 id"
+            assert bgm.name, "缺少 name"
+            assert bgm.style, f"{bgm.name} 缺少 style"
+            assert bgm.duration > 0, f"{bgm.name} 时长无效"
+
+
+# ── BGM 处理端到端测试 ──────────────────────────────────────────────────────
+
+
+class TestPrepareBGMTrack:
+    """prepare_bgm_track 端到端测试。"""
+
+    def test_bgm_without_loop_short_duration(self, ctx, bgm_audio_path):
+        """BGM 比目标时长短且不循环 → 截断到目标时长（但前面没有足够内容）。"""
+        bgm = BGMConfig(bgm_path=bgm_audio_path, volume=0.5, loop_enabled=False)
+        result = prepare_bgm_track(ctx, bgm, target_duration=3.0)
+
+        assert result.exists()
+        assert result.stat().st_size > 0
+
+    def test_bgm_with_loop_longer_duration(self, ctx, bgm_audio_path):
+        """BGM 比目标时长短，循环铺满。"""
+        bgm = BGMConfig(bgm_path=bgm_audio_path, volume=0.3, loop_enabled=True)
+        # BGM 5 秒，目标 12 秒，需要循环 3 次
+        result = prepare_bgm_track(ctx, bgm, target_duration=12.0)
+
+        assert result.exists()
+        assert result.stat().st_size > 0
+
+    def test_bgm_fade_in_and_fade_out(self, ctx, bgm_audio_path):
+        """BGM 淡入淡出效果。"""
+        bgm = BGMConfig(
+            bgm_path=bgm_audio_path,
+            volume=0.5,
+            fade_in=1.0,
+            fade_out=1.0,
+            loop_enabled=False,
+        )
+        result = prepare_bgm_track(ctx, bgm, target_duration=4.0)
+
+        assert result.exists()
+        assert result.stat().st_size > 0
+
+    def test_volume_zero(self, ctx, bgm_audio_path):
+        """音量为 0 时仍能正常处理。"""
+        bgm = BGMConfig(bgm_path=bgm_audio_path, volume=0.0, loop_enabled=False)
+        result = prepare_bgm_track(ctx, bgm, target_duration=3.0)
+
+        assert result.exists()
+        assert result.stat().st_size > 0
+
+    def test_volume_one(self, ctx, bgm_audio_path):
+        """音量为 1（最大）时正常处理。"""
+        bgm = BGMConfig(bgm_path=bgm_audio_path, volume=1.0, loop_enabled=False)
+        result = prepare_bgm_track(ctx, bgm, target_duration=3.0)
+
+        assert result.exists()
+        assert result.stat().st_size > 0
+
+
+class TestMixBGMMain:
+    """BGM + 主音频混音端到端测试。"""
+
+    def test_simple_mix(self, ctx, main_audio_path, bgm_audio_path):
+        """普通 amix 混音（无 sidechain）。"""
+        bgm = BGMConfig(
+            bgm_path=bgm_audio_path,
+            volume=0.3,
+            loop_enabled=True,
+            sidechain_enabled=False,
+        )
+        result = mix_bgm_with_main(ctx, Path(main_audio_path), bgm, target_duration=8.0)
+
+        assert result.exists()
+        assert result.stat().st_size > 0
+
+    def test_sidechain_mix(self, ctx, main_audio_path, bgm_audio_path):
+        """sidechain 人声闪避混音。"""
+        bgm = BGMConfig(
+            bgm_path=bgm_audio_path,
+            volume=0.5,
+            loop_enabled=True,
+            sidechain_enabled=True,
+            sidechain_ratio=0.3,
+            sidechain_threshold=-25.0,
+            sidechain_attack=0.02,
+            sidechain_release=0.5,
+        )
+        result = mix_bgm_with_main(ctx, Path(main_audio_path), bgm, target_duration=8.0)
+
+        assert result.exists()
+        assert result.stat().st_size > 0
+
+    def test_sidechain_max_ratio(self, ctx, main_audio_path, bgm_audio_path):
+        """sidechain 最大闪避比例。"""
+        bgm = BGMConfig(
+            bgm_path=bgm_audio_path,
+            volume=0.5,
+            loop_enabled=True,
+            sidechain_enabled=True,
+            sidechain_ratio=0.9,  # 降低 90%
+        )
+        result = mix_bgm_with_main(ctx, Path(main_audio_path), bgm, target_duration=5.0)
+
+        assert result.exists()
+        assert result.stat().st_size > 0
+
+
+class TestBuildBGMOnly:
+    """纯 BGM 模式测试。"""
+
+    def test_build_bgm_only(self, ctx, bgm_audio_path):
+        """只有 BGM、没有主音频时生成纯 BGM 音频。"""
+        bgm = BGMConfig(
+            bgm_path=bgm_audio_path,
+            volume=0.3,
+            fade_in=1.0,
+            fade_out=1.0,
+            loop_enabled=True,
+        )
+        result = build_bgm_only(ctx, bgm, target_duration=15.0)
+
+        assert result.exists()
+        assert result.stat().st_size > 0
+
+
+# ── Config Schema 集成测试 ───────────────────────────────────────────────────
+
+
+class TestConfigSchemaIntegration:
+    """config schema 与渲染配置的集成测试。"""
+
+    def test_full_bgm_config(self):
+        """完整 BGM 配置能正确解析。"""
+        from packages.domain.config_schemas import EditPlanConfigSchema, normalize_plan_config
+
+        config = normalize_plan_config(
+            {
+                "bgm": {
+                    "enabled": True,
+                    "source": "library",
+                    "asset_id": "bgm-asset-001",
+                    "volume": 0.4,
+                    "fade_in": 2.5,
+                    "fade_out": 3.0,
+                    "loop_enabled": True,
+                    "sidechain_enabled": True,
+                    "sidechain_ratio": 0.4,
+                }
+            }
+        )
+
+        bgm = config["bgm"]
+        assert bgm["enabled"] is True
+        assert bgm["volume"] == 0.4
+        assert bgm["fade_in"] == 2.5
+        assert bgm["fade_out"] == 3.0
+        assert bgm["loop_enabled"] is True
+        assert bgm["sidechain_enabled"] is True
+        assert bgm["sidechain_ratio"] == 0.4
+        # 默认值保留
+        assert bgm["sidechain_attack"] == 0.02
+        assert bgm["sidechain_release"] == 0.5
+        assert bgm["sidechain_threshold"] == -25.0
+
+    def test_bgm_disabled_by_default(self):
+        """默认 BGM 是关闭的。"""
+        from packages.domain.config_schemas import normalize_plan_config
+
+        config = normalize_plan_config({})
+        assert config["bgm"]["enabled"] is False