Trixy7/trixy_core/config/config_manager.py

# -*- 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