feat(#1196): /voices?type=clone从克隆音色档案读取(智能剪辑配音可选克隆音色) #1202

Merged
auto-approve-bot merged 3 commits from feat/voices-clone-from-profile-1196 into develop 2026-07-30 14:46:12 +08:00
3 changed files with 312 additions and 8 deletions
+46 -7
View File
@@ -134,6 +134,35 @@ def _preset_to_unified_response(preset) -> UnifiedVoiceItemResponse:
)
def _clone_profile_to_unified_response(profile) -> UnifiedVoiceItemResponse:
"""将克隆音色档案转换为统一响应格式。
注意:克隆音色是「音色模型」(可用于 TTS 合成任意文本),
不同于配音库条目(具体的配音作品)。
"""
return UnifiedVoiceItemResponse(
id=profile.id,
type="clone",
name=profile.name,
description=profile.description or "",
gender=profile.gender or "unknown",
language=profile.language or "zh-CN",
voice_id=profile.voice_id or "",
voice_provider=profile.voice_model or "cosyvoice",
audio_url="", # 克隆音色没有预合成音频,需通过 /voice-clones/{id}/preview 试听
preview_url="", # 试听需实时合成,前端调用 preview 接口
duration=0,
file_size=0,
status=profile.status.value if hasattr(profile.status, "value") else str(profile.status),
tags=[],
user_id=profile.user_id,
project_id=None,
voice_clone_profile_id=profile.id,
created_at=profile.created_at,
updated_at=profile.updated_at,
)
# ==================== 统一配音列表(预置 + 克隆)====================
@@ -170,14 +199,24 @@ def list_voices_unified(
preset_items = [_preset_to_unified_response(p) for p in PRESET_VOICES]
preset_count = len(preset_items)
# 获取克隆音色
# 获取克隆音色(从 voice_clone_profile 读取,ready 状态的克隆音色)
if has_clone:
use_case = ListVoiceLibraryUseCase(voice_repository)
clone_items_raw, clone_count = use_case.execute(user_id, status=status_filter, skip=skip, limit=limit)
# 批量查询 voice_id → profile_id 映射,填充 voice_clone_profile_id
voice_ids = [i.voice_id for i in clone_items_raw if i.voice_id]
profile_id_map = clone_profile_repository.find_profile_ids_by_voice_ids(voice_ids) if voice_ids else {}
clone_items = [_to_unified_response(i, profile_id_map, sign_url) for i in clone_items_raw]
# status_filter 映射:不传则默认只返回 ready 状态(可用的克隆音色)
# 前端可以传 status=all 获取所有状态,或传具体状态过滤
filter_status = None
if status_filter and status_filter != "all":
filter_status = status_filter
elif not status_filter:
filter_status = "ready"
clone_profiles = clone_profile_repository.list_by_user(
user_id,
status=filter_status,
limit=limit,
offset=skip,
)
clone_count = clone_profile_repository.count_by_user(user_id, status=filter_status)
clone_items = [_clone_profile_to_unified_response(p) for p in clone_profiles]
# 组装结果
if type == "preset":
@@ -642,4 +642,3 @@
font-size: 36px;
}
}
+266
View File
@@ -0,0 +1,266 @@
"""统一音色列表接口单元测试 (#1196).
验证 /voices?type=clone 从 voice_clone_profile 读取克隆音色。
"""
from __future__ import annotations
from datetime import datetime, timezone
from unittest.mock import MagicMock
import pytest
from packages.domain.voice_clone_profile import VoiceCloneProfile, VoiceCloneStatus
def _make_clone_profile(**kwargs) -> VoiceCloneProfile:
defaults = {
"id": "clone_001",
"user_id": "user_001",
"name": "我的克隆音色",
"description": "测试用克隆音色",
"source_audio_url": "https://example.com/source.mp3",
"voice_id": "cosyvoice_clone_001",
"voice_model": "cosyvoice-v3",
"language": "zh-CN",
"gender": "female",
"status": VoiceCloneStatus.READY,
"error_message": "",
"retry_count": 0,
"max_retries": 3,
"metadata": {},
"created_at": datetime(2026, 7, 30, 10, 0, 0, tzinfo=timezone.utc),
"updated_at": datetime(2026, 7, 30, 10, 5, 0, tzinfo=timezone.utc),
}
defaults.update(kwargs)
return VoiceCloneProfile(**defaults)
def _make_auth_user(user_id: str = "user_001") -> MagicMock:
user = MagicMock()
user.id = user_id
auth = MagicMock()
auth.user = user
return auth
class TestVoicesCloneFromProfile:
"""测试 /voices?type=clone 从克隆音色档案读取。"""
def _call_list_voices(
self,
type_filter=None,
status_filter=None,
skip=0,
limit=50,
clone_profiles=None,
clone_count=0,
user_id="user_001",
):
"""调用 list_voices_unified 路由函数。"""
from app.api.routes.voices import list_voices_unified
# mock 依赖
voice_repo = MagicMock()
clone_profile_repo = MagicMock()
clone_profile_repo.list_by_user.return_value = clone_profiles or []
clone_profile_repo.count_by_user.return_value = clone_count
sign_url = MagicMock(side_effect=lambda u: u + "?signed=1")
result = list_voices_unified(
type=type_filter,
status_filter=status_filter,
skip=skip,
limit=limit,
authenticated_user=_make_auth_user(user_id),
voice_repository=voice_repo,
clone_profile_repository=clone_profile_repo,
sign_url=sign_url,
)
return result, clone_profile_repo, voice_repo
def test_type_clone_returns_ready_clones(self) -> None:
"""type=clone 返回 ready 状态的克隆音色。"""
profiles = [
_make_clone_profile(id="c1", name="音色1", voice_id="v1"),
_make_clone_profile(id="c2", name="音色2", voice_id="v2"),
]
result, clone_repo, voice_repo = self._call_list_voices(
type_filter="clone",
clone_profiles=profiles,
clone_count=2,
)
assert result.total == 2
assert result.clone_count == 2
assert result.preset_count == 0
assert len(result.items) == 2
# 验证字段
item = result.items[0]
assert item.id == "c1"
assert item.type == "clone"
assert item.name == "音色1"
assert item.voice_id == "v1"
assert item.voice_clone_profile_id == "c1"
assert item.status == "ready"
assert item.gender == "female"
assert item.language == "zh-CN"
assert item.user_id == "user_001"
# 调用了 clone_profile_repository
clone_repo.list_by_user.assert_called_once()
call_kwargs = clone_repo.list_by_user.call_args
assert call_kwargs.kwargs.get("status") == "ready"
# 没有调用 voice_repository(克隆音色不从配音库取)
voice_repo.execute.assert_not_called()
def test_type_clone_default_status_ready(self) -> None:
"""不传 status 时默认只返回 ready 状态。"""
result, clone_repo, _ = self._call_list_voices(
type_filter="clone",
clone_profiles=[],
clone_count=0,
)
clone_repo.list_by_user.assert_called_once()
call_kwargs = clone_repo.list_by_user.call_args
assert call_kwargs.kwargs.get("status") == "ready"
def test_type_clone_status_filter(self) -> None:
"""status 参数可以过滤特定状态。"""
result, clone_repo, _ = self._call_list_voices(
type_filter="clone",
status_filter="processing",
clone_profiles=[],
clone_count=0,
)
clone_repo.list_by_user.assert_called_once()
call_kwargs = clone_repo.list_by_user.call_args
assert call_kwargs.kwargs.get("status") == "processing"
def test_type_clone_status_all(self) -> None:
"""status=all 返回所有状态的克隆音色。"""
result, clone_repo, _ = self._call_list_voices(
type_filter="clone",
status_filter="all",
clone_profiles=[],
clone_count=0,
)
clone_repo.list_by_user.assert_called_once()
call_kwargs = clone_repo.list_by_user.call_args
assert call_kwargs.kwargs.get("status") is None
def test_type_preset_no_clone_data(self) -> None:
"""type=preset 时不查询克隆音色。"""
result, clone_repo, _ = self._call_list_voices(
type_filter="preset",
clone_profiles=[],
clone_count=0,
)
assert result.preset_count > 0
assert result.clone_count == 0
clone_repo.list_by_user.assert_not_called()
def test_type_none_returns_both(self) -> None:
"""不传 type 时返回预置 + 克隆。"""
profiles = [_make_clone_profile(id="c1", name="克隆音色")]
result, clone_repo, _ = self._call_list_voices(
type_filter=None,
clone_profiles=profiles,
clone_count=1,
)
assert result.preset_count > 0
assert result.clone_count == 1
assert result.total == result.preset_count + 1
clone_repo.list_by_user.assert_called_once()
def test_pagination(self) -> None:
"""分页参数正确传递。"""
result, clone_repo, _ = self._call_list_voices(
type_filter="clone",
skip=10,
limit=20,
clone_profiles=[],
clone_count=50,
)
clone_repo.list_by_user.assert_called_once()
call_kwargs = clone_repo.list_by_user.call_args
assert call_kwargs.kwargs.get("offset") == 10
assert call_kwargs.kwargs.get("limit") == 20
def test_empty_clone_list(self) -> None:
"""用户没有克隆音色时返回空列表。"""
result, clone_repo, _ = self._call_list_voices(
type_filter="clone",
clone_profiles=[],
clone_count=0,
)
assert result.total == 0
assert result.clone_count == 0
assert len(result.items) == 0
def test_clone_item_has_voice_id(self) -> None:
"""克隆音色条目包含 voice_id,前端可用于 TTS 合成。"""
profile = _make_clone_profile(voice_id="my_clone_voice_123")
result, _, _ = self._call_list_voices(
type_filter="clone",
clone_profiles=[profile],
clone_count=1,
)
assert result.items[0].voice_id == "my_clone_voice_123"
assert result.items[0].voice_provider == "cosyvoice-v3"
def test_clone_item_no_audio_url(self) -> None:
"""克隆音色没有预合成 audio_url(需通过 preview 接口试听)。"""
profile = _make_clone_profile()
result, _, _ = self._call_list_voices(
type_filter="clone",
clone_profiles=[profile],
clone_count=1,
)
item = result.items[0]
assert item.audio_url == ""
assert item.preview_url == ""
assert item.duration == 0
assert item.file_size == 0
def test_clone_item_has_profile_id(self) -> None:
"""克隆音色条目携带 voice_clone_profile_id,前端可用于调用 preview 接口。"""
profile = _make_clone_profile(id="profile_abc_123")
result, _, _ = self._call_list_voices(
type_filter="clone",
clone_profiles=[profile],
clone_count=1,
)
assert result.items[0].voice_clone_profile_id == "profile_abc_123"
def test_multiple_statuses(self) -> None:
"""不同状态的克隆音色返回正确的 status 字段。"""
for status in [
VoiceCloneStatus.PENDING,
VoiceCloneStatus.PROCESSING,
VoiceCloneStatus.READY,
VoiceCloneStatus.FAILED,
]:
profile = _make_clone_profile(id=f"c_{status.value}", status=status)
result, _, _ = self._call_list_voices(
type_filter="clone",
status_filter=status.value,
clone_profiles=[profile],
clone_count=1,
)
assert result.items[0].status == status.value