Error Fingerprinting em Python Puro: O Algoritmo Que Agrupa Erros Similares Automaticamente e Mostra Qual Bug Resolver Primeiro (Sem Sentry, Sem Bugsnag)

Você já ficou olhando pro log de erros da sua aplicação às 2 da manhã, tentando entender qual dos 847 erros era o mais urgente pra resolver? Eu já. Mais vezes do que gostaria de admitir.

O problema é brutal: quando você tem centenas de erros chegando por hora, agrupar manualmente é impossível. E contratar Sentry, Bugsnag ou Rollbar? Caro demais pra quem tá começando ou rodando side projects.

Então eu construí um error fingerprinting em Python puro. Um algoritmo que lê seus logs, calcula a “impressão digital” de cada erro, agrupa os similares automaticamente e te mostra qual bug resolver primeiro baseado em frequência e impacto. Zero dependências externas. Zero mensalidade.

O Problema: Por Que Logs Brutos Não Funcionam

Deixa eu te contar uma história real. Semana passada, minha API tava jogando 1.200 erros por hora no log. Olhando rápido, parecia um massacre:

• 300 “Connection refused”
• 250 “Timeout exceeded”
• 200 “Invalid JSON”
• 150 “Database connection failed”
• E mais 300 variações de stack traces

Mas aqui está o pulo do gato: 80% desses erros eram variações do mesmo problema. O database connection failed? Era o mesmo pool de conexões estourando em 4 endpoints diferentes. O timeout? Era um serviço downstream lento afetando 3 chamadas.

Sem agrupamento inteligente, você perde horas caçando fantasmas. Com fingerprinting, você vê em 5 minutos: “Ah, é só escalar o pool de conexões e adicionar retry no serviço X.”

O Que é Error Fingerprinting (e Por Que Funciona)

Error fingerprinting é calcular um hash único pra cada “tipo” de erro, ignorando dados variáveis (timestamps, IDs, IPs) e focando na estrutura do problema.

Pensa assim: dois “Connection refused to database:5432” são o mesmo erro, mesmo que tenham timestamps diferentes. Mas “Connection refused to database:5432” e “Connection refused to redis:6379” são erros diferentes.

O algoritmo que vou te mostrar faz isso:

1. Extrai a mensagem de erro e o stack trace
2. Normaliza dados variáveis (remove números, UUIDs, timestamps)
3. Calcula um hash determinístico
4. Agrupa erros com o mesmo hash
5. Rankeia por frequência × severidade

Implementação: O Core do Algoritmo

Vamos pro código. Primeiro, a função de normalização — o coração do fingerprint:

import re
import hashlib
from datetime import datetime
from collections import defaultdict

def normalize_error(message):
    """
    Remove dados variáveis pra calcular fingerprint consistente.
    """
    # Remove timestamps (ISO 8601, Unix, etc)
    normalized = re.sub(
        r'\d{4}-\d{2}-\d{2}[T ]\d{2}:\d{2}:\d{2}[.\d]*[Z]?',
        '[TIMESTAMP]',
        message
    )
    
    # Remove UUIDs
    normalized = re.sub(
        r'[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}',
        '[UUID]',
        normalized,
        flags=re.IGNORECASE
    )
    
    # Remove números (IDs, ports, line numbers)
    normalized = re.sub(r'\b\d+\b', '[NUM]', normalized)
    
    # Remove IPs
    normalized = re.sub(
        r'\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}',
        '[IP]',
        normalized
    )
    
    # Remove paths específicos, mantém estrutura
    normalized = re.sub(
        r'(/[^/\s]+)+\.\w+',
        '[PATH]',
        normalized
    )
    
    return normalized

def calculate_fingerprint(error_message, stack_trace=""):
    """
    Calcula hash único pro tipo de erro.
    """
    normalized_msg = normalize_error(error_message)
    normalized_stack = normalize_error(stack_trace)
    
    # Combina mensagem + stack pra fingerprint único
    combined = f"{normalized_msg}||{normalized_stack}"
    
    # SHA256 dos primeiros 16 chars (suficiente pra evitar colisões)
    return hashlib.sha256(combined.encode()).hexdigest()[:16]

Testa isso:

