#!/var/www/ev-aguas/ev-aguas-proceso/venv/bin/python3 """ EV Aguas Proceso - Primera Iteración Procesador de mensajes IoT para medidores de agua Este script procesa mensajes JSON del archivo messages.json, los inserta en la base de datos y categoriza los mensajes en: - msg_procesados.json: Mensajes procesados exitosamente - msg_erroneos.json: Mensajes con errores de datos - messages.json: Solo mensajes con errores del sistema para reintentar """ import logging import sys import os import json from datetime import datetime from pathlib import Path from dotenv import load_dotenv from processors.message_processor import MessageProcessor # Obtener el directorio del script actual SCRIPT_DIR = Path(__file__).parent.absolute() PROJECT_ROOT = SCRIPT_DIR.parent # Cambiar al directorio del script para asegurar que las rutas relativas funcionen os.chdir(SCRIPT_DIR) # Cargar variables de entorno load_dotenv('.env') def setup_logging(): """Configura el sistema de logging""" logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler('proceso.log'), logging.StreamHandler(sys.stdout) ] ) def read_messages_file(file_path): """Lee todas las líneas del archivo de mensajes""" try: with open(file_path, 'r', encoding='utf-8') as file: return [line.strip() for line in file if line.strip()] except FileNotFoundError: logging.error(f"Archivo no encontrado: {file_path}") return [] except Exception as e: logging.error(f"Error leyendo archivo: {e}") return [] def append_to_processed_file(message_line): """Agrega una línea al archivo de mensajes procesados exitosamente""" try: with open('msg_procesados.json', 'a', encoding='utf-8') as file: file.write(message_line + '\n') return True except Exception as e: logging.error(f"Error escribiendo a msg_procesados.json: {e}") return False def append_to_error_file(message_line, error_reason): """Agrega una línea al archivo de mensajes erróneos con la razón del error""" try: # Crear un objeto con el mensaje original y la razón del error error_record = { "timestamp": datetime.now().isoformat(), "error_reason": error_reason, "original_message": json.loads(message_line) if message_line.strip().startswith('{') else message_line } with open('msg_erroneos.json', 'a', encoding='utf-8') as file: file.write(json.dumps(error_record, separators=(',', ':')) + '\n') return True except Exception as e: logging.error(f"Error escribiendo a msg_erroneos.json: {e}") return False def update_messages_file(remaining_messages): """Actualiza el archivo messages.json con las líneas no procesadas (errores del sistema)""" try: messages_file_path = PROJECT_ROOT / 'messages.json' with open(messages_file_path, 'w', encoding='utf-8') as file: for message in remaining_messages: file.write(message + '\n') return True except Exception as e: logging.error(f"Error actualizando messages.json: {e}") return False def main(): """Función principal""" setup_logging() # Obtener número de mensajes a procesar por lote messages_per_batch = int(os.getenv('MESSAGES_PER_BATCH', 10)) print("=" * 50) print("EV AGUAS PROCESO - Primera Iteración") print("=" * 50) print(f"📝 Configuración: Procesando hasta {messages_per_batch} mensajes por lote") print(f"📁 Directorio de trabajo: {SCRIPT_DIR}") print(f"📁 Directorio del proyecto: {PROJECT_ROOT}") # Verificar que existe el archivo de mensajes usando ruta absoluta messages_file = PROJECT_ROOT / 'messages.json' if not messages_file.exists(): print(f"❌ Error: No se encuentra el archivo {messages_file}") logging.error(f"Archivo no encontrado: {messages_file}") return # Leer todas las líneas del archivo print("Leyendo archivo de mensajes...") messages = read_messages_file(messages_file) if not messages: print("❌ No se encontraron mensajes para procesar") return # Determinar cuántos mensajes procesar messages_to_process = min(messages_per_batch, len(messages)) print(f"📄 Se encontraron {len(messages)} mensajes. Procesando {messages_to_process} mensajes...") # Inicializar procesador processor = MessageProcessor() # Contadores processed_count = 0 data_error_count = 0 system_error_count = 0 remaining_messages = [] # Procesar los primeros N mensajes for i in range(messages_to_process): message_line = messages[i] try: print(f"\n🔄 Procesando mensaje {i+1}/{messages_to_process}...") logging.info(f"Procesando mensaje {i+1}: {message_line[:100]}...") # Procesar el mensaje result = processor.process_single_message_detailed(message_line) if result['success']: # Mensaje procesado exitosamente if append_to_processed_file(message_line): processed_count += 1 print(f"✅ Mensaje {i+1} procesado exitosamente") logging.info(f"Mensaje {i+1} procesado y movido a msg_procesados.json") else: # Error guardando en procesados (error del sistema) remaining_messages.append(message_line) system_error_count += 1 print(f"⚠️ Mensaje {i+1} procesado pero no se pudo guardar") elif result['error_type'] == 'data_error': # Error de datos - mover a msg_erroneos.json if append_to_error_file(message_line, result['error_reason']): data_error_count += 1 print(f"📝 Mensaje {i+1} movido a erróneos: {result['error_reason']}") logging.info(f"Mensaje {i+1} movido a msg_erroneos.json: {result['error_reason']}") else: # Error guardando en erróneos (error del sistema) remaining_messages.append(message_line) system_error_count += 1 print(f"⚠️ Mensaje {i+1} con error de datos pero no se pudo guardar") else: # Error del sistema - mantener en archivo original para reintentar remaining_messages.append(message_line) system_error_count += 1 print(f"🔄 Mensaje {i+1} con error del sistema - se reintentará") logging.error(f"Error del sistema procesando mensaje {i+1}: {result['error_reason']}") except Exception as e: # Error inesperado - mantener en archivo original remaining_messages.append(message_line) system_error_count += 1 logging.error(f"Error inesperado procesando mensaje {i+1}: {e}") print(f"❌ Error inesperado en mensaje {i+1}: {e}") # Agregar los mensajes no procesados (desde messages_to_process en adelante) remaining_messages.extend(messages[messages_to_process:]) # Actualizar el archivo original con los mensajes no procesados print(f"\n📊 Actualizando archivo de mensajes...") if update_messages_file(remaining_messages): print(f"✅ Archivo messages.json actualizado") else: print(f"❌ Error actualizando messages.json") # Resumen final print("\n" + "=" * 50) print("RESUMEN DEL PROCESAMIENTO") print("=" * 50) print(f"📄 Total de mensajes en archivo: {len(messages)}") print(f"🔄 Mensajes procesados en esta ejecución: {messages_to_process}") print(f"✅ Procesados exitosamente: {processed_count}") print(f"📝 Errores de datos: {data_error_count}") print(f"🔄 Errores del sistema: {system_error_count}") print(f"📄 Mensajes restantes en archivo: {len(remaining_messages)}") print("=" * 50) if processed_count > 0: print(f"📁 Mensajes procesados: msg_procesados.json") if data_error_count > 0: print(f"📁 Mensajes erróneos: msg_erroneos.json") if len(remaining_messages) > 0: print(f"📁 Para reintentar: messages.json") if len(remaining_messages) > 0: print(f"\nEjecuta el programa nuevamente para procesar los siguientes {min(messages_per_batch, len(remaining_messages))} mensajes.") else: print(f"\n🎉 ¡Todos los mensajes han sido procesados!") print(f"📝 Configuración actual: {messages_per_batch} mensajes por lote (editable en config.env)") if __name__ == "__main__": main()