apps/api/app/api/routes/voices.py

@@ -5,6 +5,9 @@
 
 from __future__ import annotations
 
+import logging
+import time
+from concurrent.futures import ThreadPoolExecutor, as_completed
 from typing import Literal, Optional
 
 from app.api.routes._helpers import get_user_plan
@@ -42,6 +45,7 @@ from packages.domain.preset_voices import PRESET_VOICES, get_preset_voice_by_id
 from packages.ports.user_repository import UserRepository
 
 router = APIRouter()
+logger = logging.getLogger(__name__)
 
 # 预置音色试听音频缓存（内存缓存，减少重复TTS调用）
 # key: voice_id, value: (audio_url, timestamp)
@@ -51,6 +55,68 @@ PREVIEW_CACHE_TTL = 7 * 24 * 3600  # 7天TTL
 PREVIEW_TEMPLATE = "你好，我是{name}，很高兴认识你。"
 
 
+def _resolve_preset_preview_url(
+    voice_id: str,
+    fallback_url: str,
+    cosyvoice: CosyVoiceService,
+) -> str:
+    """为预置音色获取有效的 preview_url.
+
+    优先从内存缓存读取；缓存失效时调用 CosyVoice 重新合成；
+    合成失败时降级返回硬编码 URL（可能已过期，但不会报错）。
+    """
+    # 检查缓存
+    if voice_id in _preset_preview_cache:
+        audio_url, cached_at = _preset_preview_cache[voice_id]
+        if time.time() - cached_at < PREVIEW_CACHE_TTL:
+            return audio_url
+
+    # 缓存失效，调用 CosyVoice 合成
+    preset = get_preset_voice_by_id(voice_id)
+    if preset is None:
+        return fallback_url
+
+    preview_text = PREVIEW_TEMPLATE.format(name=preset.name)
+    try:
+        result = cosyvoice.synthesize_speech(
+            text=preview_text,
+            voice_id=voice_id,
+            format="mp3",
+            speed=1.0,
+        )
+        audio_url = result.audio_url
+        _preset_preview_cache[voice_id] = (audio_url, time.time())
+        logger.info("Preset voice preview generated: %s", voice_id)
+        return audio_url
+    except Exception as e:
+        logger.warning("Failed to generate preview for %s, using fallback: %s", voice_id, e)
+        return fallback_url
+
+
+def _resolve_all_preset_preview_urls(
+    presets: list,
+    cosyvoice: CosyVoiceService,
+) -> dict[str, str]:
+    """并行解析所有预置音色的 preview_url.
+
+    Returns:
+        voice_id -> preview_url 映射
+    """
+    result_map: dict[str, str] = {}
+    with ThreadPoolExecutor(max_workers=4) as executor:
+        future_to_voice = {
+            executor.submit(_resolve_preset_preview_url, p.voice_id, p.preview_url, cosyvoice): p.voice_id
+            for p in presets
+        }
+        for future in as_completed(future_to_voice):
+            vid = future_to_voice[future]
+            try:
+                result_map[vid] = future.result()
+            except Exception:
+                result_map[vid] = ""
+    return result_map
+
+
 def _get_voice_repository(session: Session = Depends(get_db_session)) -> SQLAlchemyVoiceLibraryRepository:
     return SQLAlchemyVoiceLibraryRepository(session)
 
@@ -118,8 +184,16 @@ def _to_unified_response(item, profile_id_map: dict | None = None, sign_url=None
     )
 
 
-def _preset_to_unified_response(preset) -> UnifiedVoiceItemResponse:
-    """将预置音色转换为统一响应格式。"""
+def _preset_to_unified_response(preset, preview_url_map: dict[str, str] | None = None) -> UnifiedVoiceItemResponse:
+    """将预置音色转换为统一响应格式。
+
+    Args:
+        preset: 预置音色对象
+        preview_url_map: voice_id -> preview_url 动态映射，优先使用
+    """
+    preview_url = preset.preview_url
+    if preview_url_map and preset.voice_id in preview_url_map:
+        preview_url = preview_url_map[preset.voice_id]
     return UnifiedVoiceItemResponse(
         id=preset.voice_id,
         type="preset",
@@ -129,7 +203,7 @@ def _preset_to_unified_response(preset) -> UnifiedVoiceItemResponse:
         language=preset.language,
         voice_id=preset.voice_id,
         voice_provider="cosyvoice",
-        preview_url=preset.preview_url,
+        preview_url=preview_url,
         tags=preset.tags or [],
     )
 
