import sys sys.path.insert(0, './') from datetime import datetime import json # Rerportería - Excel from openpyxl import Workbook from openpyxl.utils import get_column_letter from openpyxl.styles import Border, Side, PatternFill, Font, Alignment # from datetime import datetime, timedelta # from datetime import time as toTime # -- Imagenes Excel -- # ExcelIMG_import = '../img/Logo_Reportes.png' color = "d45520" # "e84e0e" # "fcd5b4" white_color = "fbffff" # Crear documento excel def generate_and_save_excel_document( all_days_data, excel_save_path ): # Creamos una nueva instancia de Workbook, que será nuestro documento Excel. wb = Workbook() # Eliminamos la hoja por defecto que crea openpyxl. wb.remove( wb.active ) # Establecemos un estilo de fuente para los encabezados de las columnas y alineación para los títulos. bold_font = Font(bold=True , color = white_color) center_aligned_text = Alignment(horizontal="center", vertical="center") # Configuramos un borde simple. thin_border = Border(left=Side(style='thin'), right=Side(style='thin'), top=Side(style='thin'), bottom=Side(style='thin')) # Color de relleno de celda fill = PatternFill(start_color=color, end_color=color, fill_type="solid") # Color naranjo # Definición de headers headers = ["Hora", "Batería"] profundidades = ["Profundidad_1", "Profundidad_2", "Profundidad_3"] data_titles = ["O2", "DO2", "Temp", "Sal"] initial_row = 2 # Define fila inicial para la tabla - Donde se posiciona nombre de estacion produnfidades_row = initial_row + 1 # Define fila inicial de titulos profundidaades headers_row = initial_row + 2 # Define fila inicial de headers # Por cada día en nuestro JSON, creamos una hoja en el documento Excel. for station_data in all_days_data: station_name = station_data["station_idx"] # Iterar sobre las llaves en el objeto de la estación for day in station_data: # Saltar las llaves '_id' y 'station_idx' if day in ['_id', 'station_idx']: continue ws = wb.create_sheet(title=f"{day}-{station_name}") # La columna inicial para la primera estación. current_column = 1 # Creamos una lista para almacenar las celdas que necesitaremos combinar más tarde. cells_to_merge = [] day_data = station_data[day] # Aseguramos que hay datos antes de proceder. if not day_data: continue # Agregamos el nombre de la estación como título de la tabla. station_title_cell = ws.cell(row=initial_row, column=current_column, value=f'Estación {station_name}') station_title_cell.font = bold_font station_title_cell.alignment = center_aligned_text # Añadimos la celda a la lista de celdas a combinar. # cells_to_merge.append((1, current_column, 1, current_column+7)) # Agregamos los encabezados de las columnas para la estación actual. for col, header in enumerate(headers, start=current_column): cell = ws.cell(row=headers_row-1, column=col, value=header) cell.font = bold_font cells_to_merge.append((headers_row-1, col, headers_row, col)) # Colocamos la columna inicial para los datos de profundidad. depth_column_start = current_column + len(headers) # Añadimos los encabezados de los parámetros. for depth_idx, depth in enumerate(profundidades, start=1): # Asignamos el valor y estilo a la celda superior izquierda. depth_cell = ws.cell(row=produnfidades_row, column=depth_column_start, value=f"{depth}") depth_cell.font = bold_font depth_cell.alignment = center_aligned_text # Añadimos las celdas de profundidad a la lista de celdas a combinar ANTES de combinar. cells_to_merge.append((produnfidades_row, depth_column_start, produnfidades_row, depth_column_start + len(data_titles) - 1)) # Añadimos los encabezados de los parámetros en la segunda fila. for i, param in enumerate( data_titles ): param_cell = ws.cell(row=produnfidades_row + 1, column=depth_column_start + i, value=param) param_cell.font = bold_font param_cell.border = thin_border param_cell.fill = fill depth_column_start += len(data_titles) # ? Combinar las celdas DESPUÉS de asignar los valores. # Combinar celdas y aplicar estilos for start_row, start_column, end_row, end_column in cells_to_merge: # Aplicar estilos a la celda superior izquierda top_left_cell = ws.cell(row=start_row, column=start_column) top_left_cell.fill = fill top_left_cell.border = thin_border top_left_cell.alignment = center_aligned_text top_left_cell.font = bold_font # Combinar las celdas ws.merge_cells(start_row=start_row, start_column=start_column, end_row=end_row, end_column=end_column) # Rellenamos las filas con los datos de cada estación. row_counter = headers_row + 1 for entry in day_data: # Extraemos la hora de la fecha y la batería. ws.cell(row=row_counter, column=current_column, value=entry["fecha"].split(" ")[1]) ws.cell(row=row_counter, column=current_column+1, value=entry["Battery"]) # Columna para los valores de profundidad. depth_column = current_column + 2 for depth in profundidades: # Extraemos los datos de cada profundidad. depth_data = entry.get(depth, {}) for param in data_titles: param_data = depth_data.get(param) # Solo escribimos el dato si no es nulo. if param_data and param_data.get("data") is not None: ws.cell(row=row_counter, column=depth_column, value=round( param_data["data"] , 2 ) ) depth_column += 1 # Pasamos a la siguiente fila. row_counter += 1 # Ajustamos el ancho de las columnas. for col in range(1, depth_column_start): ws.column_dimensions[get_column_letter(col)].width = 14 # Movemos la columna inicial para la siguiente estación. current_column = depth_column_start + 3 # 2 celdas de separación # Realizamos todos los merges necesarios después de procesar todas las estaciones. # Guardamos el documento Excel. wb.save(excel_save_path) return True # Ajuste utilizado por servidor def _adjust_server_data_format( server_DB_documents ): formatted_data = [] for dia in server_DB_documents: for estacion in dia[ "estaciones" ]: estacion_dict = { "_id": dia[ "_id" ], "station_idx": estacion[ "Nombre" ] } for dato in estacion[ "datos" ]: fecha = dato[ "fecha" ].split()[0] if fecha not in estacion_dict: estacion_dict[ fecha ] = [] estacion_dict[ fecha ].append(dato) formatted_data.append( estacion_dict ) return formatted_data def generate_report_from_server( date_range, server_db_collection, centro_id, excel_save_path ): # date_range: array de string fechas, formato YY-mm-dd, ej: 2024-01-31 # Convertir el rango de fechas en formato adecuado para la consulta start_date = datetime.strptime(date_range[0], '%Y-%m-%d') end_date = datetime.strptime(date_range[-1], '%Y-%m-%d') # Consulta para obtener todos los documentos en el rango de fechas query = { 'centroId': centro_id, 'fecha': { '$gte': start_date.strftime("%Y-%m-%d"), '$lte': end_date.strftime("%Y-%m-%d") } } server_DB_documents = list(server_db_collection.find(query)) # Verificar si se encontraron documentos if not server_DB_documents: print('No hay datos en la fecha solicitada') return False formatted_data = _adjust_server_data_format( server_DB_documents ) # excel_save_path = f'{centro_id}_{date_range[0]}_{date_range[-1]}.xlsx' return generate_and_save_excel_document( formatted_data, excel_save_path ) # bool # Ajuste utilizado por cliente def _adjust_local_data_format( formatted_local_DB_documents, _format_one_data ): formatted_data = [] for dia in formatted_local_DB_documents: try: for estacion in formatted_local_DB_documents[dia][ "estaciones" ]: estacion_dict = { "_id": formatted_local_DB_documents[dia][ "_id" ], "station_idx": estacion[ "Nombre" ] } for full_dato in estacion[ "datos" ]: print( f'full_dato: \n{full_dato}') formatted_dato = _format_one_data( dia, full_dato, estacion[ "Nombre" ] ) print(f'dato keys: \n{formatted_dato.keys()}') print(f'dato fecha: \n{formatted_dato[ "fecha" ]}') fecha = formatted_dato[ "fecha" ].split()[0] if fecha not in estacion_dict: estacion_dict[ fecha ] = [] estacion_dict[ fecha ].append(formatted_dato) formatted_data.append( estacion_dict ) except Exception as e: print(f'Error en _adjust_local_data_format, línea {e.__traceback__.tb_lineno}') return formatted_data # if __name__ == '__main__': # """ Ejemplo de uso # from generate_server_report import generate_report_from_server # """ # from pymongo import MongoClient, UpdateOne # # Acceso a BD # client = MongoClient('mongodb://localhost/', 27017) # server_db = client.SERVIDOR_AMBIENTAL # # Datos del cento a solicitar # CLIENT = 'Ambiental' # REGION, AREA, CENTRO = 'Loslagos', 'Puertomontt', 'Itg' # centro_id = f"{CENTRO}-{AREA}-{REGION}" # server_db_collection = server_db[f"{CLIENT}_sensor_data"] # date_range = [ "2024-01-19" ] # # Generar reporte # generate_report_from_server( date_range, server_db_collection, centro_id )