# Mesmo erro, timestamps diferentes
err1 = "2026-07-30T21:00:00Z Connection refused to database:5432 (attempt 42)"
err2 = "2026-07-30T21:05:23Z Connection refused to database:5432 (attempt 137)"

fp1 = calculate_fingerprint(err1)
fp2 = calculate_fingerprint(err2)

print(fp1 == fp2)  # True! Mesmo fingerprint

Agrupando Erros em Tempo Real

Agora a parte divertida: agrupar erros conforme chegam e rankear por prioridade.

class ErrorFingerprinter:
    def __init__(self):
        self.groups = defaultdict(lambda: {
            'count': 0,
            'first_seen': None,
            'last_seen': None,
            'sample_message': None,
            'severity_score': 0
        })
    
    def ingest(self, error_message, stack_trace="", severity=1):
        """
        Processa um erro e atualiza grupo correspondente.
        """
        fp = calculate_fingerprint(error_message, stack_trace)
        group = self.groups[fp]
        
        now = datetime.now()
        
        if group['first_seen'] is None:
            group['first_seen'] = now
            group['sample_message'] = error_message
        
        group['count'] += 1
        group['last_seen'] = now
        group['severity_score'] = max(group['severity_score'], severity)
        
        return fp
    
    def get_priority_ranking(self):
        """
        Rankeia grupos por prioridade: frequência × severidade × recenticidade.
        """
        scored = []
        
        for fp, data in self.groups.items():
            # Score = count × severity × recency_factor
            hours_active = (data['last_seen'] - data['first_seen']).total_seconds() / 3600
            recency = 1 / (hours_active + 1)  # Mais recente = maior peso
            
            priority_score = (
                data['count'] * 
                data['severity_score'] * 
                (1 + recency)
            )
            
            scored.append({
                'fingerprint': fp,
                'count': data['count'],
                'severity': data['severity_score'],
                'priority_score': priority_score,
                'sample': data['sample_message']
            })
        
        return sorted(scored, key=lambda x: x['priority_score'], reverse=True)

Usando na Prática: Monitorando Logs de Arquivo

Agora vamos conectar isso num log real. Imagina que você tem um arquivo app.log sendo escrito continuamente:

import time

def tail_log_file(filepath, fingerprinter):
    """
    Lê log em tempo real (como tail -f) e agrupa erros.
    """
    with open(filepath, 'r') as f:
        # Vai pro fim do arquivo
        f.seek(0, 2)
        
        while True:
            line = f.readline()
            
            if not line:
                time.sleep(0.1)
                continue
            
            # Detecta se é linha de erro
            if 'ERROR' in line or 'CRITICAL' in line:
                # Extrai mensagem (assumindo formato padrão)
                parts = line.split('ERROR')
                if len(parts) > 1:
                    message = parts[1].strip()
                    
                    # Severidade: CRITICAL = 3, ERROR = 2
                    severity = 3 if 'CRITICAL' in line else 2
                    
                    fingerprinter.ingest(message, severity=severity)
            
            # A cada 10 segundos, mostra ranking
            if int(time.time()) % 10 == 0:
                ranking = fingerprinter.get_priority_ranking()
                print("\n=== TOP 5 ERROS PRIORITÁRIOS ===")
                for i, err in enumerate(ranking[:5], 1):
                    print(f"{i}. [{err['count']}x] {err['sample'][:80]}...")
                    print(f"   Score: {err['priority_score']:.2f}")

# Uso
fp = ErrorFingerprinter()
tail_log_file('/var/log/app.log', fp)

Integração com Alertas: Notificando Só o Que Importa

Agrupar é bom, mas alertar é melhor. Vamos adicionar lógica pra notificar só quando um grupo novo aparece ou quando um grupo existente explode:

class AlertingFingerprinter(ErrorFingerprinter):
    def __init__(self, alert_callback, burst_threshold=50):
        super().__init__()
        self.alert_callback = alert_callback
        self.burst_threshold = burst_threshold
        self.known_fingerprints = set()
    
    def ingest(self, error_message, stack_trace="", severity=1):
        fp = super().ingest(error_message, stack_trace, severity)
        group = self.groups[fp]
        
        # Alerta 1: Erro novo (fingerprint nunca visto)
        if fp not in self.known_fingerprints:
            self.known_fingerprints.add(fp)
            self.alert_callback(
                type='NEW_ERROR',
                fingerprint=fp,
                message=error_message,
                severity=severity
            )
        
        # Alerta 2: Burst (muitos erros rápidos)
        elif group['count'] % self.burst_threshold == 0:
            self.alert_callback(
                type='ERROR_BURST',
                fingerprint=fp,
                count=group['count'],
                message=group['sample_message']
            )
        
        return fp

