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 watchdogLa 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 FileSystemEventHandlerLos 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 FalseLuego 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 < ventanaSi 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))
raiseAntes 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.
