playground/monitor_dir.py

#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.11"
# dependencies = [
#   "watchdog"
# ]
# ///

import argparse
import dataclasses
import logging
import os
from pathlib import Path
import re
import stat
import subprocess
import sys
import time
from typing import Self
from watchdog.observers.polling import PollingObserver
from watchdog import events

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s: %(message)s")


@dataclasses.dataclass
class EventHandler:
    pattern: re.Pattern[str]
    script_path: Path

    @classmethod
    def parse(cls, handler_spec: str) -> Self:
        try:
            pattern, script = handler_spec.rsplit(":", 1)
        except ValueError:
            raise ValueError(f"Invalid handler format: '{handler_spec}'. Expected REGEX_FILTER:/path/to/script")

        if not pattern.strip():
            raise ValueError(f"Empty filter regex in handler: '{handler_spec}'")
        if not script.strip():
            raise ValueError(f"Empty script path in handler: '{handler_spec}'")

        try:
            compiled = re.compile(pattern)
        except re.error as e:
            raise ValueError(f"Invalid regex pattern '{pattern}': {e}")

        script_path = Path(script.strip()).expanduser().resolve()

        return cls(pattern=compiled, script_path=script_path)


class ChangeDispatchHandler(events.FileSystemEventHandler):
    def __init__(self, base_dir: Path, handlers_config: dict[str, list[EventHandler]], ignore_patterns: list[re.Pattern[str]]) -> None:
        super().__init__()
        self.base_dir = base_dir
        self.handlers_config = handlers_config
        self.ignore_patterns = ignore_patterns if ignore_patterns else []

    def _execute_script(self, script_path: Path, event_name: str, src_path: Path, dest_path: Path | None = None) -> None:
        env_vars = os.environ.copy()
        env_vars["MONITORED_DIR"] = str(self.base_dir)
        env_vars["EVENT_TYPE"] = event_name.upper()
        env_vars["SRC_PATH"] = str(src_path.absolute())
        args = [
            event_name,
            str(src_path.absolute()),
            *([str(dest_path.absolute())] if dest_path else []),
        ]
        if dest_path:
            env_vars["DEST_PATH"] = str(dest_path.absolute())
        elif "DEST_PATH" in env_vars:
            del env_vars["DEST_PATH"]

        logging.info(f"Executing script '{script_path}' for event {event_name.upper()} on '{src_path}'" + (f" -> '{dest_path}'" if dest_path else ""))
        try:
            proc = subprocess.run(args=[script_path, *args], env=env_vars, capture_output=True, text=True, check=False, shell=False)
            if proc.stdout:
                logging.info(f"Script '{script_path}' STDOUT:\n{proc.stdout.strip()}")
            if proc.stderr:
                logging.warning(f"Script '{script_path}' STDERR:\n{proc.stderr.strip()}")

            if proc.returncode == 0:
                logging.info(f"Script '{script_path}' completed successfully.")
            else:
                logging.error(f"Script '{script_path}' failed with exit code {proc.returncode}.")

        except FileNotFoundError:
            logging.error(f"Script '{script_path}' not found. Ensure it's a valid path and in PATH if not absolute.")
        except PermissionError:
            logging.error(f"Permission denied for script '{script_path}'. Ensure it is executable.")
        except Exception as e:
            logging.error(f"An unexpected error occurred while running script '{script_path}': {e}")

    def _is_ignored(self, path_to_check: Path) -> bool:
        abs_path_to_check = path_to_check.absolute()
        for pattern in self.ignore_patterns:
            if pattern.search(str(abs_path_to_check)):
                return True
            # Also check relative path to monitored dir, if applicable
            try:
                relative_path = abs_path_to_check.relative_to(self.base_dir)
                if pattern.search(str(relative_path)):
                    return True
            except ValueError:
                # Path is not inside base_dir, so relative path check is not applicable
                pass
        return False

    def _process_event_type(self, event_name: str, path_to_filter_on: Path, src_path: Path, dest_path: Path | None = None) -> None:
        if self._is_ignored(path_to_filter_on):
            logging.debug(f"Path '{path_to_filter_on}' is ignored by configured patterns.")
            return

        handlers = self.handlers_config.get(event_name, [])
        matched_handlers = [it for it in handlers if it.pattern.search(str(path_to_filter_on))]
        if not matched_handlers:
            logging.debug(f"No handlers matched for event '{event_name}' on path '{path_to_filter_on}'.")
            return

        for handler in matched_handlers:
            logging.info(f"Event '{event_name}': Path '{path_to_filter_on}' matched regex '{handler.pattern.pattern}'.")
            self._execute_script(
                script_path=handler.script_path,
                event_name=event_name,
                src_path=src_path,
                dest_path=dest_path,
            )

    def on_created(self, event: events.FileSystemEvent) -> None:
        what = "directory" if event.is_directory else "file"
        logging.debug(f"Watchdog event: CREATED {what} at {event.src_path}")

        self._process_event_type(
            event_name=event.event_type,
            path_to_filter_on=Path(event.src_path),
            src_path=Path(event.src_path),
        )

    def on_deleted(self, event: events.FileSystemEvent) -> None:
        what = "directory" if event.is_directory else "file"
        logging.debug(f"Watchdog event: DELETED {what} at {event.src_path}")

        self._process_event_type(
            event_name=event.event_type,
            path_to_filter_on=Path(event.src_path),
            src_path=Path(event.src_path),
        )

    def on_moved(self, event: events.FileSystemMovedEvent) -> None:
        what = "directory" if event.is_directory else "file"
        logging.debug(f"Watchdog event: MOVED {what} from {event.src_path} to {event.dest_path}")

        src_path = Path(event.src_path)
        dest_path = Path(event.dest_path)

        self._process_event_type(
            event_name=event.event_type,
            path_to_filter_on=dest_path,
            src_path=src_path,
            dest_path=dest_path,
        )


