feat: Add dir monitor

monitor_dir.py

@@ -0,0 +1,269 @@
+#!/usr/bin/env -S uv run
+# /// script
+# requires-python = ">=3.8"
+# dependencies = [
+#   "watchdog"
+# ]
+# ///
+
+import argparse
+import dataclasses
+import logging
+import os
+from pathlib import Path
+import re
+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")
+
+
+@dataclasses.dataclass
+class EventHandler:
+    pattern: re.Pattern
+    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, EventHandler], ignore_patterns: list[re.Pattern]):
+        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):
+        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())
+        args = [
+            event_name,
+            str(Path(src_path).resolve()),
+            *([str(Path(dest_path).resolve())] if dest_path else []),
+        ]
+        if dest_path:
+            env_vars["DEST_PATH"] = str(Path(dest_path).resolve())
+        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.resolve()
+        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):
+        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):
+        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,
+            path_to_filter_on=Path(event.src_path),
+            src_path=Path(event.src_path),
+        )
+
+    def on_deleted(self, event: events.FileSystemEvent):
+        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,
+            path_to_filter_on=Path(event.src_path),
+            src_path=Path(event.src_path),
+        )
+
+    def on_moved(self, event: events.FileSystemEvent):
+        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.event_type,
+            path_to_filter_on=dest_path,
+            src_path=src_path,
+            dest_path=dest_path,
+        )
+
+
+def parse_args():
+    parser = argparse.ArgumentParser(
+        description="Monitors a folder for changes and executes handlers based on event type and regex filters.", formatter_class=argparse.RawTextHelpFormatter
+    )
+    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("--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.")
+    return parser.parse_args()
+
+
+def main():
+    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, 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:
+                handler: EventHandler
+                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}")
+
+    event_handler = ChangeDispatchHandler(
+        base_dir=args.dir.resolve(),
+        handlers_config=handlers,
+        ignore_patterns=args.ignore_patterns,
+    )
+    observer = PollingObserver(timeout=args.poll)
+    observer.schedule(event_handler, args.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()

process_new_videos.py

@@ -0,0 +1,54 @@
+#!/usr/bin/env python3
+import sys
+from pathlib import Path
+import shutil
+import subprocess
+import logging
+
+# Configure logging
+logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", datefmt="%Y-%m-%d %H:%M:%S")
+logger = logging.getLogger(__name__)
+
+TARGET_DIR = Path("/mnt/box/files/_raw/prt/")
+SCAN_SCRIPT_PATH = Path("~/_scripts/scan_vids.py").expanduser()
+VIDEO_EXTENSIONS = {".mkv", ".mp4"}
+
+
+def main():
+    # Simplified: Assume correct arguments are passed
+    event_name = sys.argv[1]
+
+    src_path = Path(sys.argv[2]).resolve()
+
+    logger.info(f"Received event: {event_name}, Source path: {src_path}")
+
+    if event_name.lower() == "moved":
+        src_path = Path(sys.argv[3]).resolve()
+
+    if not src_path.is_file() or src_path.suffix.lower() not in VIDEO_EXTENSIONS:
+        logger.info(f"Ignoring non-video file or non-existent file: {src_path.name}")
+        exit(0)
+        return
+
+    logger.info(f"Processing created video file: {src_path}")
+
+    target_path = TARGET_DIR / src_path.name
+
+    logger.info(f"Moving '{src_path}' to '{target_path}'")
+    src_path.rename(target_path)
+
+    subprocess.run(
+        [str(SCAN_SCRIPT_PATH), str(target_path)],
+        text=True,
+        check=True,
+    )
+
+    has_other_videos = any(f.is_file() and f.suffix.lower() in VIDEO_EXTENSIONS for f in src_path.parent.iterdir())
+    has_large_files = any(f.is_file() and f.stat().st_size > 30 * 1024 * 1024 for f in src_path.parent.iterdir())
+    if not (has_other_videos or has_large_files):
+        logger.info(f"No other video files or large files in '{src_path.parent}'. Deleting it.")
+        shutil.rmtree(src_path.parent)
+
+
+if __name__ == "__main__":
+    main()