## # @file Transformation.py # # @brief Script para realizar la transformación de la colección original a la colección transformada. # # @authors DopamineLabs LTDA # @date 2025 ## # Usage Example # python Transformation.py --database ScappDB --target-collection dummy_data --old-collection dummy_data_old --parameter-collection # parameters --log-file log.json from tqdm import tqdm from pymongo.errors import PyMongoError from pymongo import UpdateOne from .TransformationFunctions import document_transformer from .Logger import dump_to_log_file # from scapp_modules.Logger import main_logger def main_logger(text): print(text) DEFAULT_MONGO_URI = 'mongodb://localhost:27017/' time_series_options = { "timeseries": { "timeField": "timestamp", # "metaField": "metadata", "metaField": "camera_id", "granularity": "seconds" } } def transform_to_timeseries( database, args ): """ Args: database : Instancia de la bd de MongoDB. args (dict): { target_collection: Nombre de la coleccion objetivo old_collection : Nombre para renombrar colección objetivo log_file: archivo para guardar logs de salida verbose } """ # Si la colección data no existe, se retorna anticipadamente, luego de crearla y añadir el indice compuesto if ensure_data_ts_col_and_compose_index( database, args['target_collection'] ): return # Si la colección existe, pero ya es serie de tiempo, se revisa el índice y se retorna anticipadamente. if is_timeseries_collection( database, args['target_collection'] ): print() ensure_data_compose_index( database[args['target_collection']] ) main_logger(f"La Colección '{args['target_collection']}' ya es una serie de tiempo. Retornando anticipadamente de verificacion.") return # Load target collection current_collection = database[args['target_collection']] n_docs_current_collection = current_collection.count_documents({}) print(f"[INFO] There are {n_docs_current_collection} documents on the current collection to be transformed") # Rename the current collection print(f"[INFO] Renaming the target collection {args['target_collection']} to {args['old_collection']}") # Verificar que los nombres no sean idénticos if args['target_collection'] == args['old_collection']: args['old_collection'] += '_old' current_collection.rename(args['old_collection']) current_collection = database[args['old_collection']] # * Re estructuración de visitas visits_collection = database['data_visitas'] # Create handle for new time-series collection database.create_collection(args['target_collection'], **time_series_options) new_collection = database[args['target_collection']] # Añade el indice compuesto a la nueva colección de series de tiempo ensure_data_compose_index( new_collection ) # Transformation error array: save the _id of this documents transformation_errors = [] # Transformation associaton: Save the original _id and the resulting _id from new documents associated_ids = {} # Start the transformation process with tqdm(total=n_docs_current_collection, desc='Document progress', position=0) as pbar: for doc in current_collection.find({}): # try: _id = doc['_id'] new_docs, new_visits_docs = document_transformer(doc, args['verbose'], pbar) if new_docs: resulting_ids = new_collection.insert_many(new_docs) # Save the ids associated_ids[str(_id)] = list(resulting_ids.inserted_ids) else: # Get id from document transformation_errors.append(_id) pbar.write(f'[Dataerror] Invalid documment could not be transformed: {doc}') # * Re estructuración de visitas if new_visits_docs: try: bulk_ops = [] # Iterar sobre el buffer de visitas: claves son canales, valores son la lista de visitas for visitas in new_visits_docs.values(): # Construir la operación para setear el campo "Visitas" con la lista actualizada bulk_ops.append( UpdateOne( { "camera_id": visitas["camera_id"], "day": visitas["day"] }, {"$set": visitas}, upsert=True ) ) # Ejecutar la operación bulk_write si existen operaciones pendientes if bulk_ops: _ = visits_collection.bulk_write(bulk_ops) except Exception as e: main_logger(f"Error en bulk update set de visitas {e.__traceback__.tb_lineno}: {e}") # except Exception as e: # main_logger(f"ERROR MIGRANDO {e.__traceback__.tb_lineno}: {e}") pbar.update(1) # Save the logs onto file if args.get('log_file') is not None: print(f"[INFO] Writting errors and associated ids onto file {args['log_file']}") dump_to_log_file(args['log_file'], transformation_errors, associated_ids) """ # updata ScappDB flag parameters_collection = database[args['parameter_collection']] search_query = {'ScappDB_version': {'_id': {'$exists' : True}}} parameters_collection.update_one(search_query, {'$set': {'ScappDB_version': '20241118'}}, upsert=True) print(f"[INFO] Updating ScappDB version onto {args['parameter_collection']}") """ def is_timeseries_collection( db, nombre_coleccion ): """ Verifica si la colección especificada es una colección de series temporales. Args: db: La instancia de la base de datos MongoDB. nombre_coleccion (str): El nombre de la colección a verificar. Returns: bool: True si es una colección de series temporales, False en caso contrario. """ try: # Obtenemos información detallada de la colección mediante el comando listCollections colecciones = db.command("listCollections", filter={"name": nombre_coleccion}) if colecciones.get("cursor", {}).get("firstBatch"): info_coleccion = colecciones["cursor"]["firstBatch"][0] # Si en las opciones se encuentra la clave 'timeseries', es de series temporales if "options" in info_coleccion and "timeseries" in info_coleccion["options"]: return True except PyMongoError as e: print(f"Error al consultar la colección: {e}") return False def ensure_data_ts_col_and_compose_index(database, collection_name, index_spec=[("camera_id", 1), ("timestamp", 1)], index_name=None): """ Revisa si el índice especificado en index_spec ya existe en la colección. Si existe, muestra su configuración. En caso contrario, crea el índice. :param collection: Instancia de la colección de MongoDB. :param index_spec: Diccionario con la especificación del índice (ej.: [("camera_id", 1), ("timestamp", 1)]). :param index_name: Nombre opcional para el índice. :return: Nombre del índice existente o creado. """ created = False # Verificar si la colección existe; si no, se crea if collection_name not in database.list_collection_names(): # Create handle for new time-series collection collection = database.create_collection(collection_name, **time_series_options) print() ; main_logger(f"La Colección '{collection_name}' no existía y se ha creado como serie de tiempo. Retornando anticipadamente.") created = True else: collection = database[collection_name] # Si la colección se crea, se le añade el índice if created: ensure_data_compose_index( collection, index_spec ) return created def ensure_data_compose_index( collection, index_spec=[("camera_id", 1), ("timestamp", 1)], ): # Revisar los índices existentes en la colección created_index = False indexes = collection.index_information() for idx_name, info in indexes.items(): # Se compara la lista de campos (info["key"]) con la especificación deseada y se verifica que sea único if info.get("key") == index_spec and info.get("unique", False): #main_logger(f"El índice '{idx_name}' con especificación {index_spec} ya existe en la colección.") created_index = True break # Si el índice no existe, se crea if not created_index: created_index = collection.create_index(index_spec) main_logger(f"Índice creado en colección data: {created_index}") return # Define arguments """ parser = argparse.ArgumentParser( prog="ScappDB transformation script", description="Transforms the provided collection into the requested format and performs data integrity validation", epilog="blueOctopus" ) parser.add_argument( "--mongo-uri", default=DEFAULT_MONGO_URI, help=f"Overrides the default URI to connect to mongoDB. Default is {DEFAULT_MONGO_URI}" ) parser.add_argument( "--database", "-db", required=True, help="Database name" ) parser.add_argument( "--old-collection", "-oc", required=True, help="Old collection name" ) parser.add_argument( "--target-collection", "-tc", required=True, help="Name of the collection to migrate" ) parser.add_argument( "--parameter-collection", "-pc", required=True, help="Name of the collection that contains the db parameters" ) parser.add_argument( "--log-file", "-lf", required=False, help="File to log the transformation and validation messages", default="log.json" ) parser.add_argument( '--verbose', action='store_true', help='Show all prints.' ) args = parser.parse_args() """