import copy import os import sys import time sys.path.insert(0, './') from datetime import datetime, timedelta from io import BytesIO from pathlib import Path from pymongo import MongoClient # import numpy as np # Rerportería - Excel import openpyxl from openpyxl.utils import get_column_letter from openpyxl.styles import Border, Side, PatternFill, Font, Alignment # GradientFill, from openpyxl.utils import get_column_letter, column_index_from_string # from openpyxl.chart import LineChart, Reference from openpyxl.drawing.image import Image as ExcelImg # from openpyxl.drawing.text import CharacterProperties from openpyxl.drawing.image import Image as ExcelImg """ # Matplotlib - Gráficos import matplotlib matplotlib.use('Agg') # Backend sin interfaz gráfica import matplotlib.pyplot as plt import matplotlib.dates as mdates """ # from ctypes import windll, create_unicode_buffer from .reports_utils import ( logger, obtener_datos_por_fecha_por_alias, get_escritorio_path, add_markwater_images, get_styles, asegurar_indices_sensores_timeseries, get_historial_control_as_dict, insert_matplot_graphs, apply_cell_styles, ) # debug_iny DB_HOST = 'localhost' # -> Se sobreescribe al utilizarse como __main__ # Obtener directorio de ejecución if getattr(sys, 'frozen', False): # Modo congelado (PyInstaller): utilizar sys._MEIPASS como base, # ya que allí se encuentran los archivos de recursos. current_dir = Path(getattr(sys, '_MEIPASS')) else: # Ejecución normal: la ruta base es la del script principal current_dir = Path(__file__).resolve().parent """ home = Path.home() desktop = home / "Desktop" if not os.path.exists(desktop): logger.info(f"Escritorio inválido: {desktop}") desktop = home / "Escritorio" if os.path.exists(desktop): logger.info(f"Escritorio válido: {desktop}") else: logger.warning(f"Escritorio inválido: {desktop}. No fue posible identificar esritorio, se utilizará {current_dir}") desktop = current_dir else: logger.info(f"Escritorio identificado: {desktop}") """ def get_desktop_path() -> Path: """ Retorna la ruta al escritorio del usuario independientemente del sistema operativo. Para Windows, prueba 'Desktop' y 'Escritorio' bajo USERPROFILE. Para Linux/mac, prueba 'Desktop' y 'Escritorio' bajo home. Si no encuentra ninguno, retorna el directorio actual. """ home = Path.home() candidates = [] # Windows paths if os.name == 'nt': userprofile = os.environ.get('USERPROFILE') if userprofile: candidates.extend([ Path(userprofile) / 'Desktop', Path(userprofile) / 'Escritorio', Path(userprofile) / 'OneDrive - ITG Chile' / 'Desktop', Path(userprofile) / 'OneDrive - ITG Chile' / 'Escritorio', ]) else: # Unix-like candidates.extend([ home / 'Desktop', home / 'Escritorio', ]) for desk in candidates: if desk.exists(): logger.info(f"Escritorio identificado: {desk}") return desk cwd = Path.cwd() logger.warning(f"No fue posible identificar escritorio. Usando directorio actual: {cwd}") return cwd desktop = get_desktop_path() # base_dir = os.path.dirname(os.path.abspath(__file__)) # Directorio actual del script (database) # img_path = os.path.join(base_dir, "..", "images", "Logo_Reportes.png") # Subir un nivel y entrar a 'images' img_path = current_dir / "images" / "Logo_Reportes.png" save_dir = desktop # current_dir # Todo: CAMBIAR A ESCRITORIO # Generar reporte de datos def _gen_data_report( rango_fechas, control_type, db , excel_save_path, sensores_coll_name="sensores_timeseries", metadata_coll_name = "metadata", centro_name = None , centro_id = None , ts_request = ""): print() collection_names = db.list_collection_names() logger.info(f"{sensores_coll_name=} {(sensores_coll_name in collection_names)=}") logger.info(f"{metadata_coll_name=} {(metadata_coll_name in collection_names)=}") print() """ Funcion que genera un reporte Excel con datos de sensores y estado de inyección de O2, para un rango de fechas, en una BD. Inputs: - rango_fechas: {startDate: str, endDate: str} -> Fechas como str en formato "%d/%m/%Y" - control_type: str -> ( Valores admitidos: {PS, VFD}) - db [opcional] mongo obj db -> Puntero a la db asociada al servicio nanoox - sensores_coll_name [opcional] str -> Nombre de la colección de sensores para este centro en bd. Por defecto: "sensores" - metadata_coll_name [opcional] str -> Nombre de la colección de metadata para este centro en bd. Por defecto: "metadata" - centro_name [opcional] str -> Nombre añadido al reporte: Reporte_{centro_name}_{rango_fechas}.xslm. Por defecto "". Retorna booleano: True indica reporte generado correctamente, False error al generar. """ # ? Timer reporte completo time_total = time.time() try: # ? Timer obtener datos time_db_start = time.time() logger.info(f"generando reporte de tipo {control_type}") logger.debug(rango_fechas) t_indices_start = time.time() try: # Asegurar indices en coleccion sensores asegurar_indices_sensores_timeseries(db, sensores_coll_name) except Exception as e: logger.error(f"Error revisando indices de la colección sensores: {e}") t_indices = (time.time() - t_indices_start)*1e3 # Supongamos que 'msg' es un diccionario que contiene las claves 'FechaInicio' y 'FechaFin' # Convertir las fechas de cadena a objetos datetime start = datetime.strptime(rango_fechas['startDate'], "%d/%m/%Y") end = datetime.strptime(rango_fechas['endDate'], "%d/%m/%Y") #-- Calculo de cantidad de dias a analizar --# if len( str(end-start).split(" ") ) > 1: diasEntre = int( str(end-start).split(" ")[0] ) + 1 # para que tome en cuenta ultimo dia logger.info(f"Se han seleccionado {diasEntre} dias") else: diasEntre = 1 #* -- Crear libro nuevo en Excel-- # workbook = openpyxl.Workbook() ws_names = [] fechas = [] #-- Analisis por dia entre fechas del rango --# # logger.debug(f"{diasEntre=}") logger.info("Comenzando a generar reporte...") t_dias = [] days_with_data = [] # Definir Estilos # font, fill, fill_iny, fill_iny_man, fill_vdf, border = get_styles() font, fill, fill_iny, fill_iny_man, fill_vdf, border, fill_alarm = get_styles() time_db = (time.time() - time_db_start)*1e3 for i in range(diasEntre): # ? Timer inicio del día t_dia_init = time.time() dia = str(start + timedelta(days=i)).split(" ")[0].split("-") fecha_format = "-".join(dia) fechas.append(fecha_format) logger.debug(f"{fecha_format=}") datos_sensores_bombas = obtener_datos_por_fecha_por_alias(db, fecha_format, sensors_coll_name= sensores_coll_name, metadata_coll_name= metadata_coll_name) # ? Timer preparar datos t_prepare_data = (time.time() - t_dia_init)*1e3 process_data_text = f'\n\t\t Obtener y Preparar datos: {t_prepare_data:.2f}ms' if datos_sensores_bombas: logger.info(f"Existen datos en la fecha {fecha_format}") # ? Timer añadir datos t_put_data_start = time.time() days_with_data.append( fecha_format ) # Nombres de cámaras bomb_names = list(datos_sensores_bombas.keys()) bomb_names.sort() logger.debug(f"{bomb_names=}") rangos_iny_on = {alias_bomba: [] for alias_bomba in bomb_names } rangos_bomba_on = {alias_bomba: [] for alias_bomba in bomb_names } # Timestamp para grafico - auto rangos_iny_on_ts = {alias_bomba: [] for alias_bomba in bomb_names } rangos_bomba_on_ts = {alias_bomba: [] for alias_bomba in bomb_names } # Timestamp para grafico - manual rangos_iny_on_manual_ts = {alias_bomba: [] for alias_bomba in bomb_names } # Fecha para pintar celda - manual rangos_iny_on_manual = {alias_bomba: [] for alias_bomba in bomb_names } #cantidad de elementos por columna max_column = len(datos_sensores_bombas[bomb_names[0]]) #Creacion de pagina especifica para la fecha analizada worksheet = workbook.create_sheet(fecha_format) ws_names.append(fecha_format) #* ----- especificaciones para paginas en el excel ---- * cant_keys_tabla = 6 #keys interes (Hora , O2, SO2, TEMP, Presion) "_id": 0, v_offset = 16 #espacio vertical desde el inicio de la pagina h_offset = 2 #espacio horizontal desde el inicio de la pagina tablas_dist = cant_keys_tabla + 1 # 6 #espacio entre tablas tablas_space = [""]*(tablas_dist - 5) tablas_space = [""]*(tablas_dist - 6) # print(f"{KEYS_INTERES=}") space_titles = cant_keys_tabla - h_offset - 1 #* ------------------------------------------------------- #* -------------------------------------------- Agregar filas --------------------------------------- # for i in range(v_offset): worksheet.append([]) # Dejar celdas vacias al inicio #* -- Primera Fila : Nombres de las estanques -- # fila = [] for alias_bomba in bomb_names: #en formato fila para facilitar visualizacion fila+= [""]*h_offset ; fila.append(f"Bomba {alias_bomba}") ; fila+= [""]*space_titles fila += tablas_space worksheet.append(fila) #* -- Segunda Fila : información de las jaulas -- # estructura = ([""]*h_offset + ["Hora" ,"Oxígeno mg/L", "Saturación %" , 'Temperatura °C', "Presión bar", "Alarmas "] ) estructura += ([""]*(h_offset-1) + ["Hora" ,"Oxígeno mg/L", "Saturación %" , 'Temperatura °C', "Presión bar", "Alarmas"] )*(len(bomb_names)-1) # logger.debug(f"{estructura=}") worksheet.append( estructura ) #* ------- Filas siguientes: ------- # #* Agrega iterativamente los datos directamente desde la base de datos Rangos = {} # Rango de celdas de cada canal for idx,channel in enumerate(bomb_names): cell_inicial_Letra = get_column_letter( idx*tablas_dist + h_offset +1 ) # El ultimo +1 es por desplazamiento lateral de las tablas cell_final_Letra = get_column_letter( idx*tablas_dist + h_offset + cant_keys_tabla ) cell_graph = get_column_letter( idx*tablas_dist + h_offset +1) Rangos[channel] = { 'L_Inicial':cell_inicial_Letra, 'L_Final': cell_final_Letra, 'L_Graph': cell_graph, 'celdaInicial': str(cell_inicial_Letra + str(2 + v_offset)), 'celdaFinal': str(cell_final_Letra + str( len(datos_sensores_bombas[channel]) + 2 + v_offset ) ), } values_for_charts = {} null_data = 0 sandwich_case = 0 total_data = 0 for i in range(max_column): nueva_fila = [""]*(h_offset -1) # Para cada dato -> recorrer todos los canales for idx,channel in enumerate(bomb_names): # if channel != "101": continue dato_sensores_bomba = datos_sensores_bombas[channel] if len(dato_sensores_bomba) > i: total_data += 1 data_fila = dato_sensores_bomba[i] # print(data_fila) ts = data_fila["timestamp"].strftime("%H:%M:%S") ts_obj = data_fila["timestamp"] # Verificar datos dato_o2, dato_so2 = data_fila["O2"], data_fila["SO2"] if not dato_o2 and not dato_so2: null_data += 1 continue # Leer modo del tablero (manuaal/auto), estado de electrovalvula y bombas InStart = data_fila.pop("InStart") inyection_o2 = bool( data_fila.pop("RLY") ) in_sel = data_fila.pop("InSel") if in_sel is not None: tcb_auto = bool( int( in_sel ) ) else: # Caso de todo none, se salta dato if InStart is None: logger.debug(f"Dato con null detectado: {InStart=}, InSel={in_sel}, {data_fila}") # continue null_data += 1 tcb_auto = False # Definir que array utilizar, dependiendo de si es manual o automático if tcb_auto: data_array = rangos_bomba_on[channel] graph_array = rangos_bomba_on_ts[channel] else: data_array = rangos_iny_on_manual[channel] graph_array = rangos_iny_on_manual_ts[channel] # Obtener timestamps a graficar de 1 y 2 datos anteriores prev_graph_ts = None # prev_prev_graph_ts = None if tcb_auto: if len( rangos_iny_on_ts[channel] ) > 0 and len(graph_array) > 0: prev_graph_ts = graph_array[-1] if len( rangos_iny_on_ts[channel] ) > 1 and len(graph_array) > 1: prev_prev_graph_ts = graph_array[-2] else: if len(graph_array) > 0: prev_graph_ts = graph_array[-1] # if len(graph_array) > 1: # prev_prev_graph_ts = graph_array[-2] # Obtener timestamps de 1 y 2 datos anteriores last_ts = 0 # Evitar caso de borde: dato inicial ambos en None last_last_ts = 0 # Evitar caso de borde: dato inicial ambos en None if i >= 1: last_ts = dato_sensores_bomba[i-1]["timestamp"] if i >= 2: last_last_ts = dato_sensores_bomba[i-2]["timestamp"] # * Lógica para "arreglar" la chambonada de Nelson de leer mal la inyección, con dos enfoques: # Utilizando a la celda anterior y la corriente # Analizando caso "sandwich" corriente = data_fila.pop("Io") voltaje = data_fila.pop("Vo") alarmas = data_fila.get("alarms", []) Status = data_fila.get("Status") # Revisar si en el instante anterior hubo inyección prev_is_fill = prev_graph_ts == last_ts # correct_this_cell = prev_is_fill and ( corriente > 0 and voltaje > 0 ) correct_this_cell = Status == 9 #prev_is_fill and ( corriente > 0 and voltaje > 0 ) # Revisar si hace 2 datos hubo inyección (dato del gráfico anterior debería corresponder al de hace 2 datos de timestamp) prev_prev_is_fill = prev_graph_ts == last_last_ts # prev_prev_graph_ts == last_last_ts # Definir inyección del instante actual inyecting_vdf = correct_this_cell or ( control_type == "VDF" and InStart == "01" ) inyecting_ps = correct_this_cell or ( control_type == "PS" and ( InStart is not None and (bool( int( InStart ))) ) ) pump_running = (inyecting_vdf or inyecting_ps) and not alarmas # * Revisar caso "sandwich", inyeccion activa ahora y hace 2 datos, pero no el justo anterior if prev_prev_is_fill and not prev_is_fill and pump_running: # Corregir el dato anterior, añadiendolo a los arrays de datos graph_array.append( last_ts ) data_array.append( last_ts.strftime("%H:%M:%S") ) # print(f"Caso sandwich {channel}: {last_ts.strftime("%H:%M:%S")}") sandwich_case += 1 """ # Hardcoded debug if ts == '00:06:28': print(f"{channel}: {ts=} {last_ts=} {last_last_ts=} ; {prev_graph_ts=}") """ # * Añadir inyección del dato actual if pump_running: # En auto data_array, graph_array es bomba. En Manual son inyeccion directamente (rangos_iny_on, rangos_iny_on_ts) data_array.append(ts) graph_array.append( ts_obj ) # Modo TCB automático -> Corroborar inyección, independiente de bomba encendida if tcb_auto: if inyection_o2 and pump_running: # Creo que debería ser and, chambonazo rangos_iny_on[channel].append(ts) # Dibujo en gráfico rangos_iny_on_ts[channel].append( ts_obj ) nueva_fila.append( '' ) if data_fila["Presion"]: try: rounded_press = round(data_fila["Presion"],2) nueva_fila += [ts, data_fila["O2"], data_fila["SO2"], data_fila["TEMP"], rounded_press if rounded_press > -0.5 else None ] except: # continue nueva_fila += [ts, data_fila["O2"], data_fila["SO2"], data_fila["TEMP"], None ] else: nueva_fila += [ts, data_fila["O2"], data_fila["SO2"], data_fila["TEMP"], None ] nueva_fila+= [",".join(alarmas)] if alarmas else "-" nueva_fila += tablas_space[0:-h_offset] if channel not in values_for_charts: values_for_charts[channel] = {"ts": [], "O2": [], "SO2": []} else: if isinstance(data_fila["O2"] , float): values_for_charts[channel]["ts"].append(ts_obj) values_for_charts[channel]["O2"].append(data_fila["O2"]) values_for_charts[channel]["SO2"].append(data_fila["SO2"]) logger.debug(nueva_fila) # # Agregar a la planilla la fila con todas las columnas worksheet.append( nueva_fila ) if null_data or sandwich_case: try: day_info = f"{fecha_format}: \n\tDatos totales: {total_data}" if null_data: day_info += f"\n\tDatos con null detectados: {null_data} ({(null_data/total_data)*100:.2f}%)." if sandwich_case: day_info += f"\n\tCaso sandwich corregidos: {sandwich_case} ({(sandwich_case/total_data)*100:.2f}%)." logger.info(day_info) except: pass """ if null_data > 0: logger.warning(f"{fecha_format} -> Datos con null detectados: {null_data}.") if sandwich_case > 0: logger.info(f"{fecha_format} -> Caso sandwich corregidos: {sandwich_case}") """ logger.debug(f"{values_for_charts=}") t_put_data = (time.time() - t_put_data_start)*1e3 process_data_text += f"\n\t\t Agregar datos: {t_put_data:.2f}ms" # #* -------------------------------------------- Graficos --------------------------------------- # t_graph_start = time.time() insert_matplot_graphs( worksheet, values_for_charts, rangos_iny_on_ts, rangos_bomba_on_ts, rangos_iny_on_manual_ts, Rangos) t_graph = ( time.time() - t_graph_start )*1e3 process_data_text += f"\n\t\t Graficar: {t_graph:.2f}ms" # * ------------------------------------------- Estilos --------------------------------------- # # Marca de Agua t_styles_start = time.time() # img_path = os.path.join(os.path.dirname(__file__), 'Logo_Reportes.png') # Ruta relativa al script add_markwater_images(worksheet, img_path, start_column="C", start_row=21, repeats=len(datos_sensores_bombas.keys()), step=7) # Aplicar Estilos apply_cell_styles( worksheet, bomb_names, Rangos, cant_keys_tabla, tablas_dist, h_offset, v_offset, max_column, rangos_bomba_on, rangos_iny_on, rangos_iny_on_manual, border, font, fill, fill_iny, fill_vdf, fill_iny_man, fill_alarm ) t_styles = ( time.time() - t_styles_start )*1e3 process_data_text += f"\n\t\t Estilizar: {t_styles:.2f}ms" # * Proteger documento password = 'itg123chile' # Proteger Hoja #Al finalizar la escritura del worksheet worksheet.protection.sheet = True worksheet.protection.objects=True worksheet.protection.scenarios=True worksheet.protection.password= password # * Proteger documento #Antes de guardar y cerrar el woorkbook workbook.security.workbookPassword = password workbook.security.lockStructure =True else: logger.warning(f"No hay datos para esta fecha: {fecha_format}") t_dia = (time.time() - t_dia_init)*1e3 t_dias.append( f"{fecha_format} : {t_dia:.2f}ms {process_data_text} \n" ) if not days_with_data: logger.warning(f"No hay días con datos en el rango seleccionado.") return False, None if 'Sheet' in workbook.sheetnames: workbook.remove(workbook['Sheet']) try: t_save = time.time() # Detectar la ruta del escritorio path_escritorio = get_escritorio_path() logger.debug(f"Ruta detectada del escritorio: {path_escritorio}") # Crear la ruta completa para el archivo Excel logger.debug(f"{rango_fechas=}") logger.debug(f"{fecha_format=}") logger.debug(f"{fechas=}, {fechas[0]=}, {fechas[-1]}=") identifier = f"{fechas[0]}_{fechas[-1]}" if fechas[0] != fechas[-1] else f"{fechas[0]}" if centro_name: identifier = f"{centro_name}_{identifier}" """ dest_path = os.path.join(f"Reporte_{identifier}.xlsx" # dest_path = os.path.join(path_escritorio, f"Reporte_{identifier}.xlsx") ) # Verificar que la ruta exista if not os.path.exists(dest_path): raise FileNotFoundError(f"El escritorio no existe o no es accesible: {dest_path}") """ # dest_path = excel_save_path / f"Reporte_{identifier}.xlsx" logger.debug(f"{excel_save_path=}") dest_path = os.path.join( excel_save_path, f"Reporte_{identifier}_{ts_request}.xlsx" ) logger.info(f"Guardado archivo Excel en: {dest_path}...") # Guardar el archivo Excel workbook.save(dest_path) if hasattr(workbook, 'close'): workbook.close() logger.info(f"Archivo Excel guardado correctamente en: {dest_path}") t_save_end = time.time() t_total_end = time.time() time_info = f"Tiempo total de procesamiento: {(t_total_end - time_total)*1e3:.2f} ms\n" time_info += f"\t Asegurar índices de búsqueda en Colección sensores: {t_indices:.2f}ms\n" time_info += f"\t Preparar documento: {time_db:.2f}ms\n\t " time_info += ('\t ').join( t_dias ) time_info += f"\t Guardado de archivo: {(t_save_end - t_save)*1e3:.2f}ms\n" logger.info(time_info) except Exception as e: logger.error(f"Error al guardar el archivo ({e.__traceback__.tb_lineno}) : {e}") return True, dest_path except Exception as e: logger.error(f"Error: {e} {e.__traceback__.tb_lineno}") return False, None # Generar reporte de historial de control def _gen_reporte_historial( rango_fechas, db_host="localhost", db_name="NANOOX_TS", coll_name="historial_control", centro_name = '' ): """ Funcion que genera un reporte Excel con el historial de control para un rango de fechas, en una BD. Inputs: - rango_fechas: {startDate: str, endDate: str} -> Fechas como str en formato "%d/%m/%Y" - db_host [opcional] str -> direccion IP de bd. Por defecto: "localhost" - db_name [opcional] str -> Nombre de la base de datos. Por defecto: "NANOOX_TS" - coll_name [opcional] str -> Nombre de la colección en bd. Por defecto: "historial_control" - centro_name [opcional] str -> Nombre añadido al reporte: Reporte_{centro_name}_{rango_fechas}.xslm. Por defecto "". Retorna booleano: True indica reporte generado correctamente, False error al generar. """ # Supongamos que 'msg' es un diccionario que contiene las claves 'FechaInicio' y 'FechaFin' # Convertir las fechas de cadena a objetos datetime start = datetime.strptime(rango_fechas['startDate'], "%d/%m/%Y") end = datetime.strptime(rango_fechas['endDate'], "%d/%m/%Y") #-- Calculo de cantidad de dias a analizar --# if len( str(end-start).split(" ") ) > 1: diasEntre = int( str(end-start).split(" ")[0] ) + 1 # para que tome en cuenta ultimo dia logger.info(f"Se han seleccionado {diasEntre} dias") else: diasEntre = 1 try: #diccionario con el formato {date: [{"hora": hour, "msg": msg}, {"hora": hour, "msg": msg}, ...] } db = get_db( db_host, db_name ) dicc_logs = get_historial_control_as_dict( db, coll_name ) #* ----- especificaciones para paginas en el excel ---- * v_offset = 2 #espacio vertical desde el inicio de la pagina h_offset = 2 #espacio horizontal desde el inicio de la pagina tablas_dist = 6 #espacio entre tablas tablas_space = [""]*(tablas_dist - 4) cant_keys_tabla = 3 space_titles = cant_keys_tabla - h_offset - 1 #* -- Crear libro nuevo en Excel-- # workbook = openpyxl.Workbook() ws_names = [] # # #* ------------------------------------------------------- # Definir Estilos font, fill, fill_iny, fill_iny_man, fill_vdf, border = get_styles() # font, fill, fill_iny, fill_iny_man, fill_vdf, border, fill_alarm = get_styles() colors = dict() Rangos = {} # Rango de celdas de ca in enumerate(dicc_logs): for key,value in dicc_logs.items(): cell_inicial_Letra = get_column_letter( h_offset +1 ) # El ultimo +1 es por desplazamiento lateral de las tablas cell_final_Letra = get_column_letter( h_offset + cant_keys_tabla ) cell_graph = get_column_letter( h_offset +1) Rangos[key] = { 'L_Inicial':cell_inicial_Letra, 'L_Final': cell_final_Letra, 'L_Graph': cell_graph, 'celdaInicial': str(cell_inicial_Letra + str(1 + v_offset)), 'celdaFinal': str(cell_final_Letra + str( len(value) + 2 + v_offset ) ), } for reg in value: # logger.debug(f"{reg['color']} {key} {reg['msg']}") if key not in colors: colors[key] = [] if reg["color"] != "#FFFFFF": colors[key].append(reg["color"]) else: colors[key].append("#000000") days_with_data = [] # # #-- Analisis por dia entre fechas del rango --# fechas = [] for i in range(diasEntre): # print("Comenzando a generar reporte...") dia = str(start + timedelta(days=i)).split(" ")[0].split("-") fecha_format = "-".join(dia) fechas.append(fecha_format) if fecha_format in dicc_logs.keys() : # #Creacion de pagina especifica para la fecha analizada worksheet = workbook.create_sheet(fecha_format) ws_names.append(fecha_format) days_with_data.append(fecha_format) #* -------------------------------------------- Agregar filas --------------------------------------- # for i in range(v_offset): worksheet.append([]) # Dejar celdas vacias al inicio #* -- Primera Fila : Nombres de las estanques -- # estructura = ([""]*h_offset + ["Historial de control"]) worksheet.append( estructura ) # #* -- Segunda Fila : información de las jaulas -- # fila= [""]*h_offset + ["Hora","Categoría" ,"Descripción"] worksheet.append(fila) # #* -- Filas siguientes: -- # # #* Agrega iterativamente los datos directamente desde la base de datos sheet_data_src = dicc_logs[fecha_format] #lista de diccionarios # print(f"{sheet_data_src=}") # #cantidad de elementos por columna max_column = len(sheet_data_src) for idx, log in enumerate(sheet_data_src): # for i in range(len(sheet_data_src)): nueva_fila = [] nueva_fila.append( '' ) nueva_fila.append( '' ) try: msg = log.get("msg", "") idx = msg.find("]") ts_hh_mm_ss = log.get("date").split(" ")[1] cat = msg[:idx+1] dec = msg[idx+1:] nueva_fila += [ts_hh_mm_ss, cat, dec] except Exception as e: logger.error(f"Error: {e} {e.__traceback__.tb_lineno}") # print(msj, e) # nueva_fila += [hora, msj, msj] worksheet.append( nueva_fila ) #* -------------------------------------------- Estilos --------------------------------------- # # Marca de Agua # Construir la ruta absoluta a la imagen try: base_dir = os.path.dirname(os.path.abspath(__file__)) # Directorio actual del script (database) img_path = os.path.join(base_dir, "..", "images", "Logo_Reportes.png") # Subir un nivel y entrar a 'images' # img_path = os.path.join("Logo_Reportes.png") # Subir un nivel y entrar a 'images' # Verificar si la imagen existe if not os.path.exists(img_path): raise FileNotFoundError(f"No se encuentra la imagen en: {img_path}") # Cargar la imagen img = ExcelImg(img_path) img.anchor = 'C21' # Posición inicial de la imagen en la hoja # Ajustar tamaño de la imagen img.width = 600 # Ancho de la imagen img.height = 450 # Altura de la imagen # Agregar la imagen a la hoja de trabajo worksheet.add_image(img) logger.info("Marca de agua agregada correctamente.") except Exception as e: logger.error(f"Error al agregar la marca de agua: {e}") #-- Diseño celdas -- # Letra1 = get_column_letter( h_offset + 1 ) # El +1 es por desplazamiento lateral de las tablas Letra2 = get_column_letter( h_offset + 2 ) Letra3 = get_column_letter( h_offset + 3 ) cell_merge = worksheet[Letra1 + str(v_offset +1)] cell_merge.font = font cell_merge.fill = fill # Definicion celdas cell_1 = worksheet[Letra1 + str(v_offset + 2)] cell_2 = worksheet[Letra2 + str( v_offset +2 )] cell_3 = worksheet[Letra3 + str( v_offset +2 )] # cell_4 = worksheet[Letra4 + str( v_offset +2 )] # Font: cell_1.font = font cell_2.font = font cell_3.font = font # cell_4.font = font # Fill: cell_1.fill = fill cell_2.fill = fill cell_3.fill = fill # cell_4.fill = fill #-- Combinar celdas de titulos --# worksheet.merge_cells(start_row=v_offset + 1, start_column= h_offset +1, end_row=v_offset +1, end_column= h_offset +3 ) # fila 2 ya que se deja una vacía al inicio # #-- Aplicar el borde a un conjunto de celdas --# cell_range = worksheet[Rangos[fecha_format]['celdaInicial'] : Rangos[fecha_format]['celdaFinal'] ] for row in cell_range: for cell in row: cell.border = border # if row[0].value in colors: # cell.fill = PatternFill(fill_type="solid", start_color=colors[row[0].value][:1] ,end_color= colors[row[0].value][1:] ) # elif row[0].value in rangos_vdf[channel]: # cell.fill = fill_vdf #-- color texto --# for idx, row in enumerate( worksheet.iter_rows(min_row=v_offset +3, max_row= max_column + v_offset +2)): for cell in row: cell.font = font = Font(color=colors[fecha_format][idx][1:] )# color="00FFFFFF" -> blanco #-- alineacion de celdas -- for idx, row in enumerate( worksheet.iter_rows(min_row=v_offset +1, max_row= max_column + v_offset +2)): for cell in row: cell.alignment = Alignment(horizontal='center', vertical='center', shrink_to_fit=True, wrap_text=False) # cell.font = font = Font(color=colors[fecha_format][idx][1:] )# color="00FFFFFF" -> blanco # print(len( worksheet.iter_rows(min_row=v_offset, max_row= max_column + v_offset + 4))) # Ajustar el ancho de las columnas solo para aquellas que contienen datos for column in worksheet.columns: max_length = 0 for cell in column: try: if cell.value is not None and len(str(cell.value)) > max_length: max_length = len(cell.value) except: pass adjusted_width = (max_length + 2) if adjusted_width > 2: # Ajustar solo si hay datos en la columna worksheet.column_dimensions[column[0].column_letter].width = adjusted_width if not days_with_data: logger.warning(f"No hay días con datos en el rango seleccionado.") return False if 'Sheet' in workbook.sheetnames: workbook.remove(workbook['Sheet']) try: identifier = f"{fechas[0]}_{fechas[-1]}" if fechas[0] != fechas[-1] else f"{fechas[0]}" if centro_name: identifier = f"{centro_name}_{identifier}" dest_path = save_dir / f"Reporte_historial_control_{identifier}.xlsx" # Guardar el archivo Excel workbook.save(dest_path) if hasattr(workbook, 'close'): workbook.close() logger.info(f"Archivo Excel guardado correctamente en: {dest_path}") return True except Exception as e: logger.error(f"Error al guardar el archivo: {e}") except Exception as e: logger.error(f"Error generando reporte de historial de control: {e} {e.__traceback__.tb_lineno}") def get_db( db_host, db_name ): mongo_uri = f"mongodb://{db_host}:27017" logger.info(f"Buscando datos en colección {db_name} de {mongo_uri}.") client = MongoClient(mongo_uri) db = client[db_name] return db if __name__ == "__main__": DB_HOST = '10.0.0.53' """ current_dir = os.path.dirname(os.path.abspath(__file__)) print(f"{current_dir=}") img_source_dir = os.path.abspath(os.path.join(current_dir, '../images')) # print(f"{root_dir=}") ExcelIMG_import = os.path.join('../images', 'Logo_Reportes.png') sys.path.append(img_source_dir) """ CONTROL = "PS" rango_fechas = {"startDate": "24/04/2025", "endDate": "24/04/2025" } print(rango_fechas) _gen_data_report(rango_fechas=rango_fechas, control_type=CONTROL, db_host=DB_HOST, centro_id= None ) # _gen_reporte_historial( rango_fechas ) # print(rango_fechas) # _gen_data_report(rango_fechas=rango_fechas,centro_id= "centro")