File size: 6,522 Bytes
58b74a0 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 | """
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) |