echo-core/tools/tts.py

#!/usr/bin/env python3
"""Text-to-speech — rutează pe engine (Supertonic sau pocket-tts) pe baza catalogului de voci.

CLI:
    python3 tools/tts.py --text "Salut Marius" [--voice M1] [--lang ro]
    → stdout: {"ok": true, "path": "/tmp/echo-tts-xxx.wav", "size_bytes": 12345, "engine_used": "supertonic"}
    → stdout: {"ok": false, "error": "..."}

Module:
    from tools.tts import synthesize
    result = synthesize("text", voice="M1", lang="ro")
"""

import argparse
import json
import sys
import tempfile
from pathlib import Path

import httpx

REPO_ROOT = Path(__file__).resolve().parent.parent
if str(REPO_ROOT) not in sys.path:
    sys.path.insert(0, str(REPO_ROOT))

SUPERTONIC_URL = "http://127.0.0.1:7788"
VOICES = {"M1", "M2", "M3", "M4", "M5", "F1", "F2", "F3", "F4", "F5"}
DEFAULT_VOICE = "M2"
DEFAULT_LANG = "ro"
DEFAULT_ENGINE = "supertonic"

_VOICES_CATALOG_PATH = REPO_ROOT / "tts_voices.json"

# Punctuation Supertonic synthesis rejects with HTTP 500 (Romanian curly quotes,
# smart dashes, ellipsis, angle quotes). Mapped to ASCII so a stray „foo" in
# any caller's text doesn't kill the whole request.
_TTS_PUNCT_MAP = {
    '„': '"', '“': '"', '”': '"',
    '‘': "'", '’': "'", '‚': "'",
    '«': '"', '»': '"',
    '–': '-', '—': '-',
    '…': '...',
}


# Supertonic ONNX model hard limit: inputs longer than this trigger
# Mul node dimension mismatches in attention layers.
_MAX_TTS_CHARS = 400


class _PocketTTSUnavailable(Exception):
    """pocket-tts server e nereachable sau eșuează server-side (5xx) — caller-ul trebuie să facă fallback pe Supertonic."""


def sanitize_for_supertonic(text: str) -> str:
    """Replace Unicode punctuation and strip chars that crash Supertonic's ONNX model."""
    for src, dst in _TTS_PUNCT_MAP.items():
        text = text.replace(src, dst)
    # Strip emoji and high-codepoint chars (keep ASCII printable + Latin/Romanian diacritice)
    cleaned = []
    for ch in text:
        cp = ord(ch)
        if (32 <= cp <= 126) or (128 <= cp <= 591):
            cleaned.append(ch)
        else:
            cleaned.append(' ')
    text = ' '.join(''.join(cleaned).split())
    if len(text) > _MAX_TTS_CHARS:
        text = text[:_MAX_TTS_CHARS]
    return text


def _load_voice_catalog() -> dict:
    from src.jsonlock import read_locked

    try:
        return read_locked(str(_VOICES_CATALOG_PATH))
    except FileNotFoundError:
        return {}


def _catalog_entry(voice: str) -> dict | None:
    catalog = _load_voice_catalog()
    return catalog.get(voice) or catalog.get(voice.upper())


def _config_get(key: str, default: str) -> str:
    from src.config import Config

    return Config().get(key, default)


def _write_temp_wav(content: bytes) -> dict:
    try:
        fd, path = tempfile.mkstemp(prefix="echo-tts-", suffix=".wav")
        with open(fd, "wb") as f:
            f.write(content)
        return {"ok": True, "path": path, "size_bytes": len(content)}
    except Exception as e:
        return {"ok": False, "error": f"Scriere fișier: {e}"}