# Callback exemplo
def send_alert(type, **kwargs):
    if type == 'NEW_ERROR':
        print(f"🆕 NOVO ERRO DETECTADO: {kwargs['message'][:100]}")
    elif type == 'ERROR_BURST':
        print(f"💥 BURST: {kwargs['count']} ocorrências de {kwargs['fingerprint']}")

# Uso
alert_fp = AlertingFingerprinter(send_alert, burst_threshold=100)
tail_log_file('/var/log/app.log', alert_fp)

Otimizações: Lidando com Milhões de Erros

Se você tá processando milhões de erros por dia, algumas otimizações são essenciais:

1. Bloom Filter pra Deduplicação Rápida

Antes de calcular fingerprint completo, usa um Bloom filter pra checar se já vimos algo similar:

from pybloom_live import BloomFilter

class OptimizedFingerprinter(ErrorFingerprinter):
    def __init__(self, capacity=1000000, error_rate=0.001):
        super().__init__()
        self.bloom = BloomFilter(capacity=capacity, error_rate=error_rate)
    
    def ingest(self, error_message, stack_trace="", severity=1):
        # Check rápido: já vimos essa mensagem exata?
        if error_message in self.bloom:
            # Provavelmente duplicata, mas ainda calcula fingerprint
            pass
        else:
            self.bloom.add(error_message)
        
        return super().ingest(error_message, stack_trace, severity)

2. Time-Windowed Grouping

Agrupa só erros das últimas N horas (descarta antigos):

def cleanup_old_groups(self, max_age_hours=24):
    """Remove grupos inativos há mais de max_age_hours."""
    now = datetime.now()
    cutoff = timedelta(hours=max_age_hours)
    
    to_delete = [
        fp for fp, data in self.groups.items()
        if now - data['last_seen'] > cutoff
    ]
    
    for fp in to_delete:
        del self.groups[fp]
        self.known_fingerprints.discard(fp)

Comparação: Fingerprinting vs Soluções Pagas

Vale a pena construir isso ao invés de pagar Sentry/Bugsnag?

Critério	Error Fingerprinting DIY	Sentry/Bugsnag
Custo	Zero (tempo de dev)	$26-299/mês
Setup	30 minutos	5 minutos
Customização	Total	Limitada
Manutenção	Você	Eles
Features extras	Você constrói	Pronto (user tracking, sourcemaps, etc)

Minha recomendação: usa fingerprinting DIY se você tem:

• Side projects ou MVPs com orçamento zero
• Necessidades específicas de agrupamento
• Vontade de aprender como debugging funciona por baixo

Usa Sentry/Bugsnag se:

• Time grande, precisa de colaboração
• App em produção com SLA sério
• Não quer manter infra de logging

Próximos Passos: Onde Levar Isso

O algoritmo base tá pronto, mas tem muito pra expandir:

• Integração com Slack/Telegram: Manda ranking diário dos top 5 erros
• Dashboard web: Flask + Chart.js mostrando tendências
• Machine learning: Agrupa por similaridade semântica (não só regex)
• Distribuído: Roda em múltiplos servidores, agrega via Redis

Conclusão: Pare de Caçar Fantasmas

Error fingerprinting não é luxo — é sobrevivência. Quando você tem centenas de erros chegando, agrupar manualmente é pedir pra enlouquecer.

Com 200 linhas de Python puro, você tem:

• Agrupamento inteligente de erros similares
• Ranking automático por prioridade
• Alertas de erros novos e bursts
• Zero dependência de serviços pagos

Da próxima vez que seu log parecer um campo de batalha, lembra: não é sobre quantos erros você tem, é sobre quantos tipos diferentes de erro você tem.

E agora você sabe como descobrir isso em 5 minutos.

---

Qual automação de debugging você quer ver aqui? Log analysis com machine learning? Detecção automática de anomalias? Integração com Prometheus/Grafana? Me conta nos comentários que eu trago no próximo post.