tts-voice-creator-clone-and-design-2/core/config.py

"""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,
    _VISION_OCR_DEFAULT, _VISION_OCR_MODEL_DEFAULT,
    _VISION_OCR_FALLBACK_DEFAULT, _VISION_OCR_FALLBACK_MODEL_DEFAULT,
    _TTS_STREAM_DEFAULT,
)
from core.retry import request_with_retries
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",
    "vision_ocr_url", "vision_ocr_model", "vision_ocr_enabled",
    "vision_ocr_fallback_url", "vision_ocr_fallback_model",
    "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. The server-side gate is enabled by default; deployments
    # behind another authentication layer may explicitly opt out with
    # REQUIRE_EXTERNAL_API_KEY=0.
    "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 = request_with_retries("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 request_with_retries("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":              "",
        "vision_ocr_url":         _VISION_OCR_DEFAULT,
        "vision_ocr_model":       _VISION_OCR_MODEL_DEFAULT,
        "vision_ocr_enabled":     True,
        "vision_ocr_fallback_url":   _VISION_OCR_FALLBACK_DEFAULT,
        "vision_ocr_fallback_model": _VISION_OCR_FALLBACK_MODEL_DEFAULT,
        "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": True,
        "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")