tts-voice-creator-clone-and.../core/config.py
mARTin-B78 ea50267c30 Add unified Studio casting workflow and fix voice/casting pipeline bugs
Introduces the new Studio section (Source -> Characters -> Voices ->
Perform & Export) that reuses the existing Read Aloud/Library/Script
Rehearsal code via DOM reparenting instead of duplicating it, and rolls up
a long tail of bugs found while producing a real audiobook through it:
umlaut-eating name sanitizers, a voice picker that mispositioned itself and
capped results at 60, PDF pagination silently breaking on trimmed \f
markers, a race letting stale audio keep playing after a new line was
clicked, an alias-overlap bug that could silently redirect a voice/image
save onto the wrong character, voice design failing outright during brief
TTS backend restarts instead of retrying, sparse cast entries defaulting to
English/wrong gender, and a reassigned voice never reaching an already-open
Stage session or invalidating its cached audio. Also adds a persistent
per-line audio cache, audiobook export browsing/download, and an inline
voice-design prompt editor. Full details in CHANGELOG.md.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
2026-07-26 02:03:56 +02:00

375 lines
16 KiB
Python

"""Settings management: load, save, normalize, TTS stability helpers."""
from __future__ import annotations
import json
import re
from pathlib import Path
import requests
from core.constants import (
CONFIG_DIR, CONFIG_FILE,
_VOICES_DIR_DEFAULT, _OUTPUT_DIR_DEFAULT,
_WHISPER_DEFAULT, _TTS_DEFAULT, _TTS_STREAM_DEFAULT,
_CUSTOMVOICE_DEFAULT, _VOICE_DESIGN_DEFAULT,
_NVIDIA_ROUTER_DEFAULT, _NVIDIA_TTS_DEFAULT, _NVIDIA_ASR_DEFAULT,
_NVIDIA_CLONE_DEFAULT, _NVIDIA_ZEROSHOT_DEFAULT, _NVIDIA_FLOW_DEFAULT,
_FASTER_WHISPER_DEFAULT, _WHISPER_CPP_DEFAULT,
_KOKORO_DEFAULT, _VIBEVOICE_DEFAULT, _XTTS_DEFAULT, _FISHSPEECH_DEFAULT,
_TTS_STREAM_DEFAULT,
)
from core.database import state_get, state_put, state_updated
# ── Settings key whitelist ────────────────────────────────────────────────────
_SETTINGS_KEYS = {
"whisper_url", "tts_url", "tts_stream_url", "tts_stream_mode", "tts_backend",
"output_dir", "voices_scan_dir", "voice_design_url", "customvoice_url",
"nvidia_router_url", "nvidia_tts_url", "nvidia_asr_url", "nvidia_clone_url",
"nvidia_zeroshot_url", "nvidia_flow_url",
"faster_whisper_url", "whisper_cpp_url", "groq_api_key", "kokoro_url", "vibevoice_url", "xtts_url",
"fishspeech_url",
"whisper_api_key", "tts_api_key", "voice_design_api_key", "elevenlabs_api_key",
"tts_stability_enabled", "tts_extra_params", "tts_extra_params_by_backend",
# Captures settings
"stt_language", "stt_preferred_backend",
"auto_refine", "refine_model",
"refine_fillers", "refine_repetitions", "refine_corrections", "refine_punctuation",
"captures_default_voice",
"client_voice_bindings",
"llm_url", "llm_model", "llm_api_key",
"engine_local_urls", "engine_container_names", "engine_api_keys", "custom_engine_cards",
# Character portrait generation (cloud APIs + local ComfyUI)
"image_gen_provider", "image_gen_model",
"comfyui_url", "comfyui_workflow", "comfyui_prompt_node_id",
"comfyui_prompt_field", "comfyui_output_node_id",
# Inbound API key — gates non-browser callers (external scripts, MCP
# clients) hitting this app's own /api/* and /mcp routes. Distinct from
# every *outbound* key above, which are credentials this app sends to
# OTHER services. Off by default (external_api_key_required=False) —
# this is a live, actively-used app and the same-origin detection this
# gate relies on has never been exercised against real browser traffic,
# so enforcing it unconditionally risked locking out the working UI on
# an untested edge case. Turn it on deliberately once confirmed safe.
"external_api_key", "external_api_key_required",
# Browser-persistent UI state
"refine_llm_url", "conv_llm_url", "seed_finder_text",
"seed_finder_dir", "pt_dir", "audiobook_prompt",
}
# ── TTS stability defaults ────────────────────────────────────────────────────
_TTS_STABILITY_DEFAULT = {"temperature": 0.1, "top_p": 0.8, "seed": 0}
_TTS_STABILITY_BY_BACKEND_DEFAULT = {
"voice_clone": dict(_TTS_STABILITY_DEFAULT),
"streaming": dict(_TTS_STABILITY_DEFAULT),
"customvoice": dict(_TTS_STABILITY_DEFAULT),
# NOT voice_design: the stability block's fixed seed=0 + temperature=0.1
# exists so repeated reads of the SAME cloned voice sound consistent
# across takes — exactly backwards for voice design, where every call is
# supposed to produce a DIFFERENT voice from a different character
# prompt. Pinning the model's random draw meant the prompt text was the
# only source of variation, and low temperature flattened even that —
# confirmed live: auto-designed voices for different characters all
# sounded near-identical. Leaving this empty lets the backend use its
# own natural randomization per call, same as the other creative-voice
# backends below.
"voice_design": {},
"nvidia_magpie": {},
"nvidia_zeroshot": {},
"nvidia_flow": {},
"kokoro": {},
"vibevoice": {},
"xtts": {},
"fishspeech": {},
}
_TTS_PAYLOAD_CORE_KEYS = {"model", "input", "voice", "response_format", "instruct", "language"}
def _settings_bool(value, default: bool = True) -> bool:
if value is None:
return default
if isinstance(value, bool):
return value
return str(value).strip().lower() not in {"0", "false", "off", "no"}
def _clean_tts_param_dict(raw, fallback: dict | None = None) -> dict:
if raw in (None, ""):
raw = fallback or {}
if isinstance(raw, str):
try:
raw = json.loads(raw)
except Exception:
raw = fallback or {}
if not isinstance(raw, dict):
return {}
params = {}
for key, value in raw.items():
if key in _TTS_PAYLOAD_CORE_KEYS or str(key).startswith("_"):
continue
if isinstance(value, (str, int, float, bool)) or value is None:
params[str(key)] = value
return params
def _clean_preview_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",
"voiceclone": "voice_clone",
"voice_clone_base": "voice_clone",
"stream": "streaming",
"tts_streaming": "streaming",
"custom": "customvoice",
"custom_voice": "customvoice",
"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",
"kokoro_fastapi": "kokoro",
"kokoro_tts": "kokoro",
"kokoro_local": "kokoro",
"vibevoice_service": "vibevoice",
"vibe_voice": "vibevoice",
"vibetts": "vibevoice",
"xtts_v2": "xtts",
"xtts2": "xtts",
"coqui_xtts": "xtts",
"fish": "fishspeech",
"fish_speech": "fishspeech",
"fishaudio": "fishspeech",
"openaudio": "fishspeech",
}
key = aliases.get(key, key)
return key if key in {"voice_clone", "streaming", "customvoice", "voice_design", "nvidia_magpie", "nvidia_zeroshot", "nvidia_flow", "kokoro", "vibevoice", "xtts", "fishspeech"} else "voice_clone"
def _tts_extra_params(settings: dict, backend: str = "voice_clone") -> dict:
if not _settings_bool(settings.get("tts_stability_enabled"), True):
return {}
backend = _clean_preview_backend(backend)
by_backend = settings.get("tts_extra_params_by_backend")
if isinstance(by_backend, str):
try:
by_backend = json.loads(by_backend)
except Exception:
by_backend = None
if isinstance(by_backend, dict):
fallback = _TTS_STABILITY_BY_BACKEND_DEFAULT.get(backend, _TTS_STABILITY_DEFAULT)
if backend in by_backend:
return _clean_tts_param_dict(by_backend.get(backend), fallback)
return _clean_tts_param_dict(fallback, fallback)
return _clean_tts_param_dict(settings.get("tts_extra_params", _TTS_STABILITY_BY_BACKEND_DEFAULT.get(backend, _TTS_STABILITY_DEFAULT)), _TTS_STABILITY_BY_BACKEND_DEFAULT.get(backend, _TTS_STABILITY_DEFAULT))
def _apply_tts_extra_params(payload: dict, settings: dict, backend: str = "voice_clone") -> dict:
params = _tts_extra_params(settings, backend)
if params:
payload.update(params)
return payload
def _strip_tts_extra_params(payload: dict) -> dict:
return {k: v for k, v in payload.items() if k in _TTS_PAYLOAD_CORE_KEYS}
def _post_tts_with_fallback(endpoint: str, payload: dict, headers: dict, **kwargs) -> requests.Response:
resp = requests.post(endpoint, json=payload, headers=headers, **kwargs)
fallback = _strip_tts_extra_params(payload)
if resp.status_code in {400, 404, 415, 422} and fallback != payload:
try:
resp.close()
except Exception:
pass
import logging
logging.getLogger("uvicorn.error").warning("TTS backend rejected extra generation params; retrying without them")
return requests.post(endpoint, json=fallback, headers=headers, **kwargs)
return resp
def _preview_backend_base_url(settings: dict, backend: str) -> str:
backend = _clean_preview_backend(backend)
if backend == "streaming":
return settings.get("tts_stream_url") or settings.get("tts_url") or _TTS_STREAM_DEFAULT
if backend == "customvoice":
return settings.get("customvoice_url") or _CUSTOMVOICE_DEFAULT
if backend == "voice_design":
return settings.get("voice_design_url") or _VOICE_DESIGN_DEFAULT
if backend == "nvidia_magpie":
return settings.get("nvidia_tts_url") or settings.get("nvidia_router_url") or _NVIDIA_TTS_DEFAULT
if backend == "nvidia_zeroshot":
return settings.get("nvidia_zeroshot_url") or settings.get("nvidia_clone_url") or settings.get("nvidia_router_url") or _NVIDIA_ZEROSHOT_DEFAULT
if backend == "nvidia_flow":
return settings.get("nvidia_flow_url") or settings.get("nvidia_clone_url") or settings.get("nvidia_router_url") or _NVIDIA_FLOW_DEFAULT
if backend == "kokoro":
return settings.get("kokoro_url") or _KOKORO_DEFAULT
if backend == "vibevoice":
return settings.get("vibevoice_url") or _VIBEVOICE_DEFAULT
if backend == "xtts":
return settings.get("xtts_url") or _XTTS_DEFAULT
if backend == "fishspeech":
return settings.get("fishspeech_url") or _FISHSPEECH_DEFAULT
return settings.get("tts_url") or _TTS_DEFAULT
def _normalize_settings(s: dict) -> dict:
scan_dir = Path(s.get("voices_scan_dir", _VOICES_DIR_DEFAULT))
output_dir = Path(s.get("output_dir", _OUTPUT_DIR_DEFAULT))
legacy_output_dirs = {
Path("/voices/speakers"),
scan_dir / "speakers",
}
if output_dir in legacy_output_dirs:
s["output_dir"] = str(scan_dir / "active_voices")
if s.get("tts_stream_mode") not in {"auto", "streaming", "buffered"}:
s["tts_stream_mode"] = "auto"
if not isinstance(s.get("engine_local_urls"), dict):
s["engine_local_urls"] = {}
if not isinstance(s.get("engine_container_names"), dict):
s["engine_container_names"] = {}
if not isinstance(s.get("engine_api_keys"), dict):
s["engine_api_keys"] = {}
if not isinstance(s.get("custom_engine_cards"), list):
s["custom_engine_cards"] = []
return s
_settings_cache: dict | None = None
_settings_cache_mtime: float = -1.0
_settings_cache_db_updated: str = ""
def _load_settings() -> dict:
global _settings_cache, _settings_cache_mtime, _settings_cache_db_updated
mtime = CONFIG_FILE.stat().st_mtime if CONFIG_FILE.exists() else 0.0
db_updated = state_updated("settings")
if (
_settings_cache is not None
and mtime == _settings_cache_mtime
and db_updated == _settings_cache_db_updated
):
return dict(_settings_cache)
defaults = {
"whisper_url": _WHISPER_DEFAULT,
"tts_url": _TTS_DEFAULT,
"tts_stream_url": _TTS_STREAM_DEFAULT,
"tts_stream_mode": "auto",
"tts_backend": "openai",
"output_dir": _OUTPUT_DIR_DEFAULT,
"voices_scan_dir": _VOICES_DIR_DEFAULT,
"voice_design_url": _VOICE_DESIGN_DEFAULT,
"customvoice_url": _CUSTOMVOICE_DEFAULT,
"nvidia_router_url": _NVIDIA_ROUTER_DEFAULT,
"nvidia_tts_url": _NVIDIA_TTS_DEFAULT,
"nvidia_asr_url": _NVIDIA_ASR_DEFAULT,
"nvidia_clone_url": _NVIDIA_CLONE_DEFAULT,
"nvidia_zeroshot_url": _NVIDIA_ZEROSHOT_DEFAULT,
"nvidia_flow_url": _NVIDIA_FLOW_DEFAULT,
"faster_whisper_url": _FASTER_WHISPER_DEFAULT,
"whisper_cpp_url": _WHISPER_CPP_DEFAULT,
"groq_api_key": "",
"kokoro_url": _KOKORO_DEFAULT,
"vibevoice_url": _VIBEVOICE_DEFAULT,
"xtts_url": _XTTS_DEFAULT,
"fishspeech_url": _FISHSPEECH_DEFAULT,
"whisper_api_key": "",
"tts_api_key": "",
"voice_design_api_key": "",
"tts_stability_enabled": True,
"tts_extra_params": _TTS_STABILITY_DEFAULT,
"tts_extra_params_by_backend": _TTS_STABILITY_BY_BACKEND_DEFAULT,
# Captures
"stt_language": "",
"stt_preferred_backend": "",
"auto_refine": "off",
"refine_model": "",
"refine_fillers": True,
"refine_repetitions": True,
"refine_corrections": True,
"refine_punctuation": True,
"captures_default_voice": "",
"client_voice_bindings": {},
"llm_url": "http://localhost:11434/v1",
"llm_api_key": "",
"llm_model": "",
"engine_local_urls": {},
"engine_container_names": {},
"engine_api_keys": {},
"custom_engine_cards": [],
"image_gen_provider": "",
"image_gen_model": "",
"comfyui_url": "http://host.docker.internal:8188",
"comfyui_workflow": "",
"comfyui_prompt_node_id": "",
"comfyui_prompt_field": "text",
"comfyui_output_node_id": "",
"external_api_key": "",
"external_api_key_required": False,
"refine_llm_url": "",
"conv_llm_url": "",
"seed_finder_text": "",
"seed_finder_dir": "",
"pt_dir": "",
}
file_saved = None
if CONFIG_FILE.exists():
try:
file_saved = json.loads(CONFIG_FILE.read_text())
if isinstance(file_saved, dict):
defaults.update({k: v for k, v in file_saved.items() if k in _SETTINGS_KEYS})
except Exception:
pass
db_saved = state_get("settings", None)
if isinstance(db_saved, dict):
defaults.update({k: v for k, v in db_saved.items() if k in _SETTINGS_KEYS})
result = _normalize_settings(defaults)
if not db_updated and (file_saved is not None or CONFIG_FILE.exists()):
try:
state_put("settings", result)
db_updated = state_updated("settings")
except Exception:
pass
_settings_cache = result
_settings_cache_mtime = mtime
_settings_cache_db_updated = db_updated
return dict(result)
def _ensure_external_api_key() -> str:
"""Returns the app's inbound API key, generating + persisting one on
first use. Called lazily by the auth middleware rather than at startup,
so a fresh install doesn't need a migration step."""
import secrets
settings = _load_settings()
key = (settings.get("external_api_key") or "").strip()
if key:
return key
key = secrets.token_urlsafe(32)
settings["external_api_key"] = key
_save_settings(settings)
return key
def _save_settings(s: dict) -> None:
global _settings_cache, _settings_cache_mtime, _settings_cache_db_updated
CONFIG_DIR.mkdir(parents=True, exist_ok=True)
clean = _normalize_settings({k: v for k, v in s.items() if k in _SETTINGS_KEYS})
state_put("settings", clean)
CONFIG_FILE.write_text(json.dumps(clean, indent=2))
_settings_cache = clean
_settings_cache_mtime = CONFIG_FILE.stat().st_mtime
_settings_cache_db_updated = state_updated("settings")