Vigiando arquivos e diretórios com Python.

Seja bem-vindo ao mundo da automação com Python! Neste post, vamos desbravar o uso do pacote watchdog para vigiar diretórios e agir sobre mudanças como criação, modificação ou exclusão de arquivos. Ideal para quem busca otimizar fluxos de trabalho ou simplesmente tem curiosidade sobre como manter um olho digital em seus diretórios. Vamos lá?

Neste post, mostro como vigiar um diretório e fazer algo quando um arquivo é criado, modificado, movido, excluido ou renomeado.

Pre-requisitos

Para criar o vigia, precisamos de instalar o pacote watchdog. Para isso, no terminal, execute o seguinte comando:

pip install watchdog

(A versão utilizada neste post é a 0.9.0)

Como funciona

Vamos precisar de um handler, que é quem vai executar as ações em si; Um Observer, que vai ficar vigiando os diretórios e algo para manter o script ocupado, pois este monitor trabalha em uma thread paralela.

Criando o Handler

O que queremos fazer é vigiar um diretório, então precisamos importar o handler correto:

from watchdog.events import FileSystemEventHandler

Agora temos que criar uma classe que herde deste handler e implementar as ações que serão utilizadas. Vou fazer dois handlers: Um com métodos específicos e outro genérico. (Calma, tudo vai fazer sentido).

Handler 1: Genérico

class WatchdogHandlerAny(FileSystemEventHandler):
    def on_any_event(self, event):
        """Will be triggered if anything happen in the folder we're watching."""
        print("Something happened!")
        print(f"Event type: {event.event_type}")
        print(event)

Na classe criada acima, o método on_any_event é disparado sempre que qualquer coisa acontecer dentro da pasta que estamos vigiando.

Handler 2: Específico

class WatchdogHandler(FileSystemEventHandler):
    def on_modified(self, event):
        """Triggered when something is modified."""
        print(f"File modified: {event.src_path}")

    def on_created(self, event):
        """Triggered when a file or folder is created."""
        print(f"File created: {event.src_path}")

    def on_deleted(self, event):
        """Triggered when a file or folder is removed."""
        print(f"File deleted: {event.src_path}")

    def on_moved(self, event):
        """Triggered when a file or folder is moved.
        Note: Renaming is considered moving. Will only be considered moving if it's between watched folders."""
        print(f"File moved: From: {event.src_path} | To: {event.dest_path")

Este segundo handler possui uma classe separada para cada um dos eventos que podem ocorrer:

1. Criação de arquivo/pasta;
2. Exclusão de arquivo/pasta;
3. Alteração de arquivo/pasta;
4. Quando um arquivo/diretório é movido de um lugar para outro dentro das pastas vigiada.
   • Um lembrete: Renomear um arquivo ou diretório é o equivalente a mover. Este não é o funcionamento do pacote, é assim que o arquivo de sistemas funciona.

Criando o Observer

class FolderWatchDog:
    def __init__(self, handler, watch_path):
        """
        Constructor
        :param handler: Who will handle what happens in our watch_path.
        """
        self.handler = handler
        self.handler = watch_path
        self.observer = Observer()  # This is the class that does the actual watching.

    def start(self):
        self.observer.schedule(self.handler,
                               path=self.watch_patch,
                               recursive=True)  # If set to true, will watch every folder recursively.

        try:
            print(f"Starting to watch folder: {self.watch_patch}")
            self.observer.start()
            while True:
                sleep(5)

        except KeyboardInterrupt:
            self.stop()
        except Exception as e:
            print(f"Something went wrong...")
            print(e)

        finally:
            print("All done!")

    def stop(self):
        self.observer.stop()
        self.observer.join()

Para deixar o código encapsulado, preferi criar uma classe. Todavia, você não tem que fazer isso, pode utilizar apenas o objeto Observer (com os métodos schedule, start, stop, e join).

Explicando a classe acima:

1. Constructor:
   • Recebe o caminho que será vigiado (watch_path), a instancia do handler e o observer é instanciado;
2. Método start:
   • A primeira coisa é chamar o método schedule do observer. Nele definimos o handler, o caminho que será vigiado e se apenas o diretório informado será vigiado ou se todos os diretórios a partir dele serão vigiados recursivamente.
     • (recursive=True significa que todos os diretórios a partir do caminho informado vão ser vigiados)
   • Depois chamo o método start do observer. A partir deste momento, os caminhos estarão sendo vigiados.
   • O próximo passo é manter o script ocupado, para que ele não encerre a execução. Isso é feito através do while True com sleep.
   • Quando for para encerrar o monitoramento, o usuário vai apertar Ctrl+C e isso vai causar um KeyboardInterrupt exception. Neste catch eu chamo o método stop (descrito no item 3 desta lista)
3. Método stop:
   1. Dentro do método stop da classe que criei para encapsular o Observer, chamo o método stop dele e depois join. Encerrando a thread e o parando de vigiar os diretórios.

O próximo passo é executar este código.

Iniciando o Watchdog

if __name__ == '__main__':
    to_watch = "/path/to/watch"
    w_dog = FolderWatchDog(handler=WatchdogHandlerAny())
    w_dog.start()

Para executar é bem simples:

1. Definir o caminho que será vigiado (to_watch);
2. Instanciar a classe que criei (que encapsula o Observer), passando no construtor uma instancia do handler 1 ou 2
3. Chamar o método start da classe FolderWatchDog.
4. Pronto!

Se quiser, no meu Github tem um exemplo parecido com este.

Espero ter ajudado!