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,

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="音频时长（秒）")

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