def _synthesize_supertonic(text: str, voice: str = DEFAULT_VOICE, lang: str = DEFAULT_LANG) -> dict:
    """Call Supertonic server and save audio to a temp WAV file.

    Returns:
        {"ok": True, "path": "/tmp/echo-tts-xxx.wav", "size_bytes": N}
        {"ok": False, "error": "mesaj eroare"}
    """
    text = sanitize_for_supertonic(text)

    voice = voice.upper()
    if voice not in VOICES:
        voice = DEFAULT_VOICE

    try:
        resp = httpx.post(
            f"{SUPERTONIC_URL}/v1/audio/speech",
            json={
                "model": "supertonic-3",
                "input": text,
                "voice": voice,
                "response_format": "wav",
                "lang": lang,
            },
            timeout=60.0,
        )
        resp.raise_for_status()
    except httpx.ConnectError:
        return {
            "ok": False,
            "error": (
                "Serverul Supertonic nu rulează pe :7788. "
                "Pornește cu: systemctl --user start supertonic-tts"
            ),
        }
    except httpx.HTTPStatusError as e:
        body = e.response.text[:300]
        # Fallback: dacă lang=ro eșuează, încearcă na (language-agnostic)
        if lang != "na":
            return _synthesize_supertonic(text, voice=voice, lang="na")
        return {"ok": False, "error": f"HTTP {e.response.status_code}: {body}"}
    except Exception as e:
        return {"ok": False, "error": str(e)}

    # Salvează în fișier temp
    return _write_temp_wav(resp.content)


def _synthesize_pockettts(text: str, entry: dict) -> dict:
    """POST multipart la serverul pocket-tts (:7789/tts). Ignoră lang (modelul e monolingv per deployment).

    `entry` e intrarea din tts_voices.json (deja rezolvată de caller — evită un al doilea read+lock
    al catalogului). state_path (voce clonată) -> încarcă fișierul ca voice_wav. voice_url (voce
    predefinită, ex. 'alba') -> trimite voice_url. Nici unul -> nu trimite voice_wav/voice_url
    (serverul alege default-ul limbii).

    Raises:
        _PocketTTSUnavailable: eșec tehnic (connect error/timeout/5xx) — caller-ul face fallback.
    """
    state_path = entry.get("state_path")
    voice_url = entry.get("voice_url")

    data = {"text": text}
    files = None
    if state_path:
        full_path = REPO_ROOT / state_path
        if not full_path.exists():
            return {"ok": False, "error": f"Fișier voce lipsă: {state_path}"}
        files = {"voice_wav": (full_path.name, full_path.read_bytes())}
    elif voice_url:
        data["voice_url"] = voice_url

    try:
        resp = httpx.post(
            f"{_config_get('tts.pockettts_url', 'http://127.0.0.1:7789')}/tts",
            data=data,
            files=files,
            timeout=60.0,
        )
    except (httpx.ConnectError, httpx.TimeoutException) as e:
        raise _PocketTTSUnavailable(str(e)) from e

    if resp.status_code >= 500:
        raise _PocketTTSUnavailable(f"HTTP {resp.status_code}: {resp.text[:300]}")
    if resp.status_code >= 400:
        return {"ok": False, "error": f"HTTP {resp.status_code}: {resp.text[:300]}"}

    return _write_temp_wav(resp.content)


def synthesize(text: str, voice: str = DEFAULT_VOICE, lang: str = DEFAULT_LANG) -> dict:
    """Rutează sinteza pe engine-ul rezolvat din catalog (fallback la tts.default_engine).

    Returns:
        {"ok": True, "path": "/tmp/echo-tts-xxx.wav", "size_bytes": N, "engine_used": "pockettts"}
        {"ok": False, "error": "mesaj eroare", "engine_used": "..."}
    """
    if not text or not text.strip():
        return {"ok": False, "error": "Text gol."}

    entry = _catalog_entry(voice) or {}
    engine = entry.get("engine") or _config_get("tts.default_engine", DEFAULT_ENGINE)

    engine_used = "supertonic"
    if engine == "pockettts":
        try:
            result = _synthesize_pockettts(text, entry)
            engine_used = "pockettts"
        except _PocketTTSUnavailable:
            result = _synthesize_supertonic(text, voice=DEFAULT_VOICE, lang=DEFAULT_LANG)
    else:
        result = _synthesize_supertonic(text, voice=voice, lang=lang)

    result["engine_used"] = engine_used
    return result


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Echo Core TTS CLI")
    parser.add_argument("--text", required=True, help="Text de convertit în audio")
    parser.add_argument(
        "--voice", default=DEFAULT_VOICE,
        help="Voce din tts_voices.json (ex: M1-M5, F1-F5, 'Marius 1', 'alba'). Default: M2"
    )
    parser.add_argument(
        "--lang", default=DEFAULT_LANG,
        help="Limbă (ro, en, na) — folosită doar pe calea Supertonic. Default: ro."
    )
    args = parser.parse_args()

    result = synthesize(args.text, voice=args.voice, lang=args.lang)
    print(json.dumps(result, ensure_ascii=False))
    sys.exit(0 if result.get("ok") else 1)