pcvisit-filetracker/watch.py

from pathlib import Path

from watchdog.events import FileSystemEvent, FileSystemEventHandler
from watchdog.observers import Observer

from move_files import refresh_and_move


class XmlChangeHandler(FileSystemEventHandler):
    def __init__(
        self,
        directory: Path,
        files_csv: Path,
        clients_csv: Path,
        desktop_directory: Path,
        ignore_timestamp: bool,
    ) -> None:
        self.directory = directory
        self.files_csv = files_csv
        self.clients_csv = clients_csv
        self.desktop_directory = desktop_directory
        self.ignore_timestamp = ignore_timestamp

    def on_created(self, event: FileSystemEvent) -> None:
        self.process_xml(event)

    def on_modified(self, event: FileSystemEvent) -> None:
        self.process_xml(event)

    def on_moved(self, event: FileSystemEvent) -> None:
        self.process_xml(event)

    def process_xml(self, event: FileSystemEvent) -> None:
        if event.is_directory or not Path(event.src_path).suffix.lower() == ".xml":
            return
        refresh_and_move(
            self.directory,
            self.files_csv,
            self.clients_csv,
            self.desktop_directory,
            self.ignore_timestamp,
        )


def run(
    directory: Path,
    files_csv: Path,
    clients_csv: Path,
    desktop_directory: Path,
    ignore_timestamp: bool = False,
) -> None:
    if not directory.is_dir():
        raise ValueError(f"Das Verzeichnis existiert nicht: {directory}")

    observer = Observer()
    observer.schedule(
        XmlChangeHandler(directory, files_csv, clients_csv, desktop_directory, ignore_timestamp),
        str(directory),
        recursive=True,
    )
    print(f"Überwachung des XML-Verzeichnisses gestartet: {directory}")
    print("Mit Strg+C kann die Überwachung beendet werden.")
    observer.start()
    try:
        while observer.is_alive():
            observer.join(1)
    except KeyboardInterrupt:
        observer.stop()
        print("Überwachung wurde manuell mit Strg+C beendet.")
    observer.join()