(Clic en la imagen para agrandar. Código de fuente al final del artículo).

Watchdog es una librería multiplataforma de Python que permite monitorear eventos del sistema de archivos en tiempo real. Resulta muy útil para automatizar tareas: cuando queremos que nuestro programa ejecute alguna operación al modificarse un archivo, o eliminarse, o moverse, etc. Veamos cómo funciona creando un simple programa que registre los eventos de los archivos de una carpeta.

Antes de comenzar, instalemos el paquete vía pip, ejecutando en la terminal:

python -m pip install watchdog

Corroboremos que se haya instalado correctamente escribiendo en la consola interactiva:

>>> import watchdog

Si no arroja ningún error, la instalación fue exitosa.

Existen dos conceptos principales en Watchdog, que son el de observador y el de manejador de eventos. El observador es una clase encargada de monitorear los eventos ocurridos en uno o varios directorios. Cuando el observador detecta un evento en la carpeta que está monitoreando, lo despacha a otra clase, el manejador de eventos. Por lo general nuestra única tarea será la de implementar el manejador de eventos, para poder responder con nuestro propio código ante una creación, eliminación, modificación o movimiento de un archivo, mientras que el observador es provisto por Watchdog vía la clase watchdog.observers.Observer.

Considerando estos dos conceptos, empecemos con un código básico que imprime un mensaje en pantalla cuando un archivo es modificado en el directorio donde se está ejecutando:

from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler


class MyEventHandler(FileSystemEventHandler):

    def on_modified(self, event):
        print(event.src_path, "modificado.")


observer = Observer()
observer.schedule(MyEventHandler(), ".", recursive=False)
observer.start()
try:
    while observer.is_alive():
        observer.join(1)
except KeyboardInterrupt:
    observer.stop()
observer.join()

En las líneas 11 y 12 creamos una instancia del observador y vía el método schedule() le asignamos un manejador (MyEventHandler) para que responda a los eventos que ocurran en la carpeta actual ("."), sin considerar las subcarpetas (recursive=False). Si quisiéramos monitorear, por ejemplo, todos los eventos de todos los archivos y carpetas de un disco, podríamos decir:

# Monitorear todos los eventos del disco local C: (en Windows).
observer.schedule(MyEventHandler(), "C:\\", recursive=True)

En la línea 13 llamamos al método start(), que inicia el observador. Es importante notar que el observador corre en un hilo de ejecución secundario (de ahí el método start() propio de la clase threading.Thread, véase Cómo lanzar un hilo («thread»)), puesto que en consecuencia también nuestro manejador de eventos será invocado desde el mismo hilo secundario.

Las líneas 14-18 permiten que el observador sea detenido al presionar CTRL + D (Linux y Mac OS) o CTRL + C (Windows). La última línea 19 espera a que el hilo del observador finalice antes de terminar el programa.

En las líneas 5-8 creamos la clase MyEventHandler que hereda de watchdog.events.FileSystemEventHandler, que es una clase provista por Watchdog para que usemos como base para implementar nuestro propio manejador de eventos. El método on_modified() será invocado por Watchdog cada vez que un archivo es modificado dentro del directorio donde hemos instalado el observador (".", que representa el directorio donde se encuentra nuestro programa). El argumento event será una instancia de watchdog.events.FileSystemEvent, que principalmente tiene los siguientes atributos:

• is_directory, un booleano que indica si el objeto del evento es una carpeta;
• src_path, la ruta del archivo o carpeta del evento.

Así, en el código usamos event.src_path para imprimir en pantalla la ruta del archivo o directorio que ha sido modificado. Para saber cuándo un objeto del sistema de archivos ha sido creado, movido o eliminado, tenemos los métodos on_created(), on_moved() y on_deleted():

class MyEventHandler(FileSystemEventHandler):

    def on_modified(self, event):
        print(event.src_path, "modificado.")
    
    def on_created(self, event):
        print(event.src_path, "creado.")
    
    def on_moved(self, event):
        print(event.src_path, "movido a", event.dest_path)
    
    def on_deleted(self, event):
        print(event.src_path, "eliminado.")

Nótese que en el método on_moved(), el evento tiene un atributo adicional, dest_path, que indica la nueva ruta del archivo o carpeta movido. No obstante, on_moved() es invocado más bien cuando un archivo o carpeta es renombrado. En cambio, cuando un archivo es movido a otra ubicación, el evento recibido es on_deleted(), al igual que cuando es eliminado. Inversamente, cuando un archivo es movido desde otra ubicación al directorio que estamos monitoreando con Watchdog, el evento es on_created(), como cuando se crea un nuevo archivo.

