Analisando Logs e Texto

Scripts de operações extraem sinais de arquivos de log e saída de comandos usando I/O de streaming, regex e análise JSON opcional - sem carregar arquivos de vários gigabytes na memória.

Receita

import re
from pathlib import Path
 
PATTERN = re.compile(r"ERROR (?P<component>\w+): (?P<message>.+)")
 
for line in Path("app.log").open(encoding="utf-8", errors="replace"):
    match = PATTERN.search(line)
    if match:
        print(match.groupdict())

Quando usar isso:

Triagem de incidentes - scripts semelhantes a grep com lógica mais rica
Métricas de logs - contar códigos de erro por hora
ETL ad-hoc de formatos de texto legados
Validar arquivos de exportação antes do upload

Exemplo de Trabalho

Transmitir log de acesso do nginx, agregar códigos de status, lidar com linhas JSON de formato misto.

import json
import re
from collections import Counter
from dataclasses import dataclass
from pathlib import Path
 
NGINX = re.compile(
    r'^(?P<ip>\S+) \S+ \S+ \[(?P<time>[^\]]+)\] "(?P<method>\w+) (?P<path>\S+)'
)
 
@dataclass
class ParseResult:
    status_counts: Counter
    errors: int
 
def parse_access_log(path: Path) -> ParseResult:
    counts: Counter = Counter()
    errors = 0
    with path.open(encoding="utf-8", errors="replace") as f:
        for line in f:
            line = line.strip()
            if not line:
                continue
            if line.startswith("{"):
                try:
                    rec = json.loads(line)
                    counts[rec.get("status", "unknown")] += 1
                    continue
                except json.JSONDecodeError:
                    errors += 1
                    continue
            m = NGINX.search(line)
            if not m:
                errors += 1
                continue
            # o código de status geralmente vem depois da requisição - a demonstração simplificada usa o token do caminho
            counts["parsed"] += 1
    return ParseResult(counts, errors)
 
if __name__ == "__main__":
    result = parse_access_log(Path("access.log"))
    print(result.status_counts)
    print("erros de análise", result.errors)

O que isso demonstra:

O streaming linha por linha mantém a memória plana
errors="replace" evita travamentos em bytes UTF-8 inválidos
O caminho de linhas JSON suporta logs estruturados ao lado de texto legado

Mergulho Profundo

Estratégia de Análise

Tentar análise estruturada (JSON) primeiro
Recorrer a regex para linhas legadas
Contar linhas não analisáveis para métricas de qualidade de dados

Dicas de Regex

Use grupos nomeados (?P<name>...) para extração legível
Compile padrões uma vez fora dos loops
Ancore cuidadosamente - os formatos de log variam por versão

Notas de Python

import gzip
from pathlib import Path
 
with gzip.open(Path("app.log.gz"), "rt", encoding="utf-8", errors="replace") as f:
    for line in f:
        ...

Armadilhas

Ler o arquivo inteiro com .read() - OOM em logs de GB. Correção: iterar linhas ou usar mmap para casos avançados.
Regex ganancioso em campos variáveis - backtracking catastrófico. Correção: quantificadores possessivos ou dividir campos sem regex quando possível.
Assumir formato de log único - atualizações quebram o analisador. Correção: detecção de versão ou analisador múltiplo permissivo com contador de erros.
Fuso horário local em timestamps - buckets de agregação incorretos. Correção: analisar para UTC com datetime.timezone.
Registrar PII analisado - incidente GDPR. Correção: agregar contagens, redigir IPs na saída.

Alternativas

Alternativa	Usar Quando	Não Usar Quando
`rg`/`grep` shell	Inspeção humana rápida	Necessita de lógica de agregação em Python
ELK/Loki	Plataforma de log centralizada existente	Análise pontual em laptop
`pandas.read_csv`	Logs tabulares cabem na RAM	Arquivos de vários GB

FAQs

regex vs split?

Dividir por delimitadores quando o formato é fixo; regex quando os campos variam ou são opcionais.

