fix: /tts/preview 解析克隆音色 UUID 为 CosyVoice voice_id #1497
@@ -381,15 +381,34 @@ def preview_tts(
|
||||
request: TTSPreviewRequest,
|
||||
authenticated_user: AuthenticatedUser = Depends(get_current_user),
|
||||
cosyvoice_service: CosyVoiceService = Depends(get_cosyvoice_service),
|
||||
voice_clone_repo=Depends(get_voice_clone_profile_repository),
|
||||
) -> TTSPreviewResponse:
|
||||
"""TTS 预览(试听)——同步合成,立即返回音频 URL。
|
||||
|
||||
用于前端预览配音效果,限制文本长度 200 字以内。
|
||||
支持预设音色和克隆音色:克隆音色传的是 profile UUID,需解析为 CosyVoice voice_id。
|
||||
"""
|
||||
# 解析 voice_id:前端可能传 VoiceCloneProfile UUID 或预设音色 ID
|
||||
actual_voice_id = request.voice_id
|
||||
profile = voice_clone_repo.get(request.voice_id)
|
||||
if profile is not None:
|
||||
# 命中克隆音色 profile — 校验归属权限
|
||||
if profile.user_id != authenticated_user.user.id:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_403_FORBIDDEN,
|
||||
detail="无权访问该音色",
|
||||
)
|
||||
if not profile.voice_id:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail="音色克隆尚未完成,请稍后再试",
|
||||
)
|
||||
actual_voice_id = profile.voice_id
|
||||
|
||||
try:
|
||||
result = cosyvoice_service.synthesize_speech(
|
||||
text=request.text,
|
||||
voice_id=request.voice_id,
|
||||
voice_id=actual_voice_id,
|
||||
speed=request.speed,
|
||||
)
|
||||
except CosyVoiceError as e:
|
||||
|
||||
@@ -81,6 +81,12 @@ class TestTTSPreviewEndpoint:
|
||||
)
|
||||
app.dependency_overrides[get_cosyvoice_service] = lambda: mock_service
|
||||
|
||||
from app.dependencies import get_voice_clone_profile_repository
|
||||
|
||||
mock_clone_repo = MagicMock()
|
||||
mock_clone_repo.get.return_value = None # no profile found = preset voice
|
||||
app.dependency_overrides[get_voice_clone_profile_repository] = lambda: mock_clone_repo
|
||||
|
||||
client = self._make_client(app)
|
||||
resp = client.post(
|
||||
"/tts/preview",
|
||||
@@ -120,6 +126,12 @@ class TestTTSPreviewEndpoint:
|
||||
)
|
||||
app.dependency_overrides[get_cosyvoice_service] = lambda: mock_service
|
||||
|
||||
from app.dependencies import get_voice_clone_profile_repository
|
||||
|
||||
mock_clone_repo = MagicMock()
|
||||
mock_clone_repo.get.return_value = None # no profile found = preset voice
|
||||
app.dependency_overrides[get_voice_clone_profile_repository] = lambda: mock_clone_repo
|
||||
|
||||
client = self._make_client(app)
|
||||
resp = client.post(
|
||||
"/tts/preview",
|
||||
@@ -158,6 +170,12 @@ class TestTTSPreviewEndpoint:
|
||||
mock_service.synthesize_speech.side_effect = CosyVoiceError("API timeout")
|
||||
app.dependency_overrides[get_cosyvoice_service] = lambda: mock_service
|
||||
|
||||
from app.dependencies import get_voice_clone_profile_repository
|
||||
|
||||
mock_clone_repo = MagicMock()
|
||||
mock_clone_repo.get.return_value = None # no profile found = preset voice
|
||||
app.dependency_overrides[get_voice_clone_profile_repository] = lambda: mock_clone_repo
|
||||
|
||||
client = self._make_client(app)
|
||||
resp = client.post(
|
||||
"/tts/preview",
|
||||
@@ -186,6 +204,12 @@ class TestTTSPreviewEndpoint:
|
||||
mock_service.synthesize_speech.side_effect = ValueError("text 不能为空")
|
||||
app.dependency_overrides[get_cosyvoice_service] = lambda: mock_service
|
||||
|
||||
from app.dependencies import get_voice_clone_profile_repository
|
||||
|
||||
mock_clone_repo = MagicMock()
|
||||
mock_clone_repo.get.return_value = None # no profile found = preset voice
|
||||
app.dependency_overrides[get_voice_clone_profile_repository] = lambda: mock_clone_repo
|
||||
|
||||
client = self._make_client(app)
|
||||
resp = client.post(
|
||||
"/tts/preview",
|
||||
@@ -213,6 +237,12 @@ class TestTTSPreviewEndpoint:
|
||||
mock_service = MagicMock()
|
||||
app.dependency_overrides[get_cosyvoice_service] = lambda: mock_service
|
||||
|
||||
from app.dependencies import get_voice_clone_profile_repository
|
||||
|
||||
mock_clone_repo = MagicMock()
|
||||
mock_clone_repo.get.return_value = None # no profile found = preset voice
|
||||
app.dependency_overrides[get_voice_clone_profile_repository] = lambda: mock_clone_repo
|
||||
|
||||
client = self._make_client(app)
|
||||
resp = client.post(
|
||||
"/tts/preview",
|
||||
@@ -239,9 +269,173 @@ class TestTTSPreviewEndpoint:
|
||||
mock_service = MagicMock()
|
||||
app.dependency_overrides[get_cosyvoice_service] = lambda: mock_service
|
||||
|
||||
from app.dependencies import get_voice_clone_profile_repository
|
||||
|
||||
mock_clone_repo = MagicMock()
|
||||
mock_clone_repo.get.return_value = None # no profile found = preset voice
|
||||
app.dependency_overrides[get_voice_clone_profile_repository] = lambda: mock_clone_repo
|
||||
|
||||
client = self._make_client(app)
|
||||
resp = client.post(
|
||||
"/tts/preview",
|
||||
json={"text": "hello", "voice_id": ""},
|
||||
)
|
||||
assert resp.status_code == 422
|
||||
|
||||
def test_preview_clone_voice_resolves_to_cosyvoice_id(self):
|
||||
"""Clone voice UUID is resolved to CosyVoice voice_id."""
|
||||
from dataclasses import dataclass
|
||||
|
||||
from fastapi import FastAPI
|
||||
|
||||
app = FastAPI()
|
||||
from app.api.routes.tts import router
|
||||
|
||||
app.include_router(router, prefix="/tts")
|
||||
|
||||
from app.auth import get_current_user
|
||||
from app.dependencies import get_cosyvoice_service, get_voice_clone_profile_repository
|
||||
|
||||
fake_user = MagicMock()
|
||||
fake_user.user.id = "user-1"
|
||||
app.dependency_overrides[get_current_user] = lambda: fake_user
|
||||
|
||||
mock_service = MagicMock()
|
||||
mock_service.synthesize_speech.return_value = FakeSynthesizeResult(
|
||||
audio_url="https://x.com/cloned.mp3",
|
||||
duration=1.8,
|
||||
)
|
||||
app.dependency_overrides[get_cosyvoice_service] = lambda: mock_service
|
||||
|
||||
# Mock voice clone profile with voice_id
|
||||
mock_profile = MagicMock()
|
||||
mock_profile.user_id = "user-1"
|
||||
mock_profile.voice_id = "cosyvoice_actual_voice_123"
|
||||
mock_clone_repo = MagicMock()
|
||||
mock_clone_repo.get.return_value = mock_profile
|
||||
app.dependency_overrides[get_voice_clone_profile_repository] = lambda: mock_clone_repo
|
||||
|
||||
client = self._make_client(app)
|
||||
# Frontend sends the profile UUID as voice_id
|
||||
resp = client.post(
|
||||
"/tts/preview",
|
||||
json={"text": "克隆音色测试", "voice_id": "abc123-uuid-of-profile"},
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["audio_url"] == "https://x.com/cloned.mp3"
|
||||
|
||||
# Verify CosyVoice was called with the resolved voice_id, not the UUID
|
||||
mock_service.synthesize_speech.assert_called_once_with(
|
||||
text="克隆音色测试",
|
||||
voice_id="cosyvoice_actual_voice_123",
|
||||
speed=1.0,
|
||||
)
|
||||
# Verify repo was queried with the UUID
|
||||
mock_clone_repo.get.assert_called_once_with("abc123-uuid-of-profile")
|
||||
|
||||
def test_preview_clone_voice_incomplete_returns_400(self):
|
||||
"""Clone profile with empty voice_id returns 400."""
|
||||
from fastapi import FastAPI
|
||||
|
||||
app = FastAPI()
|
||||
from app.api.routes.tts import router
|
||||
|
||||
app.include_router(router, prefix="/tts")
|
||||
|
||||
from app.auth import get_current_user
|
||||
from app.dependencies import get_voice_clone_profile_repository
|
||||
|
||||
fake_user = MagicMock()
|
||||
fake_user.user.id = "user-1"
|
||||
app.dependency_overrides[get_current_user] = lambda: fake_user
|
||||
|
||||
# Mock voice clone profile with empty voice_id (clone not finished)
|
||||
mock_profile = MagicMock()
|
||||
mock_profile.user_id = "user-1"
|
||||
mock_profile.voice_id = ""
|
||||
mock_clone_repo = MagicMock()
|
||||
mock_clone_repo.get.return_value = mock_profile
|
||||
app.dependency_overrides[get_voice_clone_profile_repository] = lambda: mock_clone_repo
|
||||
|
||||
client = self._make_client(app)
|
||||
resp = client.post(
|
||||
"/tts/preview",
|
||||
json={"text": "测试未完成克隆", "voice_id": "abc123-uuid"},
|
||||
)
|
||||
assert resp.status_code == 400
|
||||
assert "音色克隆尚未完成" in resp.json()["detail"]
|
||||
|
||||
def test_preview_preset_voice_passthrough(self):
|
||||
"""Preset voice ID (not a profile UUID) passes through unchanged."""
|
||||
from fastapi import FastAPI
|
||||
|
||||
app = FastAPI()
|
||||
from app.api.routes.tts import router
|
||||
|
||||
app.include_router(router, prefix="/tts")
|
||||
|
||||
from app.auth import get_current_user
|
||||
from app.dependencies import get_cosyvoice_service, get_voice_clone_profile_repository
|
||||
|
||||
fake_user = MagicMock()
|
||||
fake_user.user.id = "user-1"
|
||||
app.dependency_overrides[get_current_user] = lambda: fake_user
|
||||
|
||||
mock_service = MagicMock()
|
||||
mock_service.synthesize_speech.return_value = FakeSynthesizeResult(
|
||||
audio_url="https://x.com/preset.mp3",
|
||||
duration=2.0,
|
||||
)
|
||||
app.dependency_overrides[get_cosyvoice_service] = lambda: mock_service
|
||||
|
||||
# Mock repo returns None (preset voice, not a clone profile)
|
||||
mock_clone_repo = MagicMock()
|
||||
mock_clone_repo.get.return_value = None
|
||||
app.dependency_overrides[get_voice_clone_profile_repository] = lambda: mock_clone_repo
|
||||
|
||||
client = self._make_client(app)
|
||||
resp = client.post(
|
||||
"/tts/preview",
|
||||
json={"text": "预设音色测试", "voice_id": "longxiaoxia_v3"},
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
|
||||
# Verify CosyVoice was called with the original preset voice_id
|
||||
mock_service.synthesize_speech.assert_called_once_with(
|
||||
text="预设音色测试",
|
||||
voice_id="longxiaoxia_v3",
|
||||
speed=1.0,
|
||||
)
|
||||
|
||||
def test_preview_clone_voice_wrong_user_returns_403(self):
|
||||
"""Accessing another user's clone profile returns 403."""
|
||||
from fastapi import FastAPI
|
||||
|
||||
app = FastAPI()
|
||||
from app.api.routes.tts import router
|
||||
|
||||
app.include_router(router, prefix="/tts")
|
||||
|
||||
from app.auth import get_current_user
|
||||
from app.dependencies import get_voice_clone_profile_repository
|
||||
|
||||
fake_user = MagicMock()
|
||||
fake_user.user.id = "user-1"
|
||||
app.dependency_overrides[get_current_user] = lambda: fake_user
|
||||
|
||||
# Mock profile belonging to a different user
|
||||
mock_profile = MagicMock()
|
||||
mock_profile.user_id = "user-2"
|
||||
mock_profile.voice_id = "cosyvoice_voice_xyz"
|
||||
mock_clone_repo = MagicMock()
|
||||
mock_clone_repo.get.return_value = mock_profile
|
||||
app.dependency_overrides[get_voice_clone_profile_repository] = lambda: mock_clone_repo
|
||||
|
||||
client = self._make_client(app)
|
||||
resp = client.post(
|
||||
"/tts/preview",
|
||||
json={"text": "越权测试", "voice_id": "other-user-profile-uuid"},
|
||||
)
|
||||
assert resp.status_code == 403
|
||||
assert "无权访问该音色" in resp.json()["detail"]
|
||||
|
||||
Reference in New Issue
Block a user