#!/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 import datetime 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, } IMEI = "" BATCH_SIZE = 100 # 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') if not imei: # If it's empty, ask the user for input imei = input("La variable IMEI esta vacia. Por favor ingrsarlo IMEI: ") 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 gps_reportes.`{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 = 1. 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 gps_reportes.`{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 gps_reportes.`{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 gps_reportes.`{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 gps_reportes.`{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 gps_reportes.`{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 gps_reportes.`{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, current_migrada_count): """Realiza la operación de UPDATE en la tabla de migración.""" update_sql = f""" UPDATE gps_reportes.`{MIGRATION_TABLE_NAME}` SET `fechaupdate` = CURRENT_TIMESTAMP, `tablamigrada` = %s + 1 WHERE `tabla` = %s """ params = (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): """Realiza la operación de INSERT en la tabla de migración.""" insert_sql = f""" INSERT INTO gps_reportes.`{MIGRATION_TABLE_NAME}` (`tabla`,`fecha`, `tablamigrada`) VALUES (%s,NOW(), 1) -- Para la primera inserción, el contador es 1 """ params = (table_name) 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): """ 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, current_migrada_count) else: rows_affected = _insert_migration_entry(cursor, table_name, ) 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 gps_reportes.`{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 gps_reportes.`{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 gps_reportes.`ultima_posicion` AS t where reporte_iddisp = '{IMEI}' ORDER BY t.Reporte_fechagps DESC LIMIT 500; """ 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 gps_reportes.`migracion` AS m ON t.table_name = m.tabla WHERE t.table_schema = 'gps_reportes' AND t.table_name LIKE 'LOG_{IMEI}' 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 ensure_aux_table(conn): """ Verifica si la tabla auxiliar existe, y si no, la crea. """ create_aux_table_sql = f""" CREATE TABLE IF NOT EXISTS gps_reportes.`{AUX_TABLE_NAME}` ( `Reporte_NroMensaje1` CHAR(10) NOT NULL, `Reporte_FechaGPS1` DATETIME NOT NULL, `Reporte_NroMensaje2` CHAR(10) DEFAULT NULL, `Reporte_FechaGPS2` DATETIME DEFAULT NULL, `checkeo` TINYINT(1) DEFAULT 0, PRIMARY KEY (`Reporte_NroMensaje1`, `Reporte_FechaGPS1`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; """ try: with conn.cursor() as cursor: cursor.execute(create_aux_table_sql) conn.commit() logging.info(f"Tabla auxiliar '{AUX_TABLE_NAME}' verificada/creada con éxito.") except mysql.connector.Error as err: logging.error(f"Error al crear la tabla auxiliar '{AUX_TABLE_NAME}': {err}") raise def truncate_aux_table(conn): """ Limpia todos los datos de la tabla auxiliar. """ try: with conn.cursor() as cursor: cursor.execute(f"TRUNCATE TABLE gps_reportes.`{AUX_TABLE_NAME}`") conn.commit() logging.info(f"Tabla auxiliar '{AUX_TABLE_NAME}' truncada con éxito.") except mysql.connector.Error as err: logging.error(f"Error al truncar la tabla auxiliar '{AUX_TABLE_NAME}': {err}") raise # Constante para el tamaño de lote (ejemplo) BATCH_SIZE = 1000 AUX_TABLE_NAME = "auxiliar_migracion" def copy_data_manual_reconciliation(src_conn, dest_conn, table): """ Realiza la conciliación y copia de datos faltantes del servidor de origen al de destino utilizando una tabla auxiliar para el seguimiento de cada reporte. """ total_copied_rows = 0 start_time = time.time() try: # Step 0: Ensure and truncate the auxiliary table on the DESTINATION server ensure_aux_table(dest_conn) truncate_aux_table(dest_conn) # Paso 1: Copiar los metadatos de los reportes del server de origen a la tabla auxiliar del DESTINO logging.info("Paso 1: Copiando metadatos de los reportes del servidor de origen a la tabla auxiliar en el destino.") select_src_sql = f"SELECT Reporte_NroMensaje, Reporte_FechaGPS FROM gps_reportes.`{table}`" insert_aux_sql = f"INSERT INTO gps_reportes.`{AUX_TABLE_NAME}` (`Reporte_NroMensaje1`, `Reporte_FechaGPS1`) VALUES (%s, %s)" # Read all data from the SOURCE and store it in memory with src_conn.cursor() as src_cursor: src_cursor.execute(select_src_sql) all_source_data = src_cursor.fetchall() # Now, write the data in batches to the auxiliary table on the DESTINATION with src_conn.cursor() as src_cur: for i in range(0, len(all_source_data), BATCH_SIZE): batch = all_source_data[i:i + BATCH_SIZE] if batch: src_cur.executemany(insert_aux_sql, batch) src_conn.commit() logging.info("Paso 1 completado. Metadatos de origen copiados a la tabla auxiliar.") # --- # Paso 2: Escanear el servidor destino y actualizar la tabla auxiliar en el DESTINO logging.info("Paso 2: Escaneando el servidor destino para conciliar los datos. y pego en le server de destino") select_dest_sql = f"SELECT Reporte_NroMensaje, Reporte_FechaGPS FROM gps_reportes.`{table}`" update_aux_sql = f"UPDATE gps_reportes.`{AUX_TABLE_NAME}` SET `Reporte_NroMensaje2` = %s, `Reporte_FechaGPS2` = %s, `checkeo` = 1 WHERE `Reporte_NroMensaje1` = %s AND `Reporte_FechaGPS1` = %s" # Read all data from the DESTINATION with dest_conn.cursor() as dest_cursor: dest_cursor.execute(select_dest_sql) all_dest_data = dest_cursor.fetchall() # Prepare and execute the update in batches on the DESTINATION update_batch = [(row[0], row[1], row[0], row[1]) for row in all_dest_data] with src_conn.cursor() as src_cur: for i in range(0, len(update_batch), BATCH_SIZE): batch = update_batch[i:i + BATCH_SIZE] if batch: src_cur.executemany(update_aux_sql, batch) src_conn.commit() logging.info("Paso 2 completado. Conciliación con el servidor destino finalizada.") # --- with src_conn.cursor(buffered=True) as src_cursor: src_cursor.execute(f'SELECT * FROM gps_reportes.`{table}` LIMIT 1') cols = [desc[0] for desc in src_cursor.description] logging.info("Paso 3.1: Columnas obtenidas del origen.") # Paso 3: Copiar los datos faltantes (checkeo = 0) del origen al destino logging.info("Paso 3: Copiando los reportes faltantes del servidor de origen al destino.") select_missing_sql = f""" SELECT Reporte_IdDisp, Reporte_FechaGPS, Reporte_FechaServidor, Reporte_Velocidad, Reporte_Rumbo, Reporte_Latitud, Reporte_Longitud, Reporte_Evento, Reporte_EventoDescripcion, Reporte_Entradas, Reporte_Edad, Reporte_Completo, Reporte_Puerto, Reporte_PC, Reporte_AlertaProcesada, Reporte_NroMensaje, Reporte_Val1, Reporte_Val2, Reporte_Val3, Reporte_Val4, Reporte_RUS, Reporte_Direccion, Reporte_Ignicion, Reporte_Mileage FROM `{table}` AS s JOIN gps_reportes.`{AUX_TABLE_NAME}` AS a ON s.Reporte_NroMensaje = a.Reporte_NroMensaje1 AND s.Reporte_FechaGPS = a.Reporte_FechaGPS1 WHERE a.checkeo = 0; """ logging.info(select_missing_sql) # Get column names first. This is a separate, isolated read operation. placeholders = ",".join(["%s"] * (len(cols) - 1)) insert_dest_sql = f"INSERT IGNORE INTO gps_reportes.`{table}` (Reporte_IdDisp, Reporte_FechaGPS,Reporte_FechaServidor, Reporte_Velocidad, Reporte_Rumbo, Reporte_Latitud, Reporte_Longitud,Reporte_Evento, Reporte_EventoDescripcion, Reporte_Entradas, Reporte_Edad,Reporte_Completo, Reporte_Puerto, Reporte_PC, Reporte_AlertaProcesada,Reporte_NroMensaje, Reporte_Val1, Reporte_Val2, Reporte_Val3, Reporte_Val4,Reporte_RUS, Reporte_Direccion, Reporte_Ignicion, Reporte_Mileage) VALUES ({placeholders})" logging.info(insert_dest_sql) # Read all missing data from the SOURCE with src_conn.cursor() as src_cursor: src_cursor.execute(select_missing_sql) missing_data = src_cursor.fetchall() # Insert the missing data in batches into the DESTINATION with dest_conn.cursor() as dest_cur: for i in range(0, len(missing_data), BATCH_SIZE): batch = missing_data[i:i + BATCH_SIZE] if batch: dest_cur.executemany(insert_dest_sql, batch) total_copied_rows += len(batch) logger.info(f" Lote de {len(batch)} filas insertado con éxito.") dest_conn.commit() logging.info(f"Paso 3 completado. Copia de {total_copied_rows} registros faltantes finalizada.") except mysql.connector.Error as err: logging.error(f"Error de MySQL en la conciliación manual para {table}: {err}.") dest_conn.rollback() raise except Exception as e: logging.error(f"Error inesperado en la conciliación manual para {table}: {e}.") raise 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 gps_reportes.`{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)}") # 1. Asegurar que la tabla auxiliar exista ensure_aux_table(src) # 2. Limpiar la tabla auxiliar para la nueva ejecución truncate_aux_table(src) 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 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) # 3. Ejecutar el proceso completo de conciliación y copia copy_data_manual_reconciliation(src, dst, table_name) 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) 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()