typestar

py_log_scanner.py en Python

Un escáner de logs con argparse, logging, regex y una tabla resumen.

import argparse
import logging
import re
import sys
from collections import Counter, defaultdict

LINEA = re.compile(
    r"(?P<time>\d{2}:\d{2}:\d{2})\s+"
    r"(?P<level>DEBUG|INFO|WARNING|ERROR)\s+"
    r"(?P<logger>[\w.]+):\s+(?P<message>.*)"
)

MUESTRA = """09:14:02 INFO typestar.serve: arrancó en 8080
09:14:07 DEBUG typestar.db: cargó 794 pasos
09:15:31 WARNING typestar.ingest: fragmento muy largo: big.py
09:16:02 ERROR typestar.serve: 500 sin manejar en /api/results
09:16:03 ERROR typestar.serve: 500 sin manejar en /api/results
09:17:44 INFO typestar.serve: 200 en /healthz
esto no es una línea de log
"""

registro = logging.getLogger("scanner")


def analizar(lineas):
    """Produce dicts de coincidencias y cuenta las líneas que no encajan."""
    omitidas = 0
    for linea in lineas:
        linea = linea.strip()
        if not linea:
            continue
        encontrado = LINEA.fullmatch(linea)
        if encontrado is None:
            omitidas += 1
            registro.debug("sin analizar: %s", linea)
            continue
        yield encontrado.groupdict()
    if omitidas:
        registro.warning("%d línea(s) sin analizar", omitidas)


def resumir(entradas):
    niveles = Counter()
    por_logger = defaultdict(Counter)
    repetidos = Counter()

    for entrada in entradas:
        niveles[entrada["level"]] += 1
        por_logger[entrada["logger"]][entrada["level"]] += 1
        if entrada["level"] == "ERROR":
            repetidos[entrada["message"]] += 1
    return niveles, por_logger, repetidos


def principal(argv=None):
    parser = argparse.ArgumentParser(description="resume un log")
    parser.add_argument("--level", default="INFO",
                        choices=["DEBUG", "INFO", "WARNING"])
    parser.add_argument("--top", type=int, default=3)
    args = parser.parse_args(argv)

    logging.basicConfig(level=getattr(logging, args.level),
                        format="%(levelname)s %(name)s: %(message)s")

    niveles, por_logger, repetidos = resumir(analizar(MUESTRA.splitlines()))

    print(f"{'nivel':<9}{'conteo':>6}")
    for level in ("ERROR", "WARNING", "INFO", "DEBUG"):
        if niveles[level]:
            print(f"{level:<9}{niveles[level]:>6}")

    print("\npor logger")
    for nombre in sorted(por_logger):
        conteos = por_logger[nombre]
        datos = " ".join(f"{k.lower()}={v}" for k, v in sorted(conteos.items()))
        print(f"  {nombre:<18} {datos}")

    if repetidos:
        print("\nerrores repetidos")
        for mensaje, conteo in repetidos.most_common(args.top):
            print(f"  {conteo}x {mensaje}")
    return 0 if not niveles["ERROR"] else 1


if __name__ == "__main__":
    sys.exit(principal())

Cómo funciona

  1. argparse define la interfaz, incluida una bandera de nivel.
  2. La regex nombra sus grupos, así el código los lee por nombre.
  3. Los Counter arman el resumen en una sola pasada por las líneas.

Palabras clave y builtins usados aquí

El intento, en números

Líneas
86
Caracteres a escribir
2356
Tokens
554
Ritmo de tres estrellas
110 tpm

Al ritmo de tres estrellas de 110 tokens por minuto, este intento toma unos 302 segundos.

Escribe este fragmento

Paso 3 de 3 en Bis; paso 41 de 41 en Herramientas de dominio.

← Anterior