diff --git a/apps/worker/video_processing/bgm_mixer.py b/apps/worker/video_processing/bgm_mixer.py new file mode 100755 index 000000000..a28320756 --- /dev/null +++ b/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 diff --git a/apps/worker/video_processing/render_audio.py b/apps/worker/video_processing/render_audio.py old mode 100644 new mode 100755 index 286b404fd..462108611 --- a/apps/worker/video_processing/render_audio.py +++ b/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 diff --git a/apps/worker/video_processing/unified_render_service.py b/apps/worker/video_processing/unified_render_service.py index a02c5a626..1aff27ef2 100755 --- a/apps/worker/video_processing/unified_render_service.py +++ b/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 diff --git a/apps/worker/worker_app/tasks/generation.py b/apps/worker/worker_app/tasks/generation.py index 16649aca5..04973fd1a 100755 --- a/apps/worker/worker_app/tasks/generation.py +++ b/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 diff --git a/packages/domain/config_schemas.py b/packages/domain/config_schemas.py index c8bc2d87d..32534f004 100755 --- a/packages/domain/config_schemas.py +++ b/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", } diff --git a/packages/domain/preset_bgm.py b/packages/domain/preset_bgm.py new file mode 100755 index 000000000..e02061f28 --- /dev/null +++ b/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 diff --git a/tests/unit/test_bgm_mixer.py b/tests/unit/test_bgm_mixer.py new file mode 100755 index 000000000..1891f9200 --- /dev/null +++ b/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