save code

services/tts.py

@@ -1,10 +1,13 @@
 from __future__ import annotations
 
 import logging
+import re
 from typing import Optional
 
 import numpy as np
 
+from config import settings
+
 logger = logging.getLogger(__name__)
 
 
@@ -28,12 +31,46 @@ class TTSService:
             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="zf_xiaoxiao"):
-                chunks.append(audio)
+            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