#!/usr/bin/env python3 """ Script para contar descargas de aplicaciones desde logs de Apache. Este script analiza los logs de acceso de Apache para contabilizar las descargas de archivos ejecutables (.exe, .msi) ubicados en un directorio específico. Genera un reporte que incluye: 1. Estadísticas acumuladas de los últimos 14 días. 2. Detalle de descargas del día anterior (incluyendo direcciones IP). El reporte se puede mostrar en consola (texto plano) y enviar por correo electrónico (HTML). """ import glob import gzip import os import re import subprocess import sys from datetime import datetime, timedelta from typing import List, Tuple, Dict, Generator, Optional # --- CONFIGURACIÓN --- # Prefijo de URL que identifica las descargas de aplicaciones PREFIX = "/estandares/aplicacion/" # Patrones para localizar los archivos de log de Apache (soporta rotación) LOG_PATTERNS = [ "/var/log/apache2/access.log*", "/var/log/apache2/other_vhosts_access.log*", ] # Directorio local donde residen los instaladores de las aplicaciones APPS_DIR = "/srv/repositorios/osluz/estandares/aplicacion" # Configuración de notificaciones por correo ENVIAR_CORREO = True EMAIL_TO = "871208@unizar.es, abailo@unizar.es" EMAIL_FROM = "abailo2@unizar.es" # Expresiones regulares precompiladas para optimización # Detecta extensiones de instaladores (.exe, .msi) EXT_RE = re.compile(r"\.(exe|msi)$", re.IGNORECASE) # Detecta el nombre base de la aplicación, ignorando versiones o sufijos tras '_' o '-' # Grupo 1 captura el nombre de la aplicación APPFILE_RE = re.compile(r"^([A-Za-z0-9]+)(?:[_-].*)?\.(exe|msi)$", re.IGNORECASE) def gather_logfiles() -> List[str]: """ Recopila todos los archivos de log que coinciden con los patrones definidos. Returns: List[str]: Lista ordenada de rutas absolutas a los archivos de log. """ files = [] for pat in LOG_PATTERNS: files.extend(glob.glob(pat)) return sorted(set(files)) def iter_log_lines(files: List[str]) -> Generator[Tuple[str, str], None, None]: """ Itera sobre las líneas de múltiples archivos de log, manejando compresión gzip. Abre transparentemente archivos .gz y archivos de texto plano. Maneja errores de permisos y archivos no encontrados para no interrumpir el proceso global. Args: files: Lista de rutas a archivos de log. Yields: Tuple[str, str]: Tupla conteniendo (línea del log, ruta del archivo origen). """ for path in files: # Verificar permisos de lectura antes de intentar abrir if not os.access(path, os.R_OK): print(f"Permiso denegado leyendo {path}. Ejecuta con sudo.", file=sys.stderr) continue try: if path.endswith(".gz"): fh = gzip.open(path, "rt", encoding="utf-8", errors="replace") else: fh = open(path, "rt", encoding="utf-8", errors="replace") except FileNotFoundError: continue except Exception as e: print(f"Error abriendo {path}: {e}", file=sys.stderr) continue with fh: for line in fh: yield line, path def apps_expuestas() -> List[str]: """ Identifica las aplicaciones disponibles en el directorio de repositorio. Escanea APPS_DIR buscando archivos .exe y .msi. Normaliza los nombres extrayendo la base del nombre (antes de guiones o guiones bajos) para agrupar diferentes versiones de la misma aplicación. Returns: List[str]: Lista ordenada de nombres base de aplicaciones (en minúsculas). """ apps = set() for ext in ("*.exe", "*.msi"): for p in glob.glob(os.path.join(APPS_DIR, ext)): fn = os.path.basename(p) m = APPFILE_RE.match(fn) if m: # Normaliza a minúsculas para comparaciones consistentes apps.add(m.group(1).lower()) return sorted(apps) def is_log_from_yesterday(filename: str) -> bool: """ Determina si un archivo de log corresponde específicamente a los datos de "ayer". Se basa en la convención estándar de logrotate: - access.log: log actual (hoy). - access.log.1: log rotado ayer (contiene datos completos de ayer). Args: filename: Nombre del archivo de log. Returns: bool: True si el archivo corresponde al log rotado de ayer. """ base = os.path.basename(filename) # Logrotate rota 'access.log' a 'access.log.1' (o .1.gz si comprime) if base.endswith(".1") or base.endswith(".1.gz"): return True return False def generar_html(total: int, counts: Dict[str, int], apps_list: List[str], downloads_yesterday: List[Tuple[str, str]], date_str: str) -> str: """ Genera el cuerpo del correo electrónico en formato HTML. Incluye estilos CSS embebidos para una presentación limpia. Muestra: 1. Total de descargas (últimos 14 días). 2. Tabla detallada por aplicación. 3. Tabla de descargas específicas del día anterior con IPs. Args: total: Número total de descargas. counts: Diccionario con conteos por aplicación {app: cantidad}. apps_list: Lista de todas las aplicaciones monitoreadas. downloads_yesterday: Lista de tuplas (IP, aplicación) descargadas ayer. date_str: Fecha de referencia para "ayer" (string formateado). Returns: str: Código HTML completo del reporte. """ fecha = datetime.now().strftime("%d/%m/%Y %H:%M:%S") html = f"""

