mARTin-B78
5fecbf06d4
Fish-Speech emotion tags were silently ignored on non-English books: per-line emotions are LLM-generated in the book's own language, but Fish-Speech only recognizes English [tag] markers, and a double-tagging bug was stacking a broken server-derived tag on top of the client's own. Added a DE->EN translation table and removed the double-tagging. Also wires the existing book-profile context and race_species field into character portrait prompts (previously only used for voice design), adds a recast-until-threshold loop for casting, and adds backend-aware emotion quick-picks to Read Aloud, Try a Voice, and Conversation. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
302 lines
13 KiB
Python
302 lines
13 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
|
|
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", ""), "")
|
|
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())
|
|
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))
|
|
_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"
|
|
|
|
|
|
# ── Language detection ────────────────────────────────────────────────────────
|
|
|
|
def _detect_text_language(text: str) -> str:
|
|
words = re.findall(r"\b[a-zäöüßéèêàçùœáíóúñąćęłńóśźż]+\b", text.lower())
|
|
if not words:
|
|
return "EN"
|
|
|
|
scores = {"EN": 0, "DE": 0, "FR": 0, "ES": 0, "IT": 0, "PT": 0, "NL": 0, "PL": 0}
|
|
|
|
if re.search(r"[äöüß]", text.lower()):
|
|
scores["DE"] += 3
|
|
if re.search(r"[ąćęłńóśźż]", text.lower()):
|
|
scores["PL"] += 3
|
|
if re.search(r"[ãõ]", text.lower()):
|
|
scores["PT"] += 3
|
|
if re.search(r"[ñ¿¡]", text.lower()):
|
|
scores["ES"] += 3
|
|
|
|
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"},
|
|
}
|
|
|
|
for word in words:
|
|
for lang, words_set in stopwords.items():
|
|
if word in words_set:
|
|
scores[lang] += 1
|
|
|
|
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()
|
|
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, 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,
|
|
)
|