from __future__ import annotations import math import base64 import json import socket import struct import subprocess import tempfile import urllib.error import urllib.request import wave from collections.abc import Callable from pathlib import Path from typing import Any from .config import AppConfig from .models import AudioSegment, ErrorCode, ProviderError class SentenceBuffer: def __init__(self, max_chars: int = 80) -> None: self.max_chars = max_chars self._buffer = "" def feed(self, text: str, force: bool = False) -> list[str]: if text: self._buffer += text chunks: list[str] = [] while self._should_emit(force): chunks.append(self._pop_chunk(force)) force = False return [chunk for chunk in chunks if chunk.strip()] def flush(self) -> list[str]: return self.feed("", force=True) def _should_emit(self, force: bool) -> bool: stripped = self._buffer.strip() if not stripped: return False return force or stripped.endswith(("。", "！", "？", ".", "!", "?", "\n")) or len(stripped) >= self.max_chars def _pop_chunk(self, force: bool) -> str: stripped = self._buffer.strip() if force or len(stripped) <= self.max_chars: self._buffer = "" return stripped chunk = stripped[: self.max_chars] self._buffer = stripped[self.max_chars :] return chunk class SineTtsProvider: def __init__(self, sample_rate: int = 16000) -> None: self.sample_rate = sample_rate self.loaded = False def load(self) -> None: self.loaded = True def synthesize(self, text: str) -> AudioSegment: if not self.loaded: raise ProviderError( ErrorCode.TTS_SYNTHESIS_FAILED, "TTS provider is not loaded", False, "sine-tts", "tts", ) clean = text.strip() if not clean: raise ProviderError( ErrorCode.TTS_EMPTY_AUDIO, "cannot synthesize empty text", True, "sine-tts", "tts", ) duration_ms = max(120, min(1200, len(clean) * 45)) samples = int(self.sample_rate * duration_ms / 1000) pcm = bytearray() for idx in range(samples): sample = int(math.sin(2 * math.pi * 440 * idx / self.sample_rate) * 12000) pcm.extend(struct.pack(" None: self.voice = voice self.loaded = False def load(self) -> None: self.loaded = True def synthesize(self, text: str) -> AudioSegment: clean = text.strip() if not clean: raise ProviderError( ErrorCode.TTS_EMPTY_AUDIO, "cannot synthesize empty text", True, "macos-say", "tts", ) with tempfile.TemporaryDirectory() as tmp: aiff_output = Path(tmp) / "speech.aiff" wav_output = Path(tmp) / "speech.wav" command = ["say", "-o", str(aiff_output)] if self.voice: command.extend(["-v", self.voice]) command.append(clean) try: subprocess.run(command, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE) except (FileNotFoundError, subprocess.CalledProcessError) as exc: raise ProviderError( ErrorCode.TTS_SYNTHESIS_FAILED, f"macOS say failed: {exc}", False, "macos-say", "tts", ) from exc try: subprocess.run( ["afconvert", "-f", "WAVE", "-d", "LEI16", "-c", "1", str(aiff_output), str(wav_output)], check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, ) with wave.open(str(wav_output), "rb") as handle: sample_rate = handle.getframerate() channels = handle.getnchannels() data = handle.readframes(handle.getnframes()) metadata = {"text": clean} except (FileNotFoundError, subprocess.CalledProcessError, wave.Error): data = aiff_output.read_bytes() sample_rate = 16000 channels = 1 metadata = {"text": clean, "format": "aiff"} if not data: raise ProviderError( ErrorCode.TTS_EMPTY_AUDIO, "macOS say produced empty audio", True, "macos-say", "tts", ) duration_ms = int(len(data) / max(1, sample_rate * channels * 2) * 1000) return AudioSegment(data, sample_rate, channels, 0, max(120, duration_ms, len(clean) * 45), metadata) class CloudTtsProvider: def __init__( self, config: AppConfig, timeout_s: float = 60.0, urlopen: Callable[..., Any] | None = None, ) -> None: self.config = config self.timeout_s = timeout_s self.urlopen = urlopen or urllib.request.urlopen self.loaded = False def load(self) -> None: self.config.require_llm_credentials() self.loaded = True def synthesize(self, text: str) -> AudioSegment: if not self.loaded: raise ProviderError( ErrorCode.TTS_SYNTHESIS_FAILED, "cloud TTS provider is not loaded", False, "newapi-tts", "tts", ) clean = text.strip() if not clean: raise ProviderError( ErrorCode.TTS_EMPTY_AUDIO, "cannot synthesize empty text", True, "newapi-tts", "tts", ) payload = { "model": self.config.tts_model, "messages": [{"role": "assistant", "content": clean}], "audio": {"format": "wav", "voice": self.config.tts_voice}, "stream": False, } request = urllib.request.Request( self.config.api_url("/v1/chat/completions"), data=json.dumps(payload).encode("utf-8"), headers={ "Authorization": f"Bearer {self.config.llm_api_key}", "Content-Type": "application/json", }, method="POST", ) try: with self.urlopen(request, timeout=self.timeout_s) as response: payload = json.loads(response.read().decode("utf-8")) except urllib.error.HTTPError as exc: raise ProviderError( ErrorCode.TTS_SYNTHESIS_FAILED, f"cloud TTS HTTP error {exc.code}", exc.code >= 500, "newapi-tts", "tts", ) from exc except (urllib.error.URLError, TimeoutError, socket.timeout, json.JSONDecodeError) as exc: raise ProviderError( ErrorCode.TTS_SYNTHESIS_FAILED, f"cloud TTS request failed: {exc}", True, "newapi-tts", "tts", ) from exc choices = payload.get("choices") or [] message = (choices[0].get("message") if choices else {}) or {} audio = message.get("audio") or {} encoded = audio.get("data") if not encoded: raise ProviderError( ErrorCode.TTS_EMPTY_AUDIO, "cloud TTS returned empty audio", True, "newapi-tts", "tts", ) try: data = base64.b64decode(encoded) except (TypeError, ValueError) as exc: raise ProviderError( ErrorCode.TTS_SYNTHESIS_FAILED, f"cloud TTS returned invalid base64 audio: {exc}", True, "newapi-tts", "tts", ) from exc return AudioSegment(data, 24000, 1, 0, max(120, len(clean) * 45), {"text": clean, "format": "wav"})