my_voice_assistant_v3_jamulix/app/runtime_config.py

"""Laufzeit-Konfigurationsüberschreibungen aus der Datenbank.

Einzelne Settings-Felder können zur Laufzeit via Admin-UI geändert werden,
ohne den Server neu zu starten. Die Werte liegen in der Tabelle
`config_overrides` und werden mit 30s TTL gecacht.

Nur Felder aus RUNTIME_SETTABLE sind überschreibbar — alle anderen
kommen weiterhin aus .env / TOML / pydantic-settings.
"""

import threading
import time
from typing import Any

from app.config import Settings, settings as _base

# (label, type_str, hint)
RUNTIME_SETTABLE: dict[str, tuple[str, str, str]] = {
    "default_stt_provider":          ("STT-Provider (Standard)",       "str",   "openrouter | faster-whisper"),
    "default_llm_provider":          ("LLM-Provider (Standard)",       "str",   "openrouter | local-openai-compatible"),
    "default_tts_provider":          ("TTS-Provider (Standard)",       "str",   "openrouter | piper | chatterbox — Standard, pro Nutzer überschreibbar"),
    "default_language":              ("Sprache (Standard)",            "str",   "de | en | … — Standard, pro Nutzer überschreibbar"),
    "default_language_mode":         ("Sprachmodus (Standard)",        "str",   "fix | flex — Standard, pro Nutzer überschreibbar"),
    "openrouter_llm_model":          ("LLM-Modell (OpenRouter)",       "str",   "z.B. google/gemini-3.1-flash-lite"),
    "openrouter_tts_model":          ("TTS-Modell (OpenRouter)",       "str",   "z.B. google/gemini-3.1-flash-tts-preview"),
    "openrouter_tts_voice":          ("TTS-Stimme (OpenRouter)",       "str",   "z.B. Zephyr, Puck, Kore"),
    "piper_voice":                   ("TTS-Stimme (piper)",            "str",   "z.B. de_DE-thorsten-high"),
    "local_llm_system_prompt":       ("Systemprompt (lokal)",          "str",   "Freier Text"),
    "local_llm_temperature":         ("Temperatur (lokal)",            "float", "0.0–2.0 — wirkt sofort (kein Neustart)"),
    "local_llm_top_p":               ("Top-p (lokal)",                 "float", "0.0–1.0 — wirkt sofort (kein Neustart)"),
    "local_llm_max_tokens":          ("Max. Tokens (lokal)",           "int",   "0 = kein Limit"),
    "tts_normalize_level":           ("TTS-Normalisierung",            "str",   "auto | full | light | off"),
    "audio_stream_default":          ("Audio-Streaming Standard",      "bool",  "true | false"),
    "memory_extraction_enabled":     ("Erinnerungs-Extraktion",        "bool",  "true | false"),
    "memory_extraction_every_n_turns": ("Extraktion alle N Turns",     "int",   "z.B. 3"),
    "daily_request_limit":           ("Tageskontingent (global)",      "int",   "0 = unbegrenzt"),
    "allow_cloud_stt":               ("Geräte-STT via Cloud erlauben", "bool",  "true = auch Cloud-Erkennung (z.B. Chrome-Desktop -> Audio zu Google); false = nur on-device"),
}

_TTL = 30.0
_cache: dict[str, str] = {}
_cache_time: float = 0.0
_lock = threading.Lock()


def _coerce(key: str, raw: str) -> Any:
    _, type_str, _ = RUNTIME_SETTABLE[key]
    try:
        if type_str == "bool":
            return raw.strip().lower() in ("1", "true", "yes")
        if type_str == "int":
            return int(raw)
        if type_str == "float":
            return float(raw)
    except (ValueError, AttributeError):
        pass
    return raw


def _get_cache() -> dict[str, str]:
    global _cache, _cache_time
    now = time.monotonic()
    with _lock:
        if now - _cache_time < _TTL:
            return _cache
        try:
            from app.dependencies import get_store
            _cache = get_store().get_config_overrides()
            _cache_time = now
        except Exception:
            pass
        return _cache


def invalidate_cache() -> None:
    global _cache_time
    with _lock:
        _cache_time = 0.0


class RuntimeSettings:
    """Wraps Settings; liest überschreibbare Felder aus der DB (30s TTL)."""

    def __init__(self, base: Settings):
        object.__setattr__(self, "_base", base)

    def __getattr__(self, name: str) -> Any:
        if name in RUNTIME_SETTABLE:
            overrides = _get_cache()
            if name in overrides:
                return _coerce(name, overrides[name])
        return getattr(object.__getattribute__(self, "_base"), name)

    # Delegiere Pydantic-Metadaten ans Basis-Objekt.
    @property
    def model_fields(self):
        return self._base.model_fields


runtime_settings = RuntimeSettings(_base)