From 9a3b78105ed319e4f8be7166e5e61dde939cb6ae Mon Sep 17 00:00:00 2001 From: Abdussamet Kocak Date: Sun, 4 Oct 2026 08:25:47 +0200 Subject: [PATCH] feat(monitor_dir): Add path/inode polling mode option --- monitor_dir.py | 112 +++++++++++++++++++++++++++++++++----------- monitor_dir_test.py | 91 +++++++++++++++++++++++++++++++++++ 2 files changed, 175 insertions(+), 28 deletions(-) create mode 100644 monitor_dir_test.py diff --git a/monitor_dir.py b/monitor_dir.py index 8eda146..b542f55 100755 --- a/monitor_dir.py +++ b/monitor_dir.py @@ -1,6 +1,6 @@ -#!/usr/bin/env -S uv run +#!/usr/bin/env -S uv run --script # /// script -# requires-python = ">=3.8" +# requires-python = ">=3.11" # dependencies = [ # "watchdog" # ] @@ -12,20 +12,20 @@ import logging import os from pathlib import Path import re +import stat import subprocess import sys import time from typing import Self -from watchdog import observers from watchdog.observers.polling import PollingObserver from watchdog import events -logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s: %(message)s", datefmt="%Y-%m-%d %H:%M:%S") +logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s: %(message)s") @dataclasses.dataclass class EventHandler: - pattern: re.Pattern + pattern: re.Pattern[str] script_path: Path @classmethod @@ -51,24 +51,24 @@ class EventHandler: class ChangeDispatchHandler(events.FileSystemEventHandler): - def __init__(self, base_dir: Path, handlers_config: dict[str, EventHandler], ignore_patterns: list[re.Pattern]): + 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: str, event_name: str, src_path: str, dest_path: str = None): + 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(Path(src_path).resolve()) + env_vars["SRC_PATH"] = str(src_path.absolute()) args = [ event_name, - str(Path(src_path).resolve()), - *([str(Path(dest_path).resolve())] if dest_path else []), + str(src_path.absolute()), + *([str(dest_path.absolute())] if dest_path else []), ] if dest_path: - env_vars["DEST_PATH"] = str(Path(dest_path).resolve()) + env_vars["DEST_PATH"] = str(dest_path.absolute()) elif "DEST_PATH" in env_vars: del env_vars["DEST_PATH"] @@ -93,7 +93,7 @@ class ChangeDispatchHandler(events.FileSystemEventHandler): 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.resolve() + abs_path_to_check = path_to_check.absolute() for pattern in self.ignore_patterns: if pattern.search(str(abs_path_to_check)): return True @@ -107,7 +107,7 @@ class ChangeDispatchHandler(events.FileSystemEventHandler): pass return False - def _process_event_type(self, event_name: str, path_to_filter_on: Path, src_path: Path, dest_path: Path | None = None): + 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 @@ -127,27 +127,27 @@ class ChangeDispatchHandler(events.FileSystemEventHandler): dest_path=dest_path, ) - def on_created(self, event: events.FileSystemEvent): + 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.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): + 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.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.FileSystemEvent): + 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}") @@ -155,19 +155,62 @@ class ChangeDispatchHandler(events.FileSystemEventHandler): dest_path = Path(event.dest_path) self._process_event_type( - event.event_type, + event_name=event.event_type, path_to_filter_on=dest_path, src_path=src_path, dest_path=dest_path, ) -def parse_args(): +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.RawTextHelpFormatter + 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=int, default=10, help="Polling interval in seconds.") + 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", @@ -206,10 +249,15 @@ def parse_args(): default=[], ) parser.add_argument("--debug", action="store_true", help="Enable debug logging.") - return parser.parse_args() + 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(): +def main() -> None: args = parse_args() if args.debug: @@ -220,7 +268,7 @@ def main(): logging.error(f"Monitored directory '{args.dir}' (resolved to '{args.dir.resolve()}') not found or not a directory.") sys.exit(1) - handlers: dict[str, EventHandler] = { + handlers: dict[str, list[EventHandler]] = { "created": args.created_handlers, "deleted": args.deleted_handlers, "moved": args.moved_handlers, @@ -235,7 +283,6 @@ def main(): if handlers_list: logging.info(f"Registered {len(handlers_list)} handler(s) for '{event_type}' events.") for handler in handlers_list: - handler: EventHandler logging.info(f" - Filter: '{handler.pattern.pattern}', Script: '{handler.script_path}'") if args.ignore_patterns: @@ -243,13 +290,22 @@ def main(): for pattern in args.ignore_patterns: logging.info(f" - {pattern.pattern}") + base_dir = args.dir.resolve() event_handler = ChangeDispatchHandler( - base_dir=args.dir.resolve(), + 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, args.dir, recursive=args.recursive) + 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.") diff --git a/monitor_dir_test.py b/monitor_dir_test.py new file mode 100644 index 0000000..3740a12 --- /dev/null +++ b/monitor_dir_test.py @@ -0,0 +1,91 @@ +#!/usr/bin/env -S uv run --script +# /// script +# dependencies = ["pytest", "watchdog"] +# /// + +from pathlib import Path +import stat +import sys +from types import SimpleNamespace +from unittest.mock import Mock, call, patch + +import pytest + +import monitor_dir as script + + +@pytest.mark.parametrize("recursive", [False, True]) +def test_recursive_scan_does_not_follow_symlinks(tmp_path: Path, recursive: bool) -> None: + child = tmp_path / "child" + child.mkdir() + file_path = child / "file.txt" + file_path.touch() + link = tmp_path / "loop" + link.symlink_to(target=tmp_path, target_is_directory=True) + + expected = {child: True, link: False} + if recursive: + expected[file_path] = False + assert script.take_snapshot(base_dir=tmp_path, recursive=recursive) == expected + + +def test_changing_inode_numbers_do_not_emit_events(tmp_path: Path) -> None: + file_path = tmp_path / "file.txt" + file_path.touch() + handler = Mock() + with patch.object( + target=Path, attribute="lstat", + side_effect=[ + SimpleNamespace(st_mode=stat.S_IFREG, st_ino=1), + SimpleNamespace(st_mode=stat.S_IFREG, st_ino=999), + ], + ): + previous = script.take_snapshot(base_dir=tmp_path, recursive=False) + current = script.take_snapshot(base_dir=tmp_path, recursive=False) + + script.dispatch_changes(previous=previous, current=current, event_handler=handler) + handler._process_event_type.assert_not_called() + + +def test_rename_emits_deleted_and_created() -> None: + source = Path("/watched/before.txt") + destination = Path("/watched/after.txt") + handler = Mock() + script.dispatch_changes(previous={source: False}, current={destination: False}, event_handler=handler) + assert handler._process_event_type.call_args_list == [ + call(event_name="deleted", path_to_filter_on=source, src_path=source), + call(event_name="created", path_to_filter_on=destination, src_path=destination), + ] + + +def test_failed_scan_preserves_last_snapshot() -> None: + base_dir = Path("/watched") + existing = base_dir / "existing.txt" + created = base_dir / "created.txt" + handler = Mock() + with patch.object( + target=script, attribute="take_snapshot", + side_effect=[{existing: False}, OSError("SSHFS disconnected"), {existing: False, created: False}], + ), patch.object( + target=script.time, attribute="sleep", side_effect=[None, None, KeyboardInterrupt], + ): + with pytest.raises(KeyboardInterrupt): + script.monitor_paths(base_dir=base_dir, recursive=True, interval=1, event_handler=handler) + handler._process_event_type.assert_called_once_with( + event_name="created", path_to_filter_on=created, src_path=created, + ) + + +def test_partial_scan_is_discarded(tmp_path: Path) -> None: + (tmp_path / "first.txt").touch() + (tmp_path / "second.txt").touch() + with patch.object( + target=Path, attribute="lstat", + side_effect=[SimpleNamespace(st_mode=stat.S_IFREG), PermissionError("scan interrupted")], + ): + with pytest.raises(PermissionError): + script.take_snapshot(base_dir=tmp_path, recursive=True) + + +if __name__ == "__main__": + sys.exit(pytest.main(args=[str(Path(__file__).resolve()), *sys.argv[1:]]))