def take_snapshot(base_dir: Path, recursive: bool) -> dict[Path, bool]:
    if not stat.S_ISDIR(base_dir.stat().st_mode):
        raise NotADirectoryError(f"Monitored path '{base_dir}' is not a directory")
    snapshot: dict[Path, bool] = {}
    directories = [base_dir]
    while directories:
        directory = directories.pop()
        for path in directory.iterdir():
            is_directory = stat.S_ISDIR(path.lstat().st_mode)
            snapshot[path] = is_directory
            if recursive and is_directory:
                directories.append(path)
    return snapshot


def dispatch_changes(previous: dict[Path, bool], current: dict[Path, bool], event_handler: ChangeDispatchHandler) -> None:
    deleted = previous.keys() - current.keys()
    created = current.keys() - previous.keys()
    replaced = {path for path in previous.keys() & current.keys() if previous[path] != current[path]}
    for path in sorted(deleted | replaced):
        event_handler._process_event_type(event_name="deleted", path_to_filter_on=path, src_path=path)
    for path in sorted(created | replaced):
        event_handler._process_event_type(event_name="created", path_to_filter_on=path, src_path=path)


def monitor_paths(base_dir: Path, recursive: bool, interval: float, event_handler: ChangeDispatchHandler) -> None:
    previous: dict[Path, bool] | None = None
    while True:
        try:
            current = take_snapshot(base_dir=base_dir, recursive=recursive)
        except OSError as error:
            logging.warning(f"Directory scan failed; retaining previous snapshot and retrying in {interval}s: {error}")
        else:
            if previous is not None:
                dispatch_changes(previous=previous, current=current, event_handler=event_handler)
            previous = current
        time.sleep(interval)


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Monitors a folder for changes and executes handlers based on event type and regex filters.", formatter_class=argparse.ArgumentDefaultsHelpFormatter
    )
    parser.add_argument("dir", type=Path, help="The directory to monitor.")
    parser.add_argument("--poll", type=float, default=10, help="Polling interval in seconds; must be positive.")
    parser.add_argument(
        "--polling-mode", choices=["path", "inode"], default="path",
        help="Path polling tolerates changing SSHFS inode numbers and retries failed scans. Renames emit deleted/created events. Inode polling supports moved events and requires stable inode numbers.",
    )
    parser.add_argument("--recursive", action="store_true", help="Monitor the directory recursively.")
    parser.add_argument(
        "--created",
        action="append",
        dest="created_handlers",
        type=EventHandler.parse,
        metavar="FILTER:SCRIPT",
        help="Handler for file/directory creation events.\nFormat: REGEX_FILTER:/path/to/script\nExample: '.*\\.txt$:/usr/local/bin/process_text_file.sh'",
        default=[],
    )
    parser.add_argument(
        "--deleted",
        action="append",
        dest="deleted_handlers",
        type=EventHandler.parse,
        metavar="FILTER:SCRIPT",
        help="Handler for file/directory deletion events.\nFormat: REGEX_FILTER:/path/to/script",
        default=[],
    )
    parser.add_argument(
        "--moved",
        action="append",
        dest="moved_handlers",
        type=EventHandler.parse,
        metavar="FILTER:SCRIPT",
        help="Handler for file/directory move events.\nThe filter applies to the destination path.\nFormat: REGEX_FILTER:/path/to/script",
        default=[],
    )
    parser.add_argument(
        "--ignore",
        action="append",
        dest="ignore_patterns",
        type=re.compile,
        metavar="REGEX_PATTERN",
        help="Regex pattern for paths to ignore. Can be specified multiple times.",
        default=[],
    )
    parser.add_argument("--debug", action="store_true", help="Enable debug logging.")
    args = parser.parse_args()
    if not 0 < args.poll < float("inf"):
        parser.error("--poll must be a finite positive number")
    if args.moved_handlers and args.polling_mode != "inode":
        parser.error("--moved requires --polling-mode inode; for SSHFS, use --created and --deleted instead")
    return args


