"""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": {}, # BUT distinct from the above: once a voice has already been designed # and saved, every ONGOING line of dialogue that reuses it is a # voice_clone-style read of a fixed identity, not a fresh design — it # needs the SAME stability as voice_clone, not voice_design's # deliberately-unconstrained randomness. Both used to share the single # "voice_design" backend key, so every line of an audiobook synthesized # with a designed voice ran with fully unconstrained temperature/top_p — # confirmed live as the actual cause of designed voices (unlike cloned # ones, which already got the stability block) occasionally repeating # text or drifting into unrelated words mid-book. # No "seed" key here, unlike the other stability defaults: the whole # point of the per-voice Seed Finder / Batch Seeds tools is to pin a # SPECIFIC seed per designed voice (stored server-side against that # voice_id) — a hardcoded seed=0 in every request would silently # override that per-voice choice on every single line. Temperature/top_p # alone already gives real stability without fighting the per-voice pin. "voice_design_playback": {k: v for k, v in _TTS_STABILITY_DEFAULT.items() if k != "seed"}, "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 {} # "voice_design_playback" is a pseudo-backend used only as a stability- # profile lookup key (the actual HTTP calls still go to the voice_design # engine) — it isn't a real routing target, so _clean_preview_backend # doesn't recognize it and was silently coercing it back to "voice_clone". # Confirmed live: that meant every designed-voice playback line picked up # voice_clone's persisted seed:0 default instead of voice_design_playback's # deliberately seed-less profile, permanently overriding every per-voice # pinned seed with a hardcoded 0. Only normalize real backend names. if backend not in _TTS_STABILITY_BY_BACKEND_DEFAULT: 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")