from __future__ import annotations import threading import time from collections.abc import Callable from dataclasses import dataclass from pathlib import Path from watchdog.events import FileSystemEvent, FileSystemEventHandler from watchdog.observers import Observer from app.config import ACCEPTED_SUFFIXES, Settings from app.pipeline import process_file OnBatch = Callable[[list[Path]], None] def is_ignored(path: Path) -> bool: name = path.name if name.startswith("."): return True if name.endswith(".tmp") or name.endswith(".part"): return True if ".syncthing." in name or name.startswith(".syncthing"): return True return path.suffix.lower() not in ACCEPTED_SUFFIXES @dataclass class _State: sig: tuple[int, float] last_change: float class IdleBatchWatcher: """Settle files until size+mtime are unchanged for idle_seconds, then batch.""" def __init__( self, inbox: Path, *, idle_seconds: float = 30.0, on_batch: OnBatch | None = None, ) -> None: self.inbox = Path(inbox) self.idle_seconds = idle_seconds self.on_batch = on_batch self._state: dict[Path, _State] = {} self._lock = threading.Lock() self._running = False self._observer: Observer | None = None self._thread: threading.Thread | None = None def note(self, path: Path, now: float) -> None: if not path.is_file() or is_ignored(path): return stat = path.stat() sig = (stat.st_size, stat.st_mtime) with self._lock: prev = self._state.get(path) if prev is None or prev.sig != sig: self._state[path] = _State(sig=sig, last_change=now) def tick(self, now: float | None = None) -> list[Path]: clock = time.monotonic() if now is None else now self.inbox.mkdir(parents=True, exist_ok=True) for path in self.inbox.iterdir(): self.note(path, clock) ready: list[Path] = [] with self._lock: for path, state in list(self._state.items()): if not path.is_file(): self._state.pop(path, None) continue if clock - state.last_change >= self.idle_seconds: ready.append(path) self._state.pop(path, None) if ready and self.on_batch is not None: self.on_batch(ready) return ready def start(self) -> None: if self._running: return self._running = True handler = _Handler(self) observer = Observer() observer.schedule(handler, str(self.inbox), recursive=False) observer.start() self._observer = observer self._thread = threading.Thread(target=self._loop, daemon=True) self._thread.start() def stop(self) -> None: self._running = False if self._observer is not None: self._observer.stop() self._observer.join(timeout=2) self._observer = None def _loop(self) -> None: while self._running: self.tick() time.sleep(min(0.25, max(0.05, self.idle_seconds / 4))) class _Handler(FileSystemEventHandler): def __init__(self, watcher: IdleBatchWatcher) -> None: self.watcher = watcher def on_any_event(self, event: FileSystemEvent) -> None: if event.is_directory: return path = Path(str(event.src_path)) self.watcher.note(path, time.monotonic()) def process_batch(paths: list[Path], settings: Settings) -> None: for path in paths: try: process_file(path, settings) except Exception: continue def start_inbox_watcher(settings: Settings) -> IdleBatchWatcher: watcher = IdleBatchWatcher( settings.inbox_dir, idle_seconds=settings.idle_seconds, on_batch=lambda paths: process_batch(paths, settings), ) watcher.start() return watcher