feat(#1196): /voices?type=clone从克隆音色档案读取(智能剪辑配音可选克隆音色) #1202
Regular → Executable
+46
-7
@@ -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;
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Executable
+266
@@ -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
|
||||
Reference in New Issue
Block a user