apps/api/app/api/routes/generation_tasks.py

@@ -61,6 +61,7 @@ def _to_generation_task_response(task) -> GenerationTaskResponse:
         batch_id=getattr(task, "batch_id", ""),
         video_title=getattr(task, "video_title", ""),
         resolution=getattr(task, "resolution", ""),
+        bgm_config=getattr(task, "bgm_config", {}) or {},
         logs=getattr(task, "logs", "[]"),
         status=task.status,
         progress=task.progress,
@@ -289,6 +290,7 @@ def create_generation_task(
                     batch_id=batch_id,
                     video_title=request.video_title,
                     resolution=request.resolution,
+                    bgm_config=request.bgm_config,
                     auto_retry_enabled=request.auto_retry_enabled,
                     auto_retry_max=request.auto_retry_max,
                 )

apps/api/app/schemas/generation_task.py

@@ -51,6 +51,11 @@ class CreateGenerationTaskRequest(BaseModel):
         default="",
         description="输出分辨率，格式为 WIDTHxHEIGHT，如 1280x720、1080x1920。为空使用默认 1280x720",
     )
+    # ── 自定义 BGM ──
+    bgm_config: dict = Field(
+        default_factory=dict,
+        description="自定义BGM配置，覆盖模板BGM设置。支持 enabled/source/asset_id/preset_id/audio_url/volume 等字段",
+    )
 
     @model_validator(mode="after")
     def _check_at_least_one_mode(self) -> "CreateGenerationTaskRequest":
@@ -80,6 +85,7 @@ class GenerationTaskResponse(BaseModel):
     batch_id: str = ""
     video_title: str = ""
     resolution: str = ""
+    bgm_config: dict = Field(default_factory=dict)
     status: str
     progress: float
     result_count: int

apps/worker/worker_app/tasks/generation.py

@@ -24,6 +24,8 @@ from typing import Any
 from worker_app.celery_app import celery_app
 from worker_app.db import SessionLocal
 
+from packages.domain.bgm_utils import merge_bgm_config
+
 OUTPUT_WIDTH = 1280
 OUTPUT_HEIGHT = 720
 OUTPUT_FPS = 25.0
@@ -1111,6 +1113,7 @@ def _load_task_info(task_id: str) -> dict | None:
             "user_id": getattr(gen_task, "created_by_user_id", "") or "",
             "video_title": getattr(gen_task, "video_title", "") or "",
             "resolution": getattr(gen_task, "resolution", "") or "",
+            "bgm_config": dict(getattr(gen_task, "bgm_config", {}) or {}),
         }
     finally:
         session.close()
