|
|
| import time
|
| import sys
|
| import os
|
| import json
|
| from datetime import datetime
|
| from watchdog.observers import Observer
|
| from watchdog.events import FileSystemEventHandler
|
|
|
|
|
| class Colors:
|
| RED = '\033[91m'
|
| GREEN = '\033[92m'
|
| ORANGE = '\033[93m'
|
| RESET = '\033[0m'
|
|
|
|
|
| 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
|
|
|
| def get_target_descriptions(self, event_path):
|
| """ Extracts all matching optimization targets from the path. """
|
| folder_names = set(event_path.split(os.sep))
|
| 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
|
|
|
| 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_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"
|
| )
|
|
|
|
|
| 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)
|
| with open(self.log_file, "a") as f:
|
| f.write(log_message + "\n")
|
|
|
| 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() |
