Untitled
user_5788845
plain_text
a year ago
2.1 kB
6
Indexable
import sys import time import logging from watchdog.observers import Observer from watchdog.events import FileSystemEventHandler import subprocess logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(message)s', datefmt='%Y-%m-%d %H:%M:%S') class Watcher: DIRECTORY_TO_WATCH = "/mnt/customer_data/scripts" DESTINATION_DIRECTORY = "/home/Alfredo/pruebas/" def __init__(self): self.observer = Observer() def run(self): self.sync_and_verify_directories() event_handler = Handler() self.observer.schedule(event_handler, self.DIRECTORY_TO_WATCH, recursive=True) self.observer.start() try: while True: time.sleep(5) except KeyboardInterrupt: self.observer.stop() logging.info("Observador detenido") self.observer.join() @staticmethod def sync_and_verify_directories(): logging.info("Realizando sincronización inicial y verificación de los directorios...") Handler.sync_directories() logging.info("Sincronización y verificación inicial completadas.") class Handler(FileSystemEventHandler): @staticmethod def sync_directories(): cmd = [ "rsync", "-avz", "--delete", # "--dry-run", # Eliminar o comentar esta línea para ejecutar de verdad la sincronización Watcher.DIRECTORY_TO_WATCH, Watcher.DESTINATION_DIRECTORY, ] result = subprocess.run(cmd, capture_output=True, text=True) logging.info(result.stdout) if result.stderr: logging.error(result.stderr) def on_any_event(self, event): if event.is_directory: return None elif event.event_type in ['created', 'modified', 'deleted']: logging.info(f"Evento detectado: {event.src_path} ha sido {event.event_type}") self.sync_directories() if __name__ == "__main__": w = Watcher() w.run()
Editor is loading...
Leave a Comment