xiaoxia-saas/apps/worker/services/tts_service_factory.py

"""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())