"""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 from core.database import state_get, state_put, state_updated # ── 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 # \w (Unicode) rather than an A-Za-z whitelist: the old pattern replaced # every non-ASCII letter with "_", so an app called "Hörbuch" was stored as # "H_rbuch" and a Cyrillic, Greek or Filipino voice name was reduced to # underscores — the route then never matched the name the user had typed. # Punctuation and control characters are still stripped. value = re.sub(r"[^\w\-\.\* ]+", "_", value, flags=re.UNICODE) return value[:80] or default def _clean_route_sound(value: str) -> str: value = str(value or "").strip() if not value: return "" # Same Unicode fix as _clean_route_token; "/" stays allowed because sound # values are paths under the voices folder. value = re.sub(r"[^\w\-\.\* /]+", "_", value, flags=re.UNICODE) return value[:240] def _clean_route_backend(value: str) -> str: if str(value or "").startswith("custom:"): return str(value) 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: # Any ISO-style language code is accepted, not just the eight the built-in # text detector can itself produce. The old fixed whitelist silently reset # anything else to "*" on save — confirmed live: choosing Filipino (TL) and # pressing Save snapped the row back to "Any", with no error to explain it. # The value is only ever compared for equality against a detected code or # bypassed entirely for voice-specific rules, so a wider set is harmless. lang = str(rule.get("language", "*") or "*").strip().upper() if lang not in _ROUTE_LANGS and not re.fullmatch(r"[A-Z]{2,3}(?:-[A-Z0-9]{2,4})?", lang): lang = "*" output_voice = _clean_route_token(rule.get("output_voice", ""), "") try: speed = float(rule.get("speed", 1.0) or 1.0) except (TypeError, ValueError): speed = 1.0 # Clamped to ffmpeg's atempo range for a single filter pass (0.5-2.0) — # see _apply_route_sounds, which is what actually applies this. speed = round(max(0.5, min(2.0, speed)), 2) 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", "")), "speed": speed, } _routes_cache: list[dict] | None = None _routes_cache_mtime: float = -1.0 _routes_cache_db_updated: str = "" def _load_tts_routes() -> list[dict]: global _routes_cache, _routes_cache_mtime, _routes_cache_db_updated mtime = TTS_ROUTES_FILE.stat().st_mtime if TTS_ROUTES_FILE.exists() else 0.0 db_updated = state_updated("tts_routes") if ( _routes_cache is not None and mtime == _routes_cache_mtime and db_updated == _routes_cache_db_updated ): return list(_routes_cache) result: list[dict] = [] try: if TTS_ROUTES_FILE.exists(): raw = json.loads(TTS_ROUTES_FILE.read_text(encoding="utf-8")) routes = raw.get("routes", raw) if isinstance(raw, dict) else raw if isinstance(routes, list): result = [_normalize_route(r, i) for i, r in enumerate(routes) if isinstance(r, dict)] except Exception: result = [] try: db_routes = state_get("tts_routes", None) if isinstance(db_routes, dict): db_routes = db_routes.get("routes") if isinstance(db_routes, list): routes = db_routes result = [_normalize_route(r, i) for i, r in enumerate(routes) if isinstance(r, dict)] except Exception: pass if not db_updated and result: try: state_put("tts_routes", {"routes": result}) db_updated = state_updated("tts_routes") except Exception: pass _routes_cache = result _routes_cache_mtime = mtime _routes_cache_db_updated = db_updated return list(result) def _save_tts_routes(routes: list[dict]) -> None: global _routes_cache, _routes_cache_mtime, _routes_cache_db_updated TTS_ROUTES_FILE.parent.mkdir(parents=True, exist_ok=True) clean = [_normalize_route(r, i) for i, r in enumerate(routes)] state_put("tts_routes", {"routes": clean}) TTS_ROUTES_FILE.write_text(json.dumps({"routes": clean}, indent=2, ensure_ascii=False), encoding="utf-8") _routes_cache = clean _routes_cache_mtime = TTS_ROUTES_FILE.stat().st_mtime _routes_cache_db_updated = state_updated("tts_routes") # ── 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" # Regional variants share a language. Voices are NAMED with the country code so # an accent is visible at a glance (AU_F_..., US_M_..., GB_N_...), but routing # and detection must treat them all as the one language — text alone cannot tell # Australian from Irish English, so a route labelled AU has to match detected EN. # # Note the namespace clash this creates: these are COUNTRY codes, while the # language list uses LANGUAGE codes, and "CA" means Canada here but Catalan in # ISO 639-1. Canadian English is the realistic case for this library, so CA maps # to EN and Catalan is listed as CAT in the pickers. _LANG_VARIANTS = { # ONLY bare codes that are not themselves ISO 639-1 languages. Regional # variants are BCP 47 subtags now (EN-GB, PT-BR), so this table exists purely # for backward compatibility with codes already saved in routes and baked # into voice filenames. # # Deliberately ABSENT because each collides with a real ISO 639-1 language, # which is exactly the namespace clash this rework removes: # CA=Catalan AR=Arabic IE=Interlingue BR=Breton CH=Chamorro # ZA=Zhuang UK=Ukrainian SG=Sango CO=Corsican # Brazilian Portuguese is PT-BR, Canadian English EN-CA, Swiss German DE-CH. "GB": "EN", "US": "EN", "AU": "EN", "NZ": "EN", "JM": "EN", "SCO": "EN", "WLS": "EN", "IR": "EN", "AT": "DE", "MX": "ES", "QC": "FR", "PH": "TL", "FIL": "TL", } def _canonical_lang(value: str) -> str: """Collapse a regional variant onto its base language code. "EN-GB" -> "EN", "PT-BR" -> "PT". Bare legacy country codes still resolve via _LANG_VARIANTS so previously saved routes and PH_* voices keep matching. """ v = str(value or "").strip().upper() if "-" in v: return v.split("-", 1)[0] return _LANG_VARIANTS.get(v, v) # ── Language detection ──────────────────────────────────────────────────────── # Languages whose WRITING SYSTEM identifies them outright. Checked before any # word statistics, because a script match is far stronger evidence than # stopword frequency and needs no per-language vocabulary. This is also what # makes the detector extensible for free: adding a voice in Russian, Japanese # or Arabic needs no new word lists. _SCRIPT_LANGS = [ ("JA", r"[\u3040-\u309f\u30a0-\u30ff]"), # kana — check before Han, Japanese also uses Han ("KO", r"[\uac00-\ud7af\u1100-\u11ff]"), # hangul ("ZH", r"[\u4e00-\u9fff]"), # han ("RU", r"[\u0400-\u04ff]"), # cyrillic (RU is the common case; UK/BG share it) ("EL", r"[\u0370-\u03ff]"), ("HE", r"[\u0590-\u05ff]"), ("AR", r"[\u0600-\u06ff]"), ("HI", r"[\u0900-\u097f]"), # devanagari ("BN", r"[\u0980-\u09ff]"), ("TA", r"[\u0b80-\u0bff]"), ("TE", r"[\u0c00-\u0c7f]"), ("TH", r"[\u0e00-\u0e7f]"), ("KA", r"[\u10a0-\u10ff]"), ("HY", r"[\u0530-\u058f]"), ("AM", r"[\u1200-\u137f]"), ] def _detect_text_language(text: str) -> str: raw = str(text or "") # A non-Latin script settles the question on its own. for code, pattern in _SCRIPT_LANGS: if re.search(pattern, raw): return code words = re.findall(r"[^\W\d_]+", raw.lower(), re.UNICODE) if not words: return "EN" scores = { "EN": 0, "DE": 0, "FR": 0, "ES": 0, "IT": 0, "PT": 0, "NL": 0, "PL": 0, "TL": 0, "ID": 0, "TR": 0, "SV": 0, "DA": 0, "NO": 0, "FI": 0, "RO": 0, "CS": 0, "HU": 0, "VI": 0, "CA": 0, } low = raw.lower() # Diacritics are only strong evidence when they are (near) unique to one # language. ä/ö/ü are NOT: Turkish, Swedish, Finnish, Estonian and Hungarian # all use them, and scoring them +3 for German made "bugün nasılsın" and # "tack så mycket för hjälpen" both come back as German. They now give a # weak nudge, while genuinely distinctive marks keep the strong score. if "ß" in low: scores["DE"] += 3 # unique to German if re.search(r"[äöü]", low): scores["DE"] += 1 # shared with several languages if re.search(r"[ąćęłńśźż]", low): scores["PL"] += 3 if re.search(r"[ãõ]", low): scores["PT"] += 3 if re.search(r"[ñ¿¡]", low): scores["ES"] += 3 if "å" in low: for c in ("SV", "DA", "NO"): scores[c] += 3 if re.search(r"[ışğ]", low) or "ı" in low: scores["TR"] += 3 if re.search(r"[őű]", low): scores["HU"] += 3 if re.search(r"[řěůč]", low): scores["CS"] += 3 if re.search(r"[ăşţșț]", low): scores["RO"] += 3 if re.search(r"[ĉĝĥ]", low): pass stopwords = { "EN": {"the", "be", "to", "of", "and", "a", "in", "that", "have", "i", "it", "for", "not", "on", "with", "he", "as", "you", "do", "at", "this", "but", "his", "by", "from", "they", "we", "say", "her", "she", "or", "an", "will", "my", "one", "all", "would", "there", "their", "what", "so", "up", "out", "if", "about", "who", "get", "which", "go", "me", "when", "make", "can", "like", "time", "no", "just", "him", "know", "take", "people", "into", "year", "your", "good", "some", "could", "them", "see", "other", "than", "then", "now", "look", "only", "come", "its", "over", "think", "also", "back", "after", "use", "two", "how", "our", "work", "first", "well", "way", "even", "new", "want", "because", "any", "these", "give", "day", "most", "us", "hello", "hi", "yes", "thanks", "please"}, "DE": {"und", "der", "die", "das", "ich", "nicht", "mit", "ist", "ein", "eine", "auf", "für", "ja", "nein", "gut", "morgen", "hallo", "danke", "bitte", "wie", "was", "warum", "wer", "wo", "hier", "da", "dann", "wenn", "so", "nur", "auch", "aber", "oder", "als", "um", "zu", "von", "aus", "bei", "nach", "vor", "an", "im", "am", "über", "unter", "doch", "schon", "sehr", "viel", "mehr", "immer", "wieder", "heute", "jetzt", "machen", "tun", "sagen", "gehen", "kommen", "sehen", "wissen", "sind", "wir", "ihr", "sie", "ihnen", "mir", "mich", "dir", "dich", "uns", "euch"}, "FR": {"et", "le", "la", "les", "des", "une", "avec", "pour", "est", "pas", "que", "je", "tu", "il", "elle", "nous", "vous", "ils", "elles", "qui", "quoi", "quand", "où", "pourquoi", "comment", "oui", "non", "merci", "bonjour", "bien", "très", "tout", "plus", "moins", "dans", "sur", "sous", "devant", "derrière", "avant", "après", "ici", "là", "aujourd'hui", "maintenant", "faire", "dire", "aller", "venir", "voir", "savoir", "un", "en", "au", "aux", "ce", "ces", "se", "sa", "son", "ses"}, "ES": {"el", "la", "los", "las", "una", "con", "para", "que", "pero", "está", "hola", "y", "o", "no", "sí", "gracias", "por", "favor", "bien", "mal", "muy", "mucho", "poco", "más", "menos", "todo", "nada", "algo", "aquí", "allí", "ahora", "hoy", "mañana", "ayer", "siempre", "nunca", "hacer", "decir", "ir", "venir", "ver", "saber", "poder", "querer", "tener", "ser", "estar", "un", "en", "su", "sus", "te", "me", "se", "nos"}, "IT": {"il", "lo", "gli", "una", "con", "per", "che", "ciao", "grazie", "sono", "della", "e", "o", "non", "sì", "prego", "bene", "male", "molto", "poco", "più", "meno", "tutto", "niente", "qualcosa", "qui", "lì", "ora", "oggi", "domani", "ieri", "sempre", "mai", "fare", "dire", "andare", "venire", "vedere", "sapere", "potere", "volere", "avere", "essere", "un", "in", "su", "di", "da", "al", "ai", "mi", "ti", "si", "ci", "vi"}, # PT had the same gap as NL/PL — present in the score table, no word # list, so it could only ever win on the ã/õ diacritic. "PT": {"o", "a", "os", "as", "de", "do", "da", "em", "para", "com", "que", "não", "sim", "é", "são", "um", "uma", "mas", "ou", "porque", "eu", "tu", "ele", "ela", "nós", "eles", "obrigado", "obrigada", "olá", "bom", "dia", "boa", "noite", "agora", "hoje", "amanhã", "ontem", "muito", "pouco", "tudo", "nada", "aqui", "ali", "bem", "mal", "pela", "pelo", "ajuda", "fazer", "dizer", "ver", "saber", "poder", "ter", "estar"}, # NL and PL were in the score table from the start but never had word # lists — Dutch could only ever win by accident, and "Hallo, hoe gaat # het vandaag met je?" was scored as German off the shared "hallo". "NL": {"de", "het", "een", "en", "van", "is", "op", "dat", "die", "niet", "je", "ik", "jij", "hij", "zij", "wij", "met", "voor", "aan", "zijn", "hebben", "maar", "of", "omdat", "hallo", "dank", "bedankt", "goedemorgen", "ja", "nee", "hoe", "wat", "waar", "wanneer", "waarom", "nu", "vandaag", "morgen", "gisteren", "gaat", "heel", "veel", "graag"}, "PL": {"i", "w", "na", "z", "do", "nie", "tak", "jest", "to", "że", "się", "ale", "lub", "bo", "ja", "ty", "on", "ona", "my", "wy", "oni", "dziękuję", "dzień", "dobry", "cześć", "proszę", "teraz", "dzisiaj", "jutro", "wczoraj", "bardzo", "mam", "jak", "co", "gdzie", "kiedy", "dlaczego"}, # Filipino/Tagalog — the language of the library's PH voice, and the # reason this list needed widening at all. "TL": {"ang", "ng", "sa", "mga", "ay", "na", "at", "ko", "mo", "niya", "namin", "natin", "nila", "ako", "ikaw", "siya", "kami", "tayo", "kayo", "sila", "ito", "iyan", "iyon", "dito", "diyan", "doon", "hindi", "oo", "salamat", "kumusta", "magandang", "umaga", "gabi", "po", "ba", "kung", "para", "dahil", "pero", "kasi", "lang", "din", "rin", "may", "meron", "wala", "gusto", "ayaw", "pwede", "kailangan", "ngayon", "bukas", "kahapon", "maraming"}, "ID": {"yang", "dan", "di", "ke", "dari", "untuk", "dengan", "ini", "itu", "saya", "aku", "kamu", "dia", "kami", "kita", "mereka", "tidak", "ya", "adalah", "ada", "sudah", "belum", "akan", "bisa", "harus", "juga", "atau", "tetapi", "karena", "kalau", "terima", "kasih", "selamat", "pagi", "malam", "sekarang", "besok", "kemarin"}, "TR": {"bir", "ve", "bu", "için", "ile", "olarak", "çok", "daha", "var", "yok", "evet", "hayır", "teşekkür", "merhaba", "günaydın", "ben", "sen", "biz", "siz", "onlar", "ama", "veya", "çünkü", "şimdi", "yarın", "dün", "değil", "gibi", "kadar", "sonra", "önce"}, "SV": {"och", "att", "det", "som", "en", "på", "är", "av", "för", "med", "till", "den", "har", "inte", "om", "jag", "du", "vi", "ni", "de", "hej", "tack", "ja", "nej", "men", "eller", "här", "där", "nu", "idag", "imorgon"}, "DA": {"og", "at", "det", "en", "til", "er", "som", "på", "de", "med", "han", "af", "for", "ikke", "der", "jeg", "du", "vi", "hej", "tak", "ja", "nej", "men", "eller", "her", "nu", "idag", "meget"}, "NO": {"og", "i", "jeg", "det", "at", "en", "et", "den", "til", "er", "som", "på", "de", "med", "han", "av", "ikke", "der", "du", "vi", "hei", "takk", "ja", "nei", "men", "eller", "her", "nå"}, "FI": {"ja", "on", "ei", "se", "että", "en", "ovat", "kun", "niin", "mutta", "minä", "sinä", "hän", "me", "te", "he", "kiitos", "hei", "kyllä", "tämä", "tuo", "nyt", "huomenna", "eilen", "myös", "vain", "koska"}, "RO": {"și", "de", "la", "în", "cu", "pe", "un", "o", "este", "sunt", "nu", "da", "pentru", "care", "ce", "eu", "tu", "el", "ea", "noi", "voi", "ei", "mulțumesc", "bună", "dar", "sau", "acum", "mâine", "ieri"}, "CS": {"a", "je", "se", "na", "že", "v", "to", "s", "z", "do", "ne", "ano", "ale", "nebo", "protože", "já", "ty", "on", "ona", "my", "vy", "oni", "děkuji", "dobrý", "den", "teď", "zítra", "včera", "být", "mít"}, "HU": {"és", "az", "egy", "hogy", "nem", "igen", "de", "vagy", "mert", "én", "te", "ő", "mi", "ti", "ők", "köszönöm", "szia", "jó", "napot", "most", "holnap", "tegnap", "van", "nincs", "ez", "az", "itt", "ott"}, "VI": {"và", "của", "là", "có", "không", "được", "trong", "cho", "với", "này", "đó", "tôi", "bạn", "chúng", "họ", "cảm", "ơn", "chào", "vâng", "nhưng", "hoặc", "vì", "bây", "giờ", "hôm", "nay", "mai"}, "CA": {"i", "el", "la", "els", "les", "de", "que", "no", "sí", "amb", "per", "una", "un", "és", "són", "però", "o", "perquè", "jo", "tu", "ell", "ella", "nosaltres", "vosaltres", "ells", "gràcies", "hola", "ara", "demà", "ahir"}, } # Weight 2 per matched word so a single real vocabulary hit outranks the # weak shared-diacritic nudge above (worth 1). Otherwise "Kiitos paljon # avusta tänään" tied Finnish against German on the ä alone and lost the # tie on dict ordering. Genuinely distinctive diacritics still score 3 and # so still beat one lone word. for word in words: for lang, words_set in stopwords.items(): if word in words_set: scores[lang] += 2 best_lang = max(scores, key=scores.get) if scores[best_lang] == 0: return "EN" return best_lang # ── 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() # PH (the country code used by the library's Filipino voice ids) and TL # (the ISO language code the detector returns) mean the same language; # without this a PH-labelled route could never match detected TL text. def _lang_key(v: str) -> str: return _canonical_lang(v) app_ok = r_app == "*" or _app_key(r_app) == _app_key(app) voice_ok = r_voice == "*" or r_voice.lower() == voice.lower() # When the caller's requested voice is already a SPECIFIC value (not the # generic "*"/"default" some apps always send regardless of language), # it already unambiguously identifies which rule applies — cross-checking # it against text-based language auto-detection is redundant, and on # short text (a flashcard app's single vocabulary words routinely score # zero on every language's stopword list) that detector silently # defaults to EN, wrongly failing the match and falling through to a # different rule mid-session even though the caller always explicitly # requested this exact voice (confirmed live). Only rules keyed on a # generic input_voice still need the detected language to disambiguate # which specific voice to route a "default" request to. voice_is_specific = r_voice not in {"*", "default"} lang_ok = voice_is_specific or r_lang in {"*", "AUTO"} or _lang_key(r_lang) == _lang_key(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, explicit_lang: str = "") -> tuple[str, dict | None]: lang = explicit_lang.upper() if explicit_lang else _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, )