Flowcode Support Board

Code: Select all

#!python3
import sys
import time
import os
import shutil
import logging
import threading
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler

# --- Configuration ---
# You can modify the destination directory directly here if you prefer
# not to provide it as a command-line argument every time.
DEFAULT_DESTINATION_DIR = "/tmp/file_backups/"

class ChangeHandler(FileSystemEventHandler):
    """
    This class handles the file system events.
    It will be triggered when a file is modified.
    Includes a debounce mechanism to avoid multiple copies for a single save event.
    """
    def __init__(self, source_file, dest_dir, debounce_interval=1.0):
        """
        Initializes the event handler.
        :param source_file: The file to watch.
        :param dest_dir: The directory to copy the file to.
        :param debounce_interval: The number of seconds to wait for more changes before copying.
        """
        super().__init__()
        self.source_file_path = os.path.abspath(source_file)
        self.dest_dir = dest_dir
        self.debounce_interval = debounce_interval
        self.copy_timer = None # Timer to debounce file modifications

        # Ensure the destination directory exists
        os.makedirs(self.dest_dir, exist_ok=True)
        print(f"✅ Destination directory ensured at: {self.dest_dir}")


    def on_modified(self, event):
        """
        Called when a file or directory is modified.
        This method starts or resets a timer to perform the copy operation.
        """
        # Check if the modified file is the one we are watching
        if not event.is_directory and event.src_path == self.source_file_path:
            # If there's an existing timer, cancel it to reset the debounce period
            if self.copy_timer and self.copy_timer.is_alive():
                self.copy_timer.cancel()

            # Start a new timer to perform the copy after the debounce interval
            self.copy_timer = threading.Timer(
                self.debounce_interval,
                self.copy_file,
                args=[event.src_path]
            )
            print(f"👀 Change detected in: {self.source_file_path}. Waiting {self.debounce_interval}s for more changes...")
            self.copy_timer.start()

    def copy_file(self, src_path):
        """
        Copies the source file to the destination directory.
        This method is called by the debouncing timer.
        """
        try:
            # Check if the source file still exists before copying
            if os.path.exists(src_path):
                print(f"⏳ Debounce finished. Copying {src_path} to {self.dest_dir}...")
                shutil.copy(src_path, self.dest_dir)
                print(f"✅ Successfully copied to {os.path.join(self.dest_dir, os.path.basename(src_path))}\n")
            else:
                # This can happen if a temp file is created and deleted quickly
                print(f"🤷 File {src_path} was deleted before it could be copied.\n")
        except Exception as e:
            logging.error(f"❌ Error copying file: {e}\n")


def main():
    """
    Main function to set up and run the file watcher.
    """
    # Configure basic logging
    logging.basicConfig(level=logging.INFO,
                        format='%(asctime)s - %(message)s',
                        datefmt='%Y-%m-%d %H:%M:%S')

    # --- Argument Handling ---
    # Expects: python script.py <file_to_watch> [optional: <destination_directory>]
    if len(sys.argv) < 2:
        print("❌ Error: You must provide the path to the file to watch.")
        print("Usage: python watch_and_copy.py <file_to_watch> [destination_directory]")
        sys.exit(1)

    source_file = sys.argv[1]
    # Use the second argument as destination, or the default if not provided
    dest_dir = sys.argv[2] if len(sys.argv) > 2 else DEFAULT_DESTINATION_DIR

    # --- Pre-run Checks ---
    if not os.path.exists(source_file):
        print(f"❌ Error: The source file '{source_file}' does not exist.")
        sys.exit(1)
    
    # We only watch the directory containing the file, not the file itself.
    # This is a requirement of the watchdog library.
    source_path = os.path.dirname(os.path.abspath(source_file))
    
    print("--- File Watcher Initialized ---")
    print(f"👁️  Watching file: {os.path.abspath(source_file)}")
    print(f"🎯 Destination for copies: {os.path.abspath(dest_dir)}")
    print("------------------------------------")
    print("Press Ctrl+C to stop the script.")

    # --- Observer Setup ---
    # The debounce logic is now handled inside the ChangeHandler
    event_handler = ChangeHandler(source_file, dest_dir)
    observer = Observer()
    observer.schedule(event_handler, source_path, recursive=False) # recursive=False to not watch subdirectories

    observer.start()
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        # Before stopping, cancel any pending timer to avoid an exception
        if event_handler.copy_timer:
            event_handler.copy_timer.cancel()
        observer.stop()
        print("\n🛑 Watcher stopped by user.")
    observer.join()


if __name__ == "__main__":
    main()

Code: Select all

python watcher.py d:\projects\flowcode\test_file.hex k:\

Code: Select all

watcher.py d:\projects\flowcode\test_file.hex