Reporte de Descargas - Softlibre

Últimos 14 días
Total: {total:,} descargas
""" # Ordenar por número de descargas (descendente) y luego alfabéticamente ordered = sorted(apps_list, key=lambda a: (-counts.get(a, 0), a)) for app in ordered: n = counts.get(app, 0) html += f" \n" html += """
Aplicación Descargas
{app}{n:,}
""" # Sección de descargas del día anterior html += f"""

Día anterior ({date_str})

""" if downloads_yesterday: for ip, app in downloads_yesterday: html += f" \n" else: html += " \n" html += f"""
IP Aplicación
{ip}{app}
No hubo descargas registradas ayer.
Generado: {fecha}
Ruta: {PREFIX}
Período: Últimos 14 días + Detalle de {date_str}
""" return html def generar_texto_plano(total: int, counts: Dict[str, int], apps_list: List[str], downloads_yesterday: List[Tuple[str, str]], date_str: str) -> str: """ Genera el reporte en formato texto plano para salida por consola. Replica la estructura del reporte HTML pero formateado con tablas de texto. Args: total: Número total de descargas. counts: Diccionario con conteos por aplicación. apps_list: Lista de aplicaciones. downloads_yesterday: Lista de descargas de ayer (IP, App). date_str: Fecha de ayer formateada. Returns: str: Texto formateado listo para imprimir. """ lines = [] # Calcular ancho de columna dinámico basado en el nombre de app más largo max_len = max([len("Aplicación")] + [len(a) for a in apps_list]) if apps_list else len("Aplicación") lines.append("=" * 60) lines.append(" ESTADÍSTICAS DE DESCARGAS") lines.append(" ÚLTIMOS 14 DÍAS") lines.append("=" * 60) lines.append("") lines.append(f"TOTAL: {total:,} descargas") lines.append("") lines.append(f"{'Aplicación':<{max_len}} {'Descargas':>10}") lines.append("-" * (max_len + 11)) # Ordenar y listar descargas acumuladas ordered = sorted(apps_list, key=lambda a: (-counts.get(a, 0), a)) for app in ordered: lines.append(f"{app:<{max_len}} {counts.get(app, 0):10,d}") # Sección de Detalles de Ayer lines.append("") lines.append("=" * 60) lines.append(f" DESCARGAS DE AYER ({date_str})") lines.append("=" * 60) lines.append("") if downloads_yesterday: lines.append(f"{'IP':<16} {'Aplicación':<20}") lines.append("-" * 40) for ip, app in downloads_yesterday: lines.append(f"{ip:<16} {app:<20}") else: lines.append("No hubo descargas registradas ayer.") lines.append("") lines.append("Generado: " + datetime.now().strftime("%a %d %b %Y %H:%M:%S").strip()) return "\n".join(lines) def enviar_reporte(html_content: str): """ Envía el reporte generado por correo electrónico usando sendmail local. Utiliza el binario /usr/sbin/sendmail para inyectar el correo en la cola del sistema. Requiere que el sistema tenga un MTA configurado. Args: html_content: Cuerpo del mensaje en formato HTML. """ if not ENVIAR_CORREO: return asunto = f"[softlibre] Reporte de Descargas (14 días) - {datetime.now().strftime('%d/%m/%Y')}" destinatarios = [d.strip() for d in EMAIL_TO.split(",") if d.strip()] # Construcción de cabeceras MIME msg = f"""From: {EMAIL_FROM} To: {", ".join(destinatarios)} Subject: {asunto} MIME-Version: 1.0 Content-Type: text/html; charset=UTF-8 {html_content} """ try: # -t: lee destinatarios desde las cabeceras (To:, Cc:, Bcc:) subprocess.run(["/usr/sbin/sendmail", "-t"], input=msg, text=True, check=True) print(f"\n[OK] Reporte enviado correctamente a {EMAIL_TO}") except Exception as e: print(f"\n[ERROR] Error al enviar correo: {e}") def main(): """ Función principal de orquestación. 1. Busca y lee los logs de Apache. 2. Identifica las aplicaciones disponibles. 3. Itera línea a línea sobre los logs para extraer métricas. 4. Genera reportes en texto y HTML. 5. Envía el reporte por correo. """ files = gather_logfiles() if not files: print("No se encontraron logs en /var/log/apache2/") sys.exit(1) apps_list = apps_expuestas() if not apps_list: print(f"No se encontraron .exe/.msi en {APPS_DIR}") sys.exit(1) total = 0 counts: dict[str, int] = {} # Calcular fecha de ayer para el reporte diario ayer = datetime.now() - timedelta(days=1) str_ayer = ayer.strftime("%d/%b/%Y") # Formato visual para el reporte downloads_yesterday = [] # Procesamiento de logs for line, filepath in iter_log_lines(files): # Filtrado rápido por prefijo para descartar líneas irrelevantes if PREFIX not in line: continue # Parsing manual de logs formato Common/Combined Log Format # Estructura típica: IP - - [Fecha] "GET /ruta HTTP/1.1" 200 Bytes ... parts = line.split('"') if len(parts) < 3: continue preamble = parts[0] # IP logname user [date] request = parts[1] # GET /ruta HTTP/1.x after = parts[2].strip().split() # status bytes ... if not after: continue status = after[0] # Solo contar descargas exitosas (200) if status not in ("200"): continue req_parts = request.split() if len(req_parts) < 2: continue method, url = req_parts[0], req_parts[1] # Validaciones de URL if method != "GET": continue if not url.startswith(PREFIX): continue # Limpiar cadena de consulta (query string) si existe url_clean = url.split("?", 1)[0] if not EXT_RE.search(url_clean): continue # Extracción y limpieza del nombre de la aplicación filename = url_clean.rsplit("/", 1)[-1] base = re.sub(r"\.(exe|msi)$", "", filename, flags=re.IGNORECASE) # Normaliza eliminando sufijos de versión (ej: app_v1.0 -> app) base = re.sub(r"[_-].*$", "", base).lower() # Ignorar descargas de archivos que no corresponden a apps conocidas if base not in set(apps_list): continue # 1. Acumular estadísticas generales (últimos 14 días según rotación de logs) counts[base] = counts.get(base, 0) + 1 total += 1 # 2. Registrar descargas específicas de AYER # Se usa el nombre del fichero de log (access.log.1) como proxy para la fecha if is_log_from_yesterday(filepath): ip = preamble.split()[0] downloads_yesterday.append((ip, base)) # Generar salida en consola print(generar_texto_plano(total, counts, apps_list, downloads_yesterday, str_ayer)) # Enviar reporte por correo if ENVIAR_CORREO: html = generar_html(total, counts, apps_list, downloads_yesterday, str_ayer) enviar_reporte(html) if __name__ == "__main__": main()