diff --git a/apps/api/app/api/routes/tts.py b/apps/api/app/api/routes/tts.py index 954c52c2e..eb74903ff 100755 --- a/apps/api/app/api/routes/tts.py +++ b/apps/api/app/api/routes/tts.py @@ -20,6 +20,8 @@ from app.schemas.tts import ( SaveToLibraryRequest, SaveToLibraryResponse, TTSJobResponse, + TTSPreviewRequest, + TTSPreviewResponse, TTSStatusResponse, TTSSynthesizeRequest, TTSSynthesizeResponse, @@ -31,7 +33,7 @@ from packages.adapters.sqlalchemy_impl.tts_job_repository import ( SQLAlchemyTTSJobRepository, ) from packages.adapters.sqlalchemy_impl.voice_library_repository import SQLAlchemyVoiceLibraryRepository -from packages.application.cosyvoice_service import CosyVoiceService +from packages.application.cosyvoice_service import CosyVoiceError, CosyVoiceService from packages.application.tts_job.streaming_service import TTSStreamingService from packages.application.tts_job.use_cases import ( CreateTTSJobUseCase, @@ -373,6 +375,40 @@ def save_tts_job_to_library( ) + +@router.post("/preview", response_model=TTSPreviewResponse) +def preview_tts( + request: TTSPreviewRequest, + authenticated_user: AuthenticatedUser = Depends(get_current_user), + cosyvoice_service: CosyVoiceService = Depends(get_cosyvoice_service), +) -> TTSPreviewResponse: + """TTS 预览(试听)——同步合成,立即返回音频 URL。 + + 用于前端预览配音效果,限制文本长度 200 字以内。 + """ + try: + result = cosyvoice_service.synthesize_speech( + text=request.text, + voice_id=request.voice_id, + speed=request.speed, + ) + except CosyVoiceError as e: + raise HTTPException( + status_code=status.HTTP_502_BAD_GATEWAY, + detail=f"TTS 合成失败: {e}", + ) from e + except ValueError as e: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail=str(e), + ) from e + + return TTSPreviewResponse( + audio_url=result.audio_url, + duration=result.duration if result.duration and result.duration > 0 else None, + ) + + @router.websocket("/ws/tts/stream") async def tts_websocket_stream( websocket: WebSocket, diff --git a/apps/api/app/schemas/tts.py b/apps/api/app/schemas/tts.py index d2d098be7..1e6b47513 100644 --- a/apps/api/app/schemas/tts.py +++ b/apps/api/app/schemas/tts.py @@ -101,3 +101,19 @@ class SaveToLibraryResponse(BaseModel): voice_id: str voice_name: str status: str + + +class TTSPreviewRequest(BaseModel): + """TTS 预览(试听)请求。""" + + text: str = Field(..., min_length=1, max_length=200, description="合成文本,限制 200 字") + voice_id: str = Field(..., min_length=1, description="音色 ID") + speed: float = Field(1.0, ge=0.5, le=2.0, description="语速") + pitch: float = Field(1.0, ge=0.5, le=2.0, description="音调(预留,当前未使用)") + + +class TTSPreviewResponse(BaseModel): + """TTS 预览(试听)响应。""" + + audio_url: str = Field(..., description="合成音频 URL") + duration: Optional[float] = Field(default=None, description="音频时长(秒)") diff --git a/tests/unit/test_tts_preview.py b/tests/unit/test_tts_preview.py new file mode 100644 index 000000000..4320b4938 --- /dev/null +++ b/tests/unit/test_tts_preview.py @@ -0,0 +1,247 @@ +"""Tests for POST /tts/preview endpoint.""" + +from __future__ import annotations + +from dataclasses import dataclass +from unittest.mock import MagicMock, patch + +import pytest +from pydantic import ValidationError + + +@dataclass +class FakeSynthesizeResult: + audio_url: str + duration: float = 0.0 + file_size: int = 0 + request_id: str = "" + + +class TestTTSPreviewEndpoint: + """Integration-style tests for the /tts/preview route.""" + + def _make_client(self, app): + from fastapi.testclient import TestClient + + return TestClient(app) + + def test_schema_preview_request_validation(self): + """TTSPreviewRequest rejects text > 200 chars and empty voice_id.""" + from app.schemas.tts import TTSPreviewRequest + + # Valid + req = TTSPreviewRequest(text="hello", voice_id="v1") + assert req.text == "hello" + assert req.voice_id == "v1" + assert req.speed == 1.0 + + # Empty voice_id rejected + with pytest.raises(ValidationError): + TTSPreviewRequest(text="hello", voice_id="") + + # Text > 200 chars rejected + with pytest.raises(ValidationError): + TTSPreviewRequest(text="a" * 201, voice_id="v1") + + def test_schema_preview_response(self): + """TTSPreviewResponse serialization.""" + from app.schemas.tts import TTSPreviewResponse + + resp = TTSPreviewResponse(audio_url="https://example.com/audio.mp3") + assert resp.audio_url == "https://example.com/audio.mp3" + assert resp.duration is None + + resp2 = TTSPreviewResponse(audio_url="https://x.com/a.mp3", duration=3.5) + assert resp2.duration == 3.5 + + def test_preview_success(self): + """Successful preview returns audio_url.""" + from app.schemas.tts import TTSPreviewRequest + from fastapi import FastAPI + + app = FastAPI() + + # We need to register the route with proper dependencies + from app.api.routes.tts import router + + app.include_router(router, prefix="/tts") + + # Override dependencies + from app.auth import get_current_user + from app.dependencies import get_cosyvoice_service + + 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://cosyvoice.example.com/audio.mp3", + duration=2.5, + ) + app.dependency_overrides[get_cosyvoice_service] = lambda: mock_service + + client = self._make_client(app) + resp = client.post( + "/tts/preview", + json={"text": "你好世界", "voice_id": "longxiaochun"}, + ) + assert resp.status_code == 200 + data = resp.json() + assert data["audio_url"] == "https://cosyvoice.example.com/audio.mp3" + assert data["duration"] == 2.5 + + mock_service.synthesize_speech.assert_called_once_with( + text="你好世界", + voice_id="longxiaochun", + speed=1.0, + ) + + def test_preview_with_speed(self): + """Custom speed is passed through to CosyVoice.""" + 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 + + 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/a.mp3", + duration=0.0, + ) + app.dependency_overrides[get_cosyvoice_service] = lambda: mock_service + + client = self._make_client(app) + resp = client.post( + "/tts/preview", + json={"text": "测试", "voice_id": "v1", "speed": 1.5}, + ) + assert resp.status_code == 200 + data = resp.json() + assert data["audio_url"] == "https://x.com/a.mp3" + assert data["duration"] is None # 0.0 -> None + + mock_service.synthesize_speech.assert_called_once_with( + text="测试", + voice_id="v1", + speed=1.5, + ) + + def test_preview_cosyvoice_error_returns_502(self): + """CosyVoice failure returns 502.""" + from fastapi import FastAPI + + from packages.application.cosyvoice_service import CosyVoiceError + + 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 + + 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.side_effect = CosyVoiceError("API timeout") + app.dependency_overrides[get_cosyvoice_service] = lambda: mock_service + + client = self._make_client(app) + resp = client.post( + "/tts/preview", + json={"text": "测试", "voice_id": "v1"}, + ) + assert resp.status_code == 502 + assert "TTS 合成失败" in resp.json()["detail"] + + def test_preview_value_error_returns_400(self): + """Invalid params return 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_cosyvoice_service + + 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.side_effect = ValueError("text 不能为空") + app.dependency_overrides[get_cosyvoice_service] = lambda: mock_service + + client = self._make_client(app) + resp = client.post( + "/tts/preview", + json={"text": "测试", "voice_id": "v1"}, + ) + assert resp.status_code == 400 + assert "text 不能为空" in resp.json()["detail"] + + def test_preview_text_too_long_returns_422(self): + """Text > 200 chars is rejected by Pydantic validation.""" + 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 + + fake_user = MagicMock() + fake_user.user.id = "user-1" + app.dependency_overrides[get_current_user] = lambda: fake_user + + mock_service = MagicMock() + app.dependency_overrides[get_cosyvoice_service] = lambda: mock_service + + client = self._make_client(app) + resp = client.post( + "/tts/preview", + json={"text": "a" * 201, "voice_id": "v1"}, + ) + assert resp.status_code == 422 + + def test_preview_empty_voice_id_returns_422(self): + """Empty voice_id is rejected by Pydantic validation.""" + 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 + + fake_user = MagicMock() + fake_user.user.id = "user-1" + app.dependency_overrides[get_current_user] = lambda: fake_user + + mock_service = MagicMock() + app.dependency_overrides[get_cosyvoice_service] = lambda: mock_service + + client = self._make_client(app) + resp = client.post( + "/tts/preview", + json={"text": "hello", "voice_id": ""}, + ) + assert resp.status_code == 422