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"),
    "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_base_url":            ("Lokales LLM (Basis-URL)",       "str",   "z.B. http://127.0.0.1:11434/v1 (Ollama) oder Heim-Host — wirkt sofort"),
    "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"),
    "web_search_enabled":            ("Web-Suche (Standard)",          "bool",  "true | false — global an, pro Nutzer abschaltbar"),
    "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"),
    "cost_daily_limit_usd":          ("Kostenbudget/Tag (global, $)",  "float", "0 = unbegrenzt — pro Nutzer überschreibbar"),
    "electricity_price_eur_per_kwh": ("Strompreis (€/kWh)",            "float", "Kosten lokaler LLM-Anfragen (z.B. 0.33)"),
    "emergency_alert_eur":           ("Notruf-Kosten (€/Alarm)",       "float", "SMS+Anruf pro Alarm (z.B. 0.20)"),
    "cartesia_usd_per_char":         ("Cartesia-TTS ($/Zeichen)",      "float", "z.B. 0.00005 (100k Credits ≈ 100.000 Zeichen)"),
    "openrouter_tts_usd_per_char":   ("OpenRouter-TTS ($/Zeichen)",    "float", "Schätzung, z.B. 0.000015 — an Modellpreis anpassen"),
    "emergency_cooldown_minutes":    ("Notruf-Sperre (Minuten)",       "int",   "Erneuter Alarm für N Minuten blockiert (Standard 5, 0 = aus)"),
}

_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)