hunjing/backend/app/ai_provider/audio/__init__.py

"""音频服务 — ASR / TTS 适配器外观。

用法::

    from app.ai_provider.audio import AudioService

    service = AudioService(factory)
    result = await service.transcribe(audio_bytes, format="wav")
    result = await service.synthesize("你好，魂镜", voice="default")

``AudioService`` 通过注册表中带有 ``asr`` / ``tts`` 能力标签的模型来解析
使用哪个模型，然后将创建委托给 ``ModelFactory.create_audio()``。
"""

from __future__ import annotations

from app.ai_provider.audio.base import (
    ASRProtocol,
    ASRRequest,
    ASRResponse,
    AudioServiceError,
    TTSProtocol,
    TTSRequest,
    TTSResponse,
)


class AudioService:
    """外观类，负责解析 ASR / TTS 模型并委托给 ProviderManager。

    接受 ProviderManager 或实现了以下接口的适配器：
    - ``.registry`` → 具有 ``list_capability_models(capability: str)`` 方法
    - ``.create_audio(name: str)`` → 返回 ASRProtocol / TTSProtocol 实例
    """

    def __init__(self, manager) -> None:
        self._manager = manager
        self._asr_adapter: ASRProtocol | None = None
        self._tts_adapter: TTSProtocol | None = None

    # ── Registry 兼容适配 ─────────────────────────────────────────

    @property
    def _registry(self):
        """按需获取 registry，兼容 adapter 和裸 ProviderManager/ModelRegistry。"""
        mgr = self._manager
        if hasattr(mgr, "registry"):
            return mgr.registry
        if hasattr(mgr, "model_registry"):
            return mgr.model_registry
        return mgr

    def _list_capability_models(self, tag: str):
        reg = self._registry
        # 适配器接口（list_capability_models）vs  ModelRegistry 接口（list_by_capability）
        if hasattr(reg, "list_capability_models"):
            return reg.list_capability_models(tag)
        if hasattr(reg, "list_by_capability"):
            return reg.list_by_capability(tag)
        return []

    @property
    def _factory(self):
        mgr = self._manager
        if hasattr(mgr, "create_audio"):
            return mgr
        return None

    # ── ASR ──────────────────────────────────────────────────────────────

    @property
    def asr(self) -> ASRProtocol:
        """解析并缓存 ASR 适配器（取注册表中第一个活跃 ASR 模型）。"""
        if self._asr_adapter is None:
            self._asr_adapter = self._resolve_asr()
        return self._asr_adapter

    def asr_for(self, model_name: str) -> ASRProtocol:
        """解析并缓存指定模型名的 ASR 适配器（覆盖默认第一个）。"""
        self._asr_adapter = self._factory.create_audio(model_name)
        return self._asr_adapter

    async def transcribe(
        self,
        audio_data: bytes,
        format: str = "wav",
        language: str | None = None,
        model_name: str | None = None,
    ) -> ASRResponse:
        """快捷方法：构造请求并调用 ASR 适配器。

        Args:
            model_name: 可选，指定使用哪个 ASR 模型（按注册表逻辑名）。
                省略时取注册表中第一个活跃的 ASR 模型。
        """
        if model_name is not None:
            adapter = self.asr_for(model_name)
        else:
            adapter = self.asr
        request = ASRRequest(
            audio_data=audio_data,
            format=format,
            language=language,
        )
        return await adapter.transcribe(request)

    # ── TTS ──────────────────────────────────────────────────────────────

    @property
    def tts(self) -> TTSProtocol:
        """解析并缓存 TTS 适配器。"""
        if self._tts_adapter is None:
            self._tts_adapter = self._resolve_tts()
        return self._tts_adapter

    async def synthesize(
        self,
        text: str,
        voice: str = "default",
        response_format: str = "wav",
        extra: dict | None = None,
    ) -> TTSResponse:
        """快捷方法：构造请求并调用 TTS 适配器。"""
        request = TTSRequest(
            text=text,
            voice=voice,
            response_format=response_format,
            extra=extra,
        )
        return await self.tts.synthesize(request)

    # ── 解析 ───────────────────────────────────────────────────────

    def _resolve_asr(self) -> ASRProtocol:
        """查找第一个活跃的 ASR 模型并通过工厂创建。"""
        models = self._list_capability_models("asr")
        if not models:
            raise AudioServiceError(
                status_code=503,
                code="audio_asr_no_model",
                message="No ASR model configured in the model registry.",
            )
        factory = self._factory
        if factory is None:
            raise AudioServiceError(
                status_code=501,
                code="audio_asr_no_factory",
                message="Audio adapter factory not available (provider manager missing create_audio).",
            )
        return factory.create_audio(models[0].name)

    def _resolve_tts(self) -> TTSProtocol:
        """查找第一个活跃的 TTS 模型并通过工厂创建。"""
        models = self._list_capability_models("tts")
        if not models:
            raise AudioServiceError(
                status_code=503,
                code="audio_tts_no_model",
                message="No TTS model configured in the model registry.",
            )
        factory = self._factory
        if factory is None:
            raise AudioServiceError(
                status_code=501,
                code="audio_tts_no_factory",
                message="Audio adapter factory not available (provider manager missing create_audio).",
            )
        return factory.create_audio(models[0].name)

    # ── 生命周期 ────────────────────────────────────────────────────────

    async def close(self) -> None:
        """释放适配器持有的 HTTP 连接。"""
        for adapter in (self._asr_adapter, self._tts_adapter):
            if adapter is None:
                continue
            close = getattr(adapter, "close", None)
            if close is not None:
                await close()

    async def __aenter__(self) -> AudioService:
        return self

    async def __aexit__(self, *exc_info: object) -> None:
        await self.close()


__all__ = [
    "ASRProtocol",
    "ASRRequest",
    "ASRResponse",
    "AudioService",
    "AudioServiceError",
    "TTSProtocol",
    "TTSRequest",
    "TTSResponse",
]