Folder watcher with safe triggers

The problem

A scanner drops PDFs into \\fileserver\Scans\Incoming. You want them renamed with a timestamp and moved into a dated sub-folder — but only after the scanner finishes writing. Otherwise you move a half-written file and break the downstream OCR.

This recipe uses watchdog with a small debounce window so a file has to sit quiet for N seconds before being touched.

Install

pip install watchdog

The script

import shutil
import time
from datetime import datetime
from pathlib import Path
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler

WATCH  = Path(r"\\fileserver\Scans\Incoming")
TARGET = Path(r"\\fileserver\Scans\Processed")
QUIET_SECONDS = 4  # how long a file must be untouched before we move it


class ScanHandler(FileSystemEventHandler):
    def __init__(self):
        self._pending: dict[Path, float] = {}

    def on_created(self, event):
        if event.is_directory:
            return
        self._pending[Path(event.src_path)] = time.time()

    def on_modified(self, event):
        if event.is_directory:
            return
        self._pending[Path(event.src_path)] = time.time()

    def tick(self):
        """Call this every second from the main loop."""
        now = time.time()
        due = [p for p, t in self._pending.items() if now - t >= QUIET_SECONDS]
        for p in due:
            self._pending.pop(p, None)
            try:
                self._process(p)
            except Exception as exc:
                print(f"!! failed {p.name}: {exc}")

    def _process(self, src: Path):
        if not src.exists():
            return
        day = datetime.now().strftime("%Y-%m-%d")
        dst_dir = TARGET / day
        dst_dir.mkdir(parents=True, exist_ok=True)
        stamp = datetime.now().strftime("%H%M%S")
        dst = dst_dir / f"{stamp}-{src.name}"
        shutil.move(str(src), str(dst))
        print(f"-> {dst}")


def main():
    handler = ScanHandler()
    obs = Observer()
    obs.schedule(handler, str(WATCH), recursive=False)
    obs.start()
    print(f"watching {WATCH} ...")
    try:
        while True:
            time.sleep(1)
            handler.tick()
    except KeyboardInterrupt:
        obs.stop()
    obs.join()


if __name__ == "__main__":
    main()

Run it as a service

On Windows, wrap it with NSSM (nssm install scan-watcher python.exe C:\scripts\watch.py). On Linux, a minimal systemd unit works:

[Service]
ExecStart=/usr/bin/python3 /srv/watch.py
Restart=on-failure

Why the debounce matters

Most “file was created” events from Windows fire before the app has finished writing. Naive scripts race the scanner and end up moving a truncated file. The quiet-window trick — only act once a file has stopped changing for a few seconds — costs you a few seconds of latency but eliminates the race.

Extensions

• Post a Teams/Slack notification via an incoming webhook.
• Push to Azure Blob instead of a fileshare for off-site archive.
• Add a filter by extension (if src.suffix.lower() != ".pdf": return).