dschlueter
8c152ee77c
1) Notruf-Sperre (Cooldown) gegen Mehrfach-Alarme verwirrter/verängstigter
Senioren:
- Neue Einstellung emergency_cooldown_minutes (Default 5), zur Laufzeit im
Admin-Bereich änderbar (RUNTIME_SETTABLE), 0 = aus.
- record_manual_emergency unterdrückt innerhalb der Sperre jeden weiteren
Alarm (kein zweiter Versand, kein Log-Eintrag) und liefert status=cooldown
mit minutes_ago/remaining_seconds. Neuer Endpunkt GET /api/emergency/status
für die Anzeige nach einem Seiten-Neuladen. Store: last_manual_emergency_at.
- UI: Nach dem Alarm wird der Button grün, zeigt „✓" und pulsiert für die
Dauer der Sperre. Erneuter Druck öffnet einen grünen Hinweis-Dialog:
„Ich habe Deine Helfer vor <n> Minuten alarmiert. Wahrscheinlich ist Deine
Hilfe schon unterwegs." Der grüne Zustand übersteht ein Neuladen
(Status-Abruf beim Start).
2) „Neues Gespräch" als Symbol (Stift) zurück in die Kopfzeile — zwischen
Sprachauswahl und Menü; löst dieselbe Aktion aus wie der Menüeintrag.
Tests: neue test_emergency_cooldown.py (Status/Suppression/Endpoint); zwei
bestehende Notruf-Tests an die neue Interface-Methode bzw. die Sperre
angepasst. 250 Offline-Tests grün. Frontend via Headless-Chrome verifiziert
(Alarm->grün/pulsierend, Hinweis-Dialog, Reload-Persistenz, Header-Button).
Bump app.js v=50 -> v=51.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
99 lines
4.2 KiB
Python
99 lines
4.2 KiB
Python
"""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_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"),
|
||
"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)
|