Como Criar um Monitor de Alteração em Arquivos de Log (Log File Watcher) em Python

O monitoramento de arquivos de texto e logs em tempo real é uma necessidade frequente no desenvolvimento de software e na gestão de infraestrutura. Seja para capturar falhas em servidores de aplicação, acompanhar execuções de rotinas em lote (batch processing) ou disparar alertas assim que uma exceção é gravada em disco, automatizar a observação do sistema de arquivos evita checagens manuais e reduz significativamente o tempo de resposta a incidentes.

Em Python, a solução padrão da indústria para essa tarefa é a biblioteca Watchdog. Projetada para operar de forma assíncrona e em tempo real através do sistema de eventos do sistema operacional, ela elimina a necessidade de loops contínuos de leitura com polling (que consomem alto processamento de CPU). Este guia prático apresenta desde a arquitetura fundamental do Watchdog até a construção de um Log File Watcher completo com suporte a filtros e leitura incremental.

1. Por Que Usar Eventos do Sistema de Arquivos em Vez de Polling?

Tradicionalmente, a verificação de alterações em arquivos era realizada por meio de abordagens ingênuas baseadas em intervalos fixos de tempo (polling). O script entrava em um laço de repetição infinito com time.sleep(), verificando constantemente o tamanho ou a data de modificação (mtime) do arquivo.

Embora simples de conceber, o polling apresenta desvantagens operacionais relevantes:

• Ineficiência de Recursos: A checagem constante consome ciclos desnecessários de CPU e I/O de disco, especialmente quando monitoramos diretórios com milhares de arquivos.
• Latência Controlada por Intervalo: Alterações ocorridas entre os intervalos de leitura só serão percebidas no ciclo seguinte, gerando atraso na reação a eventos críticos.
• Risco de Perda de Eventos Intermediários: Se um arquivo for criado, modificado e excluído no intervalo entre duas checagens, o script de polling simplesmente não detectará a ocorrência.

A biblioteca Watchdog contorna esses entraves utilizando as APIs nativas do kernel do sistema operacional — como inotify no Linux, FSEvents no macOS e ReadDirectoryChangesW no Windows. O sistema operacional notifica a aplicação no exato milissegundo em que o evento ocorre.

2. A Arquitetura do Watchdog: Observers e Event Handlers

O funcionamento do Watchdog está estruturado sobre dois componentes essenciais que trabalham de forma desacoplada:

1. Observer (Observador): É uma thread secundária encarregada de monitorar continuamente o diretório especificado. O Observer escuta os sinais enviados pelo sistema operacional e encaminha as ocorrências para o gerenciador de eventos correspondente.
2. FileSystemEventHandler (Gerenciador de Eventos): É a classe responsável por definir a lógica de reação quando um evento é capturado. Ela fornece métodos callback que podem ser sobrescritos pelo desenvolvedor, tais como on_modified(), on_created(), on_deleted() e on_moved().

A separação entre a thread do observador e a execução da lógica de negócios garante que o monitoramento ocorra em segundo plano sem bloquear a execução principal da aplicação.

3. Implementando o Log File Watcher Passo a Passo

Para iniciar o desenvolvimento do monitor, instale a biblioteca Watchdog no seu ambiente Python por meio do gerenciador de pacotes:

pip install watchdog

Passo 1: Filtrando Apenas Arquivos de Log e Texto

Em cenários reais, diretórios de servidores contêm arquivos temporários, backups e extensões diversas. Para evitar o disparo desnecessário de rotinas quando arquivos irrelevantes forem alterados, utiliza-se a classe PatternMatchingEventHandler. Ela permite definir padrões de nomes de arquivos através de máscaras (wildcards).

O código abaixo demonstra a criação de um gerenciador que reage exclusivamente a modificações em arquivos com extensão .log e .txt:

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

class LogFileHandler(PatternMatchingEventHandler):
    def __init__(self):
        # Configura os padrões de arquivo desejados e ignora alterações em diretórios
        super().__init__(
            patterns=["*.log", "*.txt"],
            ignore_directories=True,
            case_sensitive=False
        )

    def on_modified(self, event):
        file_name = Path(event.src_path).name
        print(f"[{time.strftime('%H:%M:%S')}] Modificação detectada no arquivo: {file_name}")

    def on_created(self, event):
        file_name = Path(event.src_path).name
        print(f"[{time.strftime('%H:%M:%S')}] Novo arquivo de log criado: {file_name}")

def iniciar_monitoramento(diretorio_alvo):
    event_handler = LogFileHandler()
    observer = Observer()
    
    # Agenda o monitoramento (recursive=False monitora apenas a raiz do diretório)
    observer.schedule(event_handler, path=diretorio_alvo, recursive=False)
    observer.start()
    
    print(f"Monitorando arquivos de log em '{diretorio_alvo}'... (Pressione Ctrl+C para parar)")
    
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
        print("\nParando o observador...")
    
    observer.join()

if __name__ == "__main__":
    iniciar_monitoramento(".")

Neste exemplo, quando qualquer arquivo .log no diretório for alterado, o método on_modified() é executado nativamente, imprimindo o nome do arquivo e a hora exata da alteração.

4. Capturando Apenas Novas Linhas Adicionadas (Efeito Tail -f)

Embora saber que um arquivo foi alterado seja útil, em um ambiente de log watcher o objetivo principal costuma ser ler o conteúdo recém-adicionado sem precisar reprocessar todo o histórico do arquivo.

