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/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 new file mode 100755 index 000000000..a47694190 --- /dev/null +++ b/tests/unit/test_voice_clone_preview.py @@ -0,0 +1,282 @@ +"""音色克隆试听接口单元测试 (#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 构造 + 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 + + 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 + GetVoiceCloneUseCase.execute = original_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_TTL, _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/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()