Como analiso traces de pilha de várias linhas?

Máquina de estados: iniciar registro em linha de timestamp, acumular até o próximo padrão de timestamp.

Como lidar com rotação gzip?

Abrir .gz com gzip.open em modo de texto; processar arquivos rotacionados em ordem classificada por mtime.

Entrada de log estruturada?

json.loads por linha (NDJSON) - o caminho mais rápido quando os aplicativos emitem logs JSON.

Como testo analisadores?

Arquivos de fixture com contagens esperadas de "golden" em parametrização pytest.

Desempenho em arquivos enormes?

A iteração de linha geralmente é suficiente; considere multiprocessing por arquivo ao agrupar milhares de arquivos.

E logs binários?

Permanecer em modo binário e decodificar apenas regiões de texto conhecidas - não assumir UTF-8 em todo o arquivo.

Como extraio timestamps ISO?

datetime.fromisoformat para logs JSON; strptime para formatos legados com string de formato explícita.

Posso usar pathlib com gzip?

Passe o caminho para gzip.open - objetos pathlib funcionam onde a biblioteca padrão aceita pathlike.

Quando enviar o analisador para o pipeline de produção?

Quando o script ad-hoc estabiliza - promova para um trabalho agendado com testes e monitoramento na taxa de erro de análise.

Relacionado

Automação de Arquivos e Diretórios - caminhos de arquivos de log
Logging Estruturado - produção de logs JSON
Scripts Robustos - códigos de saída em limites de falha de análise
subprocess & Interoperabilidade com Shell - saída de comando de análise
Melhores Práticas de Script de Sistema - higiene de scripts de operações

Versões de Stack: Esta página foi escrita para Python 3.14.0 (estável 3.14, manutenção 3.13), FastAPI 0.115+, Django 5.2, Flask 3.1, Pydantic 2, PyTorch 2.6+, pandas 2.2+, Polars 1.x, ruff 0.9+ e uv 0.6+.

Analisando Logs e Texto

Scripts de operações extraem sinais de arquivos de log e saída de comandos usando I/O de streaming, regex e análise JSON opcional - sem carregar arquivos de vários gigabytes na memória.

Receita

import re
from pathlib import Path
 
PATTERN = re.compile(r"ERROR (?P<component>\w+): (?P<message>.+)")
 
for line in Path("app.log").open(encoding="utf-8", errors="replace"):
    match = PATTERN.search(line)
    if match:
        print(match.groupdict())

Quando usar isso:

Triagem de incidentes - scripts semelhantes a grep com lógica mais rica
Métricas de logs - contar códigos de erro por hora
ETL ad-hoc de formatos de texto legados
Validar arquivos de exportação antes do upload

Exemplo de Trabalho

Transmitir log de acesso do nginx, agregar códigos de status, lidar com linhas JSON de formato misto.

import json
import re
from collections import Counter
from dataclasses import dataclass
from pathlib import Path
 
NGINX = re.compile(
    r'^(?P<ip>\S+) \S+ \S+ \[(?P<time>[^\]]+)\] "(?P<method>\w+) (?P<path>\S+)'
)
 
@dataclass
class ParseResult:
    status_counts: Counter
    errors: int
 
def parse_access_log(path: Path) -> ParseResult:
    counts: Counter = Counter()
    errors = 0
    with path.open(encoding="utf-8", errors="replace") as f:
        for line in f:
            line = line.strip()
            if not line:
                continue
            if line.startswith("{"):
                try:
                    rec = json.loads(line)
                    counts[rec.get("status", "unknown")] += 1
                    continue
                except json.JSONDecodeError:
                    errors += 1
                    continue
            m = NGINX.search(line)
            if not m:
                errors += 1
                continue
            # o código de status geralmente vem depois da requisição - a demonstração simplificada usa o token do caminho
            counts["parsed"] += 1
    return ParseResult(counts, errors)
 
if __name__ == "__main__":
    result = parse_access_log(Path("access.log"))
    print(result.status_counts)
    print("erros de análise", result.errors)

