knowledge_agent/watcher.py

"""
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)