83 lines
2.7 KiB
Python
83 lines
2.7 KiB
Python
from __future__ import annotations
|
|
|
|
import logging
|
|
import re
|
|
from typing import Optional
|
|
|
|
import numpy as np
|
|
|
|
from config import settings
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class TTSService:
|
|
def __init__(self) -> None:
|
|
self._pipeline = None
|
|
self._ready = False
|
|
self._attempted = False
|
|
self._init_error: Optional[str] = None
|
|
|
|
def _ensure_loaded(self) -> None:
|
|
if self._ready or self._attempted:
|
|
return
|
|
self._attempted = True
|
|
try:
|
|
import kokoro
|
|
|
|
self._pipeline = kokoro.KPipeline(lang_code="z")
|
|
self._ready = True
|
|
except Exception as exc: # pragma: no cover
|
|
self._init_error = str(exc)
|
|
logger.warning("TTS fallback mode: %s", exc)
|
|
|
|
@staticmethod
|
|
def _normalize_text(text: str) -> str:
|
|
cleaned = (text or "").strip()
|
|
if not cleaned:
|
|
return ""
|
|
cleaned = cleaned.replace("\r", "\n")
|
|
cleaned = re.sub(r"[\t\f\v]+", " ", cleaned)
|
|
cleaned = re.sub(r"\s*([，。！？；：、,.!?;:])\s*", r"\1", cleaned)
|
|
cleaned = re.sub(r"(?<=[\u4e00-\u9fff])\s+(?=[\u4e00-\u9fff])", "", cleaned)
|
|
cleaned = re.sub(r"\n{2,}", "\n", cleaned)
|
|
cleaned = re.sub(r"\s{2,}", " ", cleaned)
|
|
return cleaned.strip()
|
|
|
|
@staticmethod
|
|
def _apply_edge_fade(audio: np.ndarray, sr: int) -> np.ndarray:
|
|
if audio.size == 0:
|
|
return audio
|
|
fade_samples = max(1, min(int(sr * max(settings.tts_fade_ms, 0) / 1000.0), audio.shape[0] // 8))
|
|
if fade_samples <= 1:
|
|
return audio
|
|
out = np.array(audio, copy=True)
|
|
ramp = np.linspace(0.0, 1.0, fade_samples, dtype=np.float32)
|
|
out[:fade_samples] *= ramp
|
|
out[-fade_samples:] *= ramp[::-1]
|
|
return out
|
|
|
|
async def synthesize(self, text: str) -> tuple[np.ndarray, int]:
|
|
text = self._normalize_text(text)
|
|
if not text:
|
|
return np.zeros(1, dtype=np.float32), 24000
|
|
self._ensure_loaded()
|
|
if self._ready and self._pipeline is not None:
|
|
chunks = []
|
|
for _, _, audio in self._pipeline(
|
|
text,
|
|
voice=settings.tts_voice,
|
|
speed=max(0.8, min(1.1, settings.tts_speed)),
|
|
split_pattern=r"\n+",
|
|
):
|
|
chunks.append(self._apply_edge_fade(np.asarray(audio, dtype=np.float32), 24000))
|
|
if chunks:
|
|
return np.concatenate(chunks), 24000
|
|
|
|
# 0.5s silence fallback for flow verification
|
|
return np.zeros(12000, dtype=np.float32), 24000
|
|
|
|
@property
|
|
def health(self) -> dict:
|
|
return {"ready": self._ready, "attempted": self._attempted, "error": self._init_error}
|