#!/usr/bin/env python3 """ send_reports_loop_mejorado.py Script optimizado para el envío de reportes de forma más eficiente. """ import mysql.connector import socket import sys import time # ---- Configuración de la base de datos ---- DB_CONFIG = { 'host': '10.2.12.220', 'user': 'gps', 'password': 'q1w2e3r4', 'database': 'gps_reportes', 'port': 3306 } # ---- Configuración del balanceador de carga ---- #BALANCER_HOST = 'nlb-recepcion-tcp-98af84506c9b42a6.elb.us-east-1.amazonaws.com' BALANCER_HOST = 'gpsaws3.divisiongps.com.ar' BALANCER_PORT = 7001 # ---- Parámetros del script ---- REPORT_LIMIT = 200 # Cuántos reportes traer en cada iteración SLEEP_INTERVAL = 10 # Segundos de espera entre cada bucle completo SLEEP_BETWEEN_BATCHES = 10 # Segundos de espera entre lotes del mismo ciclo def get_total_records(): """Obtiene el total de registros en la tabla ultima_posicion.""" try: conn = mysql.connector.connect(**DB_CONFIG) cursor = conn.cursor() cursor.execute("SELECT COUNT(*) FROM ultima_posicion") total = cursor.fetchone()[0] cursor.close() conn.close() return total except mysql.connector.Error as e: print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] [ERROR] Error al contar registros: {e}", file=sys.stderr) return 0 def fetch_reports_batch(offset=0, limit=REPORT_LIMIT): """Obtiene un lote de reportes usando LIMIT y OFFSET para paginación.""" print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] Obteniendo lote {offset//limit + 1}: registros {offset + 1} a {offset + limit}...") start_time = time.time() try: conn = mysql.connector.connect(**DB_CONFIG) cursor = conn.cursor() query = """ SELECT Reporte_Completo FROM ultima_posicion ORDER BY Reporte_Fechagps DESC LIMIT %s OFFSET %s """ cursor.execute(query, (limit, offset)) rows = cursor.fetchall() cursor.close() conn.close() end_time = time.time() print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] Recuperados {len(rows)} reportes en {end_time - start_time:.2f}s.") return [row[0] for row in rows] except mysql.connector.Error as e: print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] [ERROR] Fallo al conectar a la BD: {e}", file=sys.stderr) return [] def send_reports_optimized(reports): """Envía los reportes usando una única conexión TCP persistente.""" total = len(reports) if not reports: print("No hay reportes para enviar. Saltando el envío.") return print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] Conectando a {BALANCER_HOST}:{BALANCER_PORT}...") try: # ---- CAMBIO CLAVE: Se establece la conexión una sola vez ---- with socket.create_connection((BALANCER_HOST, BALANCER_PORT), timeout=5) as sock: print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] Conexión establecida. Iniciando envío de {total} reportes...") start_time = time.time() # Se envía cada reporte utilizando la misma conexión for i, report in enumerate(reports, start=1): try: sock.sendall(report.encode('utf-8')) # Opcional: imprimir el progreso cada 1000 reportes para evitar saturar la consola if i % 1000 == 0 or i == total: print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] Enviado {i}/{total}") except Exception as e: print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] [ERROR] Fallo el envío del reporte {i}/{total}: {e}", file=sys.stderr) # Si falla un envío, la conexión puede estar rota. Se sale del bucle. break end_time = time.time() duration = end_time - start_time print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] Envío completado. Tiempo total: {duration:.2f} segundos.") except Exception as e: print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] [ERROR] No se pudo establecer la conexión TCP: {e}", file=sys.stderr) def main_loop(): """Bucle principal de ejecución con envío completo de la tabla por lotes.""" print("Iniciando bucle de envío de reportes. Presiona Ctrl+C para detener.") try: while True: print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] ===== INICIANDO NUEVO CICLO COMPLETO =====") # Obtener total de registros total_records = get_total_records() if total_records == 0: print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] No hay registros en la tabla. Esperando...") time.sleep(SLEEP_INTERVAL) continue print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] Total de registros en la tabla: {total_records}") total_batches = (total_records + REPORT_LIMIT - 1) // REPORT_LIMIT # Ceiling division print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] Se procesarán {total_batches} lotes de {REPORT_LIMIT} registros") # Procesar todos los lotes cycle_start_time = time.time() total_sent = 0 for batch_num in range(total_batches): offset = batch_num * REPORT_LIMIT print(f"\n[{time.strftime('%Y-%m-%d %H:%M:%S')}] ----- LOTE {batch_num + 1}/{total_batches} -----") reports = fetch_reports_batch(offset, REPORT_LIMIT) if reports: send_reports_optimized(reports) total_sent += len(reports) print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] Lote {batch_num + 1} enviado. Total enviado hasta ahora: {total_sent}") # Pequeña pausa entre lotes para no saturar if batch_num < total_batches - 1: # No pausar después del último lote time.sleep(SLEEP_BETWEEN_BATCHES) else: print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] No hay reportes en el lote {batch_num + 1}") cycle_end_time = time.time() cycle_duration = cycle_end_time - cycle_start_time print(f"\n[{time.strftime('%Y-%m-%d %H:%M:%S')}] ===== CICLO COMPLETO TERMINADO =====") print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] Total enviado: {total_sent} registros en {cycle_duration:.2f} segundos") print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] Esperando {SLEEP_INTERVAL} segundos antes del próximo ciclo...\n") time.sleep(SLEEP_INTERVAL) except KeyboardInterrupt: print("\nProceso interrumpido por el usuario. Saliendo...") sys.exit(0) if __name__ == "__main__": main_loop()