Compare commits
3 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| b23bd2a942 | |||
| d6dfb48687 | |||
| 1437120634 |
@@ -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,
|
||||
)
|
||||
|
||||
Regular → Executable
+22
@@ -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
|
||||
"""文件大小(字节)"""
|
||||
|
||||
@@ -642,4 +642,3 @@
|
||||
font-size: 36px;
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Executable
+282
@@ -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()
|
||||
Reference in New Issue
Block a user