lib.watcher

  1"""
  2Observador reactivo de ficheros YAML.
  3
  4Usa watchdog (inotify en Linux, FSEvents en macOS, ReadDirectoryChanges en Windows)
  5para detectar cambios en tiempo real sin polling.
  6
  7Uso:
  8    from lib.watcher import start_watch, WatchdogNotAvailable
  9
 10    def on_change(yaml_path):
 11        rebuild(yaml_path)
 12
 13    try:
 14        start_watch([Path('courses/java/content')], on_change)
 15    except WatchdogNotAvailable:
 16        print("instala watchdog: uv sync --group watch")
 17"""
 18
 19from __future__ import annotations
 20
 21import threading
 22from pathlib import Path
 23from typing import Callable
 24
 25try:
 26    from watchdog.observers import Observer
 27    from watchdog.events import FileSystemEventHandler, FileModifiedEvent
 28    WATCHDOG_AVAILABLE = True
 29except ImportError:
 30    WATCHDOG_AVAILABLE = False
 31    Observer = None                    # type: ignore[assignment,misc]
 32    FileSystemEventHandler = object    # type: ignore[assignment,misc]
 33
 34
 35class WatchdogNotAvailable(RuntimeError):
 36    """Se lanza si watchdog no está instalado."""
 37
 38
 39class _YamlChangeHandler(FileSystemEventHandler):
 40    """Dispara un callback con debounce cuando cambia un YAML vigilado."""
 41
 42    def __init__(
 43        self,
 44        yaml_set: set[Path],
 45        on_change: Callable[[Path], None],
 46        debounce_s: float = 0.3,
 47    ) -> None:
 48        super().__init__()
 49        self._yaml_set   = yaml_set
 50        self._on_change  = on_change
 51        self._debounce   = debounce_s
 52        self._pending: set[Path] = set()
 53        self._lock  = threading.Lock()
 54        self._timer: threading.Timer | None = None
 55
 56    def on_modified(self, event) -> None:
 57        if getattr(event, 'is_directory', False):
 58            return
 59        p = Path(event.src_path).resolve()
 60        if p in self._yaml_set:
 61            self._schedule(p)
 62
 63    def _schedule(self, path: Path) -> None:
 64        with self._lock:
 65            self._pending.add(path)
 66            if self._timer is not None:
 67                self._timer.cancel()
 68            self._timer = threading.Timer(self._debounce, self._flush)
 69            self._timer.start()
 70
 71    def _flush(self) -> None:
 72        with self._lock:
 73            to_build = list(self._pending)
 74            self._pending.clear()
 75        for p in to_build:
 76            try:
 77                self._on_change(p)
 78            except Exception as exc:
 79                import sys
 80                print(f'[watch] Error en {p.name}: {exc}', file=sys.stderr)
 81
 82    def cancel_pending(self) -> None:
 83        with self._lock:
 84            if self._timer is not None:
 85                self._timer.cancel()
 86
 87
 88def start_watch(
 89    yaml_paths: list[Path],
 90    on_change: Callable[[Path], None],
 91    debounce_s: float = 0.3,
 92) -> None:
 93    """Inicia el observer y bloquea hasta Ctrl+C.
 94
 95    Args:
 96        yaml_paths: Lista de ficheros YAML a vigilar.
 97        on_change:  Callback llamado con el Path del YAML modificado.
 98        debounce_s: Segundos de espera para agrupar cambios rápidos.
 99
100    Raises:
101        WatchdogNotAvailable: si watchdog no está instalado.
102    """
103    if not WATCHDOG_AVAILABLE:
104        raise WatchdogNotAvailable(
105            "watchdog no está instalado.\n"
106            "Instalar con: uv sync --group watch"
107        )
108
109    resolved = [p.resolve() for p in yaml_paths]
110    yaml_set = set(resolved)
111    dirs     = {p.parent for p in resolved}
112
113    handler  = _YamlChangeHandler(yaml_set, on_change, debounce_s)
114    observer = Observer()
115    for d in dirs:
116        observer.schedule(handler, str(d), recursive=False)
117
118    observer.start()
119    try:
120        observer.join()
121    except KeyboardInterrupt:
122        observer.stop()
123        handler.cancel_pending()