@@ -179,6 +253,7 @@ def list_voices_unified(
     voice_repository: SQLAlchemyVoiceLibraryRepository = Depends(_get_voice_repository),
     clone_profile_repository: SQLAlchemyVoiceCloneProfileRepository = Depends(_get_clone_profile_repository),
     sign_url=Depends(get_audio_url_signer),
+    cosyvoice: CosyVoiceService = Depends(get_cosyvoice_service),
 ) -> UnifiedVoiceListResponse:
     """获取配音列表（预置音色 + 用户克隆音色）。
 
@@ -194,9 +269,10 @@ def list_voices_unified(
     has_preset = type is None or type == "preset"
     has_clone = type is None or type == "clone"
 
-    # 获取预置音色
+    # 获取预置音色（动态生成 preview_url）
     if has_preset:
-        preset_items = [_preset_to_unified_response(p) for p in PRESET_VOICES]
+        preview_url_map = _resolve_all_preset_preview_urls(PRESET_VOICES, cosyvoice)
+        preset_items = [_preset_to_unified_response(p, preview_url_map) for p in PRESET_VOICES]
         preset_count = len(preset_items)
 
     # 获取克隆音色（从 voice_clone_profile 读取，ready 状态的克隆音色）
@@ -243,11 +319,15 @@ def list_voices_unified(
 
 
 @router.get("/presets", response_model=PresetVoiceListResponse)
-def list_preset_voices() -> PresetVoiceListResponse:
+def list_preset_voices(
+    cosyvoice: CosyVoiceService = Depends(get_cosyvoice_service),
+) -> PresetVoiceListResponse:
     """获取预置音色列表。
 
     不需要认证，返回所有系统预置的 CosyVoice 音色。
+    preview_url 通过 CosyVoice 动态生成，不依赖硬编码的过期 URL。
     """
+    preview_url_map = _resolve_all_preset_preview_urls(PRESET_VOICES, cosyvoice)
     items = [
         PresetVoiceItemResponse(
             voice_id=p.voice_id,
@@ -255,7 +335,7 @@ def list_preset_voices() -> PresetVoiceListResponse:
             description=p.description,
             gender=p.gender,
             language=p.language,
-            preview_url=p.preview_url,
+            preview_url=preview_url_map.get(p.voice_id, p.preview_url),
             tags=p.tags or [],
         )
         for p in PRESET_VOICES
@@ -275,8 +355,6 @@ def get_preset_voice_preview(
     - 相同 voice_id 重复调用直接返回缓存的音频URL
     - 可传入自定义 text 参数试听不同文本
     """
-    import time
-
     preset = get_preset_voice_by_id(voice_id)
     if preset is None:
         raise HTTPException(status_code=404, detail=f"预置音色不存在: {voice_id}")

tests/unit/test_preset_voice_dynamic_preview.py

