lib.watcher

  1# Copyright (C) 2026  David Vaquero <pepesan@gmail.com>
  2#
  3# This program is free software: you can redistribute it and/or modify
  4# it under the terms of the GNU General Public License as published by
  5# the Free Software Foundation, either version 3 of the License, or
  6# (at your option) any later version.
  7#
  8# This program is distributed in the hope that it will be useful,
  9# but WITHOUT ANY WARRANTY; without even the implied warranty of
 10# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 11# GNU General Public License for more details.
 12#
 13# You should have received a copy of the GNU General Public License
 14# along with this program.  If not, see <https://www.gnu.org/licenses/>.
 15
 16"""
 17Observador reactivo de ficheros YAML.
 18
 19Usa watchdog (inotify en Linux, FSEvents en macOS, ReadDirectoryChanges en Windows)
 20para detectar cambios en tiempo real sin polling.
 21
 22Uso:
 23    from lib.watcher import start_watch, WatchdogNotAvailable
 24
 25    def on_change(yaml_path):
 26        rebuild(yaml_path)
 27
 28    try:
 29        start_watch([Path('courses/java/content')], on_change)
 30    except WatchdogNotAvailable:
 31        print("instala watchdog: uv sync --group watch")
 32"""
 33
 34from __future__ import annotations
 35
 36import threading
 37from pathlib import Path
 38from typing import Callable
 39
 40try:
 41    from watchdog.observers import Observer
 42    from watchdog.events import FileSystemEventHandler, FileModifiedEvent
 43    WATCHDOG_AVAILABLE = True
 44except ImportError:
 45    WATCHDOG_AVAILABLE = False
 46    Observer = None                    # type: ignore[assignment,misc]
 47    FileSystemEventHandler = object    # type: ignore[assignment,misc]
 48
 49
 50class WatchdogNotAvailable(RuntimeError):
 51    """Se lanza si watchdog no está instalado."""
 52
 53
 54class _YamlChangeHandler(FileSystemEventHandler):
 55    """Dispara un callback con debounce cuando cambia un YAML vigilado."""
 56
 57    def __init__(
 58        self,
 59        yaml_set: set[Path],
 60        on_change: Callable[[Path], None],
 61        debounce_s: float = 0.3,
 62    ) -> None:
 63        super().__init__()
 64        self._yaml_set   = yaml_set
 65        self._on_change  = on_change
 66        self._debounce   = debounce_s
 67        self._pending: set[Path] = set()
 68        self._lock  = threading.Lock()
 69        self._timer: threading.Timer | None = None
 70
 71    def _check(self, path: str) -> None:
 72        p = Path(path).resolve()
 73        if p in self._yaml_set:
 74            self._schedule(p)
 75
 76    def on_modified(self, event) -> None:
 77        if not getattr(event, 'is_directory', False):
 78            self._check(event.src_path)
 79
 80    def on_created(self, event) -> None:
 81        # Algunos editores guardan escribiendo un fichero nuevo
 82        if not getattr(event, 'is_directory', False):
 83            self._check(event.src_path)
 84
 85    def on_moved(self, event) -> None:
 86        # Guardado atómico: escribe en tmp y renombra al path original (vim, gedit, PyCharm…)
 87        if not getattr(event, 'is_directory', False):
 88            self._check(event.dest_path)
 89
 90    def _schedule(self, path: Path) -> None:
 91        with self._lock:
 92            self._pending.add(path)
 93            if self._timer is not None:
 94                self._timer.cancel()
 95            self._timer = threading.Timer(self._debounce, self._flush)
 96            self._timer.start()
 97
 98    def _flush(self) -> None:
 99        with self._lock:
100            to_build = list(self._pending)
101            self._pending.clear()
102        for p in to_build:
103            try:
104                self._on_change(p)
105            except Exception as exc:
106                import sys
107                print(f'[watch] Error en {p.name}: {exc}', file=sys.stderr)
108
109    def cancel_pending(self) -> None:
110        with self._lock:
111            if self._timer is not None:
112                self._timer.cancel()
113
114
115def start_watch(
116    yaml_paths: list[Path],
117    on_change: Callable[[Path], None],
118    debounce_s: float = 0.3,
119) -> None:
120    """Inicia el observer y bloquea hasta Ctrl+C.
121
122    Args:
123        yaml_paths: Lista de ficheros YAML a vigilar.
124        on_change:  Callback llamado con el Path del YAML modificado.
125        debounce_s: Segundos de espera para agrupar cambios rápidos.
126
127    Raises:
128        WatchdogNotAvailable: si watchdog no está instalado.
129    """
130    if not WATCHDOG_AVAILABLE:
131        raise WatchdogNotAvailable(
132            "watchdog no está instalado.\n"
133            "Instalar con: uv sync --group watch"
134        )
135
136    resolved = [p.resolve() for p in yaml_paths]
137    yaml_set = set(resolved)
138    dirs     = {p.parent for p in resolved}
139
140    handler  = _YamlChangeHandler(yaml_set, on_change, debounce_s)
141    observer = Observer()
142    for d in dirs:
143        observer.schedule(handler, str(d), recursive=False)
144
145    observer.start()
146    try:
147        while observer.is_alive():
148            observer.join(timeout=1)
149    except KeyboardInterrupt:
150        pass
151    finally:
152        observer.stop()
153        handler.cancel_pending()
154        observer.join()