Trixy7/trixy_core/plugins/hotreload/watcher.py

# -*- coding: utf-8 -*-
"""
Dateiüberwachung für Plugin Hot-Reload.

Beobachtet Plugin-Verzeichnisse auf Änderungen.
"""
from __future__ import annotations

import asyncio
import os
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum, auto
from pathlib import Path
from typing import Any, Callable


class WatchEventType(Enum):
    """Typ eines Datei-Events."""
    CREATED = auto()    # Datei erstellt
    MODIFIED = auto()   # Datei geändert
    DELETED = auto()    # Datei gelöscht
    MOVED = auto()      # Datei verschoben


@dataclass
class WatchEvent:
    """
    Ein Datei-Änderungs-Event.

    Attributes:
        event_type: Art der Änderung
        path: Betroffener Pfad
        is_directory: Ist ein Verzeichnis
        timestamp: Zeitpunkt
        old_path: Alter Pfad bei MOVED
    """
    event_type: WatchEventType
    path: Path
    is_directory: bool = False
    timestamp: datetime = field(default_factory=datetime.now)
    old_path: Path | None = None

    def to_dict(self) -> dict[str, Any]:
        """Konvertiert zu Dictionary."""
        return {
            "event_type": self.event_type.name,
            "path": str(self.path),
            "is_directory": self.is_directory,
            "timestamp": self.timestamp.isoformat(),
            "old_path": str(self.old_path) if self.old_path else None,
        }


# Callback-Typ
WatchCallback = Callable[[WatchEvent], None]
AsyncWatchCallback = Callable[[WatchEvent], Any]  # Koroutine


