#codigo hecho por; EL BANDU from pprint import pprint import requests from copy import deepcopy from bson import ObjectId import json from threading import Thread import time from datetime import datetime from ServerSender.utils_scapp import sanitize_data import logging def get_logger(name: str = "mi_logger", level=logging.DEBUG): logger = logging.getLogger(name) logger.setLevel(level) if not logger.handlers: # Evita agregar múltiples handlers console_handler = logging.StreamHandler() formatter = logging.Formatter("logger.%(levelname)s: %(message)s") console_handler.setFormatter(formatter) logger.addHandler(console_handler) return logger # -- COMUNICACIÓN CON SERVIDOR HTTP REMOTO -- # # -- CONFIGURACIÓN -- # REQUEST_TIMEOUT = 10 CHUNK_SIZE = 50 # Número de documentos a enviar por lote #logger = logging.basicConfig(level=logging.info) class RemoteServerSender: def __init__(self, EMPRESA, REGION, AREA, CENTRO, centro, scapp_analytic_db ): self.EMPRESA = EMPRESA self.REGION = REGION self.AREA = AREA self.CENTRO = CENTRO # Objeto centro self.centro = centro self.coleccion_no_enviados = scapp_analytic_db["pendientes"] self.connected = False self.server_url = None self.data_endpoint = f"/scapp/mirror_analytics/{REGION}/{AREA}/{CENTRO}" self.headers = { "Content-Type": "application/json", "Accept": "application/json" } # Datos para enviar al servidor self.data_to_post = None # Flag para evitar llamar nuevamente a find_server_url self.finding_server = False # Flag para saber si ya existe una rutina de posteo de datos activa self.posting_data = False self.find_server_url_thread() self.logger = get_logger("mi_logger") self.logger.info(f"----- Clase RemoteServerSender inicializada -----") def __call__( self, data_to_post ): """Maneja el envío de datos al servidor respetando la cronología de los pendientes.""" self.logger.debug(f"Intentando enviar datos al servidor...") # Sanear los datos antes de cualquier operación self.data_to_post = sanitize_data(data_to_post) # Llamar al hilo de posteo de datos self.post_data_thread() def find_server_url_thread( self ): """ Inicia un hilo para buscar conexión con el servidor """ # Si ya se está intentando conectar al servidor se retorna anticipadamente if self.finding_server: return self.finding_server = True # Inicia hilo de buscar conexión a servidor self.url_thread = Thread(target=self.find_server_url, daemon = True).start() return def find_server_url( self ): """ Bucle que busca respuesta del servidor cada 5s. Al encontrar conexión setea los atributos self.connected y self.server_url. """ while (not self.connected or not self.server_url): try: server_url = self.get_active_server_url() # Si la url no es correcta, duerme antes de volver a intentarlo if not bool( server_url ): # Duerme 5s antes de volver a reintentar conectar al servidor time.sleep(5) self.connected = False self.server_url = None self.logger.error(f"Error: no hay conexión con el servidor.") continue self.logger.info(f"URL conectado exitosamente en {server_url}") self.server_url = server_url # f"{server_url}/nanoox/new_data/{self.REGION}/{self.AREA}/{self.CENTRO}" self.connected = True self.finding_server = False break except Exception as e: self.logger.warning(f"Error en find_server_url ({e.__traceback__.tb_lineno}): {e}") time.sleep(5) # Establecer conexión def get_active_server_url( self, ): """ Corrobora que exista conexión al servidor. @return - server_url: str / None """ if self.EMPRESA.lower() == "local": server_url = "http://172.16.1.118:7777" # "http://127.0.0.1:4024" else: server_url = f"http://{self.EMPRESA.lower()}.itgchile.com" if self.check_server_status( server_url ): self.logger.info(f"Conectado a {server_url} exitosamente\n") return server_url else: print("Error en conexión. No se pudo conectar a ninguno de los servidores.\n") self.logger.info("Error en conexión. No se pudo conectar a ninguno de los servidores.\n") return def check_server_status( self, url, timeout=3 ): """ Verifica si el servidor está activo solicitando su versión. @Args: - url (str): URL del servidor a verificar. - timeout (int): Tiempo en segundos para esperar una respuesta. @Returns: - bool: True si el servidor está activo, False en caso contrario. """ _server_url = f"{url}/check_server_status" try: response = requests.head(_server_url, timeout=timeout) response.raise_for_status() # Esto arrojará un error si la respuesta tiene un código de error HTTP. if response.status_code == 200: return True except requests.RequestException as e: self.logger.error(f"Error al consultar la versión del servidor en {url}: {e}") return False def post_data_thread( self ): """ Inicia hilo para postear datos al servidor o almacenarlos en la colección de pendientes """ self.post_thread = Thread(target=self.post_data_routine, daemon = True).start() def post_data_routine( self ): """ Rutina de posteo de pendientes y datos nuevos al servidor. Si no hay conexión retorna anticipadamente. self.data_to_post : Datos ya saneados """ if not bool( self.data_to_post ): self.logger.warning(f"Warning post_data_routine: no hay datos que postear {self.all_station_data=}, retornando anticipadamente.") return data_to_post = self.data_to_post # Corroborar conexión a servidor if (not self.connected) or (self.server_url is None) or self.posting_data: # self.coleccion_no_enviados.insert_one(data_to_post) self.coleccion_no_enviados.insert_many(data_to_post) if (not self.connected) or (self.server_url is None): self.logger.warning(f"Intento de envio fallido, no hay conexión al servidor. Pendientes acumulados: {self.pendientes_status()} ") else: self.logger.warning(f"Intento de envio fallido, ya existe un proceso de envio activo. Pendientes restantes: {self.pendientes_status()} ") return # -- Inicio de rutina de postear datos -- # self.posting_data = True # Si hay pendientes, agregamos primero el nuevo dato a la base de datos pendientes = self.pendientes_status() if pendientes: self.logger.info(f"Existen {pendientes} pendientes. Guardando el nuevo dato en la cola antes de enviar.") # self.coleccion_no_enviados.insert_one(data_to_post) self.coleccion_no_enviados.insert_many(data_to_post) else: # Si no hay pendientes, intentar enviar directamente el nuevo dato # envio_exitoso = self.enviar_datos_al_servidor([data_to_post]) # Enviar como lista con un solo elemento envio_exitoso = self.enviar_datos_al_servidor(data_to_post) # Enviar datos no como lista if envio_exitoso: self.posting_data = False self.logger.info(f"Envio de datos sin pendientes exitoso.") # return # Salir si el envío fue exitoso # Si falla al enviar se asume que se perdió la conexión -> Inicia hilo de conexión a servidor else: self.logger.warning("Fallo en el envío. Guardando en pendientes.") # self.coleccion_no_enviados.insert_one(data_to_post) self.coleccion_no_enviados.insert_many(data_to_post) self.connected = False self.server_url = None self.find_server_url_thread() self.posting_data = False # return # -- Rutina de envio de pendientes -- # # Ahora procesamos los pendientes en lotes if self.pendientes_status(): try: self.logger.info("Hay datos pendientes. Intentando reenviar en lotes...") while self.pendientes_status(): # Seguir enviando mientras haya pendientes pendientes = list(self.coleccion_no_enviados.find().limit(CHUNK_SIZE)) # Obtener un chunk if not pendientes: break # Preparar datos sin _id para el envío datos_para_enviar = [] ids_para_eliminar = [] for doc in pendientes: doc_aux = deepcopy(doc) doc_id = doc_aux.pop('_id', None) # Eliminar el ID para envío datos_para_enviar.append(doc_aux) if isinstance(doc_id, ObjectId): ids_para_eliminar.append(doc_id) if self.enviar_datos_al_servidor(datos_para_enviar): # Si el envío fue exitoso, eliminar los documentos de la base de datos self.coleccion_no_enviados.delete_many({'_id': {'$in': ids_para_eliminar}}) self.logger.info(f"Enviados y eliminados {len(ids_para_eliminar)} documentos pendientes.") # Error de conexión con servidor else: self.logger.warning("No se pudo enviar el batch de pendientes. Se mantendrán en la base de datos.") self.connected = False self.server_url = None self.find_server_url_thread() break # Salir del bucle si hay un fallo time.sleep(.100) # Espera para no colapsar servidor except Exception as e: self.logger.warning(f"Error en rutina de envío de pendientes ({e.__traceback__.tb_lineno}): {e}") self.connected = False self.server_url = None self.find_server_url_thread() self.posting_data = False return def pendientes_status( self, ): """Verifica si hay documentos pendientes en la base de datos.""" return self.coleccion_no_enviados.count_documents({}) # Funciones de envio de datos def enviar_datos_al_servidor( self, datos ): """Envía una lista de datos al servidor y devuelve True si fue exitoso.""" serialized_data = self.serialize_data( datos ) if serialized_data is None: return False return self.post_data( serialized_data ) def serialize_data(self, data): """Verifica que los objetos datetime se conviertan a formato ISO para luego serializar todos los datos a enviar y formatear para dict o list""" def default_serialize(obj): if isinstance(obj, datetime): return obj.isoformat() # Convertir datetime a string ISO 8601 raise TypeError(f"Object of type {type(obj)} is not JSON serializable") #self.logger.error(f"Object of type {type(obj)} is not JSON serializable") try: # Convertir objetos datetime antes de serializar serialized_data = json.dumps(data, default=default_serialize) if isinstance(serialized_data, dict): serialized_data = [serialized_data] return serialized_data except (TypeError, ValueError) as e: self.logger.error(f"Error en datos no serializables: {e}") return None def serialize_data_original( self, data ): try: # Verificar si el objeto es serializable antes de enviarlo return json.dumps(data) except (TypeError, ValueError) as e: return def post_data( self, data ): try: data_url = f"{self.server_url}{self.data_endpoint}" response = requests.post( data_url, headers=self.headers, json=data, timeout=REQUEST_TIMEOUT) if response.status_code == 200: self.logger.info(f"Datos enviados correctamente al servidor.") return True else: self.logger.warning(f"Error en respuesta del servidor: {response.text}") return False except requests.Timeout: self.logger.error("Timeout al intentar enviar datos al servidor.") except requests.RequestException as e: self.logger.error(f"Error en la solicitud HTTP: {e}") return False def post_umbrales_centro( self ): """ Actualiza los umbrales locales en el servidor remoto """ umbrales = self.centro.center_configurations.get("umbrales", {}) self.logger.info(f"En post umbrales centro clase sender{umbrales=}") centro_id = self.centro.center_id umbrales_url = f"{self.server_url}{self.umbrales_endpoint}" self.logger.info(f"Se enviarn datos a la ruta {umbrales_url}") headers = { "Content-Type": "application/json", "Accept": "application/json" } data = { "centro": centro_id, "data": umbrales } try: response = requests.post(umbrales_url, headers=headers, json=data, timeout= 1) if response.status_code == 200: self.logger.info(f"Umbrales enviados correctamente a {self.server_url}.") return True else: self.logger.info(f"Error al enviar umrbales al servidor. Respuesta: {response.text}") return False except requests.RequestException as e: self.logger.info(f"Error al enviar datos al servidor: \n{e}")