# -*- coding: utf-8 -*- """ ConfigManager für das Laden und Verwalten von Konfigurationen. Unterstützt: - JSON-Laden mit Typ-Validierung - Dateiüberwachung mit Auto-Reload - Default + Override Merging - Thread-sichere Zugriffsmodi """ import json import threading import time from dataclasses import dataclass, fields, is_dataclass, asdict from pathlib import Path from typing import TypeVar, Type, Any, Callable from watchdog.observers import Observer from watchdog.events import FileSystemEventHandler, FileModifiedEvent from trixy_core.utils.debug import pinfo, pdebug, perror, pwarn T = TypeVar("T") class ConfigFileHandler(FileSystemEventHandler): """Handler für Dateiänderungen.""" def __init__(self, callback: Callable[[str], None]) -> None: super().__init__() self.callback = callback self._last_modified: dict[str, float] = {} self._debounce_seconds = 0.5 def on_modified(self, event: FileModifiedEvent) -> None: if event.is_directory: return path = str(event.src_path) now = time.time() # Debouncing if path in self._last_modified: if now - self._last_modified[path] < self._debounce_seconds: return self._last_modified[path] = now self.callback(path) class ConfigManager: """ Verwaltet Konfigurationen für Trixy. Unterstützt das Laden von JSON-Dateien in typisierte Dataclasses, Dateiüberwachung mit Auto-Reload und Thread-sicheren Zugriff. """ def __init__(self) -> None: """Initialisiert den ConfigManager.""" self._configs: dict[str, Any] = {} self._config_paths: dict[str, Path] = {} self._config_types: dict[str, Type] = {} self._callbacks: dict[str, list[Callable[[Any], None]]] = {} self._lock = threading.RLock() self._observer: Observer | None = None self._watching = False def load( self, config_path: str | Path, config_type: Type[T], name: str | None = None, auto_reload: bool = False ) -> T: """ Lädt eine Konfiguration aus einer JSON-Datei. Args: config_path: Pfad zur JSON-Datei config_type: Ziel-Dataclass-Typ name: Optionaler Name für die Konfiguration auto_reload: Automatisches Neuladen bei Änderungen Returns: Geladene Konfiguration als Dataclass Raises: FileNotFoundError: Wenn die Datei nicht existiert ValueError: Bei ungültigem JSON oder Typ-Fehlern """ path = Path(config_path) config_name = name or path.stem if not path.exists(): pwarn(f"Konfigurationsdatei nicht gefunden: {path}, verwende Standardwerte") config = config_type() else: config = self._load_file(path, config_type) with self._lock: self._configs[config_name] = config self._config_paths[config_name] = path self._config_types[config_name] = config_type if auto_reload: self._start_watching(path, config_name) pinfo(f"Konfiguration geladen: {config_name}") return config def _load_file(self, path: Path, config_type: Type[T]) -> T: """ Lädt eine JSON-Datei und konvertiert sie in eine Dataclass. Args: path: Pfad zur Datei config_type: Ziel-Typ Returns: Konfiguration als Dataclass """ try: with open(path, "r", encoding="utf-8") as f: data = json.load(f) except json.JSONDecodeError as e: perror(f"Ungültiges JSON in {path}: {e}") raise ValueError(f"Ungültiges JSON: {e}") return self._dict_to_dataclass(data, config_type) def _dict_to_dataclass(self, data: dict[str, Any], dc_type: Type[T]) -> T: """ Konvertiert ein Dictionary rekursiv in eine Dataclass. Args: data: Eingabe-Dictionary dc_type: Ziel-Dataclass-Typ Returns: Instanz der Dataclass """ if not is_dataclass(dc_type): return data field_values: dict[str, Any] = {} for field in fields(dc_type): field_name = field.name field_type = field.type if field_name in data: value = data[field_name] # Rekursive Verarbeitung für verschachtelte Dataclasses if is_dataclass(field_type) and isinstance(value, dict): value = self._dict_to_dataclass(value, field_type) # Handle list[...] inkl. list[Dataclass] elif hasattr(field_type, "__origin__"): origin = getattr(field_type, "__origin__") if origin is list and isinstance(value, list): args = getattr(field_type, "__args__", ()) if args and is_dataclass(args[0]): item_type = args[0] value = [ self._dict_to_dataclass(item, item_type) if isinstance(item, dict) else item for item in value ] # else: Listen primitiver Typen direkt übernehmen field_values[field_name] = value return dc_type(**field_values) def get(self, name: str) -> Any: """ Gibt eine geladene Konfiguration zurück. Args: name: Name der Konfiguration Returns: Konfiguration oder None """ with self._lock: return self._configs.get(name) def get_typed(self, name: str, config_type: Type[T]) -> T | None: """ Gibt eine typisierte Konfiguration zurück. Args: name: Name der Konfiguration config_type: Erwarteter Typ Returns: Typisierte Konfiguration oder None """ with self._lock: config = self._configs.get(name) if config is not None and isinstance(config, config_type): return config return None def save(self, name: str, config: Any | None = None) -> None: """ Speichert eine Konfiguration in die Datei. Args: name: Name der Konfiguration config: Optionale neue Konfiguration """ with self._lock: if config is not None: self._configs[name] = config cfg = self._configs.get(name) path = self._config_paths.get(name) if cfg is None or path is None: perror(f"Konfiguration '{name}' nicht gefunden") return try: data = asdict(cfg) if is_dataclass(cfg) else cfg path.parent.mkdir(parents=True, exist_ok=True) with open(path, "w", encoding="utf-8") as f: json.dump(data, f, indent=2, ensure_ascii=False) pinfo(f"Konfiguration gespeichert: {name}") except Exception as e: perror(f"Fehler beim Speichern von '{name}': {e}") def reload(self, name: str) -> Any | None: """ Lädt eine Konfiguration neu. Args: name: Name der Konfiguration Returns: Neu geladene Konfiguration oder None """ with self._lock: path = self._config_paths.get(name) config_type = self._config_types.get(name) if path is None or config_type is None: perror(f"Konfiguration '{name}' nicht registriert") return None try: config = self._load_file(path, config_type) self._configs[name] = config # Callbacks ausführen if name in self._callbacks: for callback in self._callbacks[name]: try: callback(config) except Exception as e: perror(f"Callback-Fehler für '{name}': {e}") pinfo(f"Konfiguration neu geladen: {name}") return config except Exception as e: perror(f"Fehler beim Neuladen von '{name}': {e}") return None def on_reload(self, name: str, callback: Callable[[Any], None]) -> None: """ Registriert einen Callback für Konfigurations-Reloads. Args: name: Name der Konfiguration callback: Callback-Funktion """ with self._lock: if name not in self._callbacks: self._callbacks[name] = [] self._callbacks[name].append(callback) def _start_watching(self, path: Path, name: str) -> None: """Startet die Dateiüberwachung.""" if self._observer is None: self._observer = Observer() self._watching = True def on_change(changed_path: str) -> None: if Path(changed_path).resolve() == path.resolve(): pdebug(f"Konfigurationsänderung erkannt: {name}") self.reload(name) handler = ConfigFileHandler(on_change) # Überwache das Verzeichnis der Datei self._observer.schedule(handler, str(path.parent), recursive=False) if not self._observer.is_alive(): self._observer.start() pdebug(f"Dateiüberwachung gestartet für: {path}") def stop_watching(self) -> None: """Stoppt die Dateiüberwachung.""" if self._observer is not None: self._observer.stop() self._observer.join() self._observer = None self._watching = False pdebug("Dateiüberwachung gestoppt") def create_default(self, config_type: Type[T], path: str | Path) -> T: """ Erstellt eine Standard-Konfigurationsdatei. Args: config_type: Konfigurationstyp path: Zielpfad Returns: Standard-Konfiguration """ config = config_type() file_path = Path(path) file_path.parent.mkdir(parents=True, exist_ok=True) data = asdict(config) if is_dataclass(config) else {} with open(file_path, "w", encoding="utf-8") as f: json.dump(data, f, indent=2, ensure_ascii=False) pinfo(f"Standard-Konfiguration erstellt: {file_path}") return config