"""TTS 服务工厂. 根据配置创建对应的 TTS 服务实例。 """ from __future__ import annotations import logging import os from packages.ports.tts_service import TtsService logger = logging.getLogger(__name__) # 可用的 provider 映射 _PROVIDERS: dict[str, type[TtsService]] = {} def register_provider(name: str, cls: type[TtsService]) -> None: """注册 TTS 供应商.""" _PROVIDERS[name] = cls def get_tts_service(provider: str | None = None, **kwargs) -> TtsService: """获取 TTS 服务实例. Args: provider: 供应商名称(None 则从环境变量读取 TTS_PROVIDER) **kwargs: 传递给服务构造函数的参数 Returns: TTS 服务实例 Raises: ValueError: 不支持的供应商 """ if provider is None: provider = os.environ.get("TTS_PROVIDER", "") if not provider: # 自动检测:配置了 CosyVoice API Key 则默认用 cosyvoice,否则用 mock try: from packages.shared.config import get_shared_settings settings = get_shared_settings() if getattr(settings, "cosyvoice_api_key", ""): provider = "cosyvoice" else: provider = "mock" except Exception: provider = "mock" provider = provider.lower() if provider not in _PROVIDERS: # 延迟导入避免循环依赖 if provider == "mock": from packages.adapters.tts.mock_tts_service import MockTtsService _PROVIDERS["mock"] = MockTtsService elif provider in ("cosyvoice", "aliyun", "dashscope"): from packages.adapters.tts.cosyvoice_tts_service import CosyVoiceTtsService _PROVIDERS["cosyvoice"] = CosyVoiceTtsService _PROVIDERS["aliyun"] = CosyVoiceTtsService _PROVIDERS["dashscope"] = CosyVoiceTtsService provider = "cosyvoice" else: logger.warning("未知 TTS provider: %s,回退到 mock", provider) from packages.adapters.tts.mock_tts_service import MockTtsService _PROVIDERS["mock"] = MockTtsService provider = "mock" cls = _PROVIDERS[provider] return cls(**kwargs) def available_providers() -> list[str]: """获取可用的供应商列表.""" # 确保 mock 已注册 if "mock" not in _PROVIDERS: from packages.adapters.tts.mock_tts_service import MockTtsService _PROVIDERS["mock"] = MockTtsService return list(_PROVIDERS.keys())