@@ -1170,6 +1173,7 @@ def _render_video(
     temp_path: Path,
     output_name: str,
     resolution: str = "",
+    bgm_config: dict | None = None,
 ) -> tuple[Path, float]:
     """渲染视频（含配音混音）。
 
@@ -1201,6 +1205,20 @@ def _render_video(
                 list(template_config.keys()),
             )
 
+    # 用户自定义 BGM 覆盖模板 BGM（用户指定优先级最高）
+    if bgm_config:
+        plan_cfg = virtual_plan.config or {}
+        template_bgm = plan_cfg.get("bgm", {}) or {}
+        merged_bgm = merge_bgm_config(template_bgm, bgm_config)
+        plan_cfg["bgm"] = merged_bgm
+        virtual_plan.config = plan_cfg
+        logger.info(
+            "[task_id=%s] [渲染] 用户自定义BGM已合并: enabled=%s source=%s",
+            task_id,
+            merged_bgm.get("enabled", False),
+            merged_bgm.get("source", ""),
+        )
+
     # 确保输出分辨率配置存在
     # 优先级：用户指定 > 模板配置 > 默认 1280x720
     plan_cfg = virtual_plan.config or {}
@@ -1453,6 +1471,7 @@ def generate_video(self, task_id: str) -> dict:
                 temp_path=temp_path,
                 output_name=output_name,
                 resolution=task_info.get("resolution", ""),
+                bgm_config=task_info.get("bgm_config", {}),
             )
 
             if gen_task:

migrations/005_custom_bgm_generation.sql

@@ -0,0 +1,7 @@
+-- 一键生成支持自定义BGM
+-- 为 generation_tasks 表添加 bgm_config 字段，用于存储用户自定义BGM配置
+-- 创建时间: 2026-07-23
+
+ALTER TABLE generation_tasks ADD COLUMN IF NOT EXISTS bgm_config JSON NOT NULL DEFAULT '{}'::json;
+
+COMMENT ON COLUMN generation_tasks.bgm_config IS '自定义BGM配置，覆盖模板BGM设置';

packages/adapters/sqlalchemy_impl/generation_task_repository.py

@@ -35,6 +35,7 @@ def _to_domain(model: GenerationTaskModel) -> GenerationTask:
         batch_id=model.batch_id or "",
         video_title=getattr(model, "video_title", "") or "",
         resolution=getattr(model, "resolution", "") or "",
+        bgm_config=dict(getattr(model, "bgm_config", {}) or {}),
         logs=model.logs or "[]",
         created_at=model.created_at,
         updated_at=model.updated_at,
@@ -72,6 +73,7 @@ class SQLAlchemyGenerationTaskRepository:
             batch_id=task.batch_id or "",
             video_title=task.video_title or "",
             resolution=task.resolution or "",
+            bgm_config=task.bgm_config or {},
             logs=task.logs,
             created_at=task.created_at,
             updated_at=task.updated_at,
@@ -234,6 +236,8 @@ class SQLAlchemyGenerationTaskRepository:
             model.video_title = task.video_title or ""
         if hasattr(model, "resolution"):
             model.resolution = task.resolution or ""
+        if hasattr(model, "bgm_config"):
+            model.bgm_config = task.bgm_config or {}
         model.logs = task.logs
         self.session.commit()
         return task

packages/adapters/sqlalchemy_impl/models.py

@@ -292,6 +292,7 @@ class GenerationTaskModel(Base):
     batch_id = Column(String(36), nullable=False, default="", index=True)
     video_title = Column(String(255), nullable=False, default="")
     resolution = Column(String(20), nullable=False, default="")
+    bgm_config = Column(JSON, nullable=False, default=dict)
     extra_meta = Column("metadata", JSON, nullable=False, default=dict)
     logs = Column(Text, nullable=False, default="[]", server_default="[]")
     created_at = Column(DateTime, nullable=False, default=lambda: datetime.now(timezone.utc))

packages/application/generation_tasks.py

@@ -23,6 +23,7 @@ class CreateGenerationTaskCommand:
     batch_id: str = ""
     video_title: str = ""
     resolution: str = ""
+    bgm_config: dict = field(default_factory=dict)
     auto_retry_enabled: bool = False
     auto_retry_max: int = 0
 
@@ -52,6 +53,7 @@ class CreateGenerationTaskUseCase:
             batch_id=command.batch_id,
             video_title=command.video_title,
             resolution=command.resolution,
+            bgm_config=command.bgm_config,
             auto_retry_enabled=command.auto_retry_enabled,
             auto_retry_max=command.auto_retry_max,
         )

packages/domain/bgm_utils.py

@@ -0,0 +1,32 @@
+"""BGM配置相关的领域工具函数。"""
+
+from __future__ import annotations
+
+
+def merge_bgm_config(template_bgm: dict, user_bgm: dict) -> dict:
+    """合并模板BGM配置与用户自定义BGM配置。
+
+    用户配置优先级高于模板配置：
+    - 用户显式指定的字段覆盖模板对应字段
+    - 用户未指定的字段保留模板值
+    - enabled 特殊处理：只有用户显式传了才覆盖，否则保留模板状态
+
+    Args:
+        template_bgm: 模板BGM配置
+        user_bgm: 用户自定义BGM配置
+
+    Returns:
+        合并后的BGM配置字典
+    """
+    if not user_bgm:
+        return dict(template_bgm)
+    if not template_bgm:
+        return dict(user_bgm)
+
+    merged = {**template_bgm, **user_bgm}
+
+    # enabled 特殊处理：用户没传就保留模板的
+    if "enabled" not in user_bgm and "enabled" in template_bgm:
+        merged["enabled"] = template_bgm["enabled"]
+
+    return merged

packages/domain/generation_task.py

@@ -92,6 +92,7 @@ class GenerationTask:
     batch_id: str = ""
     video_title: str = ""
     resolution: str = ""
+    bgm_config: dict = field(default_factory=dict)
     logs: str = "[]"
     created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
     updated_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
@@ -114,6 +115,7 @@ class GenerationTask:
         batch_id: str = "",
         video_title: str = "",
         resolution: str = "",
+        bgm_config: dict | None = None,
         auto_retry_enabled: bool = False,
         auto_retry_max: int = 0,
     ) -> "GenerationTask":
@@ -137,6 +139,7 @@ class GenerationTask:
             batch_id=batch_id,
             video_title=video_title.strip(),
             resolution=resolution.strip(),
+            bgm_config=dict(bgm_config) if bgm_config else {},
             auto_retry_enabled=auto_retry_enabled,
             auto_retry_max=auto_retry_max,
         )

tests/unit/test_bgm_config_merge.py

@@ -0,0 +1,160 @@
+"""BGM配置合并逻辑单元测试。"""
+import pytest
+
+from packages.domain.bgm_utils import merge_bgm_config
+
+
+class TestMergeBgmConfig:
+    """merge_bgm_config 单元测试。"""
+
+    def test_template_empty_user_has_config(self):
+        """模板为空，用户有配置 → 返回用户配置副本。"""
+        template = {}
+        user = {"audio_url": "http://example.com/bgm.mp3", "volume": 0.8}
+
+        result = merge_bgm_config(template, user)
+
+        assert result == {"audio_url": "http://example.com/bgm.mp3", "volume": 0.8}
+        # 确保返回的是新对象，不是同一引用
+        assert result is not user
+
+    def test_user_empty_template_has_config(self):
+        """用户为空，模板有配置 → 返回模板配置副本。"""
+        template = {"audio_url": "http://example.com/default.mp3", "enabled": True}
+        user = {}
+
+        result = merge_bgm_config(template, user)
+
+        assert result == {"audio_url": "http://example.com/default.mp3", "enabled": True}
+        assert result is not template
+
+    def test_both_empty(self):
+        """都为空 → 返回空dict。"""
+        result = merge_bgm_config({}, {})
+        assert result == {}
+
+    def test_user_overrides_single_field(self):
+        """用户覆盖单个普通字段。"""
+        template = {"audio_url": "http://template.com/default.mp3", "volume": 0.5}
+        user = {"audio_url": "http://user.com/custom.mp3"}
+
+        result = merge_bgm_config(template, user)
+
+        assert result["audio_url"] == "http://user.com/custom.mp3"
+        assert result["volume"] == 0.5  # 模板值保留
+
+    def test_user_overrides_multiple_fields(self):
+        """用户覆盖多个字段。"""
+        template = {
+            "audio_url": "http://template.com/default.mp3",
+            "volume": 0.5,
+            "fade_in": 1.0,
+            "fade_out": 1.0,
+        }
+        user = {
+            "audio_url": "http://user.com/custom.mp3",
+            "volume": 0.9,
+        }
+
+        result = merge_bgm_config(template, user)
+
+        assert result["audio_url"] == "http://user.com/custom.mp3"
+        assert result["volume"] == 0.9
+        assert result["fade_in"] == 1.0
+        assert result["fade_out"] == 1.0
+
+    def test_enabled_not_in_user_keep_template_true(self):
+        """enabled特殊处理：用户没传，模板enabled=True → 保留True。"""
+        template = {"audio_url": "http://template.com/default.mp3", "enabled": True}
+        user = {"audio_url": "http://user.com/custom.mp3"}
+
+        result = merge_bgm_config(template, user)
+
+        assert result["enabled"] is True
+        assert result["audio_url"] == "http://user.com/custom.mp3"
+
+    def test_enabled_not_in_user_keep_template_false(self):
+        """enabled特殊处理：用户没传，模板enabled=False → 保留False。"""
+        template = {"audio_url": "http://template.com/default.mp3", "enabled": False}
+        user = {"audio_url": "http://user.com/custom.mp3"}
+
+        result = merge_bgm_config(template, user)
+
+        assert result["enabled"] is False
+
+    def test_enabled_explicit_true_overrides_template_false(self):
+        """enabled特殊处理：用户显式传True，覆盖模板False。"""
+        template = {"audio_url": "http://template.com/default.mp3", "enabled": False}
+        user = {"enabled": True}
+
+        result = merge_bgm_config(template, user)
+
+        assert result["enabled"] is True
+
+    def test_enabled_explicit_false_overrides_template_true(self):
+        """enabled特殊处理：用户显式传False，覆盖模板True。"""
+        template = {"audio_url": "http://template.com/default.mp3", "enabled": True}
+        user = {"enabled": False}
+
+        result = merge_bgm_config(template, user)
+
+        assert result["enabled"] is False
+
+    def test_nested_fields_shallow_merge(self):
+        """普通字段（包括嵌套dict）按浅层合并处理。"""
+        template = {
+            "audio_url": "http://template.com/default.mp3",
+            "effects": {"fade_in": 1.0, "fade_out": 1.0},
+        }
+        user = {
+            "effects": {"fade_in": 3.0},  # 整个覆盖，不是深合并
+        }
+
+        result = merge_bgm_config(template, user)
+
+        # 浅层合并：用户的effects完全替换模板的effects
+        assert result["effects"] == {"fade_in": 3.0}
+
+    def test_user_none_value_fields(self):
+        """用户传None值的字段，会覆盖模板值为None。"""
+        template = {"audio_url": "http://template.com/default.mp3", "volume": 0.5}
+        user = {"audio_url": None}
+
+        result = merge_bgm_config(template, user)
+
+        assert result["audio_url"] is None
+        assert result["volume"] == 0.5
+
+    def test_does_not_mutate_inputs(self):
+        """合并操作不修改原始输入对象。"""
+        template = {"audio_url": "http://template.com/default.mp3", "enabled": True, "volume": 0.5}
+        user = {"volume": 0.9}
+        template_copy = dict(template)
+        user_copy = dict(user)
+
+        merge_bgm_config(template, user)
+
+        assert template == template_copy
+        assert user == user_copy
+
+    def test_user_adds_new_field(self):
+        """用户新增模板中没有的字段。"""
+        template = {"audio_url": "http://template.com/default.mp3"}
+        user = {"loop": True, "start_time": 5.0}
+
+        result = merge_bgm_config(template, user)
+
+        assert result["audio_url"] == "http://template.com/default.mp3"
+        assert result["loop"] is True
+        assert result["start_time"] == 5.0
+
+    def test_asset_id_and_preset_id_fields(self):
+        """asset_id和preset_id字段正常合并。"""
+        template = {"preset_id": "preset_default", "volume": 0.5}
+        user = {"asset_id": "asset_user_123"}
+
+        result = merge_bgm_config(template, user)
+
+        assert result["preset_id"] == "preset_default"
+        assert result["asset_id"] == "asset_user_123"
+        assert result["volume"] == 0.5

tests/unit/test_generation_task_domain.py

@@ -92,6 +92,37 @@ class TestGenerationTaskCreate:
         assert task.auto_retry_enabled is True
         assert task.auto_retry_max == 3
 
+    def test_create_with_bgm_config(self):
+        """测试创建带自定义BGM配置"""
+        bgm_cfg = {
+            "enabled": True,
+            "source": "asset",
+            "asset_id": "bgm-asset-001",
+            "volume": 0.6,
+        }
+        task = GenerationTask.create(
+            project_id="",
+            asset_library_id="",
+            template_id="tmpl-001",
+            asset_ids=["asset-1"],
+            bgm_config=bgm_cfg,
+        )
+        assert task.bgm_config == bgm_cfg
+        # 确保是副本，不是引用
+        bgm_cfg["volume"] = 0.8
+        assert task.bgm_config["volume"] == 0.6
+
+    def test_create_default_bgm_config_empty(self):
+        """测试默认BGM配置为空dict"""
+        task = GenerationTask.create(
+            project_id="",
+            asset_library_id="",
+            template_id="tmpl-001",
+            asset_ids=["asset-1"],
+        )
+        assert task.bgm_config == {}
+        assert isinstance(task.bgm_config, dict)
+
     def test_create_with_template_id_only(self):
         """测试只提供 template_id 不提供 project_id（应通过校验）"""
         task = GenerationTask.create(