tts-voice-creator-clone-and-design-2/core/routing.py

"""TTS routing rules: load/save/resolve routes, language detection, routing log."""
from __future__ import annotations

import json
import re

from fastapi import Request

from core.constants import TTS_ROUTES_FILE, _routing_log_add

# ── Route token / backend validation ─────────────────────────────────────────

_ROUTE_LANGS = {"AUTO", "*", "EN", "DE", "FR", "ES", "IT", "PT", "NL", "PL"}
_ROUTE_BACKENDS = {"voice_clone", "streaming", "voice_design", "nvidia_magpie", "nvidia_zeroshot", "nvidia_flow"}


def _clean_route_token(value: str, default: str = "*") -> str:
    value = str(value or "").strip()
    if not value:
        return default
    value = re.sub(r"[^A-Za-z0-9_\-\.\* ]+", "_", value)
    return value[:80] or default


def _clean_route_sound(value: str) -> str:
    value = str(value or "").strip()
    if not value:
        return ""
    value = re.sub(r"[^A-Za-z0-9_\-\.\* /]+", "_", value)
    return value[:240]


def _clean_route_backend(value: str) -> str:
    key = re.sub(r"[^a-z0-9]+", "_", str(value or "voice_clone").lower()).strip("_")
    aliases = {
        "clone": "voice_clone",
        "base": "voice_clone",
        "tts": "voice_clone",
        "standard": "voice_clone",
        "voiceclone": "voice_clone",
        "voice_clone_base": "voice_clone",
        "stream": "streaming",
        "tts_streaming": "streaming",
        "voice_design": "voice_design",
        "voicedesign": "voice_design",
        "design": "voice_design",
        "nvidia": "nvidia_magpie",
        "magpie": "nvidia_magpie",
        "nvidia_tts": "nvidia_magpie",
        "nvidia_magpie_tts": "nvidia_magpie",
        "nvidia_clone": "nvidia_zeroshot",
        "nvidia_zeroshot_tts": "nvidia_zeroshot",
        "magpie_zeroshot": "nvidia_zeroshot",
        "zeroshot": "nvidia_zeroshot",
        "zero_shot": "nvidia_zeroshot",
        "nvidia_flow_tts": "nvidia_flow",
        "magpie_flow": "nvidia_flow",
        "flow": "nvidia_flow",
    }
    key = aliases.get(key, key)
    return key if key in _ROUTE_BACKENDS else "voice_clone"


def _normalize_route(rule: dict, idx: int = 0) -> dict:
    lang = str(rule.get("language", "*") or "*").strip().upper()
    if lang not in _ROUTE_LANGS:
        lang = "*"
    output_voice = _clean_route_token(rule.get("output_voice", ""), "")
    return {
        "id": _clean_route_token(rule.get("id", f"route_{idx+1}"), f"route_{idx+1}"),
        "enabled": bool(rule.get("enabled", True)),
        "app": _clean_route_token(rule.get("app", "Open WebUI"), "Open WebUI"),
        "input_voice": _clean_route_token(rule.get("input_voice", "default"), "default"),
        "language": lang,
        "backend": _clean_route_backend(rule.get("backend", "voice_clone")),
        "output_voice": output_voice,
        "before_sound": _clean_route_sound(rule.get("before_sound", "")),
        "after_sound": _clean_route_sound(rule.get("after_sound", "")),
    }


_routes_cache: list[dict] | None = None
_routes_cache_mtime: float = -1.0


def _load_tts_routes() -> list[dict]:
    global _routes_cache, _routes_cache_mtime
    mtime = TTS_ROUTES_FILE.stat().st_mtime if TTS_ROUTES_FILE.exists() else 0.0
    if _routes_cache is not None and mtime == _routes_cache_mtime:
        return list(_routes_cache)
    if not TTS_ROUTES_FILE.exists():
        _routes_cache = []
        _routes_cache_mtime = 0.0
        return []
    try:
        raw = json.loads(TTS_ROUTES_FILE.read_text())
        routes = raw.get("routes", raw) if isinstance(raw, dict) else raw
        if not isinstance(routes, list):
            result: list[dict] = []
        else:
            result = [_normalize_route(r, i) for i, r in enumerate(routes) if isinstance(r, dict)]
    except Exception:
        result = []
    _routes_cache = result
    _routes_cache_mtime = mtime
    return list(result)


def _save_tts_routes(routes: list[dict]) -> None:
    global _routes_cache, _routes_cache_mtime
    TTS_ROUTES_FILE.parent.mkdir(parents=True, exist_ok=True)
    clean = [_normalize_route(r, i) for i, r in enumerate(routes)]
    TTS_ROUTES_FILE.write_text(json.dumps({"routes": clean}, indent=2))
    _routes_cache = clean
    _routes_cache_mtime = TTS_ROUTES_FILE.stat().st_mtime


# ── App name helpers ──────────────────────────────────────────────────────────

def _app_key(value: str) -> str:
    return re.sub(r"[^a-z0-9]+", "", str(value or "").lower())