@@ -0,0 +1,230 @@
+"""Tests for dynamic preview URL resolution in voices routes.
+
+覆盖 _resolve_preset_preview_url 和 _resolve_all_preset_preview_urls 的逻辑：
+- 缓存命中时直接返回
+- 缓存失效时调用 CosyVoice 合成
+- CosyVoice 失败时降级到硬编码 URL
+- 并行解析多个预置音色
+"""
+
+from __future__ import annotations
+
+import time
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from packages.application.cosyvoice_service import CosyVoiceError, SynthesizeResult
+from packages.domain.preset_voices import PRESET_VOICES
+
+# ── fixtures ──────────────────────────────────────────────────────────
+
+
+@pytest.fixture()
+def mock_cosyvoice():
+    """构造一个模拟的 CosyVoiceService。"""
+    svc = MagicMock()
+    svc.synthesize_speech.return_value = SynthesizeResult(
+        audio_url="https://fresh-url.example.com/preview.mp3",
+        duration=2.5,
+        file_size=40000,
+        request_id="test-req-id",
+    )
+    return svc
+
+
+@pytest.fixture()
+def failing_cosyvoice():
+    """构造一个总是失败的 CosyVoiceService。"""
+    svc = MagicMock()
+    svc.synthesize_speech.side_effect = CosyVoiceError("TTS API unavailable")
+    return svc
+
+
+@pytest.fixture(autouse=True)
+def clear_cache():
+    """每个测试前清空预览缓存。"""
+    from apps.api.app.api.routes import voices
+
+    voices._preset_preview_cache.clear()
+    yield
+    voices._preset_preview_cache.clear()
+
+
+# ── _resolve_preset_preview_url ──────────────────────────────────────
+
+
+class TestResolvePresetPreviewUrl:
+    """测试单个预置音色的 preview_url 解析。"""
+
+    def test_cache_hit_returns_cached_url(self, mock_cosyvoice):
+        """缓存有效时直接返回，不调用 CosyVoice。"""
+        from apps.api.app.api.routes.voices import _preset_preview_cache, _resolve_preset_preview_url
+
+        voice_id = "longxiaochun_v3"
+        _preset_preview_cache[voice_id] = ("https://cached-url.example.com/audio.mp3", time.time())
+
+        result = _resolve_preset_preview_url(voice_id, "fallback.mp3", mock_cosyvoice)
+
+        assert result == "https://cached-url.example.com/audio.mp3"
+        mock_cosyvoice.synthesize_speech.assert_not_called()
+
+    def test_cache_expired_triggers_synthesis(self, mock_cosyvoice):
+        """缓存过期时重新调用 CosyVoice 合成。"""
+        from apps.api.app.api.routes.voices import _preset_preview_cache, _resolve_preset_preview_url
+
+        voice_id = "longxiaochun_v3"
+        # 设置一个过期的缓存（10天前）
+        _preset_preview_cache[voice_id] = ("https://old-url.mp3", time.time() - 10 * 86400)
+
+        result = _resolve_preset_preview_url(voice_id, "fallback.mp3", mock_cosyvoice)
+
+        assert result == "https://fresh-url.example.com/preview.mp3"
+        mock_cosyvoice.synthesize_speech.assert_called_once()
+        # 新 URL 应该被缓存
+        assert voice_id in _preset_preview_cache
+        cached_url, _ = _preset_preview_cache[voice_id]
+        assert cached_url == "https://fresh-url.example.com/preview.mp3"
+
+    def test_cache_miss_triggers_synthesis(self, mock_cosyvoice):
+        """缓存不存在时调用 CosyVoice 合成。"""
+        from apps.api.app.api.routes.voices import _resolve_preset_preview_url
+
+        voice_id = "longsanshu_v3"
+        result = _resolve_preset_preview_url(voice_id, "fallback.mp3", mock_cosyvoice)
+
+        assert result == "https://fresh-url.example.com/preview.mp3"
+        mock_cosyvoice.synthesize_speech.assert_called_once()
+
+    def test_synthesis_failure_returns_fallback(self, failing_cosyvoice):
+        """CosyVoice 合成失败时降级返回硬编码 URL。"""
+        from apps.api.app.api.routes.voices import _resolve_preset_preview_url
+
+        voice_id = "longyue_v3"
+        fallback = "https://hardcoded-fallback.example.com/audio.mp3"
+        result = _resolve_preset_preview_url(voice_id, fallback, failing_cosyvoice)
+
+        assert result == fallback
+        failing_cosyvoice.synthesize_speech.assert_called_once()
+
+    def test_nonexistent_voice_returns_fallback(self, mock_cosyvoice):
+        """voice_id 不存在时直接返回 fallback。"""
+        from apps.api.app.api.routes.voices import _resolve_preset_preview_url
+
+        result = _resolve_preset_preview_url("nonexistent_voice_id", "fallback.mp3", mock_cosyvoice)
+
+        assert result == "fallback.mp3"
+        mock_cosyvoice.synthesize_speech.assert_not_called()
+
+    def test_synthesis_call_uses_correct_params(self, mock_cosyvoice):
+        """验证调用 CosyVoice 时参数正确。"""
+        from apps.api.app.api.routes.voices import _resolve_preset_preview_url
+
+        voice_id = "longshu_v3"
+        _resolve_preset_preview_url(voice_id, "fallback.mp3", mock_cosyvoice)
+
+        mock_cosyvoice.synthesize_speech.assert_called_once_with(
+            text="你好，我是龙书，很高兴认识你。",
+            voice_id="longshu_v3",
+            format="mp3",
+            speed=1.0,
+        )
+
+
+# ── _resolve_all_preset_preview_urls ─────────────────────────────────
+
+
+class TestResolveAllPresetPreviewUrls:
+    """测试批量并行解析所有预置音色的 preview_url。"""
+
+    def test_all_voices_resolved(self, mock_cosyvoice):
+        """所有 8 个预置音色都能解析出 preview_url。"""
+        from apps.api.app.api.routes.voices import _resolve_all_preset_preview_urls
+
+        result = _resolve_all_preset_preview_urls(PRESET_VOICES, mock_cosyvoice)
+
+        assert len(result) == 8
+        for preset in PRESET_VOICES:
+            assert preset.voice_id in result
+            assert result[preset.voice_id] == "https://fresh-url.example.com/preview.mp3"
+
+    def test_partial_failure_still_returns_all(self, mock_cosyvoice):
+        """部分合成失败时仍返回所有 voice_id（成功的用新 URL，失败的用 fallback）。"""
+        from apps.api.app.api.routes.voices import _resolve_all_preset_preview_urls
+
+        # 让第一个音色失败，其余成功
+        call_count = 0
+        original_side_effect = mock_cosyvoice.synthesize_speech.side_effect
+
+        def side_effect_fn(*args, **kwargs):
+            nonlocal call_count
+            call_count += 1
+            if call_count == 1:
+                raise CosyVoiceError("first call fails")
+            return SynthesizeResult(audio_url="https://ok.mp3", duration=1.0)
+
+        mock_cosyvoice.synthesize_speech.side_effect = side_effect_fn
+
+        result = _resolve_all_preset_preview_urls(PRESET_VOICES, mock_cosyvoice)
+
+        assert len(result) == 8
+        # 第一个失败的音色应该返回 fallback（硬编码 URL）
+        first_voice = PRESET_VOICES[0]
+        assert result[first_voice.voice_id] == first_voice.preview_url
+
+    def test_all_fail_returns_empty_strings_or_fallbacks(self, failing_cosyvoice):
+        """全部合成失败时，每个 voice_id 都有 fallback URL。"""
+        from apps.api.app.api.routes.voices import _resolve_all_preset_preview_urls
+
+        result = _resolve_all_preset_preview_urls(PRESET_VOICES, failing_cosyvoice)
+
+        assert len(result) == 8
+        for preset in PRESET_VOICES:
+            # 失败时返回硬编码的 fallback URL
+            assert result[preset.voice_id] == preset.preview_url
+
+    def test_empty_presets_returns_empty(self, mock_cosyvoice):
+        """空列表时返回空字典。"""
+        from apps.api.app.api.routes.voices import _resolve_all_preset_preview_urls
+
+        result = _resolve_all_preset_preview_urls([], mock_cosyvoice)
+
+        assert result == {}
+
+
+# ── _preset_to_unified_response with preview_url_map ─────────────────
+
+
+class TestPresetToUnifiedResponseWithMap:
+    """测试 _preset_to_unified_response 支持 preview_url_map 参数。"""
+
+    def test_with_preview_url_map(self):
+        """传入 preview_url_map 时优先使用 map 中的 URL。"""
+        from apps.api.app.api.routes.voices import _preset_to_unified_response
+
+        preset = PRESET_VOICES[0]
+        url_map = {preset.voice_id: "https://dynamic-url.example.com/audio.mp3"}
+
+        result = _preset_to_unified_response(preset, url_map)
+
+        assert result.preview_url == "https://dynamic-url.example.com/audio.mp3"
+
+    def test_without_map_uses_hardcoded(self):
+        """不传 preview_url_map 时使用硬编码 URL。"""
+        from apps.api.app.api.routes.voices import _preset_to_unified_response
+
+        preset = PRESET_VOICES[0]
+        result = _preset_to_unified_response(preset)
+
+        assert result.preview_url == preset.preview_url
+
+    def test_map_missing_voice_falls_back(self):
+        """preview_url_map 中没有对应 voice_id 时使用硬编码 URL。"""
+        from apps.api.app.api.routes.voices import _preset_to_unified_response
+
+        preset = PRESET_VOICES[0]
+        url_map = {"other_voice_id": "https://other.mp3"}
+
+        result = _preset_to_unified_response(preset, url_map)
+
+        assert result.preview_url == preset.preview_url