Python implements file system monitoring based on watchdog library

import sys
# Import logging module to record the log information when the program is runningimport logging
# Import the Observer class from the module to monitor changes in the file systemfrom  import Observer
# Import the LoggingEventHandler class from the module to handle file system events and log logsfrom  import LoggingEventHandler

if __name__ == "__main__":
    # Configure basic settings for logging    # level= indicates that only log information at INFO level and above is recorded    # format='%(asctime)s - %(message)s' specifies the output format of the log, including timestamps and log messages    # datefmt='%Y-%m-%d %H:%M:%S' Specify the format of the timestamp    (level=,
                        format='%(asctime)s - %(message)s',
                        datefmt='%Y-%m-%d %H:%M:%S')
    # Specify the folder path to monitor, which must exist    path = "demo_folder"
    # Create a LoggingEventHandler object to handle file system events and log logs    event_handler = LoggingEventHandler()
    # Create an Observer object to monitor changes in the file system    observer = Observer()
    # Arrange observers to monitor file system events under specified paths    # event_handler is an event handler    # path is the path to be monitored    # recursive=True means recursively monitor all subfolders under this path    (event_handler, path, recursive=True)
    # Start the observer and start monitoring file system changes    ()
    try:
        # Continuous loop when the observer is active        while observer.is_alive():
            # Wait for 1 second to give observers time to process file system events            (1)
    finally:
        # Stop the observer and end the file system monitoring        ()
        # Wait for the observer thread to end        ()

# This class is used to handle file system events, such as file creation, deletion, modification, etc.from  import FileSystemEventHandler
from  import Observer

# Define a custom file event handling class, inherited from FileSystemEventHandlerclass FileEventHandler(FileSystemEventHandler):
    def __init__(self):
        FileSystemEventHandler.__init__(self)
        # Initialize an empty list to store file names        self.file_names = []  

    # Trigger this method when a file or directory is moved    def on_moved(self, event):
        # Determine whether it is a directory move        if event.is_directory:
            # Print directory movement information            print(f"Directory from {event.src_path} Move to {event.dest_path}")
        else:
            # Print file movement information            print(f"File from {event.src_path} Move to {event.dest_path}")

    # This method is triggered when a file or directory is created    def on_created(self, event):
        # Determine whether it is directory creation        if event.is_directory:
            # Print directory creation information            print(f"Directory created: {event.src_path}")
        else:
            # Print the information about file creation            print(f"File created: {event.src_path}")
            # Extract file name from the full path of the file            file_full_name = str(event.src_path.split('/')[-1])
            if file_full_name.endswith('.csv'):
                # If it is a csv file, add it to the file name list                self.file_names.append(file_full_name)
            # Print a list of file names            print(self.file_names)

    # This method is triggered when a file or directory is deleted    def on_deleted(self, event):
        # Determine whether it is directory deletion        if event.is_directory:
            # Print directory deletion information            print(f"Directory has been deleted: {event.src_path}")
        else:
            # Print file deletion information            print(f"File deleted: {event.src_path}")

    # This method is triggered when a file or directory is modified. The comment code indicates that the modification event is not triggered.    # def on_modified(self, event):
    #     pass

if __name__ == "__main__":
    # Create an Observer object to monitor changes in the file system    observer = Observer()
    # Create a custom file event processing object    event_handler = FileEventHandler()
    # Define the directory to monitor    watch_directory = "demo_folder"
    # Add the event processing object and the directory to be monitored to the observer and set it to recursively monitor    (event_handler, watch_directory, recursive=True)
    # Start the observer and start monitoring file system changes    ()
    try:
        # Continuous loop when the observer is active        while observer.is_alive():
            # Wait for 1 second to give observers time to process file system events            (1)
    finally:
        # Stop the observer and end the file system monitoring        ()
        # Wait for the observer thread to end        ()

import time
import os
import threading
from  import Observer
from  import *
from  import DirectorySnapshot, DirectorySnapshotDiff


class FileChangeHandler(FileSystemEventHandler):
    def __init__(self, monitored_path, delay=0.5):
        # Call the parent class constructor        super().__init__()
        # The path to the monitored directory        self.monitored_path = monitored_path
         = delay
        # Timer object for delay checking        self.delay_timer = None
        # Snapshot is used to record the status of the specified directory at a certain point in time, including information about all files and subdirectories in the directory.        # Record the initial snapshot of the directory        self.initial_snapshot = DirectorySnapshot(self.monitored_path)

    def on_any_event(self, event):
        # Reduce unnecessary checks through timers        # If the timer already exists, cancel it        if self.delay_timer:
            self.delay_timer.cancel()
        # Create a new timer and perform the operation to check the snapshot after delaying it        # Detect file system changes through snapshots        self.delay_timer = (, self._check_directory_changes)
        self.delay_timer.start()

    def _check_directory_changes(self):
        # Get a new snapshot of the current directory        new_snapshot = DirectorySnapshot(self.monitored_path)
        # Calculate the difference between old and new snapshots        snapshot_difference = DirectorySnapshotDiff(self.initial_snapshot, new_snapshot)
        # Update the initial snapshot to a new snapshot        self.initial_snapshot = new_snapshot
        # Clear the timer        self.delay_timer = None
        # Print directory change information        self._print_changes(snapshot_difference)

    def _print_changes(self, diff):
        # Print creation, deletion, modification and movement of files and directories        print("Created file:", diff.files_created)
        print("Deleted File:", diff.files_deleted)
        print("Modified file:", diff.files_modified)
        print("Moving file:", diff.files_moved)
        print("Modified Directory:", diff.dirs_modified)
        print("Moving Directory:", diff.dirs_moved)
        print("Deleted Directory:", diff.dirs_deleted)
        print("Created Directory:", diff.dirs_created)

class DirectoryMonitor:
    def __init__(self, monitored_path):
        # The path to the monitored directory        self.monitored_path = monitored_path
        # Create an observer object        self.directory_observer = Observer()

    def start_monitoring(self):
        # Create file change processing object        change_handler = FileChangeHandler(self.monitored_path)
        # Arrange observers to monitor the specified directory and recursively monitor the subdirectories        self.directory_observer.schedule(change_handler, self.monitored_path, recursive=True)
        # Start the observer        self.directory_observer.start()

    def stop_monitoring(self):
        # Stop the observer        self.directory_observer.stop()
        # Wait for the observer thread to end        self.directory_observer.join()

if __name__ == "__main__":
    monitor = DirectoryMonitor("demo_folder")
    # Start monitoring    monitor.start_monitoring()
    try:
        while True:
            (1)
    except KeyboardInterrupt:
        monitor.stop_monitoring()

from  import Observer
from  import FileSystemEventHandler
class CustomFileEventHandler(FileSystemEventHandler):
    def on_created(self, event):
            patterns = [“*.jpg”， “*.png”， “*.gif”]  # Only monitor image files def on_modified(self, event): print(f "The image file has changed: {event.src_path}")        entity = "directory" if event.is_directory else "file"
        print(f"{entity} on_created: {event.src_path}")

if __name__ == "__main__":
    file_observer = Observer()
    monitored_directories = ['demo_folder', 'demo_folder2']
    file_event_handler = CustomFileEventHandler()
    for directory in monitored_directories:
        # Register event handlers for each directory        file_observer.schedule(file_event_handler, directory, recursive=True)
    file_observer.start()
    try:
        # Continuous loop when the observer is active        while file_observer.is_alive():
            # Wait for 1 second to give observers time to process file system events            file_observer.join(1)
    finally:
        # Stop the observer and end the file system monitoring        file_observer.stop()
        # Wait for the observer thread to end        file_observer.join()