File size: 4,060 Bytes
2edb151 | 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 | 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
|