def main() -> None:
    args = parse_args()

    if args.debug:
        logging.getLogger().setLevel(logging.DEBUG)
        logging.debug("Debug logging enabled.")

    if not args.dir.is_dir():
        logging.error(f"Monitored directory '{args.dir}' (resolved to '{args.dir.resolve()}') not found or not a directory.")
        sys.exit(1)

    handlers: dict[str, list[EventHandler]] = {
        "created": args.created_handlers,
        "deleted": args.deleted_handlers,
        "moved": args.moved_handlers,
    }

    if not any(lst for lst in handlers.values()):
        logging.error("No valid handler scripts provided.")
        sys.exit(1)

    logging.info(f"Starting filesystem monitor for directory: '{args.dir}'")
    for event_type, handlers_list in handlers.items():
        if handlers_list:
            logging.info(f"Registered {len(handlers_list)} handler(s) for '{event_type}' events.")
            for handler in handlers_list:
                logging.info(f"  - Filter: '{handler.pattern.pattern}', Script: '{handler.script_path}'")

    if args.ignore_patterns:
        logging.info("Ignoring paths matching the following patterns:")
        for pattern in args.ignore_patterns:
            logging.info(f"  - {pattern.pattern}")

    base_dir = args.dir.resolve()
    event_handler = ChangeDispatchHandler(
        base_dir=base_dir,
        handlers_config=handlers,
        ignore_patterns=args.ignore_patterns,
    )
    if args.polling_mode == "path":
        logging.info("Monitor is now running with path polling. Press Ctrl+C to stop.")
        try:
            monitor_paths(base_dir=base_dir, recursive=args.recursive, interval=args.poll, event_handler=event_handler)
        except KeyboardInterrupt:
            logging.info("Monitor has been shut down.")
        return

    observer = PollingObserver(timeout=args.poll)
    observer.schedule(event_handler=event_handler, path=str(base_dir), recursive=args.recursive)
    observer.start()
    logging.info("Monitor is now running. Press Ctrl+C to stop.")

    try:
        while observer.is_alive():
            observer.join(timeout=1)
    except KeyboardInterrupt:
        logging.info("Keyboard interrupt received. Shutting down monitor...")
    finally:
        if observer.is_alive():
            observer.stop()
        observer.join()
        logging.info("Monitor has been shut down.")


if __name__ == "__main__":
    main()