Para simular o comportamento do comando tail -f do Linux, podemos manter o estado do ponteiro de leitura de cada arquivo monitorado em um dicionário Python. Quando o evento on_modified é acionado, abrimos o arquivo, posicionamos o ponteiro no deslocamento (offset) gravado anteriormente e lemos apenas as novas linhas.

Abaixo está a implementação completa dessa funcionalidade:

import os
import time
from pathlib import Path
from watchdog.observers import Observer
from watchdog.events import PatternMatchingEventHandler

class RealTimeLogTailHandler(PatternMatchingEventHandler):
    def __init__(self):
        super().__init__(
            patterns=["*.log"],
            ignore_directories=True,
            case_sensitive=False
        )
        # Guarda a posição do ultimo byte lido para cada arquivo
        self.file_positions = {}

    def _ler_novas_linhas(self, file_path):
        caminho_str = str(file_path)
        
        if not os.path.exists(caminho_str):
            return

        # Se for a primeira vez que vemos o arquivo, iniciamos no final dele
        if caminho_str not in self.file_positions:
            self.file_positions[caminho_str] = os.path.getsize(caminho_str)
            return

        posicao_anterior = self.file_positions[caminho_str]
        tamanho_atual = os.path.getsize(caminho_str)

        # Trata o caso em que o log sofreu rotação ou foi limpo (truncated)
        if tamanho_atual < posicao_anterior:
            posicao_anterior = 0

        with open(caminho_str, "r", encoding="utf-8", errors="replace") as file:
            file.seek(posicao_anterior)
            novas_linhas = file.readlines()
            self.file_positions[caminho_str] = file.tell()

        # Processa e exibe as novas linhas capturadas
        for linha in novas_linhas:
            linha_limpa = linha.strip()
            if linha_limpa:
                nome_arquivo = Path(file_path).name
                print(f"[{nome_arquivo}] -> {linha_limpa}")

    def on_modified(self, event):
        self._ler_novas_linhas(event.src_path)

    def on_created(self, event):
        # Quando um novo arquivo é criado, registramos e lemos eventual conteúdo inicial
        self._ler_novas_linhas(event.src_path)

def executar_log_watcher(pasta_logs):
    handler = RealTimeLogTailHandler()
    observer = Observer()
    observer.schedule(handler, path=pasta_logs, recursive=False)
    observer.start()

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

if __name__ == "__main__":
    executar_log_watcher("./logs")

Destaques do Algoritmo Incremental:

1. Controle de Rotação de Logs: Se o tamanho atual do arquivo for menor que a posição anterior registrada (tamanho_atual < posicao_anterior), o script detecta que o log foi limpo ou truncado e reinicia a leitura do byte zero.
2. Atualização de Offset com file.tell(): Após a leitura das novas linhas, a posição final do ponteiro é armazenada para garantir continuidade no próximo evento.
3. Tratamento de Encodings: O uso de errors="replace" previne que erros de caracteres malformatados em textos e logs interrompam a execução do script.

5. Comparativo de Classes e Métodos da Biblioteca Watchdog

A tabela a seguir resume as principais opções disponíveis no ecossistema Watchdog para manipulação de eventos e filtragem:

6. Integrando o Watchdog com Aplicações e GUIs Usando Filas (Queue)

Como o Observer roda em uma thread secundária, modificar diretamente a interface gráfica (como Tkinter ou PyQt) ou disparar operações pesadas diretamente dentro dos métodos on_modified() pode gerar condições de corrida (race conditions) ou congelamento de tela.

A boa prática para arquiteturar sistemas robustos consiste em utilizar uma fila thread-safe (queue.Queue) como canal de comunicação intermediário:

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

class QueueLogHandler(FileSystemEventHandler):
    def __init__(self, fila_comunicacao):
        super().__init__()
        self.fila = fila_comunicacao

    def on_modified(self, event):
        if not event.is_directory and event.src_path.endswith(".log"):
            # Envia os dados do evento para a fila assíncrona
            self.fila.put({
                "arquivo": event.src_path,
                "evento": "modificado",
                "timestamp": time.time()
            })

# Consumidor em thread principal
def processar_eventos_da_fila(fila):
    while True:
        try:
            item = fila.get_nowait()
            print(f"Processando log: {item['arquivo']} às {time.ctime(item['timestamp'])}")
        except queue.Empty:
            break

Com este padrão:

• A thread do Watchdog é liberada imediatamente após colocar o evento na fila.
• A thread principal (GUI ou serviço de background) consome a fila no seu próprio ritmo sem risco de travamentos.

7. Boas Práticas e Considerações de Produção

Ao implantar um monitor de arquivos de log em ambientes de produção, considere as seguintes recomendações:

1. Atenção aos Arquivos Ocultos e Temporários: Editores de texto e sistemas criam arquivos temporários (como .log.tmp ou .~lock). Sempre utilize ignore_patterns ou ignore_directories=True para filtrar ruídos.
2. Limite de Descritores de Arquivos no Linux: Em diretórios muito extensos com monitoramento recursivo (recursive=True), o Linux pode atingir o limite do inotify. Se necessário, ajuste a variável de sistema fs.inotify.max_user_watches.
3. Desalocação de Recursos: Sempre encerre o observador adequadamente invocando observer.stop() e observer.join() dentro de blocos try...finally para liberar os descritores do sistema operacional.