life-echo/api/app/adapters/tts/openai_tts.py

"""OpenAI TTS adapter — implements TTSProvider port."""

import asyncio
from io import BytesIO

from openai import OpenAI

from app.core.logging import get_logger

logger = get_logger(__name__)


class OpenAITTSProvider:
    def __init__(self, api_key: str, model: str = "tts-1"):
        self._client = OpenAI(api_key=api_key) if api_key else None
        self._model = model

    def _synthesize_sync(self, text: str, voice: str) -> bytes:
        if not self._client:
            return b""
        response = self._client.audio.speech.create(
            model=self._model,
            voice=voice,
            input=text,
        )
        buf = BytesIO()
        for chunk in response.iter_bytes():
            buf.write(chunk)
        return buf.getvalue()

    async def synthesize(
        self,
        text: str,
        voice: str = "alloy",
        *,
        language: str = "zh",  # noqa: ARG002 — OpenAI TTS auto-detects language
    ) -> bytes:
        if not self._client:
            return b""
        try:
            return await asyncio.to_thread(self._synthesize_sync, text, voice)
        except Exception as e:
            logger.warning("OpenAI TTS synthesize failed: {}", e)
            return b""