class FileWatcher:
    """
    Überwacht Verzeichnisse auf Änderungen.

    Verwendet Polling für maximale Kompatibilität.
    Optional kann watchdog verwendet werden wenn verfügbar.

    Example:
        watcher = FileWatcher()

        # Verzeichnis hinzufügen
        watcher.add_directory(Path("plugins"))

        # Callback registrieren
        @watcher.on_change
        def handle_change(event: WatchEvent):
            print(f"{event.event_type.name}: {event.path}")

        # Überwachung starten
        await watcher.start()

        # ... später ...
        await watcher.stop()
    """

    DEFAULT_POLL_INTERVAL = 1.0  # Sekunden

    def __init__(
        self,
        poll_interval: float = DEFAULT_POLL_INTERVAL,
        recursive: bool = True,
        extensions: list[str] | None = None,
    ) -> None:
        """
        Initialisiert den FileWatcher.

        Args:
            poll_interval: Intervall für Polling in Sekunden
            recursive: Unterverzeichnisse einbeziehen
            extensions: Nur diese Dateiendungen (z.B. [".py", ".json"])
        """
        self._poll_interval = poll_interval
        self._recursive = recursive
        self._extensions = set(extensions) if extensions else None

        self._directories: set[Path] = set()
        self._callbacks: list[WatchCallback] = []
        self._async_callbacks: list[AsyncWatchCallback] = []

        self._running = False
        self._task: asyncio.Task | None = None

        # Snapshot für Änderungserkennung
        self._snapshot: dict[Path, float] = {}  # path -> mtime

    @property
    def is_running(self) -> bool:
        """Läuft die Überwachung?"""
        return self._running

    @property
    def watched_directories(self) -> list[Path]:
        """Liste überwachter Verzeichnisse."""
        return list(self._directories)

    def add_directory(self, path: Path) -> bool:
        """
        Fügt ein Verzeichnis zur Überwachung hinzu.

        Args:
            path: Zu überwachendes Verzeichnis

        Returns:
            True wenn hinzugefügt
        """
        if not path.exists() or not path.is_dir():
            return False

        self._directories.add(path)
        self._update_snapshot(path)
        return True

    def remove_directory(self, path: Path) -> bool:
        """
        Entfernt ein Verzeichnis aus der Überwachung.

        Args:
            path: Zu entfernendes Verzeichnis

        Returns:
            True wenn entfernt
        """
        if path in self._directories:
            self._directories.discard(path)

            # Snapshot bereinigen
            to_remove = [p for p in self._snapshot if str(p).startswith(str(path))]
            for p in to_remove:
                del self._snapshot[p]

            return True
        return False

    def on_change(self, callback: WatchCallback) -> WatchCallback:
        """
        Decorator zum Registrieren eines Callbacks.

        Args:
            callback: Callback-Funktion

        Returns:
            Der registrierte Callback
        """
        self._callbacks.append(callback)
        return callback

    def on_change_async(
        self,
        callback: AsyncWatchCallback,
    ) -> AsyncWatchCallback:
        """Decorator für async Callbacks."""
        self._async_callbacks.append(callback)
        return callback

    def add_callback(self, callback: WatchCallback) -> None:
        """Fügt einen Callback hinzu."""
        self._callbacks.append(callback)

    def add_async_callback(self, callback: AsyncWatchCallback) -> None:
        """Fügt einen async Callback hinzu."""
        self._async_callbacks.append(callback)

    def remove_callback(self, callback: WatchCallback) -> bool:
        """Entfernt einen Callback."""
        try:
            self._callbacks.remove(callback)
            return True
        except ValueError:
            return False

    async def start(self) -> None:
        """Startet die Überwachung."""
        if self._running:
            return

        self._running = True

        # Initial-Snapshot erstellen
        for directory in self._directories:
            self._update_snapshot(directory)

        self._task = asyncio.create_task(self._watch_loop())

    async def stop(self) -> None:
        """Stoppt die Überwachung."""
        self._running = False

        if self._task:
            self._task.cancel()
            try:
                await self._task
            except asyncio.CancelledError:
                pass
            self._task = None

    async def _watch_loop(self) -> None:
        """Haupt-Überwachungs-Loop."""
        while self._running:
            try:
                await asyncio.sleep(self._poll_interval)

                for directory in list(self._directories):
                    if directory.exists():
                        await self._check_directory(directory)

            except asyncio.CancelledError:
                break
            except Exception:
                pass

    async def _check_directory(self, directory: Path) -> None:
        """Prüft ein Verzeichnis auf Änderungen."""
        current_files: dict[Path, float] = {}

        # Alle Dateien sammeln
        for path in self._get_files(directory):
            if self._should_watch(path):
                try:
                    current_files[path] = path.stat().st_mtime
                except OSError:
                    pass

        # Änderungen erkennen
        old_paths = {
            p for p in self._snapshot
            if str(p).startswith(str(directory))
        }
        new_paths = set(current_files.keys())

        # Gelöschte Dateien
        for path in old_paths - new_paths:
            event = WatchEvent(
                event_type=WatchEventType.DELETED,
                path=path,
                is_directory=False,
            )
            del self._snapshot[path]
            await self._notify(event)

        # Neue Dateien
        for path in new_paths - old_paths:
            event = WatchEvent(
                event_type=WatchEventType.CREATED,
                path=path,
                is_directory=path.is_dir(),
            )
            self._snapshot[path] = current_files[path]
            await self._notify(event)

        # Geänderte Dateien
        for path in new_paths & old_paths:
            if current_files[path] != self._snapshot[path]:
                event = WatchEvent(
                    event_type=WatchEventType.MODIFIED,
                    path=path,
                    is_directory=path.is_dir(),
                )
                self._snapshot[path] = current_files[path]
                await self._notify(event)

    def _get_files(self, directory: Path) -> list[Path]:
        """Gibt alle Dateien im Verzeichnis zurück."""
        files = []

        if self._recursive:
            for root, dirs, filenames in os.walk(directory):
                # Versteckte Verzeichnisse überspringen
                dirs[:] = [d for d in dirs if not d.startswith(".")]

                for filename in filenames:
                    files.append(Path(root) / filename)
        else:
            for item in directory.iterdir():
                if item.is_file():
                    files.append(item)

        return files

    def _should_watch(self, path: Path) -> bool:
        """Prüft, ob eine Datei überwacht werden soll."""
        # Versteckte Dateien ignorieren
        if path.name.startswith("."):
            return False

        # __pycache__ ignorieren
        if "__pycache__" in str(path):
            return False

        # Extension-Filter
        if self._extensions:
            return path.suffix.lower() in self._extensions

        return True

    def _update_snapshot(self, directory: Path) -> None:
        """Aktualisiert den Snapshot für ein Verzeichnis."""
        for path in self._get_files(directory):
            if self._should_watch(path):
                try:
                    self._snapshot[path] = path.stat().st_mtime
                except OSError:
                    pass

    async def _notify(self, event: WatchEvent) -> None:
        """Benachrichtigt alle Callbacks."""
        for callback in self._callbacks:
            try:
                callback(event)
            except Exception:
                pass

        for callback in self._async_callbacks:
            try:
                result = callback(event)
                if asyncio.iscoroutine(result):
                    await result
            except Exception:
                pass

    def get_snapshot(self) -> dict[str, float]:
        """Gibt den aktuellen Snapshot zurück."""
        return {str(k): v for k, v in self._snapshot.items()}

    def clear_callbacks(self) -> None:
        """Entfernt alle Callbacks."""
        self._callbacks.clear()
        self._async_callbacks.clear()