453 lines
25 KiB
Python
453 lines
25 KiB
Python
"""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:
|
||
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,
|
||
)
|