""" Genera un resumen del día de todos los logs en /logs/*/ Uso: python3 resumen_logs.py # logs de hoy python3 resumen_logs.py 2026-05-27 # fecha específica """ import re import sys from datetime import date from pathlib import Path LOGS_DIR = Path(__file__).resolve().parent / "logs" # Nuevo formato: 2026-05-29T14:23:01-03:00 [INFO ] [proceso] [run:abc123] mensaje FMT_NUEVO = re.compile( r"^(\d{4}-\d{2}-\d{2})T\d{2}:\d{2}:\d{2}[+-]\d{2}:\d{2} \[(\w+)\s*\] \[([^\]]+)\] \[run:([^\]]+)\] (.+)$" ) # Formato legado: [2026-05-29 12:00:00] INFO: mensaje FMT_LEGADO_1 = re.compile(r"^\[(\d{4}-\d{2}-\d{2}) \d{2}:\d{2}:\d{2}\] (\w+): (.+)$") # Formato legado: 12:00:00 [INFO] mensaje FMT_LEGADO_2 = re.compile(r"^\d{2}:\d{2}:\d{2} \[(\w+)\] (.+)$") TIMEOUT_RE = re.compile(r"(timed out|ConnectTimeout|Max retries exceeded)", re.IGNORECASE) DESCARGADO_RE = re.compile(r"Descargado:\s*\S+/([^/\s]+\.pdf)") SCRIPT_RE = re.compile(r"\[([^\]]+\.py)\]") FALLO_RE = re.compile(r"FALLO (.+?\.py)") OK_RE = re.compile(r"^OK\s+(.+\.py)$") EXITOSOS_RE = re.compile(r"Exitosos\s+\((\d+)\):\s*(.+)") CON_ERROR_RE = re.compile(r"Con error\s+\((\d+)\):\s*(.+)") def _candidatos_log(proc_dir: Path, fecha: str) -> list[tuple[Path, str]]: """ Retorna lista de (path, modo) para leer. Modo: 'nuevo' | 'legado1' | 'legado2' Prioriza el archivo de fecha (nuevo formato) sobre proceso.log (legado). """ candidatos = [] fecha_log = proc_dir / f"{fecha}.log" if fecha_log.exists(): candidatos.append((fecha_log, "nuevo")) proceso_log = proc_dir / "proceso.log" if proceso_log.exists(): candidatos.append((proceso_log, "legado")) return candidatos def _parse_line(line: str, fecha: str, modo: str) -> tuple[str, str] | None: """Retorna (level, msg) si la línea pertenece al día indicado, sino None.""" if modo == "nuevo": m = FMT_NUEVO.match(line) if m and m.group(1) == fecha: return m.group(2).strip(), m.group(5).strip() return None # Legado: intentar FMT1 primero (tiene fecha), luego FMT2 (sin fecha, se acepta todo) m = FMT_LEGADO_1.match(line) if m: return (m.group(2), m.group(3).strip()) if m.group(1) == fecha else None m = FMT_LEGADO_2.match(line) if m: return m.group(1), m.group(2).strip() return None def _extraer_lineas_legado(lines: list[str], fecha: str) -> tuple[list[str], str]: """ Para logs legado sin fecha por línea (FMT2), extrae el bloque del día. Retorna (lineas_del_dia, submodo). """ tiene_fmt1_hoy = any( (m := FMT_LEGADO_1.match(l)) and m.group(1) == fecha for l in lines ) if tiene_fmt1_hoy: return lines, "legado1" # FMT2: buscar bloque por INICIO que mencione la fecha inicio_idx = None for i, line in enumerate(lines): if fecha in line and re.search(r"INICIO", line, re.IGNORECASE): inicio_idx = i break if inicio_idx is None: return [], "legado2" bloque = [] for line in lines[inicio_idx:]: m = FMT_LEGADO_1.match(line) if m and m.group(1) != fecha: break bloque.append(line) return bloque, "legado2" def analizar_proceso(proc_dir: Path, fecha: str) -> dict | None: candidatos = _candidatos_log(proc_dir, fecha) if not candidatos: return None stats = { "inicio": False, "fin": False, "scripts_ok": [], "scripts_fallo": [], "descargados": [], "errores": [], "warnings": [], "timeouts": [], "resumen_exitosos": None, "resumen_errores": None, "run_ids": set(), } encontro_algo = False for path, modo in candidatos: try: lines = path.read_text(encoding="utf-8").splitlines() except Exception as e: return {"error_lectura": str(e)} if modo == "legado": lines, modo = _extraer_lineas_legado(lines, fecha) if not lines: continue for line in lines: parsed = _parse_line(line, fecha, modo) if not parsed: continue level, msg = parsed encontro_algo = True # Extraer run_id si está presente (formato nuevo) if modo == "nuevo": m = FMT_NUEVO.match(line) if m: stats["run_ids"].add(m.group(4)) if re.search(r"INICIO", msg, re.IGNORECASE): stats["inicio"] = True if re.match(r"^FIN$", msg, re.IGNORECASE) or "PROCESO_FINALIZADO" in msg: stats["fin"] = True if m := OK_RE.match(msg): stats["scripts_ok"].append(m.group(1)) if m := FALLO_RE.search(msg): stats["scripts_fallo"].append(m.group(1)) if m := DESCARGADO_RE.search(msg): nombre_pdf = m.group(1) if fecha.replace("-", "") in nombre_pdf or fecha in nombre_pdf: stats["descargados"].append(nombre_pdf) if TIMEOUT_RE.search(msg): script = SCRIPT_RE.search(msg) nombre = script.group(1) if script else "desconocido" if nombre not in stats["timeouts"]: stats["timeouts"].append(nombre) if m := EXITOSOS_RE.search(msg): stats["resumen_exitosos"] = (int(m.group(1)), m.group(2).strip()) if m := CON_ERROR_RE.search(msg): stats["resumen_errores"] = (int(m.group(1)), m.group(2).strip()) if level in ("ERROR", "CRITICAL") and not any( x in msg for x in ["STDERR", "Parametros", "Ejecutando"] ): if not msg.startswith("File ") and "line" not in msg[:20]: stats["errores"].append(msg[:150]) if level == "WARNING": stats["warnings"].append(msg[:150]) # Si encontramos datos en el archivo de fecha (nuevo), no seguir con legado if encontro_algo and modo == "nuevo": break return stats if encontro_algo else None def estado_proceso(stats: dict) -> str: if stats.get("error_lectura"): return "ERROR DE LECTURA" if not stats["inicio"]: return "SIN ACTIVIDAD HOY" if stats["scripts_fallo"]: return "CON FALLOS" if stats["errores"]: return "CON ERRORES" if stats["inicio"] and not stats["fin"]: return "INCOMPLETO / EN CURSO" return "OK" def imprimir_resumen(fecha: str) -> None: procesos = sorted( p for p in LOGS_DIR.iterdir() if p.is_dir() and p.name != "_auditoria" and p.name != "test" ) resultados = [] for proc_dir in procesos: stats = analizar_proceso(proc_dir, fecha) if stats: resultados.append((proc_dir.name, stats)) if not resultados: print(f"\nNo hay actividad registrada para el {fecha}.") return print(f"\n{'='*60}") print(f" RESUMEN DE LOGS — {fecha}") print(f"{'='*60}") for nombre, stats in resultados: if stats.get("error_lectura"): print(f"\n ✗ {nombre.upper()}: no se pudo leer ({stats['error_lectura']})") continue estado = estado_proceso(stats) icono = "OK" if estado == "OK" else "!!" if "ERROR" in estado or "FALLO" in estado else "--" run_str = f" [{len(stats['run_ids'])} run(s)]" if stats["run_ids"] else "" print(f"\n [{icono}] {nombre.upper()} — {estado}{run_str}") if stats["resumen_exitosos"] is not None: ok_n, ok_lista = stats["resumen_exitosos"] err_n, err_lista = stats["resumen_errores"] or (0, "ninguno") print(f" Scripts OK ({ok_n}): {ok_lista}") if err_n > 0: print(f" Scripts con error ({err_n}): {err_lista}") if stats["descargados"]: print(f" Descargados ({len(stats['descargados'])}): {', '.join(stats['descargados'])}") if stats["timeouts"]: print(f" Timeouts: {', '.join(stats['timeouts'])}") if stats["errores"]: print(f" Errores ({len(stats['errores'])}):") for e in stats["errores"][:3]: print(f" - {e}") if len(stats["errores"]) > 3: print(f" ... y {len(stats['errores']) - 3} más") if stats["warnings"]: print(f" Warnings ({len(stats['warnings'])}):") for w in stats["warnings"][:2]: print(f" - {w}") if len(stats["warnings"]) > 2: print(f" ... y {len(stats['warnings']) - 2} más") print(f"\n{'='*60}\n") if __name__ == "__main__": fecha = sys.argv[1] if len(sys.argv) > 1 else str(date.today()) imprimir_resumen(fecha)