From 1437120634ac85586f106b7b80b6f185a9f2c27e Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Thu, 30 Jul 2026 13:56:42 +0800 Subject: [PATCH 1/3] =?UTF-8?q?feat(#1195):=20=E6=96=B0=E5=A2=9E=E5=85=8B?= =?UTF-8?q?=E9=9A=86=E9=9F=B3=E8=89=B2=E8=AF=95=E5=90=AC=E6=8E=A5=E5=8F=A3?= =?UTF-8?q?=20/voice-clones/{id}/preview?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 新增 GET /voice-clones/{clone_id}/preview 端点 - 支持默认试听文本（缓存7天）和自定义文本 - 状态校验：仅 ready 状态可试听 - 10 个单元测试覆盖成功/失败/缓存/权限等场景 --- apps/api/app/api/routes/voice_clones.py | 85 +++++++- apps/api/app/schemas/voice_clone.py | 22 ++ tests/unit/test_voice_clone_preview.py | 271 ++++++++++++++++++++++++ 3 files changed, 377 insertions(+), 1 deletion(-) mode change 100644 => 100755 apps/api/app/schemas/voice_clone.py create mode 100755 tests/unit/test_voice_clone_preview.py diff --git a/apps/api/app/api/routes/voice_clones.py b/apps/api/app/api/routes/voice_clones.py index 08a201718..cc513b615 100755 --- a/apps/api/app/api/routes/voice_clones.py +++ b/apps/api/app/api/routes/voice_clones.py @@ -11,6 +11,7 @@ from app.dependencies import get_cosyvoice_service, get_voice_clone_profile_repo from app.schemas.voice_clone import ( CreateVoiceCloneRequest, ListVoiceCloneResponse, + VoiceClonePreviewResponse, VoiceCloneProfileResponse, VoiceCloneStatusResponse, ) @@ -19,7 +20,7 @@ from fastapi import APIRouter, Depends, HTTPException, Query, Response, status from packages.adapters.sqlalchemy_impl.voice_clone_profile_repository import ( SQLAlchemyVoiceCloneProfileRepository, ) -from packages.application.cosyvoice_service import CosyVoiceService +from packages.application.cosyvoice_service import CosyVoiceError, CosyVoiceService from packages.application.voice_clone.use_cases import ( DeleteVoiceCloneUseCase, GetVoiceCloneStatusUseCase, @@ -36,6 +37,13 @@ logger = logging.getLogger(__name__) router = APIRouter() +# 克隆音色试听缓存（减少重复TTS调用） +# key: clone_id, value: (audio_url, duration, file_size, text, timestamp) +_clone_preview_cache: dict[str, tuple[str, float, int, str, float]] = {} +CLONE_PREVIEW_CACHE_TTL = 7 * 24 * 3600 # 7天TTL +# 默认试听文本 +CLONE_PREVIEW_TEMPLATE = "你好，这是我的克隆音色，很高兴能为你配音。" + def _to_response(profile) -> VoiceCloneProfileResponse: # source_audio_url 是用户传入的原始 URL（可能是外部地址），不做预签名转换 @@ -223,3 +231,78 @@ def retry_voice_clone( logger.error(f"Failed to mark profile as failed after dispatch error: {inner_e}") return _to_response(profile) + + +@router.get("/{clone_id}/preview", response_model=VoiceClonePreviewResponse) +def get_voice_clone_preview( + clone_id: str, + text: str = Query("", description="自定义试听文本，为空则使用默认示例"), + authenticated_user: AuthenticatedUser = Depends(get_current_user), + repository: SQLAlchemyVoiceCloneProfileRepository = Depends(get_voice_clone_profile_repository), + cosyvoice: CosyVoiceService = Depends(get_cosyvoice_service), +) -> VoiceClonePreviewResponse: + """获取克隆音色试听音频（实时 TTS 合成）。 + + - 克隆音色必须处于 ready 状态 + - 使用默认试听文本时，结果缓存 7 天 + - 可传入自定义 text 参数试听不同文本 + """ + import time + + use_case = GetVoiceCloneUseCase(repository) + try: + profile = use_case.execute(clone_id, authenticated_user.user.id) + except VoiceCloneNotFoundError as _e: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Voice clone not found") from _e + + if not profile.is_ready: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail=f"Voice clone is not ready (current status: {profile.status})", + ) + + # 有自定义文本时不缓存 + use_cache = not text.strip() + + if use_cache and clone_id in _clone_preview_cache: + audio_url, duration, file_size, cached_text, cached_at = _clone_preview_cache[clone_id] + if time.time() - cached_at < CLONE_PREVIEW_CACHE_TTL: + return VoiceClonePreviewResponse( + clone_id=clone_id, + voice_id=profile.voice_id, + audio_url=audio_url, + text=cached_text, + duration=duration, + file_size=file_size, + ) + + # 合成试听音频 + preview_text = text.strip() or CLONE_PREVIEW_TEMPLATE + try: + result = cosyvoice.synthesize_speech( + text=preview_text, + voice_id=profile.voice_id, + format="mp3", + speed=1.0, + ) + except CosyVoiceError as e: + raise HTTPException(status_code=502, detail=f"TTS 合成失败: {e}") from e + + # 缓存（仅默认试听文本） + if use_cache: + _clone_preview_cache[clone_id] = ( + result.audio_url, + result.duration, + result.file_size, + preview_text, + time.time(), + ) + + return VoiceClonePreviewResponse( + clone_id=clone_id, + voice_id=profile.voice_id, + audio_url=result.audio_url, + text=preview_text, + duration=result.duration, + file_size=result.file_size, + ) diff --git a/apps/api/app/schemas/voice_clone.py b/apps/api/app/schemas/voice_clone.py old mode 100644 new mode 100755 index 1ae0fc40c..b13517366 --- a/apps/api/app/schemas/voice_clone.py +++ b/apps/api/app/schemas/voice_clone.py @@ -63,3 +63,25 @@ class ListVoiceCloneResponse(BaseModel): items: List[VoiceCloneProfileResponse] total: int + + +class VoiceClonePreviewResponse(BaseModel): + """克隆音色试听响应。""" + + clone_id: str + """音色克隆档案 ID""" + + voice_id: str + """CosyVoice 音色 ID""" + + audio_url: str + """试听音频 URL""" + + text: str + """试听文本""" + + duration: float = 0.0 + """音频时长（秒）""" + + file_size: int = 0 + """文件大小（字节）""" diff --git a/tests/unit/test_voice_clone_preview.py b/tests/unit/test_voice_clone_preview.py new file mode 100755 index 000000000..3bf4ac00b --- /dev/null +++ b/tests/unit/test_voice_clone_preview.py @@ -0,0 +1,271 @@ +"""音色克隆试听接口单元测试 (#1195). + +直接测试路由函数逻辑，mock 所有依赖。 +""" + +from __future__ import annotations + +import time +from datetime import datetime, timezone +from unittest.mock import MagicMock + +import pytest +from fastapi import HTTPException + +from packages.application.cosyvoice_service import CosyVoiceError, SynthesizeResult +from packages.domain.voice_clone_profile import VoiceCloneProfile, VoiceCloneStatus + + +def _make_profile(**kwargs) -> VoiceCloneProfile: + defaults = { + "id": "clone_001", + "user_id": "user_001", + "name": "测试音色", + "description": "", + "source_audio_url": "", + "voice_id": "clone_voice_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.now(timezone.utc), + "updated_at": datetime.now(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 TestVoiceClonePreview: + """克隆音色试听接口测试。""" + + def _call_preview(self, profile, cosyvoice_mock, text="", user_id="user_001"): + """调用路由函数，模拟 FastAPI 注入依赖。""" + from app.api.routes.voice_clones import get_voice_clone_preview + + repo = MagicMock() + if profile is None: + from packages.application.voice_clone.use_cases import VoiceCloneNotFoundError + repo.get.return_value = None + use_case_instance = MagicMock() + use_case_instance.execute.side_effect = VoiceCloneNotFoundError("not found") + else: + repo.get.return_value = profile + use_case_instance = MagicMock() + use_case_instance.execute.return_value = profile + + # 替换 use case 构造 + original_init = None + from packages.application.voice_clone.use_cases import GetVoiceCloneUseCase + + original_init = GetVoiceCloneUseCase.__init__ + + def mock_init(self, repository): + self.repository = repository + + def mock_execute(self, clone_id, uid): + if profile is None: + from packages.application.voice_clone.use_cases import VoiceCloneNotFoundError + raise VoiceCloneNotFoundError("not found") + if profile.user_id != uid: + from packages.application.voice_clone.use_cases import VoiceCloneNotFoundError + raise VoiceCloneNotFoundError("not found") + return profile + + GetVoiceCloneUseCase.__init__ = mock_init + GetVoiceCloneUseCase.execute = mock_execute + + try: + result = get_voice_clone_preview( + clone_id=profile.id if profile else "nonexistent", + text=text, + authenticated_user=_make_auth_user(user_id), + repository=repo, + cosyvoice=cosyvoice_mock, + ) + return result + finally: + GetVoiceCloneUseCase.__init__ = original_init + if hasattr(GetVoiceCloneUseCase, 'execute'): + delattr(GetVoiceCloneUseCase, 'execute') + + def test_preview_success_ready_clone(self) -> None: + """就绪的克隆音色可以正常试听。""" + from app.api.routes.voice_clones import _clone_preview_cache + _clone_preview_cache.clear() + + profile = _make_profile() + cosyvoice = MagicMock() + cosyvoice.synthesize_speech.return_value = SynthesizeResult( + audio_url="https://oss.example.com/preview/clone_001.mp3", + duration=3.5, + file_size=56000, + request_id="req_001", + ) + + result = self._call_preview(profile, cosyvoice) + + assert result.clone_id == "clone_001" + assert result.voice_id == "clone_voice_001" + assert result.audio_url == "https://oss.example.com/preview/clone_001.mp3" + assert result.duration == 3.5 + assert result.file_size == 56000 + assert "克隆音色" in result.text + cosyvoice.synthesize_speech.assert_called_once() + call_kwargs = cosyvoice.synthesize_speech.call_args + assert call_kwargs.kwargs["voice_id"] == "clone_voice_001" + assert call_kwargs.kwargs["format"] == "mp3" + + def test_preview_custom_text(self) -> None: + """自定义试听文本。""" + from app.api.routes.voice_clones import _clone_preview_cache + _clone_preview_cache.clear() + + profile = _make_profile() + cosyvoice = MagicMock() + cosyvoice.synthesize_speech.return_value = SynthesizeResult( + audio_url="https://oss.example.com/preview/custom.mp3", + duration=2.0, + file_size=32000, + ) + + custom_text = "大家好，这是自定义试听文本。" + result = self._call_preview(profile, cosyvoice, text=custom_text) + + assert result.text == custom_text + call_kwargs = cosyvoice.synthesize_speech.call_args + assert call_kwargs.kwargs["text"] == custom_text + + def test_preview_not_found(self) -> None: + """克隆音色不存在时返回 404。""" + cosyvoice = MagicMock() + + with pytest.raises(HTTPException) as exc_info: + self._call_preview(None, cosyvoice) + + assert exc_info.value.status_code == 404 + cosyvoice.synthesize_speech.assert_not_called() + + def test_preview_not_ready_pending(self) -> None: + """pending 状态的克隆音色不能试听。""" + from app.api.routes.voice_clones import _clone_preview_cache + _clone_preview_cache.clear() + + profile = _make_profile(status=VoiceCloneStatus.PENDING, voice_id="") + cosyvoice = MagicMock() + + with pytest.raises(HTTPException) as exc_info: + self._call_preview(profile, cosyvoice) + + assert exc_info.value.status_code == 400 + assert "not ready" in exc_info.value.detail.lower() or "未就绪" in exc_info.value.detail + cosyvoice.synthesize_speech.assert_not_called() + + def test_preview_not_ready_processing(self) -> None: + """processing 状态的克隆音色不能试听。""" + from app.api.routes.voice_clones import _clone_preview_cache + _clone_preview_cache.clear() + + profile = _make_profile(status=VoiceCloneStatus.PROCESSING) + cosyvoice = MagicMock() + + with pytest.raises(HTTPException) as exc_info: + self._call_preview(profile, cosyvoice) + + assert exc_info.value.status_code == 400 + cosyvoice.synthesize_speech.assert_not_called() + + def test_preview_not_ready_failed(self) -> None: + """failed 状态的克隆音色不能试听。""" + from app.api.routes.voice_clones import _clone_preview_cache + _clone_preview_cache.clear() + + profile = _make_profile(status=VoiceCloneStatus.FAILED, error_message="超时") + cosyvoice = MagicMock() + + with pytest.raises(HTTPException) as exc_info: + self._call_preview(profile, cosyvoice) + + assert exc_info.value.status_code == 400 + cosyvoice.synthesize_speech.assert_not_called() + + def test_preview_tts_failure(self) -> None: + """TTS 合成失败返回 502。""" + from app.api.routes.voice_clones import _clone_preview_cache + _clone_preview_cache.clear() + + profile = _make_profile() + cosyvoice = MagicMock() + cosyvoice.synthesize_speech.side_effect = CosyVoiceError("API 调用失败") + + with pytest.raises(HTTPException) as exc_info: + self._call_preview(profile, cosyvoice) + + assert exc_info.value.status_code == 502 + assert "TTS" in exc_info.value.detail + + def test_preview_cache_default_text(self) -> None: + """默认试听文本使用缓存。""" + from app.api.routes.voice_clones import _clone_preview_cache, CLONE_PREVIEW_CACHE_TTL + _clone_preview_cache.clear() + + profile = _make_profile() + cosyvoice = MagicMock() + cosyvoice.synthesize_speech.return_value = SynthesizeResult( + audio_url="https://oss.example.com/preview/cached.mp3", + duration=3.0, + file_size=48000, + ) + + # 第一次调用：合成 + result1 = self._call_preview(profile, cosyvoice) + assert cosyvoice.synthesize_speech.call_count == 1 + + # 第二次调用：走缓存，不重复合成 + result2 = self._call_preview(profile, cosyvoice) + assert cosyvoice.synthesize_speech.call_count == 1 + assert result2.audio_url == result1.audio_url + assert result2.duration == result1.duration + + def test_preview_custom_text_no_cache(self) -> None: + """自定义文本不使用缓存。""" + from app.api.routes.voice_clones import _clone_preview_cache + _clone_preview_cache.clear() + + profile = _make_profile() + cosyvoice = MagicMock() + cosyvoice.synthesize_speech.side_effect = [ + SynthesizeResult(audio_url="https://example.com/1.mp3", duration=2.0, file_size=32000), + SynthesizeResult(audio_url="https://example.com/2.mp3", duration=2.5, file_size=40000), + ] + + result1 = self._call_preview(profile, cosyvoice, text="自定义文本一") + result2 = self._call_preview(profile, cosyvoice, text="自定义文本二") + + assert cosyvoice.synthesize_speech.call_count == 2 + assert result1.audio_url != result2.audio_url + + def test_preview_wrong_user(self) -> None: + """非所有者不能访问他人克隆音色的试听。""" + from app.api.routes.voice_clones import _clone_preview_cache + _clone_preview_cache.clear() + + profile = _make_profile(user_id="user_001") + cosyvoice = MagicMock() + + with pytest.raises(HTTPException) as exc_info: + self._call_preview(profile, cosyvoice, user_id="user_002") + + assert exc_info.value.status_code == 404 + cosyvoice.synthesize_speech.assert_not_called() -- 2.54.0 From d6dfb48687b37a1d3ab091f1eb39c37b7d700c09 Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Thu, 30 Jul 2026 14:29:29 +0800 Subject: [PATCH 2/3] =?UTF-8?q?fix(test):=20=E4=BF=AE=E5=A4=8Dvoice=5Fclon?= =?UTF-8?q?e=5Fpreview=E6=B5=8B=E8=AF=95=E6=B1=A1=E6=9F=93GetVoiceCloneUse?= =?UTF-8?q?Case=E7=B1=BB=E7=9A=84=E9=97=AE=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 测试中用delattr删除了类原生execute方法，导致后续测试文件 test_voice_clone_use_cases.py中GetVoiceCloneUseCase.execute丢失。 改为保存原始方法并在finally中恢复。 --- tests/unit/test_voice_clone_preview.py | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/tests/unit/test_voice_clone_preview.py b/tests/unit/test_voice_clone_preview.py index 3bf4ac00b..6f8605dfd 100755 --- a/tests/unit/test_voice_clone_preview.py +++ b/tests/unit/test_voice_clone_preview.py @@ -66,10 +66,10 @@ class TestVoiceClonePreview: use_case_instance.execute.return_value = profile # 替换 use case 构造 - original_init = None from packages.application.voice_clone.use_cases import GetVoiceCloneUseCase original_init = GetVoiceCloneUseCase.__init__ + original_execute = GetVoiceCloneUseCase.execute def mock_init(self, repository): self.repository = repository @@ -97,8 +97,7 @@ class TestVoiceClonePreview: return result finally: GetVoiceCloneUseCase.__init__ = original_init - if hasattr(GetVoiceCloneUseCase, 'execute'): - delattr(GetVoiceCloneUseCase, 'execute') + GetVoiceCloneUseCase.execute = original_execute def test_preview_success_ready_clone(self) -> None: """就绪的克隆音色可以正常试听。""" -- 2.54.0 From b23bd2a94224514da59c7521e89ec1c7c4dc763d Mon Sep 17 00:00:00 2001 From: CI Bot Date: Thu, 30 Jul 2026 06:33:54 +0000 Subject: [PATCH 3/3] style: auto-format with black + isort + prettier [skip ci-format-check] --- .../components/voice/CloneModal/clone-modal.css | 1 - tests/unit/test_voice_clone_preview.py | 14 +++++++++++++- 2 files changed, 13 insertions(+), 2 deletions(-) diff --git a/apps/web/src/components/voice/CloneModal/clone-modal.css b/apps/web/src/components/voice/CloneModal/clone-modal.css index 540853183..35952976d 100644 --- a/apps/web/src/components/voice/CloneModal/clone-modal.css +++ b/apps/web/src/components/voice/CloneModal/clone-modal.css @@ -642,4 +642,3 @@ font-size: 36px; } } - diff --git a/tests/unit/test_voice_clone_preview.py b/tests/unit/test_voice_clone_preview.py index 6f8605dfd..a47694190 100755 --- a/tests/unit/test_voice_clone_preview.py +++ b/tests/unit/test_voice_clone_preview.py @@ -57,6 +57,7 @@ class TestVoiceClonePreview: repo = MagicMock() if profile is None: from packages.application.voice_clone.use_cases import VoiceCloneNotFoundError + repo.get.return_value = None use_case_instance = MagicMock() use_case_instance.execute.side_effect = VoiceCloneNotFoundError("not found") @@ -77,9 +78,11 @@ class TestVoiceClonePreview: def mock_execute(self, clone_id, uid): if profile is None: from packages.application.voice_clone.use_cases import VoiceCloneNotFoundError + raise VoiceCloneNotFoundError("not found") if profile.user_id != uid: from packages.application.voice_clone.use_cases import VoiceCloneNotFoundError + raise VoiceCloneNotFoundError("not found") return profile @@ -102,6 +105,7 @@ class TestVoiceClonePreview: def test_preview_success_ready_clone(self) -> None: """就绪的克隆音色可以正常试听。""" from app.api.routes.voice_clones import _clone_preview_cache + _clone_preview_cache.clear() profile = _make_profile() @@ -129,6 +133,7 @@ class TestVoiceClonePreview: def test_preview_custom_text(self) -> None: """自定义试听文本。""" from app.api.routes.voice_clones import _clone_preview_cache + _clone_preview_cache.clear() profile = _make_profile() @@ -159,6 +164,7 @@ class TestVoiceClonePreview: def test_preview_not_ready_pending(self) -> None: """pending 状态的克隆音色不能试听。""" from app.api.routes.voice_clones import _clone_preview_cache + _clone_preview_cache.clear() profile = _make_profile(status=VoiceCloneStatus.PENDING, voice_id="") @@ -174,6 +180,7 @@ class TestVoiceClonePreview: def test_preview_not_ready_processing(self) -> None: """processing 状态的克隆音色不能试听。""" from app.api.routes.voice_clones import _clone_preview_cache + _clone_preview_cache.clear() profile = _make_profile(status=VoiceCloneStatus.PROCESSING) @@ -188,6 +195,7 @@ class TestVoiceClonePreview: def test_preview_not_ready_failed(self) -> None: """failed 状态的克隆音色不能试听。""" from app.api.routes.voice_clones import _clone_preview_cache + _clone_preview_cache.clear() profile = _make_profile(status=VoiceCloneStatus.FAILED, error_message="超时") @@ -202,6 +210,7 @@ class TestVoiceClonePreview: def test_preview_tts_failure(self) -> None: """TTS 合成失败返回 502。""" from app.api.routes.voice_clones import _clone_preview_cache + _clone_preview_cache.clear() profile = _make_profile() @@ -216,7 +225,8 @@ class TestVoiceClonePreview: def test_preview_cache_default_text(self) -> None: """默认试听文本使用缓存。""" - from app.api.routes.voice_clones import _clone_preview_cache, CLONE_PREVIEW_CACHE_TTL + from app.api.routes.voice_clones import CLONE_PREVIEW_CACHE_TTL, _clone_preview_cache + _clone_preview_cache.clear() profile = _make_profile() @@ -240,6 +250,7 @@ class TestVoiceClonePreview: def test_preview_custom_text_no_cache(self) -> None: """自定义文本不使用缓存。""" from app.api.routes.voice_clones import _clone_preview_cache + _clone_preview_cache.clear() profile = _make_profile() @@ -258,6 +269,7 @@ class TestVoiceClonePreview: def test_preview_wrong_user(self) -> None: """非所有者不能访问他人克隆音色的试听。""" from app.api.routes.voice_clones import _clone_preview_cache + _clone_preview_cache.clear() profile = _make_profile(user_id="user_001") -- 2.54.0