def _canonical_app_name(value: str) -> str:
    key = _app_key(value)
    if key in {"openwebui", "openwebuiapp"}:
        return "Open WebUI"
    if key == "sillytavern":
        return "SillyTavern"
    if key in {"homeassistant", "ha"}:
        return "Home Assistant"
    return str(value or "").strip() or "Open WebUI"


def _request_app_name(request: Request) -> str:
    explicit = request.headers.get("x-tts-app") or request.headers.get("x-client-app")
    if explicit:
        return _canonical_app_name(explicit)
    header = (
        request.headers.get("x-openwebui-app")
        or request.headers.get("x-openwebui-user-name")
        or request.headers.get("referer")
        or request.headers.get("origin")
        or request.headers.get("user-agent")
        or ""
    )
    h = _app_key(header)
    if "openwebui" in h:
        return "Open WebUI"
    if "sillytavern" in h:
        return "SillyTavern"
    if "homeassistant" in h:
        return "Home Assistant"
    return "Open WebUI"


# ── Language detection ────────────────────────────────────────────────────────

def _detect_text_language(text: str) -> str:
    low = f" {text.lower()} "
    if re.search(r"[äöüß]", low) or re.search(r"\b(und|der|die|das|ich|nicht|mit|ist|ein|eine|auf|für)\b", low):
        return "DE"
    if re.search(r"[éèêàçùœ]", low) or re.search(r"\b(et|le|la|les|des|une|avec|pour|est|pas|que)\b", low):
        return "FR"
    if re.search(r"[áéíóúñ¿¡]", low) or re.search(r"\b(el|la|los|las|una|con|para|que|pero|está|hola)\b", low):
        return "ES"
    if re.search(r"\b(il|lo|gli|una|con|per|che|ciao|grazie|sono|della)\b", low):
        return "IT"
    if re.search(r"[ãõç]", low) or re.search(r"\b(com|para|uma|que|não|está|obrigado)\b", low):
        return "PT"
    if re.search(r"\b(het|een|niet|met|voor|zijn|maar|dank|goede)\b", low):
        return "NL"
    if re.search(r"[ąćęłńóśźż]", low) or re.search(r"\b(jest|nie|tak|dla|oraz|dzień|dziękuję)\b", low):
        return "PL"
    return "EN"


# ── Route matching ────────────────────────────────────────────────────────────

def _route_specificity(rule: dict, app: str, voice: str, lang: str) -> tuple[int, int, int, int] | None:
    if not rule.get("enabled", True) or not rule.get("output_voice"):
        return None
    r_app = str(rule.get("app", "*"))
    r_voice = str(rule.get("input_voice", "*"))
    r_lang = str(rule.get("language", "*")).upper()
    app_ok = r_app == "*" or _app_key(r_app) == _app_key(app)
    voice_ok = r_voice == "*" or r_voice.lower() == voice.lower()
    lang_ok = r_lang in {"*", "AUTO"} or r_lang == lang
    if not (app_ok and voice_ok and lang_ok):
        return None
    return (
        1 if r_app != "*" else 0,
        1 if r_voice != "*" else 0,
        1 if r_lang not in {"*", "AUTO"} else 0,
        0,
    )


def _resolve_tts_route(app: str, voice: str, text: str) -> tuple[str, dict | None]:
    lang = _detect_text_language(text)
    best: tuple[tuple[int, int, int, int], dict] | None = None
    for idx, rule in enumerate(_load_tts_routes()):
        spec = _route_specificity(rule, app, voice, lang)
        if spec is None:
            continue
        spec = (spec[0], spec[1], spec[2], -idx)
        if best is None or spec > best[0]:
            best = (spec, rule)
    if not best:
        return voice, None
    routed = dict(best[1])
    routed["detected_language"] = lang
    routed["requested_voice"] = voice
    routed["app"] = app
    return str(routed["output_voice"]), routed


def _route_backend(route: dict | None, voice: str = "") -> str:
    from core.presets import _resolve_virtual_voice
    if _resolve_virtual_voice(voice):
        return "voice_design"
    return _clean_route_backend((route or {}).get("backend", "voice_clone"))


# ── Routing log request helper ────────────────────────────────────────────────

def _routing_log_request(
    request: Request,
    *,
    status: str,
    app: str,
    requested_voice: str,
    routed_voice: str,
    backend: str,
    route: dict | None,
    response_format: str,
    text: str,
    **extra,
) -> None:
    route = route or {}
    _routing_log_add(
        kind="proxy",
        status=status,
        app=app,
        requested_voice=requested_voice,
        routed_voice=routed_voice,
        backend=backend,
        language=str(route.get("detected_language", "")) or (_detect_text_language(text) if text else ""),
        matched=bool(route),
        route_id=str(route.get("id", "")),
        response_format=response_format,
        text_preview=text[:160],
        client=request.client.host if request.client else "",
        user_agent=str(request.headers.get("user-agent", ""))[:160],
        **extra,
    )