app/watcher.py

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