""" watcher.py — File-system watcher for auto re-indexing ====================================================== Watches the documents directory for new, modified, or deleted files and automatically updates the ChromaDB index. Run standalone: python watcher.py Or import: from watcher import start_watcher_thread """ import os import time import threading from pathlib import Path try: from watchdog.observers import Observer from watchdog.events import FileSystemEventHandler, FileSystemEvent WATCHDOG_AVAILABLE = True except ImportError: WATCHDOG_AVAILABLE = False print("[Watcher] watchdog not installed — falling back to polling watcher.") from knowledge_api import index_single, delete_doc DOCS_DIR = os.getenv("DOCS_DIR", "./documents") SUPPORTED = {".pdf", ".docx", ".doc", ".txt", ".md", ".markdown"} # Debounce time in seconds (avoid re-indexing for rapid successive saves) DEBOUNCE_SECONDS = 2.0 # ── Watchdog handler ────────────────────────────────────────────────────────── class DocumentHandler(FileSystemEventHandler): """Handles file-system events and triggers indexing.""" def __init__(self): self._timers: dict = {} # path → Timer (debounce) def _debounced_index(self, path: str): """Cancel any pending timer for this path and set a new one.""" if path in self._timers: self._timers[path].cancel() timer = threading.Timer(DEBOUNCE_SECONDS, self._do_index, args=[path]) self._timers[path] = timer timer.start() @staticmethod def _do_index(path: str): print(f"[Watcher] Indexing: {Path(path).name}") try: index_single(path) print(f"[Watcher] ✓ Done: {Path(path).name}") except Exception as e: print(f"[Watcher] ✗ Error indexing {path}: {e}") def on_created(self, event: FileSystemEvent): if not event.is_directory and Path(event.src_path).suffix.lower() in SUPPORTED: print(f"[Watcher] New file detected: {Path(event.src_path).name}") self._debounced_index(event.src_path) def on_modified(self, event: FileSystemEvent): if not event.is_directory and Path(event.src_path).suffix.lower() in SUPPORTED: print(f"[Watcher] File modified: {Path(event.src_path).name}") self._debounced_index(event.src_path) def on_deleted(self, event: FileSystemEvent): if not event.is_directory and Path(event.src_path).suffix.lower() in SUPPORTED: fname = Path(event.src_path).name print(f"[Watcher] File deleted: {fname} — removing from index.") try: delete_doc(fname) except Exception as e: print(f"[Watcher] Error removing {fname}: {e}") def on_moved(self, event: FileSystemEvent): # Treat as delete old + create new if not event.is_directory: if Path(event.src_path).suffix.lower() in SUPPORTED: delete_doc(Path(event.src_path).name) if Path(event.dest_path).suffix.lower() in SUPPORTED: self._debounced_index(event.dest_path) # ── Polling fallback ────────────────────────────────────────────────────────── class PollingWatcher: """ Simple polling watcher for when watchdog isn't available. Checks file modification times every POLL_INTERVAL seconds. """ POLL_INTERVAL = 10 # seconds def __init__(self, docs_dir: str): self.docs_dir = Path(docs_dir) self._seen: dict = {} # path → mtime def run(self): print(f"[Watcher] Polling {self.docs_dir} every {self.POLL_INTERVAL}s …") while True: self._poll() time.sleep(self.POLL_INTERVAL) def _poll(self): current = {} for p in self.docs_dir.rglob("*"): if p.is_file() and p.suffix.lower() in SUPPORTED: current[str(p)] = p.stat().st_mtime # New or modified for path, mtime in current.items(): if path not in self._seen or self._seen[path] != mtime: print(f"[Watcher] Change detected: {Path(path).name}") try: index_single(path) except Exception as e: print(f"[Watcher] Error: {e}") # Deleted for path in set(self._seen) - set(current): fname = Path(path).name print(f"[Watcher] Deleted: {fname}") try: delete_doc(fname) except Exception as e: print(f"[Watcher] Error removing {fname}: {e}") self._seen = current # ── Public API ──────────────────────────────────────────────────────────────── def start_watcher(docs_dir: str = DOCS_DIR, blocking: bool = True): """ Start the file watcher. Set blocking=False to run in a background thread (used by web_app.py). """ os.makedirs(docs_dir, exist_ok=True) if WATCHDOG_AVAILABLE: handler = DocumentHandler() observer = Observer() observer.schedule(handler, path=docs_dir, recursive=True) observer.start() print(f"[Watcher] Watching {docs_dir} with watchdog …") try: if blocking: while True: time.sleep(1) # Non-blocking: caller is responsible for keeping process alive except KeyboardInterrupt: observer.stop() print("[Watcher] Stopped.") observer.join() else: pw = PollingWatcher(docs_dir) if blocking: pw.run() else: t = threading.Thread(target=pw.run, daemon=True) t.start() return t def start_watcher_thread(docs_dir: str = DOCS_DIR) -> threading.Thread: """ Start the watcher in a background daemon thread. Returns the thread (for monitoring only — it's a daemon so it stops with the process). """ t = threading.Thread(target=start_watcher, args=(docs_dir, True), daemon=True) t.start() return t if __name__ == "__main__": start_watcher(DOCS_DIR, blocking=True)