Saltar al contenido

Generando múltiples observadores con Python watchdog

Hacemos una revisión profunda cada una de las reseñas en nuestra página web con el objetivo de mostrarte siempre la información con la mayor veracidad y actualizada.

Solución:

Gran pregunta. Este hilo es más antiguo, pero lo encontré mientras buscaba exactamente y amplí su trabajo y agregué la capacidad de pasar un archivo con una lista de directorios para ver. Por defecto no busco recursivamente, dejo que alguien más lo pruebe. Esperemos que esto ayude a cualquiera que busque el mismo tema. ¡Buen trabajo!

Ejecutar usando el nombre de archivo python watcher.py

Donde watcher.py es lo que llamé mi script y filename es el nombre del archivo con mis rutas.

Enumero las rutas completas en el archivo y están separadas por saltos de línea, es decir,

C:ruta1
C:ruta2subruta1
C:RUTA3

import logging
import sys
import time
from watchdog.observers import Observer
from watchdog.events import LoggingEventHandler


# Attach a logging event AKA FileSystemEventHandler
event_handler = LoggingEventHandler()
# Create Observer to watch directories
observer = Observer()
# take in list of paths.  If none given, watch CWD
paths = open(sys.argv[1], 'r') if len(sys.argv) > 1 else '.'
# Empty list of observers .
observers = []
# Base logging configuration
logging.basicConfig(level=logging.INFO,
                    format='%(asctime)s - %(message)s',
                    datefmt='%Y-%m-%d %H:%M:%S')

# iterate through paths and attach observers
for line in paths:
    # convert line into string and strip newline character
    targetPath = str(line).rstrip()
    # Schedules watching of a given path
    observer.schedule(event_handler, targetPath)
    # Add observable to list of observers
    observers .append(observer)

# start observer
observer.start()

try:
    while True:
        # poll every second
        time.sleep(1)
except KeyboardInterrupt:
    for o in observers:
        o.unschedule_all()
        # stop observer if interrupted
        o.stop()
for o in observers:
    # Wait until the thread terminates before exit
    o.join()

El código de ejemplo aquí muestra una función llamada start, no start_new_thread. ¿Has probado eso?

https://pypi.python.org/pypi/perro guardián

Además, probablemente deberías llamar start solo una vez, después del ciclo for, no dentro de él.

Solo quiero agregar algunas notas:

La lista de subprocesos y bibliotecas de subprocesos en el código puede ser un poco confusa para las personas que recién comienzan a usar Watchdog (incluyéndome a mí). En realidad, no son necesarios en la solución. Una forma sencilla de explicarlo es simplemente:

  • crear un observador
  • programar múltiples “eventos de observación”
  • y comienza el observador.

Eso es.

¡Haz clic para puntuar esta entrada!
(Votos: 0 Promedio: 0)



Utiliza Nuestro Buscador

Deja una respuesta

Tu dirección de correo electrónico no será publicada. Los campos obligatorios están marcados con *