#!/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. """ 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': 'base220.c8lcuo0a2bu6.us-east-1.rds.amazonaws.com', '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' # Puedes cambiar logging.INFO a logging.DEBUG para ver las consultas SQL en el log 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 def get_table_list(conn): """ Obtiene la lista de todas las tablas EXCEPTO las que empiezan con 'LOG_' del esquema dado, ordenadas alfabéticamente. Mantiene el casing original de los nombres de tabla. """ cursor = conn.cursor() # Cambiado a NOT LIKE 'LOG_%%' para excluir las tablas LOG_ sql_query = "SELECT table_name FROM information_schema.tables WHERE table_schema = %s AND table_name NOT LIKE 'LOG_%%' ORDER BY table_name ASC" logger.debug(f"Ejecutando SQL: {sql_query} con params: {(conn.database,)}") cursor.execute(sql_query, (conn.database,)) # No convertir a mayúsculas para tablas generales tables = [row[0] for row in cursor] cursor.close() return tables 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 create_migration_table_if_not_exists(conn): """ Crea la tabla 'migracion' en la base de datos de origen si no existe. Incluye la columna 'tablamigrada'. """ 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, `ultimoidcopiado` BIGINT(20) DEFAULT NULL, -- Cambiado de ultidreporte a ultimoidcopiado `fechaupdate` DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, `tablamigrada` TINYINT(1) DEFAULT 0, -- Nueva columna para indicar si la tabla está completamente migrada 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 update_migration_progress(src_conn, table_name, last_copied_id, is_completed=False): """ Inserta o actualiza el progreso de la copia de una tabla en la tabla 'migracion'. El parámetro `is_completed` indica si la tabla ya no tiene más datos para pasar. """ # Usar 'ultimoidcopiado' en lugar de 'ultidreporte' insert_update_sql = f""" INSERT INTO `{MIGRATION_TABLE_NAME}` (`tabla`, `ultimoidcopiado`, `tablamigrada`) VALUES (%s, %s, %s) ON DUPLICATE KEY UPDATE `ultimoidcopiado` = %s, `fechaupdate` = CURRENT_TIMESTAMP, `tablamigrada` = %s; """ cursor = src_conn.cursor() try: completed_flag = 1 if is_completed else 0 params = (table_name, last_copied_id, completed_flag, last_copied_id, completed_flag) logger.debug(f"Ejecutando SQL: {insert_update_sql} con params: {params}") cursor.execute(insert_update_sql, params) src_conn.commit() logger.debug(f"Progreso guardado para {table_name}: Último ID copiado {last_copied_id}, Completada: {is_completed}") 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_migration_status_from_tracker(src_conn, table_name): """ Obtiene el estado de migración (ultimoidcopiado y tablamigrada) de la tabla 'migracion' para una tabla dada. Retorna (ultimoidcopiado, tablamigrada) o (None, None) si no se encuentra la entrada. """ cursor = src_conn.cursor() try: # Usar 'ultimoidcopiado' en lugar de 'ultidreporte' sql_query = f"SELECT ultimoidcopiado, tablamigrada FROM `{MIGRATION_TABLE_NAME}` WHERE tabla = %s" logger.debug(f"Ejecutando SQL: {sql_query} con params: {(table_name,)}") cursor.execute(sql_query, (table_name,)) result = cursor.fetchone() if result: return result[0], bool(result[1]) # Retorna ultimoidcopiado y el booleano de tablamigrada return None, None except mysql.connector.Error as err: logger.error(f"Error al consultar el estado de migración para {table_name}: {err}") return None, None finally: cursor.close() def get_primary_key_info(conn, table): """ Obtiene el nombre de la columna de la clave primaria y verifica si es auto-incremental y de tipo entero. Retorna (pk_column_name, is_auto_increment) o (None, False) si no se encuentra una PK adecuada. """ cursor = conn.cursor() try: # Query INFORMATION_SCHEMA to get primary key column name and if it's auto_increment sql_query = """ SELECT COLUMN_NAME, EXTRA FROM INFORMATION_SCHEMA.KEY_COLUMN_USAGE WHERE TABLE_SCHEMA = %s AND TABLE_NAME = %s AND CONSTRAINT_NAME = 'PRIMARY' """ logger.debug(f"Ejecutando SQL: {sql_query} con params: {(conn.database, table)}") cursor.execute(sql_query, (conn.database, table)) pk_info = cursor.fetchone() if pk_info: pk_column_name = pk_info[0] is_auto_increment = 'auto_increment' in pk_info[1].lower() if pk_info[1] else False # Further check: ensure it's an integer type for comparison col_type_query = f""" SELECT DATA_TYPE FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_SCHEMA = %s AND TABLE_NAME = %s AND COLUMN_NAME = %s """ logger.debug(f"Ejecutando SQL: {col_type_query} con params: {(conn.database, table, pk_column_name)}") cursor.execute(col_type_query, (conn.database, table, pk_column_name)) col_type = cursor.fetchone() if col_type and col_type[0] in ['tinyint', 'smallint', 'mediumint', 'int', 'bigint']: return pk_column_name, is_auto_increment else: logger.warning(f"La clave primaria '{pk_column_name}' de la tabla '{table}' no es un tipo entero adecuado para el seguimiento incremental.") return None, False return None, False except mysql.connector.Error as err: logger.error(f"Error al obtener información de la clave primaria para {table}: {err}") return None, False finally: cursor.close() def get_max_pk_value(conn, table, pk_column_name): """ Obtiene el valor máximo de la columna de clave primaria dada de una tabla. Retorna 0 si la tabla está vacía o si pk_column_name es None. """ if not pk_column_name: return 0 # No PK column to track cursor = conn.cursor() try: sql_query = f"SELECT MAX(`{pk_column_name}`) 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.error(f"Error al obtener MAX({pk_column_name}) de la tabla {table}: {err}") return 0 finally: cursor.close() def create_table_on_dest(src_conn, dest_conn, table): """ Obtiene la sentencia CREATE TABLE de la tabla en el origen y la ejecuta en el destino. Esto copia la estructura, índices y propiedades exactas. Retorna True si la creación fue exitosa, False en caso contrario. """ src_cursor = src_conn.cursor() dest_cursor = dest_conn.cursor() try: # Obtener la sentencia CREATE TABLE del origen show_create_sql = """ CREATE TABLE `{table}` ( `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`), UNIQUE 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;""" logger.debug(f"Ejecutando SQL en origen: {show_create_sql}") src_cursor.execute(show_create_sql) result = src_cursor.fetchone() if not result: logger.error(f"No se pudo obtener la sentencia CREATE TABLE para {table} en el origen.") return False # Falló la obtención del DDL create_table_statement = result[1] # La sentencia CREATE TABLE está en la segunda columna # Ejecutar la sentencia CREATE TABLE obtenida en el destino logger.info(f"Creando tabla destino '{table}' con esquema obtenido del origen.") logger.debug(f"Ejecutando SQL en destino: {create_table_statement}") dest_cursor.execute(create_table_statement) dest_conn.commit() logger.info(f"Tabla '{table}' creada exitosamente en el destino.") return True except mysql.connector.Error as err: logger.error(f"Error MySQL al crear la tabla {table} en destino: {err}") return False except Exception as e: logger.error(f"Error inesperado al crear la tabla {table} en destino: {e}") return False finally: src_cursor.close() dest_cursor.close() def copy_data(src_conn, dest_conn, table): # Obtener información de la clave primaria para la tabla actual pk_column_name, is_auto_increment_pk = get_primary_key_info(src_conn, table) if not pk_column_name or not is_auto_increment_pk: logger.warning(f"La tabla '{table}' no tiene una clave primaria auto-incremental de tipo entero adecuada para la copia incremental. Saltando tabla.") return # Determinar el último valor PK en la tabla de destino para la copia incremental last_pk_dest = get_max_pk_value(dest_conn, table, pk_column_name) start_id_for_copy = last_pk_dest + 1 logger.info(f"Copiando datos de '{table}' desde {pk_column_name} >= {start_id_for_copy}.") # Obtiene total de filas para progreso desde el punto de inicio en el origen count_cursor = src_conn.cursor() # Usar el nombre de columna PK dinámico en la cláusula WHERE count_sql = f"SELECT COUNT(*) FROM `{table}` WHERE `{pk_column_name}` >= %s" logger.debug(f"Ejecutando SQL: {count_sql} con params: {(start_id_for_copy,)}") count_cursor.execute(count_sql, (start_id_for_copy,)) total_rows_to_copy = count_cursor.fetchone()[0] count_cursor.close() # Cursor para COUNT(*) cerrado. if total_rows_to_copy == 0: logger.info(f"No hay nuevos datos para copiar en la tabla {table} (desde {pk_column_name} {start_id_for_copy}).") # Actualiza el progreso con el último PK del destino. update_migration_progress(src_conn, table, last_pk_dest, is_completed=True) return logger.info(f"Copiando datos de {table}: {total_rows_to_copy} nuevas filas en total.") start_time = time.time() copied = 0 # Usar el nombre de columna PK dinámico en ORDER BY y WHERE select_sql = f"SELECT * FROM `{table}` WHERE `{pk_column_name}` >= %s ORDER BY `{pk_column_name}` ASC" src_cursor = src_conn.cursor(buffered=False) # Cursor no-buffered. try: logger.debug(f"Ejecutando SQL: {select_sql} con params: {(start_id_for_copy,)}") src_cursor.execute(select_sql, (start_id_for_copy,)) cols = [desc[0] for desc in src_cursor.description] try: # Obtener el índice de la columna de clave primaria identificada pk_col_index = cols.index(pk_column_name) except ValueError: logger.error(f"La columna de clave primaria '{pk_column_name}' no se encontró en la selección de la tabla {table}. Esto no debería ocurrir.") return placeholders = ",".join(["%s"] * len(cols)) insert_sql = f"INSERT INTO `{table}` ({', '.join(cols)}) VALUES ({placeholders})" logger.debug(f"Preparando INSERT SQL: {insert_sql}") batch = [] last_id_in_current_batch = last_pk_dest # Inicializar con el último PK del destino for row in src_cursor: batch.append(row) last_id_in_current_batch = row[pk_col_index] # Obtener el valor PK de la fila actual if len(batch) >= BATCH_SIZE: with dest_conn.cursor() as dest_cur: dest_cur.executemany(insert_sql, batch) dest_conn.commit() copied += len(batch) 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 ({pk_column_name}): {last_id_in_current_batch}. ETA: {eta:.1f}s" if eta else f" Copiadas {copied}/{total_rows_to_copy}. Último ID copiado ({pk_column_name}): {last_id_in_current_batch}.") # Procesa batch restante después de que el cursor de origen ha sido completamente leído if batch: with dest_conn.cursor() as dest_cur: dest_cur.executemany(insert_sql, batch) dest_conn.commit() copied += len(batch) except mysql.connector.Error as err: logger.error(f"Error MySQL durante la copia de datos para {table}: {err}") raise # Re-lanzar la excepción para que el main la capture except Exception as e: logger.error(f"Error inesperado durante la copia de datos para {table}: {e}") raise finally: # Asegurarse de que el cursor de origen se cierre siempre src_cursor.close() # Actualización final del progreso para esta tabla, después de que el src_cursor está cerrado is_table_completed_for_run = (copied == total_rows_to_copy) # Usar el valor PK copiado correcto para el tracker update_migration_progress(src_conn, table, last_id_in_current_batch, is_completed=is_table_completed_for_run) logger.info(f"Tabla {table} completada en {time.time() - start_time:.1f}s. Último ID copiado ({pk_column_name}): {last_id_in_current_batch}.") 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.") # Asegurar que la tabla de migración existe en el servidor de origen create_migration_table_if_not_exists(src) # Obtener la lista de tablas (excluyendo LOG_) del origen, ordenadas tables_to_process = get_table_list(src) logger.info(f"Tablas (excluyendo 'LOG_') encontradas en origen para procesar: {len(tables_to_process)}") for table in tables_to_process: logger.info(f"Procesando tabla: {table}") # Obtener información de la clave primaria para la tabla actual pk_column_name, is_auto_increment_pk = get_primary_key_info(src, table) if not pk_column_name or not is_auto_increment_pk: logger.warning(f"Tabla '{table}' no tiene una clave primaria auto-incremental de tipo entero adecuada para el seguimiento incremental. Omitiendo tabla.") continue # Saltar a la siguiente tabla # Obtener el MAX(PK) actual de la tabla en el origen y destino current_max_pk_source = get_max_pk_value(src, table, pk_column_name) current_max_pk_dest = get_max_pk_value(dst, table, pk_column_name) if current_max_pk_source > current_max_pk_dest: logger.info(f"Tabla {table}: Nuevos datos encontrados en origen (MAX PK Origen: {current_max_pk_source}, MAX PK Destino: {current_max_pk_dest}). Procediendo a copiar.") # Si la tabla no existe en el destino, la crea antes de copiar if not table_exists(dst, table): logger.info(f"Tabla {table} no existe en destino. Intentando crearla.") # create_table_on_dest ahora retorna True o False if not create_table_on_dest(src, dst, table): logger.error(f"Fallo al crear la tabla {table} en destino. Saltando a la siguiente tabla.") continue # Saltar a la siguiente tabla si la creación falló # Procede a copiar datos (la lógica incremental está dentro de copy_data) copy_data(src, dst, table) else: # current_max_pk_source <= current_max_pk_dest # La tabla está sincronizada o el destino está más actualizado (caso improbable) # Asegurarse de que el flag 'tablamigrada' en la tabla de seguimiento esté a 1 _, tracker_tablamigrada = get_migration_status_from_tracker(src, table) if tracker_tablamigrada == 1: logger.info(f"Tabla {table} está sincronizada (MAX PK Origen: {current_max_pk_source}, MAX PK Destino: {current_max_pk_dest}) y ya marcada como migrada. Omitiendo.") else: # Si no estaba marcada como completada, pero ahora lo está, actualizar el tracker. # Se usa current_max_pk_dest como el último ID copiado para el tracker. update_migration_progress(src, table, current_max_pk_dest, is_completed=True) logger.info(f"Tabla {table} ahora está sincronizada (MAX PK Origen: {current_max_pk_source}, MAX PK Destino: {current_max_pk_dest}). Marcada como migrada y omitiendo.") continue # Saltar a la siguiente tabla total_time = time.time() - overall_start logger.info(f"Proceso completado en {total_time:.1f}s.") except mysql.connector.Error as err: logger.error(f"Error de MySQL: {err}") sys.exit(1) except Exception as e: logger.error(f"Error inesperado: {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()