feat: rewrite /api/mc/tts with full ttstoy pipeline

- SFX emoji interleaving (inline 🎉💀🔥 etc.) - Voice switching via [mode|voice] tags - Chatterbox turbo tokens ([laugh], [cough]) passthrough - Multi-segment audio concatenation via ffmpeg - Supports all modes: chatterbox, clone, dectalk, morshu, vox
2026-06-09 11:19:39 -05:00
parent 47afd9c7c4
commit 7c0af41670
1 changed files with 196 additions and 36 deletions
@@ -651,68 +651,228 @@ def api_mc_voices():

@app.route("/api/mc/tts", methods=["POST"])
 def api_mc_tts():
-    """Generate TTS audio. Returns WAV bytes.
-    JSON body: {"text": "...", "mode": "chatterbox|dectalk|morshu", "voice": "Emily.wav"}
+    """Generate TTS audio with full ttstoy features. Returns WAV bytes.
+    JSON body: {"text": "...", "mode": "chatterbox", "voice": "Emily.wav"}
+
+    Supports:
+    - SFX emojis inline (🎉, 💀, etc.) → interleaved SFX audio
+    - Voice switching: [dectalk|Harry]hello [chatterbox|Emily]world
+    - Chatterbox turbo tokens: [laugh], [cough], etc. (passed through to model)
    """
+    import io
+    import re
+    import wave
+    import random
+    import subprocess as _sp
+
    data = request.get_json(force=True) if request.is_json else {}
    text = (data.get("text") or "").strip()
    if not text:
        return jsonify({"error": "Missing text"}), 400

-    mode = data.get("mode", "chatterbox")
-    voice = data.get("voice", "Emily.wav")
+    default_mode = data.get("mode", "chatterbox")
+    default_voice = data.get("voice", "Emily.wav")
+    gcfg = get_global_config()

+    # --- SFX setup ---
+    sfx_root = Path(__file__).resolve().parent.parent / "sfx"
+    emoji_map_path = sfx_root / "emoji_map.json"
    try:
-        if mode == "chatterbox":
+        emoji_map = json.load(open(emoji_map_path)) if emoji_map_path.exists() else {}
+    except Exception:
+        emoji_map = {}
+
+    EMOJI_RE = re.compile(
+        r"(<a?:\w+:\d+>|:[A-Za-z0-9_]+:|[\U0001F300-\U0001FAFF\U00002600-\U000026FF\U00002700-\U000027BF][\uFE0E\uFE0F]?(?:\u200D[\U0001F300-\U0001FAFF\U00002600-\U000026FF\U00002700-\U000027BF][\uFE0E\uFE0F]?)*)"
+    )
+    VOICE_SWITCH_RE = re.compile(r"\[([a-zA-Z]+)(?:\|([^\]]+))?\]")
+    VALID_MODES = {"chatterbox", "clone", "dectalk", "morshu", "vox"}
+    # Turbo tokens — these look like voice switches but should pass through to chatterbox
+    TURBO_TOKENS = {"laugh", "chuckle", "sigh", "gasp", "cough", "cry", "groan", "yawn", "sniff"}
+
+    def normalize_trigger(token):
+        if token.startswith("<") and token.endswith(">"):
+            parts = token.strip("<>").split(":")
+            if len(parts) == 3:
+                return f":{parts[1]}:"
+        return token
+
+    def get_sfx_file(trigger):
+        norm = normalize_trigger(trigger)
+        folder_name = emoji_map.get(norm)
+        if not folder_name:
+            return None
+        folder = sfx_root / folder_name
+        files = sorted(folder.glob("*.mp3")) + sorted(folder.glob("*.wav")) + sorted(folder.glob("*.ogg"))
+        return random.choice(files) if files else None
+
+    def generate_for_mode(seg_text, mode, voice):
+        """Generate WAV bytes for a text segment with given mode/voice."""
+        if mode == "chatterbox" or mode == "clone":
            chatterbox_url = get_chatterbox_url()
            payload = {
-                "text": text,
-                "voice_mode": "predefined",
-                "predefined_voice_id": voice,
+                "text": seg_text,
+                "voice_mode": "clone" if mode == "clone" else "predefined",
+                ("reference_audio_filename" if mode == "clone" else "predefined_voice_id"): voice or default_voice,
                "output_format": "wav",
                "split_text": False,
            }
            r = requests.post(f"{chatterbox_url}/tts", json=payload, timeout=120)
            r.raise_for_status()
-            return r.content, 200, {"Content-Type": "audio/wav"}
-
-        elif mode == "clone":
-            chatterbox_url = get_chatterbox_url()
-            payload = {
-                "text": text,
-                "voice_mode": "clone",
-                "reference_audio_filename": voice,
-                "output_format": "wav",
-                "split_text": False,
-            }
-            r = requests.post(f"{chatterbox_url}/tts", json=payload, timeout=120)
-            r.raise_for_status()
-            return r.content, 200, {"Content-Type": "audio/wav"}
-
+            return r.content
        elif mode == "dectalk":
            dectalk_url = get_dectalk_url()
-            r = requests.get(f"{dectalk_url}/say", params={"text": text}, timeout=30)
+            # Resolve voice name to DECTalk command
+            dectalk_voices = {"paul": "[:np]", "betty": "[:nb]", "harry": "[:nh]", "frank": "[:nf]",
+                            "dennis": "[:nd]", "kit": "[:nk]", "ursula": "[:nu]", "rita": "[:nr]", "wendy": "[:nw]"}
+            prefix = dectalk_voices.get((voice or "").lower(), "")
+            r = requests.get(f"{dectalk_url}/say", params={"text": f"{prefix}{seg_text}"}, timeout=30)
            r.raise_for_status()
-            return r.content, 200, {"Content-Type": "audio/wav"}
-
+            return r.content
        elif mode == "morshu":
-            r = requests.get("http://127.0.0.1:33002/say", params={"text": text}, timeout=60)
+            r = requests.get("http://127.0.0.1:33002/say", params={"text": seg_text}, timeout=60)
            r.raise_for_status()
-            return r.content, 200, {"Content-Type": "audio/wav"}
-
+            return r.content
        elif mode == "vox":
-            gcfg = get_global_config()
            pack = gcfg.get("vox_pack", "vox")
-            r = requests.get("http://127.0.0.1:33003/say", params={"text": text, "pack": pack}, timeout=60)
+            r = requests.get("http://127.0.0.1:33003/say", params={"text": seg_text, "pack": pack}, timeout=60)
            r.raise_for_status()
-            return r.content, 200, {"Content-Type": "audio/wav"}
-
+            return r.content
        else:
-            return jsonify({"error": f"Unknown mode: {mode}"}), 400
+            raise RuntimeError(f"Unknown mode: {mode}")

-    except requests.exceptions.ConnectionError:
-        return jsonify({"error": f"{mode} server not reachable"}), 502
+    def sfx_to_wav(sfx_path):
+        """Convert any audio file to WAV bytes via ffmpeg."""
+        result = _sp.run(
+            ["ffmpeg", "-y", "-i", str(sfx_path), "-ac", "1", "-ar", "24000", "-f", "wav", "-"],
+            capture_output=True
+        )
+        if result.returncode != 0:
+            return None
+        return result.stdout
+
+    def ensure_wav_format(audio_bytes):
+        """Normalize audio bytes to consistent WAV format (mono 24kHz 16-bit) via ffmpeg."""
+        result = _sp.run(
+            ["ffmpeg", "-y", "-i", "pipe:0", "-ac", "1", "-ar", "24000", "-sample_fmt", "s16", "-f", "wav", "pipe:1"],
+            input=audio_bytes, capture_output=True
+        )
+        return result.stdout if result.returncode == 0 else audio_bytes
+
+    def concat_wavs(wav_list):
+        """Concatenate multiple WAV byte arrays into one."""
+        if len(wav_list) == 1:
+            return wav_list[0]
+        # Use ffmpeg to concat
+        import tempfile, shutil
+        tmp_dir = tempfile.mkdtemp()
+        try:
+            paths = []
+            for i, wav_bytes in enumerate(wav_list):
+                p = os.path.join(tmp_dir, f"{i:03}.wav")
+                with open(p, "wb") as f:
+                    f.write(wav_bytes)
+                paths.append(p)
+            list_file = os.path.join(tmp_dir, "list.txt")
+            with open(list_file, "w") as f:
+                for p in paths:
+                    f.write(f"file '{p}'\n")
+            out_path = os.path.join(tmp_dir, "out.wav")
+            _sp.run(["ffmpeg", "-y", "-f", "concat", "-safe", "0", "-i", list_file,
+                     "-ac", "1", "-ar", "24000", "-sample_fmt", "s16", out_path],
+                    capture_output=True, check=True)
+            with open(out_path, "rb") as f:
+                return f.read()
+        finally:
+            shutil.rmtree(tmp_dir, ignore_errors=True)
+
+    try:
+        # Step 1: Split on voice switch tags (but NOT turbo tokens)
+        voice_segments = []
+        last_end = 0
+        for m in VOICE_SWITCH_RE.finditer(text):
+            tag_mode = m.group(1).lower()
+            tag_voice = (m.group(2) or "").strip()
+            # Skip turbo tokens — they stay in the text
+            if tag_mode in TURBO_TOKENS:
+                continue
+            if tag_mode not in VALID_MODES:
+                continue
+            before = text[last_end:m.start()]
+            if before:
+                if not voice_segments:
+                    voice_segments.append((default_mode, default_voice, before))
+                else:
+                    prev = voice_segments[-1]
+                    voice_segments[-1] = (prev[0], prev[1], prev[2] + before)
+            voice_segments.append((tag_mode, tag_voice or default_voice, ""))
+            last_end = m.end()
+        tail = text[last_end:]
+        if tail:
+            if not voice_segments:
+                voice_segments.append((default_mode, default_voice, tail))
+            else:
+                prev = voice_segments[-1]
+                voice_segments[-1] = (prev[0], prev[1], prev[2] + tail)
+        if not voice_segments:
+            voice_segments = [(default_mode, default_voice, text)]
+
+        # Step 2: For each voice segment, split on SFX emojis and generate
+        all_wav_parts = []
+        for seg_mode, seg_voice, seg_text in voice_segments:
+            seg_text = seg_text.strip()
+            if not seg_text:
+                continue
+
+            # Split segment into TTS and SFX parts
+            parts = []
+            cursor = 0
+            text_buf = []
+            for match in EMOJI_RE.finditer(seg_text):
+                if match.start() > cursor:
+                    text_buf.append(seg_text[cursor:match.start()])
+                token = match.group(0)
+                if normalize_trigger(token) in emoji_map:
+                    if text_buf:
+                        parts.append(("tts", "".join(text_buf)))
+                        text_buf = []
+                    parts.append(("sfx", token))
+                else:
+                    text_buf.append(token)
+                cursor = match.end()
+            if cursor < len(seg_text):
+                text_buf.append(seg_text[cursor:])
+            if text_buf:
+                parts.append(("tts", "".join(text_buf)))
+
+            if not parts:
+                parts = [("tts", seg_text)]
+
+            # Generate audio for each part
+            for kind, value in parts:
+                if kind == "tts":
+                    value = value.strip()
+                    if not value:
+                        continue
+                    wav = generate_for_mode(value, seg_mode, seg_voice)
+                    all_wav_parts.append(ensure_wav_format(wav))
+                else:
+                    sfx_file = get_sfx_file(value)
+                    if sfx_file:
+                        wav = sfx_to_wav(sfx_file)
+                        if wav:
+                            all_wav_parts.append(wav)
+
+        if not all_wav_parts:
+            return jsonify({"error": "No audio generated"}), 400
+
+        final_wav = concat_wavs(all_wav_parts) if len(all_wav_parts) > 1 else all_wav_parts[0]
+        return final_wav, 200, {"Content-Type": "audio/wav"}
+
+    except requests.exceptions.ConnectionError as e:
+        return jsonify({"error": f"TTS server not reachable: {e}"}), 502
    except Exception as e:
+        log.exception(f"mc/tts error: {e}")
        return jsonify({"error": str(e)}), 500