# Renombra la base de datos de Scapp de pelletdb a ScappDB # Optimiza las colecciones data y data detailed # ! Script Obsoleto: # Ya no existen clientes con estas versiones de BD. VERBOSE = False # -- Refactorización -- # def Scapp_refactorice_db(client): # Determinar si existe BD Antigua db_list = client.list_database_names() if 'pelletdb' in db_list: # ! Antiguas credenciales old_db = client.pelletdb # Database old_collection = old_db.pelletdb # Colección pelletdb, dentro de pelletdb # Mover cámaras a nueva colección y DB copiar_y_eliminar_documentos_de_camaras( old_collection ) # Mover parámetros a nueva colección y DB copiar_y_eliminar_parametros( old_collection ) # Mover datos de detecciones a nueva colección y DB copiar_y_eliminar_detecciones( old_collection ) # Eliminar BD Antigua client.drop_database('pelletdb') # Optimizar data_detailed Scapp_update_data_detailed() # ----- Funciones de apoyo ----- # def copiar_y_eliminar_documento( coleccion_origen, filtro, coleccion_destino): """ # ? Copia un documento de una colección a otra y luego lo elimina de la colección original. :param db: Base de datos de MongoDB. :param coleccion_origen: Nombre de la colección de origen. :param filtro: Filtro para seleccionar el documento a copiar. :param coleccion_destino: Nombre de la colección de destino. """ # Buscar el documento en la colección original documento = coleccion_origen.find_one(filtro) if documento: # Copiar el documento a la colección de destino coleccion_destino.insert_one(documento) # Eliminar el documento de la colección original coleccion_origen.delete_one(filtro) print(f"Documento copiado y eliminado exitosamente: {filtro}") else: if VERBOSE: print(f"No se encontró el documento con el filtro proporcionado: {filtro}") def copiar_y_eliminar_documento_modificado( coleccion_origen, filtro, coleccion_destino, nuevo_filtro ): """ # ? Copia un documento de una colección a otra y luego lo elimina de la colección original. :param db: Base de datos de MongoDB. :param coleccion_origen: Nombre de la colección de origen. :param filtro: Filtro para seleccionar el documento a copiar. :param coleccion_destino: Nombre de la colección de destino. :param nuevo_filtro: Modificaciones a aplicar al documento antes de copiarlo. """ # Buscar el documento en la colección original documento = coleccion_origen.find_one( filtro ) if documento: # Crear un nuevo documento excluyendo las claves presentes en filtro documento_actualizado = {k: documento[k] for k in documento if k not in filtro} # Actualizar el documento con las claves y valores de nuevo_filtro documento_actualizado.update(nuevo_filtro) # Copiar el documento a la colección de destino coleccion_destino.insert_one( documento_actualizado ) # Eliminar el documento de la colección original coleccion_origen.delete_one( filtro ) print(f"Documento copiado y eliminado exitosamente: {filtro}") else: if VERBOSE: print(f"No se encontró el documento con el filtro a modificar proporcionado: {filtro}") def copiar_y_eliminar_documentos_con_clave(coleccion_origen, clave, coleccion_destino): """ # ? Copia todos los documentos de una colección a otra que contienen una clave específica y luego los elimina de la colección original. :param coleccion_origen: Colección de origen. :param clave: Clave para seleccionar los documentos a copiar. :param coleccion_destino: Colección de destino. """ # Buscar documentos en la colección original que contienen la clave documentos = coleccion_origen.find({clave: {"$exists": True}}) # Contador para los documentos procesados contador = 0 for documento in documentos: # Copiar cada documento a la colección de destino coleccion_destino.insert_one(documento) # Eliminar el documento de la colección original coleccion_origen.delete_one({'_id': documento['_id']}) # Incrementar el contador contador += 1 if contador > 0: print(f"{contador} documentos copiados y eliminados exitosamente.") else: if VERBOSE: print(f"{contador} documentos copiados y eliminados exitosamente.") # -- Funciones de copiar y eliminar DBs -- # def copiar_y_eliminar_documentos_de_camaras( old_collection ): # Filtros de búsquedas de cámaras y registros históricos camera_filter = {'Cameras':'AllCameras'} camera_backup_filter = {'RespaldoCameras':'AllCameras'} historicos_filter = {'Historical_Index' : "Cameras_Index"} documentos_a_mover = [ camera_filter, historicos_filter] # Colecciones de origen y destino coleccion_origen = old_collection coleccion_destino = cameras_collection # Copiar y eliminar documentos for filtro in documentos_a_mover: copiar_y_eliminar_documento( coleccion_origen, filtro, coleccion_destino) # Copiar y eliminar documento backup modificado camera_backup_filter_nuevo = {'CamerasBackup':'AllCameras'} copiar_y_eliminar_documento_modificado( coleccion_origen, camera_backup_filter, coleccion_destino, camera_backup_filter_nuevo ) def copiar_y_eliminar_parametros( old_collection ): # Filtros de búsquedas de cámaras y registros históricos always_active_filter = {'Active' : "Always"} reporteria_filter = {'ScappReporter': 'AutomaticReport'} documentos_a_mover = [ always_active_filter, reporteria_filter] # Colecciones de origen y destino coleccion_origen = old_collection coleccion_destino = parameters_collection # Copiar y eliminar documentos for filtro in documentos_a_mover: copiar_y_eliminar_documento( coleccion_origen, filtro, coleccion_destino) def copiar_y_eliminar_detecciones( old_collection ): # LLave de búsqueda clave = 'dia' # Colecciones de origen y destino coleccion_origen = old_collection coleccion_destino = data_collection # Copiar y eliminar documentos copiar_y_eliminar_documentos_con_clave(coleccion_origen, clave, coleccion_destino) # -- Funciones para optimizar coleccion data_detailed -- # #verificacion de existencia del documento y estado de la base de datos def _verify_data_detailed_status(parameters_collection): ret = list( parameters_collection.find( {'data_detailed_status' : True} ) ) if ret: print("Base de datos detallada ya optimizada") return True else: print("Se optimizará la base de datos detallada") return False def _overwrite_colection(db, collection_name, new_collection): try: # Obtén la colección existente collection = db[collection_name] # Elimina todos los documentos existentes en la colección collection.delete_many({}) # Inserta los nuevos documentos en la colección collection.insert_many(new_collection) print(f'Colección "{collection_name}" en la base de datos sobrescrita.') return True except Exception as e: print(f'Ocurrió un error intentando optimizar la base de datos en línea {e.__traceback__.tb_lineno}: \n{e}') return False #manipula los registros de la colleccion data_detailed original def _update_reg_format(reg): array = reg['detecciones_array'] new_format_array_reg_detections = [] new_array_det_reg = [0] + list(filter(lambda x: x != 0,array)) + [0] fact = 1000/len(new_array_det_reg)%30*30 old_ts_format = datetime.strptime(reg["hora"], "%m/%d/%y %H:%M:%S") new_ts_format = old_ts_format.strftime("%Y-%m-%d %H:%M:%S") new_ts = [new_ts_format + "." + str(((i+ 1)*fact))[:3] for i in range(len(new_array_det_reg)) ] for i in range( len(new_array_det_reg) ): elem = new_array_det_reg[i] ts = new_ts[i] new_reg = {"hora": ts, "pellet": reg["pellet"], "detecciones": elem} new_format_array_reg_detections.append(new_reg) return new_format_array_reg_detections # Efectúa la optimización de la coleccion data_detailed y retorna una colección con el nuevo formato def _optimize_data_detailed(db): data_detailed = db["data_detailed"].find() optimized_bd = [] # Iterar sobre documentos de la bd for document in data_detailed: new_document = {} # Iteración sobre las cámaras en el documento for key in document: if key not in ["_id", "dia"]: array_reg_detections = document[key] new_format_array_reg_detections = [] # Iteración sobre el arreglo de detecciones de la cámara for i in range(len(array_reg_detections)): reg = array_reg_detections[i] if i != 0: if sum(reg['detecciones_array']) == 0 : if reg["pellet"] != array_reg_detections[i-1]["pellet"]: new_format_array_reg_detections+=_update_reg_format(reg) else: # valid_array_reg.append(reg) new_format_array_reg_detections+= _update_reg_format(reg) new_document[key] = new_format_array_reg_detections elif key=='dia': try: old_format_date = datetime.strptime(document[key], "%m/%d/%y") new_format_date = old_format_date.strftime("%Y-%m-%d") new_document[key] = new_format_date except: new_document[key] = document[key] else: # pass new_document[key] = document[key] optimized_bd.append(new_document) return optimized_bd def _change_date_format_dataDB(db): data_detailed = db["data"].find() optimized_bd = [] # Iterar sobre documentos de la bd for document in data_detailed: new_document = {} # Iteración sobre las cámaras en el documento for key in document: if key not in ["_id", "dia"]: # Cámaras array_reg_detections = document[key] new_format_array_reg_detections = [] # Iteración sobre el arreglo de detecciones de la cámara for i in range(len(array_reg_detections)): reg = array_reg_detections[i] old_ts_format = datetime.strptime(reg["hora"], "%m/%d/%y %H:%M:%S") new_ts_format = old_ts_format.strftime("%Y-%m-%d %H:%M:%S") reg['hora'] = new_ts_format new_format_array_reg_detections.append( reg ) new_document[key] = new_format_array_reg_detections elif key=='dia': try: old_format_date = datetime.strptime(document[key], "%m/%d/%y") new_format_date = old_format_date.strftime("%Y-%m-%d") new_document[key] = new_format_date # Formato no coincide con el esperado except: new_document[key] = document[key] else: # pass new_document[key] = document[key] optimized_bd.append(new_document) return optimized_bd def Scapp_update_data_detailed(): # Verifica si la base de datos ya esta optimizada if not _verify_data_detailed_status(parameters_collection): try: new_collection_data_detailed = _optimize_data_detailed(scapp_db) # Actualiza los timestamp de data new_collection_data = _change_date_format_dataDB(scapp_db) # Nombre de la nueva colección collection_name = 'data' collection_detailed_name = 'data_detailed' # Sobreescribir pases de datos overwrite_data_detailed = _overwrite_colection(scapp_db, collection_detailed_name, new_collection_data_detailed) overwrite_data = _overwrite_colection(scapp_db, collection_name, new_collection_data) if overwrite_data_detailed and overwrite_data: parameters_collection.insert_one( {'data_detailed_status' : True} ) print("Estado de la base de datos actualizado correctamente") except: pass else: print("la base de datos ya se encuentra optimizada") def _update_reg_format_save_on_db(reg): # Recibe un arreglo de detecciones, genera un json para cada detección con marca de tiempo # reg contiene: hora, pellet, detecciones_array array = reg['detecciones_array'] new_format_array_reg_detections = [] # Añade ceros al principio y final, para mantener forma de "peaks" new_array_det_reg = [0] + list(filter(lambda x: x != 0,array)) + [0] # Calcula el espaciado de los ms fact = 1000/len(new_array_det_reg)%30*30 # Genera el str de hora con ms new_ts = [reg['hora'] + "." + str(((i+ 1)*fact))[:3] for i in range(len(new_array_det_reg)) ] # Separa cada elemento en un json y almacena cada json en new_format_array_reg_detections for i in range( len(new_array_det_reg) ): elem = new_array_det_reg[i] ts = new_ts[i] new_reg = {"hora": ts, "pellet": reg["pellet"], "detecciones": elem} new_format_array_reg_detections.append(new_reg) return new_format_array_reg_detections if __name__ == '__main__': Scapp_refactorice_db()