#!/usr/bin/env python3 """ Script para replicar tablas de gps_reportes de un MySQL origen a un MySQL destino. - Crea tablas nuevas sin DROP/TRUNCATE. - Copia datos en lotes con estimación de tiempo. - Usa transacciones cortas y READ COMMITTED para minimizar locks. - Logging con timestamps y flush para debug en caliente. - Mantiene registro de progreso en la tabla 'migracion' en el servidor de origen, usando 'tablamigrada' como un contador de revisiones para priorización. """ import sys import time import logging import mysql.connector from mysql.connector import errorcode # Configuración de conexiones con credenciales proporcionadas SOURCE_CONFIG = { 'host': '10.2.12.226', 'user': 'gps', 'password': 'q1w2e3r4', 'database': 'gps_reportes', 'charset': 'utf8mb4', 'use_pure': True, } DEST_CONFIG = { 'host': '10.2.12.220', 'user': 'gps', 'password': 'q1w2e3r4', 'database': 'gps_reportes', 'charset': 'utf8mb4', 'use_pure': True, } BATCH_SIZE = 1000 # Número de filas a copiar por lote LOG_FORMAT = '%(asctime)s %(levelname)s: %(message)s' logging.basicConfig(level=logging.INFO, format=LOG_FORMAT, stream=sys.stdout) logger = logging.getLogger('replicator') MIGRATION_TABLE_NAME = 'migracion' # Nombre de la tabla de seguimiento de progreso # Plantilla de la sentencia CREATE TABLE para las tablas LOG_ # Se usa un placeholder {table_name} para insertar el nombre de la tabla dinámicamente. CREATE_TABLE_TEMPLATE = """ CREATE TABLE `{table_name}` ( `IdReporte` int(10) unsigned NOT NULL AUTO_INCREMENT, `Reporte_IdDisp` varchar(30) DEFAULT NULL, `Reporte_FechaGPS` datetime DEFAULT NULL, `Reporte_FechaServidor` datetime DEFAULT NULL, `Reporte_Velocidad` decimal(18,2) DEFAULT NULL, `Reporte_Rumbo` int(11) DEFAULT NULL, `Reporte_Latitud` varchar(50) DEFAULT NULL, `Reporte_Longitud` varchar(50) DEFAULT NULL, `Reporte_Evento` varchar(2) DEFAULT NULL, `Reporte_EventoDescripcion` varchar(100) DEFAULT NULL, `Reporte_Entradas` varchar(20) DEFAULT NULL, `Reporte_Edad` varchar(4) DEFAULT NULL, `Reporte_Completo` varchar(1000) DEFAULT NULL, `Reporte_Puerto` varchar(45) DEFAULT NULL, `Reporte_PC` varchar(45) DEFAULT NULL, `Reporte_AlertaProcesada` int(11) DEFAULT NULL, `Reporte_NroMensaje` varchar(45) DEFAULT NULL, `Reporte_Val1` varchar(45) DEFAULT NULL, `Reporte_Val2` varchar(45) DEFAULT NULL, `Reporte_Val3` varchar(45) DEFAULT NULL, `Reporte_Val4` varchar(300) DEFAULT NULL, `Reporte_RUS` int(11) DEFAULT NULL, `Reporte_Direccion` varchar(400) DEFAULT NULL, `Reporte_Ignicion` varchar(2) DEFAULT NULL, `Reporte_Mileage` decimal(10,1) DEFAULT NULL, PRIMARY KEY (`IdReporte`), KEY `id_disp` (`Reporte_IdDisp`), UNIQUE KEY `Reporte_NroMensaje_Reporte_FechaGPS` (`Reporte_NroMensaje`,`Reporte_FechaGPS`), KEY `fecha1` (`Reporte_FechaServidor`), KEY `fecha2` (`Reporte_FechaGPS`), KEY `puerto` (`Reporte_Puerto`), KEY `rus` (`Reporte_RUS`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8; """ def create_table_if_not_exists_on_dest(dest_conn, table_name): """ Verifica si una tabla existe en el destino y, si no, la crea usando la plantilla. """ cursor = dest_conn.cursor() # Usamos INFORMATION_SCHEMA para una verificación más robusta cursor.execute("SELECT COUNT(*) FROM information_schema.tables WHERE table_schema = %s AND table_name = %s", (DEST_CONFIG['database'], table_name)) exists = cursor.fetchone()[0] > 0 cursor.close() if not exists: logger.info(f"Tabla '{table_name}' no existe en el destino. Creándola...") # Construir la sentencia CREATE TABLE usando la plantilla create_sql = CREATE_TABLE_TEMPLATE.format(table_name=table_name) dest_cursor_create = dest_conn.cursor() try: dest_cursor_create.execute(create_sql) dest_conn.commit() logger.info(f"Tabla '{table_name}' creada exitosamente en el destino.") except mysql.connector.Error as err: logger.error(f"Error al crear la tabla '{table_name}' en el destino: {err}") # Re-lanza la excepción para detener el script si la creación falla raise finally: dest_cursor_create.close() else: logger.info(f"Tabla '{table_name}' ya existe en el destino. Saltando la creación.") def normalize_single_table_id(conn, table): """ Normaliza una única tabla si cumple con las condiciones: - Un solo registro. - IdReporte = 2711. Retorna True si la normalización se realizó, False en caso contrario. """ cursor = conn.cursor() normalization_performed = False logger.info(f"Iniciando la normalización para la tabla '{table}'...") try: # Contar registros y verificar el IdReporte query_count_and_id = f""" SELECT COUNT(*) AS count, MAX(IdReporte) AS max_id FROM `{table}` """ cursor.execute(query_count_and_id) result = cursor.fetchone() if result and result[0] == 1 and result[1] == 2711: logger.warning(f"Se encontró una tabla para normalizar: '{table}' con 1 registro y IdReporte = 2711.") # Actualizar el IdReporte a 1 try: update_sql = f"UPDATE `{table}` SET IdReporte = 1 WHERE IdReporte = 2711" cursor.execute(update_sql) conn.commit() logger.info(f" --> IdReporte en '{table}' actualizado de 2711 a 1.") normalization_performed = True except mysql.connector.Error as err: logger.error(f" --> Error al actualizar IdReporte en '{table}': {err}") conn.rollback() # Deshacer si falla # Normalizar el valor de AUTO_INCREMENT try: alter_sql = f"ALTER TABLE `{table}` AUTO_INCREMENT = 2" cursor.execute(alter_sql) conn.commit() logger.info(f" --> AUTO_INCREMENT en '{table}' normalizado a 2.") except mysql.connector.Error as err: logger.error(f" --> Error al normalizar AUTO_INCREMENT en '{table}': {err}") conn.rollback() # Deshacer si falla else: logger.debug(f"Tabla '{table}' no requiere normalización o no cumple los criterios.") except mysql.connector.Error as err: logger.error(f"Error MySQL durante la normalización de la tabla '{table}': {err}") finally: cursor.close() logger.info(f"Normalización de la tabla '{table}' completada.") return normalization_performed def table_exists(conn, table): """ Verifica si una tabla existe en la base de datos de una conexión dada. """ cursor = conn.cursor() sql_query = "SELECT COUNT(*) FROM information_schema.tables WHERE table_schema = %s AND table_name = %s" logger.debug(f"Ejecutando SQL: {sql_query} con params: {(conn.database, table)}") cursor.execute(sql_query, (conn.database, table)) exists = cursor.fetchone()[0] > 0 cursor.close() return exists def is_table_empty(conn, table): """ Verifica si una tabla existe y, si existe, si tiene registros. Retorna True si la tabla no existe o está vacía, False si tiene registros. """ if not table_exists(conn, table): logger.info(f"Tabla '{table}' no existe. Se considera vacía.") return True cursor = conn.cursor() try: sql_query = f"SELECT COUNT(*) FROM `{table}`" cursor.execute(sql_query) count = cursor.fetchone()[0] if count == 0: logger.info(f"Tabla '{table}' existe pero no tiene registros.") return True else: logger.info(f"Tabla '{table}' tiene {count} registros.") return False except mysql.connector.Error as err: logger.error(f"Error al verificar si la tabla '{table}' está vacía: {err}. Asumiendo que está vacía para evitar errores.") return True finally: cursor.close() def create_migration_table_if_not_exists(conn): """ Crea la tabla 'migracion' en la base de datos de origen si no existe. La columna 'tablamigrada' se establece como INT para ser un contador. """ create_sql = f""" CREATE TABLE IF NOT EXISTS `{MIGRATION_TABLE_NAME}` ( `id` INT(11) UNSIGNED NOT NULL AUTO_INCREMENT, `tabla` VARCHAR(64) COLLATE utf8_unicode_ci NOT NULL, -- Aumentado a 64 para nombres de tabla más largos `fecha` DATETIME DEFAULT CURRENT_TIMESTAMP, `ultidreporte` BIGINT(20) DEFAULT NULL, `fechaupdate` DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, `tablamigrada` INT(11) DEFAULT 0, -- Cambiado a INT(11) para ser un contador PRIMARY KEY (`id`), UNIQUE KEY `uq_tabla` (`tabla`) ) ENGINE=INNODB DEFAULT CHARSET=utf8 COLLATE=utf8_unicode_ci; """ cursor = conn.cursor() try: logger.debug(f"Ejecutando SQL: {create_sql}") cursor.execute(create_sql) conn.commit() logger.info(f"Tabla '{MIGRATION_TABLE_NAME}' asegurada en el origen.") except mysql.connector.Error as err: logger.error(f"Error al crear/verificar tabla '{MIGRATION_TABLE_NAME}': {err}") raise # Re-lanza la excepción para detener el script si falla la creación crítica finally: cursor.close() def _check_migration_entry_exists(cursor, table_name): """Verifica si una entrada para la tabla ya existe en la tabla de migración.""" select_sql = f"SELECT tablamigrada FROM `{MIGRATION_TABLE_NAME}` WHERE tabla = %s " logger.debug(f"Verificando existencia de entrada para {table_name} en {MIGRATION_TABLE_NAME}...") cursor.execute(select_sql, (table_name,)) return cursor.fetchone() def _update_migration_entry(cursor, table_name, last_copied_id, current_migrada_count): """Realiza la operación de UPDATE en la tabla de migración.""" update_sql = f""" UPDATE `{MIGRATION_TABLE_NAME}` SET `ultidreporte` = %s, `fechaupdate` = CURRENT_TIMESTAMP, `tablamigrada` = %s + 1 WHERE `tabla` = %s """ params = (last_copied_id, current_migrada_count, table_name) logger.debug(f"Ejecutando SQL (UPDATE): {update_sql} con params: {params}") cursor.execute(update_sql, params) return cursor.rowcount def _insert_migration_entry(cursor, table_name, last_copied_id): """Realiza la operación de INSERT en la tabla de migración.""" insert_sql = f""" INSERT INTO `{MIGRATION_TABLE_NAME}` (`tabla`,`fecha`, `ultidreporte`, `tablamigrada`) VALUES (%s,NOW(), %s, 1) -- Para la primera inserción, el contador es 1 """ params = (table_name, last_copied_id) logger.debug(f"Ejecutando SQL (INSERT): {insert_sql} con params: {params}") cursor.execute(insert_sql, params) return cursor.rowcount def update_migration_progress(src_conn, table_name, last_copied_id): """ Inserta o actualiza el progreso de la copia de una tabla en la tabla 'migracion'. El contador 'tablamigrada' se incrementa en cada llamada. """ cursor = src_conn.cursor() try: result = _check_migration_entry_exists(cursor, table_name) rows_affected = 0 if result: current_migrada_count = result[0] rows_affected = _update_migration_entry(cursor, table_name, last_copied_id, current_migrada_count) else: rows_affected = _insert_migration_entry(cursor, table_name, last_copied_id) src_conn.commit() logger.info(f"Progreso guardado y contador 'tablamigrada' actualizado para {table_name}: IdReporte {last_copied_id}. Filas afectadas: {rows_affected}") except mysql.connector.Error as err: logger.error(f"Error al actualizar el progreso de la migración para {table_name}: {err}") finally: cursor.close() def get_max_idreporte_from_dest(dest_conn, table): """ Obtiene el valor máximo de 'IdReporte' de una tabla en la base de datos de destino. Retorna 0 si la tabla no existe o está vacía, o el valor máximo encontrado. """ cursor = dest_conn.cursor() try: sql_query = f"SELECT MAX(IdReporte) FROM `{table}`" logger.debug(f"Ejecutando SQL: {sql_query}") cursor.execute(sql_query) max_id = cursor.fetchone()[0] return max_id if max_id is not None else 0 except mysql.connector.Error as err: logger.warning(f"No se pudo obtener MAX(IdReporte) para la tabla {table} en destino: {err}. Asumiendo 0.") return 0 finally: cursor.close() def get_max_idreporte_from_source(src_conn, table): """ Obtiene el valor máximo de 'IdReporte' de una tabla en la base de datos de origen. Retorna 0 si la tabla no existe o está vacía. """ cursor = src_conn.cursor() try: sql_query = f"SELECT MAX(IdReporte) FROM `{table}`" logger.debug(f"Ejecutando SQL: {sql_query}") cursor.execute(sql_query) result = cursor.fetchone() max_id = result[0] if result and result[0] is not None else 0 return max_id except mysql.connector.Error as err: logger.error(f"Error al obtener MAX(IdReporte) de la tabla {table} en origen: {err}") return 0 finally: cursor.close() def get_ordered_log_tables_for_migration(conn): """ Obtiene la lista de tablas LOG_ del origen, ordenadas por prioridad de migración: 1. Tablas que no existen en la tabla 'migracion' (nuevas). 2. Tablas que existen en 'migracion' con el menor valor de 'tablamigrada' (las menos revisadas). Luego, alfabéticamente para desempate. """ cursor = conn.cursor() #all_tables_query = "SELECT table_name FROM information_schema.tables WHERE table_schema = %s ORDER BY table_name ASC;" all_tables_query = f""" SELECT CONCAT('LOG_', t.Reporte_Iddisp) AS table_name FROM `ultima_posicion` AS t ORDER BY t.Reporte_fechagps DESC LIMIT 5000; """ logger.debug(f"Ejecutando SQL: {all_tables_query} con params: {(conn.database,)}") cursor.execute(all_tables_query) all_tables = [row[0] for row in cursor] logger.debug(f"Todas las tablas encontradas en la base de datos '{conn.database}': {', '.join(all_tables)}") sql_query = f""" SELECT t.table_name, (CASE WHEN m.tablamigrada IS NULL THEN 0 ELSE 1 END) AS migrado, m.tablamigrada AS current_migrada_count, m.ultidreporte AS last_copied_id_tracker FROM information_schema.tables AS t LEFT JOIN `migracion` AS m ON t.table_name = m.tabla WHERE t.table_schema = 'gps_reportes' AND t.table_name LIKE 'LOG_%' ORDER BY (CASE WHEN m.tablamigrada IS NULL THEN 0 ELSE 1 END) ASC, t.table_name ASC; """ logger.info(f"Ejecutando SQL para tablas LOG_: {sql_query} con params: {(conn.database,)}") # Se añade multi=True porque el error lo exige en tu entorno. cursor.execute(sql_query) tables_info = [] current_result_set = cursor.fetchall() if current_result_set: tables_info = [(row[0], row[1], row[2]) for row in current_result_set] while cursor.nextset(): cursor.fetchall() cursor.close() log_tables_found = [info[0] for info in tables_info] logger.info(f"Tablas que coinciden con el patrón 'LOG_%': {', '.join(log_tables_found) if log_tables_found else 'Ninguna'}") return tables_info def copy_data(src_conn, dest_conn, table): """ Copia datos de la tabla de origen a la de destino incrementalmente por IdReporte. Retorna el último IdReporte copiado, o el último IdReporte en destino si no hay datos nuevos. """ # --- NUEVA LÓGICA DE COPIA COMPLETA VS. INCREMENTAL --- if is_table_empty(dest_conn, table): logger.info(f"La tabla de destino '{table}' está vacía. Copiando la tabla completa desde el origen.") start_id_for_copy = 1 select_sql = f"SELECT * FROM `{table}` ORDER BY IdReporte ASC" last_id_in_current_batch = 0 # Valor inicial para el tracker else: last_id_dest = get_max_idreporte_from_dest(dest_conn, table) start_id_for_copy = last_id_dest + 1 logger.info(f"La tabla de destino '{table}' tiene datos. Copiando datos de '{table}' desde IdReporte >= {start_id_for_copy}.") select_sql = f"SELECT * FROM `{table}` WHERE IdReporte >= %s ORDER BY IdReporte ASC" last_id_in_current_batch = last_id_dest count_cursor = src_conn.cursor() # Usamos la misma lógica de conteo para ambos casos count_sql = f"SELECT COUNT(*) FROM `{table}`" if 'WHERE' in select_sql: count_sql += " WHERE IdReporte >= %s" count_cursor.execute(count_sql, (start_id_for_copy,)) else: count_cursor.execute(count_sql) total_rows_to_copy = count_cursor.fetchone()[0] count_cursor.close() if total_rows_to_copy == 0: logger.info(f"No hay nuevos datos para copiar en la tabla {table} (desde IdReporte {start_id_for_copy}).") return last_id_in_current_batch logger.info(f"Copiando datos de {table}: {total_rows_to_copy} nuevas filas en total.") start_time = time.time() copied = 0 try: with src_conn.cursor() as src_cursor: # La ejecución del SQL ahora es condicional a si se pasa el parámetro de start_id if 'WHERE' in select_sql: logger.debug(f"Ejecutando SQL: {select_sql} con params: {(start_id_for_copy,)}") src_cursor.execute(select_sql, (start_id_for_copy,)) else: logger.debug(f"Ejecutando SQL: {select_sql}") src_cursor.execute(select_sql) # Obtener nombres de columnas cols = [desc[0] for desc in src_cursor.description] try: id_reporte_col_index = cols.index('IdReporte') except ValueError: logger.error(f"La columna 'IdReporte' no se encontró en la tabla {table}. No se puede realizar un seguimiento de progreso incremental.") return last_id_in_current_batch placeholders = ",".join(["%s"] * len(cols)) # Usamos INSERT IGNORE para manejar entradas duplicadas sin detener el proceso insert_sql = f"INSERT IGNORE INTO `{table}` ({', '.join(cols)}) VALUES ({placeholders})" logger.info(f"Preparando INSERT SQL: {insert_sql}") batch = [] for row in src_cursor: batch.append(row) last_id_in_current_batch = row[id_reporte_col_index] if len(batch) >= BATCH_SIZE: logger.info(f" Procesando un lote de {len(batch)} filas...") try: with dest_conn.cursor() as dest_cur: dest_cur.executemany(insert_sql, batch) dest_conn.commit() copied += len(batch) logger.info(f" Lote de {len(batch)} filas insertado con éxito.") except mysql.connector.Error as err: logger.warning(f"Error al insertar lote en {table}: {err}. Deshaciendo cambios y continuando con el próximo lote.") dest_conn.rollback() # Deshacer el lote fallido finally: batch.clear() elapsed = time.time() - start_time rate = copied / elapsed if elapsed > 0 else 0 eta = (total_rows_to_copy - copied) / rate if rate > 0 else None logger.info(f" Copiadas {copied}/{total_rows_to_copy}. Último ID copiado: {last_id_in_current_batch}. ETA: {eta:.1f}s" if eta else f" Copiadas {copied}/{total_rows_to_copy}. Último ID copiado: {last_id_in_current_batch}.") if batch: logger.debug(f" Procesando el lote final de {len(batch)} filas...") try: with dest_conn.cursor() as dest_cur: dest_cur.executemany(insert_sql, batch) dest_conn.commit() copied += len(batch) logger.debug(f" Lote final de {len(batch)} filas insertado con éxito.") except mysql.connector.Error as err: # Capturamos el error para el lote final logger.warning(f"Error al insertar el lote final en {table}: {err}. Ignorando lote.") dest_conn.rollback() # Deshacer el lote fallido except mysql.connector.Error as err: logger.error(f"Error de MySQL durante la copia de datos para {table}: {err}. Finalizando copia de esta tabla.") # No relanzamos la excepción, simplemente retornamos para que el bucle principal continúe # con la siguiente tabla. return last_id_in_current_batch except Exception as e: logger.error(f"Error inesperado durante la copia de datos para {table}: {e}. Finalizando copia de esta tabla.") return last_id_in_current_batch logger.info(f"Tabla {table} completada en {time.time() - start_time:.1f}s. Total de filas copiadas: {copied}. Último ID copiado: {last_id_in_current_batch}.") return last_id_in_current_batch def get_migration_status_from_tracker(conn, table_name): """ Obtiene el ultimo ID reportado y el estado de 'tablamigrada' de la tabla 'migracion'. Devuelve (ultidreporte, tablamigrada) o (None, None) si la tabla no se encuentra en el tracker. """ cursor = conn.cursor() query = f"SELECT ultidreporte, tablamigrada FROM `{MIGRATION_TABLE_NAME}` WHERE tabla = %s;" logger.debug(f"Ejecutando SQL: {query} con params: {(table_name,)}") try: cursor.execute(query, (table_name,)) result = cursor.fetchone() if result: return result[0], result[1] else: return None, None except mysql.connector.Error as err: logger.error(f"Error al obtener estado de migración de {table_name}: {err}") raise finally: cursor.close() def main(): overall_start = time.time() try: logger.info("Conectando a origen...") src = mysql.connector.connect(**SOURCE_CONFIG) logger.info("Conectado al origen.") logger.info("Conectando a destino...") dst = mysql.connector.connect(**DEST_CONFIG) logger.info("Conectado al destino.") create_migration_table_if_not_exists(src) tables_info = get_ordered_log_tables_for_migration(src) logger.debug(f"Tablas 'LOG_' encontradas en origen para procesar (ordenadas por prioridad): {len(tables_info)}") for table_name, current_migrada_count, last_copied_id_tracker in tables_info: logger.info(f"\n--- Iniciando revisión y procesamiento para tabla: {table_name} (Migraciones previas: {current_migrada_count}) ---") # Normalizar la tabla individualmente antes de la migración. # normalization_performed = normalize_single_table_id(src, table_name) # Obtener el MAX(IdReporte) actual de la tabla en el origen current_max_id_source = get_max_idreporte_from_source(src, table_name) current_max_id_dest = last_copied_id_tracker if last_copied_id_tracker is not None else 0 last_id_for_tracker = current_max_id_dest migration_succeeded = False # Bandera para rastrear si la migración fue exitosa try: # CREACIÓN DE LA TABLA EN DESTINO ANTES DE CUALQUIER COPIA create_table_if_not_exists_on_dest(dst, table_name) # Modificación clave: si se realizó una normalización y el MAX ID es 1, # forzamos la copia reestableciendo el destino a 0 para esta iteración. #if normalization_performed and current_max_id_source == 1 and current_max_id_dest == 1: # logger.warning(f"Normalización detectada para la tabla {table_name}. Forzando la copia del registro (IdReporte=1).") # current_max_id_dest = 0 logger.info("dato del id max en source {current_max_id_source} dato del id ax en dest {current_max_id_dest} ") if current_max_id_source > current_max_id_dest: logger.info(f"Tabla {table_name}: Nuevos datos encontrados en origen (MAX ID Origen: {current_max_id_source}, MAX ID Destino: {current_max_id_dest}). Procediendo a copiar.") copied_id = copy_data(src, dst, table_name) if copied_id is not None: last_id_for_tracker = copied_id migration_succeeded = True # La migración de esta tabla se completó else: logger.info(f"Tabla {table_name} está sincronizada (MAX ID Origen: {current_max_id_source}, MAX ID Destino: {current_max_id_dest}). No se necesitan copiar nuevos datos.") migration_succeeded = True # Consideramos que la tabla está "sincronizada" si no hay nada que copiar except mysql.connector.Error as err: logger.error(f"Error de MySQL al procesar tabla {table_name}: {err}. Continuará con la siguiente tabla.") migration_succeeded = False # Se produjo un error fatal, no actualizar el progreso except Exception as e: logger.error(f"Error inesperado al procesar tabla {table_name}: {e}. Continuará con la siguiente tabla.") migration_succeeded = False # Se produjo un error fatal, no actualizar el progreso finally: if migration_succeeded: update_migration_progress(src, table_name, last_id_for_tracker) logger.info(f"--- Revisión y contador para tabla {table_name} finalizado en este ciclo. ---") else: logger.info(f"--- La migración de la tabla {table_name} falló. No se actualizará el progreso en 'migracion'. ---") total_time = time.time() - overall_start logger.info(f"Proceso de migración de tablas LOG_ completado en {total_time:.1f}s.") except mysql.connector.Error as err: logger.error(f"Error fatal de MySQL en el proceso principal: {err}") sys.exit(1) except Exception as e: logger.error(f"Error inesperado fatal en el proceso principal: {e}") sys.exit(1) finally: if 'src' in locals() and src.is_connected(): src.close() if 'dst' in locals() and dst.is_connected(): dst.close() logger.info("Conexiones a base de datos cerradas.") if __name__ == '__main__': main()