Una operación común es querer obtener únicamente el nombre del archivo del evento, sin la ruta completa. Para ello podemos usar las funciones de la clase estándar pathlib.Path:

from pathlib import Path

from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler


class MyEventHandler(FileSystemEventHandler):

    def on_modified(self, event):
        filename = Path(event.src_path).name
        print(filename, "modificado.")

O para obtener solo la ruta, sin el nombre del archivo:

    def on_modified(self, event):
        path = str(Path(event.src_path).parent)
        print(path, "modificado.")

Existe, además, el método on_any_event(), que es ejecutado, como el nombre lo indica, cuando ocurre cualquiera de los eventos anteriores:

class MyEventHandler(FileSystemEventHandler):

    def on_any_event(self, event):
        print("Ha ocurrido un evento.")

Este método es útil cuando queremos ejecutar alguna acción común a todos los eventos. El atributo event.event_type indica el tipo de evento registrado, cuyos valores pueden ser:

• watchdog.events.EVENT_TYPE_CREATED
• watchdog.events.EVENT_TYPE_DELETED
• watchdog.events.EVENT_TYPE_MODIFIED
• watchdog.events.EVENT_TYPE_MOVED

No es posible cancelar ni modificar ninguno de los eventos registrados. Watchdog solo permite monitorear las operaciones del sistema de archivos, no alterarlas.

Código de ejemplo: registrar operaciones con Tk

El siguiente código (vista previa al principio del artículo) registra las operaciones ejecutadas en la carpeta donde se ejecuta el programa y las muestra en una vista de árbol de Tk.

from pathlib import Path
from tkinter import ttk
import datetime
import queue
import tkinter as tk

from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
from watchdog.events import (
    EVENT_TYPE_CREATED,
    EVENT_TYPE_DELETED,
    EVENT_TYPE_MODIFIED,
    EVENT_TYPE_MOVED
)


class MyEventHandler(FileSystemEventHandler):

    def __init__(self, q):
        # Guardar referencia a la cola para poder utilizarla
        # en on_any_event().
        self._q = q
        super().__init__()
    
    def on_any_event(self, event):
        # Determinar el nombre de la operación.
        action = {
            EVENT_TYPE_CREATED: "Creado",
            EVENT_TYPE_DELETED: "Eliminado",
            EVENT_TYPE_MODIFIED: "Modificado",
            EVENT_TYPE_MOVED: "Movido",
        }[event.event_type]
        # Si es un movimiento, agregar la ruta de destino.
        if event.event_type == EVENT_TYPE_MOVED:
            action += f" ({event.dest_path})"
        # Agregar la información del evento a la cola, para que sea
        # procesada por loop_observer() en el hilo principal.
        # (No es conveniente modificar un control de Tk desde
        # un hilo secundario).
        self._q.put((
            # Nombre del archivo modificado.
            Path(event.src_path).name,
            # Acción ejecutada sobre ese archivo.
            action,
            # Hora en que se ejecuta la acción.
            datetime.datetime.now().strftime("%H:%M:%S")
        ))


def process_events(observer, q, modtree):
    # Chequear que el observador esté aún corriendo.
    if not observer.is_alive():
        return
    try:
        # Intentar obtener un evento de la cola.
        new_item = q.get_nowait()
    except queue.Empty:
        # Si no hay ninguno, continuar normalmente.
        pass
    else:
        # Si se pudo obtener un evento, agregarlo a la vista de árbol.
        modtree.insert("", 0, text=new_item[0], values=new_item[1:])
    # Volver a chequear dentro de medio segundo (500 ms).
    root.after(500, process_events, observer, q, modtree)


root = tk.Tk()
root.config(width=600, height=500)
root.columnconfigure(0, weight=1)
root.rowconfigure(0, weight=1)
root.title("Registro de modificaciones en tiempo real")

modtree = ttk.Treeview(columns=("action", "time",))
modtree.heading("#0", text="Archivo")
modtree.heading("action", text="Acción")
modtree.heading("time", text="Hora")
modtree.grid(column=0, row=0, sticky="nsew")

# Observador de eventos de Watchdog.
observer = Observer()
# Cola para comunicación entre el observador y la aplicación de Tk.
# Para una explicación más detallada sobre la cola y Tk, véase
# https://www.recursospython.com/guias-y-manuales/tareas-en-segundo-plano-con-tcl-tk-tkinter/.
q = queue.Queue()
observer.schedule(MyEventHandler(q), ".", recursive=False)
observer.start()
# Programar función que procesa los eventos del observador.
# Para la función after(), véase
# https://www.recursospython.com/guias-y-manuales/la-funcion-after-en-tkinter/.
root.after(1, process_events, observer, q, modtree)
root.mainloop()
observer.stop()
observer.join()