Watchdog en Python: monitoriza carpetas en tiempo real

Watchdog es una biblioteca de Python que permite detectar cambios en archivos y carpetas mientras ocurren. En lugar de revisar una ruta cada pocos segundos y comparar su contenido manualmente, Watchdog se apoya en los mecanismos de notificación del sistema operativo para reaccionar ante archivos creados, modificados, movidos o eliminados. Esto resulta útil para automatizar copias, procesar documentos recién recibidos, validar cargas, regenerar informes o iniciar una tarea cuando aparece un archivo nuevo.

Antes de construir un monitor conviene dominar el manejo de rutas con pathlib, el módulo os de Python, el sistema de logging y la creación de funciones reutilizables. Estas herramientas ayudan a que el observador sea portable, fácil de probar y seguro.

Instalar Watchdog

Crea primero un entorno virtual para aislar la dependencia:

python -m venv .venv
# Windows
.venv\Scripts\activate
# Linux o macOS
source .venv/bin/activate

python -m pip install watchdog

La documentación oficial de Watchdog explica los observadores y manejadores disponibles. También puedes consultar la página oficial del paquete en PyPI para revisar versiones y requisitos.

Conceptos principales

Watchdog separa el trabajo en dos piezas. El Observer supervisa una ruta. El manejador, normalmente una clase derivada de FileSystemEventHandler, recibe los eventos. Esta separación permite reutilizar la misma lógica con distintas carpetas.

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

Los métodos más usados son on_created, on_modified, on_moved y on_deleted. Cada evento incluye información como la ruta y si el elemento es un directorio.

Primer monitor de carpeta

import time
from pathlib import Path
from watchdog.events import FileSystemEventHandler
from watchdog.observers import Observer

class Manejador(FileSystemEventHandler):
    def on_created(self, event):
        if event.is_directory:
            return
        print(f"Archivo creado: {event.src_path}")

ruta = Path.home() / "Downloads"
observador = Observer()
observador.schedule(Manejador(), str(ruta), recursive=False)
observador.start()

try:
    while True:
        time.sleep(1)
except KeyboardInterrupt:
    observador.stop()

observador.join()

El bucle mantiene vivo el proceso. Cuando el usuario presiona Ctrl+C, el observador se detiene de forma ordenada. El parámetro recursive=False limita la vigilancia a la carpeta principal. Usa True solamente cuando también necesites supervisar subcarpetas.

Filtrar por extensión

Un monitor real no debería procesar todos los archivos. Puedes aceptar únicamente formatos concretos:

EXTENSIONES = {".csv", ".xlsx", ".json"}

class ManejadorDatos(FileSystemEventHandler):
    def on_created(self, event):
        if event.is_directory:
            return

        archivo = Path(event.src_path)
        if archivo.suffix.lower() not in EXTENSIONES:
            return

        procesar_archivo(archivo)

Esta validación evita ejecutar tareas pesadas sobre miniaturas, archivos temporales o documentos irrelevantes.

Esperar a que el archivo termine de copiarse

Un evento de creación puede ocurrir antes de que una descarga o copia haya finalizado. Si el programa abre el archivo inmediatamente, puede leer contenido incompleto. Una técnica sencilla consiste en comprobar que el tamaño se mantenga estable:

import time

def esperar_estabilidad(archivo, intentos=10, pausa=1):
    anterior = -1
    for _ in range(intentos):
        actual = archivo.stat().st_size
        if actual == anterior:
            return True
        anterior = actual
        time.sleep(pausa)
    return False

Luego llama a la función antes de procesar. También conviene ignorar extensiones como .part, .tmp y .crdownload.

Evitar eventos duplicados

Algunos editores guardan un documento mediante varias operaciones: crean un temporal, lo modifican y después lo renombran. Por eso una sola acción humana puede producir varios eventos. Para evitar trabajo repetido, guarda el instante de la última ejecución por ruta y aplica una ventana de espera.

from time import monotonic

ULTIMOS = {}

def evento_reciente(ruta, ventana=2):
    ahora = monotonic()
    previo = ULTIMOS.get(ruta, 0)
    ULTIMOS[ruta] = ahora
    return ahora - previo < ventana

Si evento_reciente() devuelve True, omite el evento. En sistemas distribuidos, utiliza una base de datos o una cola en lugar de memoria local.

Registrar errores y acciones

import logging

logging.basicConfig(
    filename="watchdog.log",
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(message)s",
    encoding="utf-8"
)

def procesar_archivo(archivo):
    try:
        logging.info("Procesando %s", archivo)
        # lógica del proyecto
    except Exception:
        logging.exception("Error procesando %s", archivo)

El registro es esencial cuando el monitor se ejecuta sin una terminal visible. Evita capturar excepciones sin guardar detalles, porque los fallos silenciosos son difíciles de investigar.

Mover archivos después de procesarlos

Una práctica común consiste en separar archivos pendientes, procesados y fallidos:

import shutil

BASE = Path("entrada")
PROCESADOS = Path("procesados")
ERRORES = Path("errores")

PROCESADOS.mkdir(exist_ok=True)
ERRORES.mkdir(exist_ok=True)

def procesar_archivo(archivo):
    try:
        # validar y transformar
        destino = PROCESADOS / archivo.name
        shutil.move(str(archivo), str(destino))
    except Exception:
        destino = ERRORES / archivo.name
        shutil.move(str(archivo), str(destino))
        raise

Antes de mover, verifica que el destino no exista para evitar sobrescrituras. También puedes generar nombres únicos o incluir una marca de tiempo.

Ejecutar tareas pesadas fuera del manejador

El método del evento debería terminar rápido. Si convierte videos, analiza grandes CSV o llama a una API, envía el trabajo a una cola. Un ThreadPoolExecutor puede servir para tareas de entrada y salida:

from concurrent.futures import ThreadPoolExecutor

pool = ThreadPoolExecutor(max_workers=4)

class Manejador(FileSystemEventHandler):
    def on_created(self, event):
        if not event.is_directory:
            pool.submit(procesar_archivo, Path(event.src_path))

Para trabajo intensivo de CPU, considera procesos separados. Limita el número de tareas para no saturar el equipo.

Probar el monitor

No dependas únicamente de pruebas manuales. Separa la lógica del manejador y llama a las funciones con archivos temporales. Comprueba extensiones permitidas, archivos vacíos, nombres duplicados, errores de permisos y movimientos incompletos. También prueba qué ocurre cuando la carpeta observada no existe o deja de estar disponible.

Buenas prácticas de producción

Valida la ruta antes de iniciar, ejecuta el proceso con los permisos mínimos y no observes todo el disco sin necesidad. Añade límites de tamaño, filtros de nombre y una estrategia para reintentos. Si el monitor procesa archivos externos, trata su contenido como no confiable. No ejecutes comandos construidos directamente con nombres de archivo.

Para mantenerlo activo, puedes configurarlo como servicio del sistema, tarea programada o contenedor. Asegúrate de que el proceso se reinicie después de un fallo y de que los registros tengan rotación para no ocupar todo el almacenamiento.

Conclusión

Watchdog convierte los cambios del sistema de archivos en eventos que Python puede manejar inmediatamente. El patrón básico combina un observador, un manejador y una función de procesamiento. La confiabilidad aparece al añadir filtros, espera de estabilidad, deduplicación, logging, manejo de errores y pruebas. Con estas medidas puedes crear automatizaciones que reaccionen a nuevos documentos, imágenes, informes o datos sin revisar carpetas manualmente.