57/watcher.py

#!/usr/bin/env python3
import time
import sys
import os
import json
from datetime import datetime
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler

# ANSI color codes
class Colors:
    RED = '\033[91m'
    GREEN = '\033[92m'
    ORANGE = '\033[93m'
    RESET = '\033[0m'

# Load optimization target mappings from models.txt
def load_optimization_targets(filename="models.txt"):
    with open(filename, "r") as file:
        content = file.read()
        start_index = content.find("{")
        end_index = content.rfind("}") + 1
        json_data = content[start_index:end_index]
        return {k: v.replace("OPTIMIZATION_TARGET_", "") for k, v in json.loads(json_data).items()}

OPTIMIZATION_TARGETS = load_optimization_targets()

class MyHandler(FileSystemEventHandler):
    def __init__(self, log_file, base_path):
        self.log_file = log_file
        self.base_path = base_path
        self.suppress_log = log_file  # Avoid infinite loop

    def get_target_descriptions(self, event_path):
        """ Extracts all matching optimization targets from the path. """
        folder_names = set(event_path.split(os.sep))  # Get all path components
        matched_targets = [
            (folder, OPTIMIZATION_TARGETS[folder])
            for folder in folder_names if folder in OPTIMIZATION_TARGETS
        ]
        return matched_targets if matched_targets else [("Unknown", "[Unknown]")]

    def format_path(self, path):
        """ Returns relative path from base directory. """
        return f".{path.replace(self.base_path, '')}" if path.startswith(self.base_path) else path

    def get_event_color(self, event_type):
        """ Returns the appropriate color for each event type """
        if "DELETED" in event_type:
            return Colors.RED
        elif "NEW" in event_type:
            return Colors.GREEN
        else:
            return Colors.ORANGE

    def log_event(self, event_type, event_path):
        if os.path.abspath(event_path) == os.path.abspath(self.suppress_log):
            return  # Ignore log file modifications

        matched_targets = self.get_target_descriptions(event_path)
        timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        formatted_path = self.format_path(event_path)
        color = self.get_event_color(event_type)

        # Log each event separately in a stacked format
        log_entries = []
        colored_entries = []
        
        for folder, target in matched_targets:
            base_entry = (
                f"------------------------------\n"
                f"Timestamp: {timestamp}\n"
                f"Event: {event_type}\n"
                f"Folder: {folder}\n"
                f"Optimization Target: {target}\n"
                f"Path: {formatted_path}\n"
                f"------------------------------\n"
            )
            
            # Create colored version for console
            colored_entry = (
                f"------------------------------\n"
                f"Timestamp: {timestamp}\n"
                f"Event: {color}{event_type}{Colors.RESET}\n"
                f"Folder: {folder}\n"
                f"Optimization Target: {target}\n"
                f"Path: {formatted_path}\n"
                f"------------------------------\n"
            )
            
            log_entries.append(base_entry)
            colored_entries.append(colored_entry)

        log_message = "\n".join(log_entries)
        colored_message = "\n".join(colored_entries)

        print(colored_message)  # Print colored version to console
        with open(self.log_file, "a") as f:
            f.write(log_message + "\n")  # Write uncolored version to file

    def on_created(self, event):
        what = "NEW FOLDER" if event.is_directory else "NEW FILE"
        self.log_event(what, event.src_path)

    def on_deleted(self, event):
        what = "FOLDER DELETED" if event.is_directory else "FILE DELETED"
        self.log_event(what, event.src_path)

    def on_modified(self, event):
        what = "FOLDER MODIFIED" if event.is_directory else "FILE MODIFIED"
        self.log_event(what, event.src_path)

    def on_moved(self, event):
        what = "FOLDER MOVED" if event.is_directory else "FILE MOVED"
        self.log_event(f"{what}: from {self.format_path(event.src_path)} to {self.format_path(event.dest_path)}", event.src_path)

if __name__ == "__main__":
    path = sys.argv[1] if len(sys.argv) > 1 else os.getcwd()
    log_file = "log.txt"

    event_handler = MyHandler(log_file, path)
    observer = Observer()
    observer.schedule(event_handler, path, recursive=True)
    observer.start()
    
    print(f"Monitoring folder: {os.path.abspath(path)}")
    with open(log_file, "a") as f:
        f.write(f"Monitoring folder: {os.path.abspath(path)}\n")
    
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()