O que isso demonstra:

O streaming linha por linha mantém a memória plana
errors="replace" evita travamentos em bytes UTF-8 inválidos
O caminho de linhas JSON suporta logs estruturados ao lado de texto legado

Mergulho Profundo

Estratégia de Análise

Tentar análise estruturada (JSON) primeiro
Recorrer a regex para linhas legadas
Contar linhas não analisáveis para métricas de qualidade de dados

Dicas de Regex

Use grupos nomeados (?P<name>...) para extração legível
Compile padrões uma vez fora dos loops
Ancore cuidadosamente - os formatos de log variam por versão

Notas de Python

import gzip
from pathlib import Path
 
with gzip.open(Path("app.log.gz"), "rt", encoding="utf-8", errors="replace") as f:
    for line in f:
        ...

Armadilhas

Ler o arquivo inteiro com .read() - OOM em logs de GB. Correção: iterar linhas ou usar mmap para casos avançados.
Regex ganancioso em campos variáveis - backtracking catastrófico. Correção: quantificadores possessivos ou dividir campos sem regex quando possível.
Assumir formato de log único - atualizações quebram o analisador. Correção: detecção de versão ou analisador múltiplo permissivo com contador de erros.
Fuso horário local em timestamps - buckets de agregação incorretos. Correção: analisar para UTC com datetime.timezone.
Registrar PII analisado - incidente GDPR. Correção: agregar contagens, redigir IPs na saída.

Alternativas

Alternativa	Usar Quando	Não Usar Quando
`rg`/`grep` shell	Inspeção humana rápida	Necessita de lógica de agregação em Python
ELK/Loki	Plataforma de log centralizada existente	Análise pontual em laptop
`pandas.read_csv`	Logs tabulares cabem na RAM	Arquivos de vários GB

FAQs

regex vs split?

Dividir por delimitadores quando o formato é fixo; regex quando os campos variam ou são opcionais.

Como analiso traces de pilha de várias linhas?

Máquina de estados: iniciar registro em linha de timestamp, acumular até o próximo padrão de timestamp.

Como lidar com rotação gzip?

Abrir .gz com gzip.open em modo de texto; processar arquivos rotacionados em ordem classificada por mtime.

Entrada de log estruturada?

json.loads por linha (NDJSON) - o caminho mais rápido quando os aplicativos emitem logs JSON.

Como testo analisadores?

Arquivos de fixture com contagens esperadas de "golden" em parametrização pytest.

Desempenho em arquivos enormes?

A iteração de linha geralmente é suficiente; considere multiprocessing por arquivo ao agrupar milhares de arquivos.

E logs binários?

Permanecer em modo binário e decodificar apenas regiões de texto conhecidas - não assumir UTF-8 em todo o arquivo.

Como extraio timestamps ISO?

datetime.fromisoformat para logs JSON; strptime para formatos legados com string de formato explícita.

Posso usar pathlib com gzip?

Passe o caminho para gzip.open - objetos pathlib funcionam onde a biblioteca padrão aceita pathlike.

Quando enviar o analisador para o pipeline de produção?

Quando o script ad-hoc estabiliza - promova para um trabalho agendado com testes e monitoramento na taxa de erro de análise.

Relacionado

Automação de Arquivos e Diretórios - caminhos de arquivos de log
Logging Estruturado - produção de logs JSON
Scripts Robustos - códigos de saída em limites de falha de análise
subprocess & Interoperabilidade com Shell - saída de comando de análise
Melhores Práticas de Script de Sistema - higiene de scripts de operações

Versões de Stack: Esta página foi escrita para Python 3.14.0 (estável 3.14, manutenção 3.13), FastAPI 0.115+, Django 5.2, Flask 3.1, Pydantic 2, PyTorch 2.6+, pandas 2.2+, Polars 1.x, ruff 0.9+ e uv 0.6+.

Analisando Logs e Texto

Buscar na documentação

Analisando Logs e Texto