feat: BGM背景音乐管理 - 配置API + 预设库API #395

Merged
xiaoxia merged 2 commits from feat/bgm-management into develop 2026-07-17 09:35:21 +08:00
2 changed files with 384 additions and 0 deletions
+170
View File
@@ -520,6 +520,176 @@ def copy_plan(
return _to_response(new_plan)
# ── BGM 背景音乐 ───────────────────────────────────────────────────────────
class BGMConfigUpdateRequest(BaseModel):
"""更新BGM配置请求体"""
enabled: Optional[bool] = Field(default=None, description="是否启用 BGM")
source: Optional[str] = Field(default=None, description="BGM 来源: library/upload/ai_recommend")
asset_id: Optional[str] = Field(default=None, max_length=64, description="BGM 素材 ID")
preset_id: Optional[str] = Field(default=None, max_length=64, description="预设 BGM ID")
audio_url: Optional[str] = Field(default=None, max_length=500, description="BGM 音频 URL")
volume: Optional[float] = Field(default=None, ge=0.0, le=1.0, description="音量 (0.0 ~ 1.0)")
fade_in: Optional[float] = Field(default=None, ge=0.0, le=30.0, description="淡入时长(秒)")
fade_out: Optional[float] = Field(default=None, ge=0.0, le=30.0, description="淡出时长(秒)")
loop_enabled: Optional[bool] = Field(default=None, description="是否循环播放")
sidechain_enabled: Optional[bool] = Field(default=None, description="是否启用人声闪避")
sidechain_ratio: Optional[float] = Field(default=None, ge=0.0, le=1.0, description="闪避音量降低比例")
@router.get(
"/{plan_id}/bgm",
response_model=dict[str, Any],
summary="获取剪辑计划的 BGM 配置",
)
def get_plan_bgm(
plan_id: str,
current_user: AuthenticatedUser = Depends(get_current_user),
db: Session = Depends(get_db_session),
project_repo=Depends(get_project_repository),
) -> dict[str, Any]:
"""获取指定剪辑计划的 BGM 配置。"""
service = EditPlanService(db)
plan = service.get_plan(plan_id)
if plan is None:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"剪辑计划不存在: {plan_id}",
)
if plan.project_id:
check_project_access(project_repo, current_user, plan.project_id)
config = plan.config or {}
bgm_config = config.get("bgm", {})
return {
"plan_id": plan.id,
"bgm": bgm_config,
}
@router.put(
"/{plan_id}/bgm",
response_model=dict[str, Any],
summary="更新剪辑计划的 BGM 配置",
)
def update_plan_bgm(
plan_id: str,
body: BGMConfigUpdateRequest,
current_user: AuthenticatedUser = Depends(get_current_user),
db: Session = Depends(get_db_session),
project_repo=Depends(get_project_repository),
) -> dict[str, Any]:
"""更新剪辑计划的 BGM 配置。
支持部分更新,只传需要修改的字段即可。
启用 BGM 后需要指定来源(asset_id / preset_id / audio_url 三选一)。
"""
service = EditPlanService(db)
plan = service.get_plan(plan_id)
if plan is None:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"剪辑计划不存在: {plan_id}",
)
if plan.project_id:
check_project_access(project_repo, current_user, plan.project_id)
# 读取当前 BGM 配置,合并更新
config = dict(plan.config) if plan.config else {}
current_bgm = dict(config.get("bgm", {}))
update_data = body.model_dump(exclude_none=True)
current_bgm.update(update_data)
# 校验:启用 BGM 时至少有一个有效来源
if current_bgm.get("enabled"):
has_source = any(current_bgm.get(key) for key in ("asset_id", "preset_id", "audio_url") if current_bgm.get(key))
if not has_source:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="启用 BGM 时需要指定素材来源(asset_id / preset_id / audio_url",
)
# 保存到 plan.config.bgm
config["bgm"] = current_bgm
updated_plan = service.update_plan_config(plan_id, config)
logger.info(
"更新BGM配置: plan_id=%s enabled=%s by user=%s",
plan_id,
current_bgm.get("enabled", False),
current_user.user.id,
)
return {
"plan_id": updated_plan.id,
"bgm": current_bgm,
}
# ── BGM 预设库 ────────────────────────────────────────────────────────────────
@router.get(
"/bgm/presets",
response_model=dict[str, Any],
summary="获取预设 BGM 列表",
)
def list_bgm_presets(
style: Optional[str] = Query(default=None, description="按风格筛选"),
keyword: Optional[str] = Query(default=None, description="关键词搜索"),
skip: int = Query(default=0, ge=0, description="分页偏移"),
limit: int = Query(default=50, ge=1, le=200, description="每页数量"),
) -> dict[str, Any]:
"""获取预设 BGM 列表,支持按风格筛选和关键词搜索。
风格可选: upbeat(轻快)、relax(治愈)、tech(科技)、commerce(电商)、
emotional(情感)、cinematic(电影)
"""
from packages.domain.preset_bgm import (
BGM_STYLES,
PRESET_BGM_LIBRARY,
list_preset_bgm_by_style,
search_preset_bgm,
)
bgm_list = PRESET_BGM_LIBRARY
if keyword:
bgm_list = search_preset_bgm(keyword)
elif style:
bgm_list = list_preset_bgm_by_style(style)
total = len(bgm_list)
paged = bgm_list[skip : skip + limit]
return {
"total": total,
"skip": skip,
"limit": limit,
"styles": BGM_STYLES,
"items": [
{
"id": bgm.id,
"name": bgm.name,
"style": bgm.style,
"style_label": BGM_STYLES.get(bgm.style, bgm.style),
"duration": bgm.duration,
"artist": bgm.artist,
"description": bgm.description,
"tags": bgm.tags,
"audio_url": bgm.audio_url,
}
for bgm in paged
],
}
# ── 保存为模板 ────────────────────────────────────────────────────────────────
+214
View File
@@ -507,3 +507,217 @@ class TestDeletePlan:
resp = c.delete("/api/v1/edit-plans/nonexistent")
assert resp.status_code == 404
assert "剪辑计划不存在" in resp.json()["detail"]
# ---------------------------------------------------------------------------
# BGM 配置测试
# ---------------------------------------------------------------------------
class TestBGMConfig:
"""BGM 配置 API 测试"""
def test_get_bgm_default_empty(self, client):
"""新计划 BGM 默认为空"""
c, repo = client
plan = EditPlan.create("tpl-001", "测试")
repo.create(plan)
resp = c.get(f"/api/v1/edit-plans/{plan.id}/bgm")
assert resp.status_code == 200
data = resp.json()
assert data["plan_id"] == plan.id
assert data["bgm"] == {}
def test_update_bgm_volume(self, client):
"""更新 BGM 音量"""
c, repo = client
plan = EditPlan.create("tpl-001", "测试")
repo.create(plan)
resp = c.put(
f"/api/v1/edit-plans/{plan.id}/bgm",
json={"volume": 0.5, "fade_in": 2.0, "fade_out": 3.0},
)
assert resp.status_code == 200
data = resp.json()
assert data["bgm"]["volume"] == 0.5
assert data["bgm"]["fade_in"] == 2.0
assert data["bgm"]["fade_out"] == 3.0
def test_enable_bgm_with_preset(self, client):
"""启用 BGM 并指定 preset_id"""
c, repo = client
plan = EditPlan.create("tpl-001", "测试")
repo.create(plan)
resp = c.put(
f"/api/v1/edit-plans/{plan.id}/bgm",
json={
"enabled": True,
"source": "library",
"preset_id": "bgm_upbeat_001",
"volume": 0.3,
},
)
assert resp.status_code == 200
data = resp.json()
assert data["bgm"]["enabled"] is True
assert data["bgm"]["preset_id"] == "bgm_upbeat_001"
def test_enable_bgm_without_source_returns_400(self, client):
"""启用 BGM 但不指定来源,返回 400"""
c, repo = client
plan = EditPlan.create("tpl-001", "测试")
repo.create(plan)
resp = c.put(
f"/api/v1/edit-plans/{plan.id}/bgm",
json={"enabled": True, "volume": 0.3},
)
assert resp.status_code == 400
assert "素材来源" in resp.json()["detail"]
def test_enable_bgm_with_asset_id(self, client):
"""启用 BGM 并指定 asset_id"""
c, repo = client
plan = EditPlan.create("tpl-001", "测试")
repo.create(plan)
resp = c.put(
f"/api/v1/edit-plans/{plan.id}/bgm",
json={
"enabled": True,
"source": "upload",
"asset_id": "asset-audio-001",
"loop_enabled": True,
},
)
assert resp.status_code == 200
data = resp.json()
assert data["bgm"]["enabled"] is True
assert data["bgm"]["asset_id"] == "asset-audio-001"
assert data["bgm"]["loop_enabled"] is True
def test_update_bgm_not_found(self, client):
"""不存在的计划返回 404"""
c, _ = client
resp = c.put(
"/api/v1/edit-plans/nonexistent/bgm",
json={"volume": 0.5},
)
assert resp.status_code == 404
def test_get_bgm_not_found(self, client):
"""不存在的计划返回 404"""
c, _ = client
resp = c.get("/api/v1/edit-plans/nonexistent/bgm")
assert resp.status_code == 404
def test_partial_update_preserves_existing(self, client):
"""部分更新保留原有配置"""
c, repo = client
plan = EditPlan.create("tpl-001", "测试")
plan.config = {"bgm": {"volume": 0.5, "fade_in": 1.0}}
repo.create(plan)
# 只改音量
resp = c.put(
f"/api/v1/edit-plans/{plan.id}/bgm",
json={"volume": 0.8},
)
assert resp.status_code == 200
data = resp.json()
assert data["bgm"]["volume"] == 0.8
assert data["bgm"]["fade_in"] == 1.0 # 保留
def test_sidechain_config(self, client):
"""人声闪避配置更新"""
c, repo = client
plan = EditPlan.create("tpl-001", "测试")
repo.create(plan)
resp = c.put(
f"/api/v1/edit-plans/{plan.id}/bgm",
json={
"enabled": True,
"preset_id": "bgm_relax_001",
"sidechain_enabled": True,
"sidechain_ratio": 0.4,
},
)
assert resp.status_code == 200
data = resp.json()
assert data["bgm"]["sidechain_enabled"] is True
assert data["bgm"]["sidechain_ratio"] == 0.4
# ---------------------------------------------------------------------------
# BGM 预设库测试
# ---------------------------------------------------------------------------
class TestBGMPresets:
"""BGM 预设列表 API 测试"""
def test_list_all_presets(self, client):
"""获取所有预设 BGM"""
c, _ = client
resp = c.get("/api/v1/edit-plans/bgm/presets")
assert resp.status_code == 200
data = resp.json()
assert "items" in data
assert "total" in data
assert "styles" in data
assert data["total"] >= 10 # 至少有 10 首预设
assert len(data["items"]) == data["total"]
def test_filter_by_style(self, client):
"""按风格筛选"""
c, _ = client
resp = c.get("/api/v1/edit-plans/bgm/presets?style=upbeat")
assert resp.status_code == 200
data = resp.json()
assert data["total"] >= 3
for item in data["items"]:
assert item["style"] == "upbeat"
def test_search_by_keyword(self, client):
"""关键词搜索"""
c, _ = client
resp = c.get("/api/v1/edit-plans/bgm/presets?keyword=钢琴")
assert resp.status_code == 200
data = resp.json()
assert data["total"] >= 1
for item in data["items"]:
has_piano = (
"钢琴" in item["name"] or "钢琴" in item["description"] or any("钢琴" in tag for tag in item["tags"])
)
assert has_piano
def test_pagination(self, client):
"""分页功能"""
c, _ = client
resp = c.get("/api/v1/edit-plans/bgm/presets?skip=0&limit=3")
assert resp.status_code == 200
data = resp.json()
assert len(data["items"]) == 3
assert data["skip"] == 0
assert data["limit"] == 3
def test_preset_structure(self, client):
"""预设条目字段完整"""
c, _ = client
resp = c.get("/api/v1/edit-plans/bgm/presets?limit=1")
assert resp.status_code == 200
item = resp.json()["items"][0]
assert "id" in item
assert "name" in item
assert "style" in item
assert "style_label" in item
assert "duration" in item
assert "artist" in item
assert "description" in item
assert "tags" in item
assert isinstance(item["tags"], list)