diff --git a/apps/api/app/api/routes/voices.py b/apps/api/app/api/routes/voices.py index 7b47b19f8..ed0a9f996 100755 --- a/apps/api/app/api/routes/voices.py +++ b/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}") diff --git a/tests/unit/test_preset_voice_dynamic_preview.py b/tests/unit/test_preset_voice_dynamic_preview.py new file mode 100644 index 000000000..25087276f --- /dev/null +++ b/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