#!/usr/bin/env python3 """ Script para sincronizar tablas específicas desde AWS hacia servidor local. - Sincroniza: asignacion_integracion, alertas, moviles, dispositivos - Actualiza, borra o agrega registros según claves primarias - Mantiene integridad referencial entre tablas relacionadas """ import mysql.connector import logging import sys import time from mysql.connector import errorcode # Configuración de conexiones SOURCE_CONFIG = { # Origen AWS RDS 'host': 'base220.c8lcuo0a2bu6.us-east-1.rds.amazonaws.com', 'user': 'gps', 'password': 'q1w2e3r4', 'database': 'gps', 'charset': 'utf8mb4', 'collation': 'utf8mb4_unicode_ci', 'autocommit': True, 'sql_mode': 'TRADITIONAL', 'connect_timeout': 60 } DEST_CONFIG = { # Servidor local MariaDB 10.4.10 #'host': '127.0.0.1', #'user': 'root', #'password': '', 'host': '10.2.12.220', 'user': 'gps', 'password': 'q1w2e3r4', 'database': 'gps', 'charset': 'utf8mb4', 'collation': 'utf8mb4_unicode_ci', 'autocommit': True, 'sql_mode': 'TRADITIONAL', 'connect_timeout': 30, 'init_command': "SET SESSION sql_mode='TRADITIONAL'" } BATCH_SIZE = 1000 LOG_FORMAT = '%(asctime)s %(levelname)s: %(message)s' logging.basicConfig(level=logging.INFO, format=LOG_FORMAT, stream=sys.stdout) logger = logging.getLogger('integrador_sync') # Definición de tablas a sincronizar con sus claves primarias y estructura TABLES_CONFIG = { 'asignacion_integracion': { 'primary_key': ['IdAsignacionIntegracion'], 'columns': ['IdAsignacionIntegracion', 'AsgIntIdMovil', 'AsgIntIdEmpresa', 'AsgIdUsuarioAlta', 'AsgFechaAlta'] }, 'alertas': { 'primary_key': ['IdAlerta'], 'columns': ['IdAlerta', 'Alerta_DispID', 'Alerta_FechaGPS', 'Alerta_FechaServidor', 'Alerta_EventoDescripcion', 'Alerta_Evento', 'Alerta_Latitud', 'Alerta_Longitud', 'Alerta_Velocidad', 'Alerta_IdReporte', 'Alerta_ReporteCompleto', 'Alerta_Leido', 'Alerta_FechaLeido', 'Alerta_Direccion', 'Alerta_BloqueadaPor'] }, 'moviles': { 'primary_key': ['IdMovil'], 'columns': ['IdMovil', 'Movil_IdCliente', 'Movil_Tipo', 'Movil_Dominio', 'Movil_Nombre', 'Movil_Marca', 'Movil_Modelo', 'Movil_Anio', 'Movil_Estado', 'Movil_Observaciones', 'Movil_Icono', 'Movil_FechaAlta', 'Movil_UsuAlta', 'Movil_FechaUltRep', 'Movil_IdMovilOtroSistema', 'Movil_Distancia', 'Movil_Servicio', 'Movil_FechaInstalacion', 'Movil_Empresa', 'Movil_FechaDesinstalacion', 'Movil_IdCateqMovil'] }, 'dispositivos': { 'primary_key': ['IdDispositivo'], 'columns': ['IdDispositivo', 'Disp_Marca', 'Disp_Modelo', 'Disp_ID', 'Disp_Linea', 'Disp_Linea2', 'Disp_NumeroSerie', 'Disp_Sim1NumeroSerie', 'Disp_Sim2NumeroSerie', 'Disp_EmpresaLinea1', 'Disp_EmpresaLinea2', 'Disp_Estado', 'Disp_Observaciones', 'Disp_FechaAlta', 'Disp_UsuAlta', 'Disp_AccesorioCAN'] }, 'clientes': { 'primary_key': ['IdCliente'], 'columns': ['IdCliente', 'Clie_RazonSocial', 'Clie_TipoDocumento', 'Clie_Documento', 'Clie_Pais', 'Clie_Provincia', 'Clie_Direccion', 'Clie_Localidad', 'Clie_CodPostal', 'Clie_Provincia2', 'Clie_Direccion2', 'Clie_Localidad2', 'Clie_CodPostal2', 'Clie_Tel', 'Clie_Tel2', 'Clie_Email', 'Clie_Estado', 'Clie_Observaciones', 'Clie_FechaAlta', 'Clie_UsuAlta', 'Clie_NumeroOtroSistema', 'Clie_Flota', 'Clie_lbutton', 'Clie_Empresa', 'Clie_ClaveVisualizacion', 'Clie_CodSanJuanIntegrador'] }, 'control_alertas': { 'primary_key': ['IdControlAlerta'], 'columns': ['IdControlAlerta', 'ControlAlerta_Nombre', 'ControlAlerta_IdCliente', 'ControlAlerta_IdInforme', 'ControlAlerta_IdUsuario', 'ControlAlerta_Sonido', 'ControlAlerta_FechaCreacion', 'ControlAlerta_RangoDesde', 'ControlAlerta_RangoHasta', 'ControlAlerta_Estado', 'ControlAlerta_Dias', 'ControlAlerta_HastaLeido', 'ControlAlerta_Combinado'] }, 'criterios': { 'primary_key': ['IdCriterio'], 'columns': ['IdCriterio', 'Crit_IdTipoAlerta', 'Crit_Nombre', 'Crit_Val1', 'Crit_Val2', 'Crit_Val3', 'Crit_Val4_IdZona', 'Crit_Cliente', 'Crit_Usuario', 'Crit_Fecha', 'Crit_Accion'] }, 'destinatarios': { 'primary_key': ['IdDestinatario'], 'columns': ['IdDestinatario', 'Destinatario_Nombre', 'Destinatario_Email', 'Destinatario_IdCliente', 'Destinatario_IdUsuario', 'Destinatario_Estado'] }, 'dispositivos_configuracion': { 'primary_key': ['IdDispConf'], 'columns': ['IdDispConf', 'DispConf_IdDispositivo', 'DispConf_DispID', 'DispConf_Texto', 'DispConf_NroEvento', 'DispConf_Alarma_IdTipoEvento', 'DispConf_Prioridad', 'DispConf_Sonido', 'DispConf_Esconder'] }, 'asig_estadistico_destinatario': { 'primary_key': ['IdAsigEstDest'], 'columns': ['IdAsigEstDest', 'AsigEstDestIdEstadistico', 'AsigEstDestDestinatario'] }, 'asig_estadistico_movil': { 'primary_key': ['IdAsigEstMovil'], 'columns': ['IdAsigEstMovil', 'AsigEstMovilIdEstadistico', 'AsigEstMovilIdMovil'] }, 'asig_movil_sensor': { 'primary_key': ['id'], 'columns': ['id', 'movil', 'sensor'] }, 'asignacion_controles': { 'primary_key': ['IdAsignacionControles'], 'columns': ['IdAsignacionControles', 'AsignacionControles_IdControl', 'AsignacionControles_IdMovil', 'AsignacionControles_IdCriterio', 'AsignacionControles_IdInforme'] }, 'asignacion_destinatarios': { 'primary_key': ['IdAsigDest'], 'columns': ['IdAsigDest', 'AsignacionDestinatarios_IdControl', 'AsignacionDestinatarios_IdDestinatario'] }, 'asignacion_dispositivo': { 'primary_key': ['IdAsignacionDispositivo'], 'columns': ['IdAsignacionDispositivo', 'AsigDisp_IdMovil', 'AsigDisp_IdDispositivo', 'AsigDisp_Estado', 'AsigDisp_FechaAlta', 'AsigDisp_FechaBaja', 'AsigDisp_TecnicoAlta', 'AsigDisp_TecnicoBaja', 'AsigDisp_DetalleAlta', 'AsigDisp_DetalleBaja'] }, 'asignacion_emails_conductor': { 'primary_key': ['IdAsignacion'], 'columns': ['IdAsignacion', 'Asignacion_IdConductor', 'Asignacion_Email'] }, 'asignacion_emails_vencimientos': { 'primary_key': ['IdAsignacion'], 'columns': ['IdAsignacion', 'Asignacion_IdDatosMoviles', 'Asignacion_Email'] }, 'asignacion_movil_conductor': { 'primary_key': ['IdAsignacionMovilConductor'], 'columns': ['IdAsignacionMovilConductor', 'AsignacionMovCond_IdConductor', 'AsignacionMovCond_IdMovil', 'AsignacionMovCond_Idbutton', 'AsignacionMovCond_FechaAlta', 'AsignacionMovCond_FechaBaja', 'AsignacionMovCond_UsuaAlta', 'AsignacionMovCond_Cliente'] }, 'asignacion_moviles_usuario': { 'primary_key': ['IdAsigMovUsu'], 'columns': ['IdAsigMovUsu', 'IdMovil', 'IdUsuario'] }, 'asignacion_servicios': { 'primary_key': ['IdAsigServicio'], 'columns': ['IdAsigServicio', 'AsigServ_IdCliente', 'AsigServ_IdServicio'] }, 'autorizados_clientes': { 'primary_key': ['IdAu'], 'columns': ['IdAu', 'Au_IdCliente', 'Au_IdMovil', 'Au_Dominio', 'Au_Nombre', 'Au_Apellido', 'Au_Telefono', 'Au_Mail', 'Au_Pin', 'Au_IDAUTOR', 'Au_EnvioMail'] }, 'integradores_configuracion': { 'primary_key': ['IdConfIntegrador'], 'columns': ['IdConfIntegrador', 'ConfInEmpresa', 'ConfTipo', 'ConfHost', 'ConfPuerto', 'ConfUsuApi', 'ConfPassApi', 'ConfEstado', 'ConfNombreVariable', 'ConfNombreClase', 'ConfNombreTxtField', 'ConfObservacion', 'ConfNumeroModulo'] }, 'integradores_empresas': { 'primary_key': ['IdInEmpresa'], 'columns': ['IdInEmpresa', 'InEmpresaNombre', 'InEmpresaClave', 'InEmpresaFechaToken', 'InEmpresaToken'] } } def get_table_structure(conn, table_name): """Obtiene la estructura completa de una tabla.""" cursor = conn.cursor() try: cursor.execute(f"DESCRIBE `{table_name}`") columns = [] for row in cursor.fetchall(): columns.append({ 'name': row[0], 'type': row[1], 'null': row[2], 'key': row[3], 'default': row[4], 'extra': row[5] }) return columns finally: cursor.close() def create_table_if_not_exists(dest_conn, src_conn, table_name): """Crea la tabla en destino si no existe, copiando la estructura del origen.""" dest_cursor = dest_conn.cursor() src_cursor = src_conn.cursor() try: # Verificar si la tabla existe en destino dest_cursor.execute(f""" SELECT COUNT(*) FROM information_schema.tables WHERE table_schema = %s AND table_name = %s """, (dest_conn.database, table_name)) if dest_cursor.fetchone()[0] == 0: logger.info(f"Creando tabla '{table_name}' en destino...") # Obtener CREATE TABLE del origen src_cursor.execute(f"SHOW CREATE TABLE `{table_name}`") create_sql = src_cursor.fetchone()[1] dest_cursor.execute(create_sql) dest_conn.commit() logger.info(f"Tabla '{table_name}' creada exitosamente.") else: logger.info(f"Tabla '{table_name}' ya existe en destino.") except mysql.connector.Error as err: logger.error(f"Error creando tabla '{table_name}': {err}") raise finally: dest_cursor.close() src_cursor.close() def get_all_records(conn, table_name, columns): """Obtiene todos los registros de una tabla.""" cursor = conn.cursor() try: columns_str = ', '.join([f"`{col}`" for col in columns]) sql = f"SELECT {columns_str} FROM `{table_name}`" cursor.execute(sql) return cursor.fetchall() finally: cursor.close() def build_where_clause(primary_keys, record, columns): """Construye la cláusula WHERE para las claves primarias.""" where_parts = [] values = [] for pk in primary_keys: pk_index = columns.index(pk) where_parts.append(f"`{pk}` = %s") values.append(record[pk_index]) return " AND ".join(where_parts), values def sync_table(src_conn, dest_conn, table_name, config): """Sincroniza una tabla completa desde origen a destino.""" logger.info(f"\n=== Iniciando sincronización de tabla '{table_name}' ===") start_time = time.time() # Verificar estructura real de la tabla en origen logger.info(f"Verificando estructura de '{table_name}' en origen...") src_structure = get_table_structure(src_conn, table_name) src_columns = [col['name'] for col in src_structure] logger.info(f"Columnas reales en origen: {src_columns}") # Crear tabla si no existe create_table_if_not_exists(dest_conn, src_conn, table_name) # Verificar estructura real de la tabla en destino logger.info(f"Verificando estructura de '{table_name}' en destino...") dest_structure = get_table_structure(dest_conn, table_name) dest_columns = [col['name'] for col in dest_structure] logger.info(f"Columnas reales en destino: {dest_columns}") # Ajustar configuración a las columnas reales real_columns = [col for col in config['columns'] if col in src_columns and col in dest_columns] if len(real_columns) != len(config['columns']): logger.warning(f"Ajustando columnas de configuración a columnas reales.") logger.warning(f"Configuradas: {config['columns']}") logger.warning(f"Reales disponibles: {real_columns}") config = dict(config) # Hacer una copia config['columns'] = real_columns # Obtener todos los datos del origen logger.info(f"Obteniendo datos del origen...") src_data = get_all_records(src_conn, table_name, config['columns']) logger.info(f"Registros en origen: {len(src_data)}") # Obtener todos los datos del destino logger.info(f"Obteniendo datos del destino...") dest_data = get_all_records(dest_conn, table_name, config['columns']) logger.info(f"Registros en destino: {len(dest_data)}") # Crear diccionarios para comparación rápida src_dict = {} dest_dict = {} # Construir diccionarios con claves primarias for record in src_data: key_values = [] for pk in config['primary_key']: pk_index = config['columns'].index(pk) key_values.append(record[pk_index]) key = tuple(key_values) src_dict[key] = record for record in dest_data: key_values = [] for pk in config['primary_key']: pk_index = config['columns'].index(pk) key_values.append(record[pk_index]) key = tuple(key_values) dest_dict[key] = record # Determinar operaciones necesarias to_insert = [] # Registros que están en origen pero no en destino to_update = [] # Registros que están en ambos pero son diferentes # Comparar registros for key, src_record in src_dict.items(): if key not in dest_dict: to_insert.append(src_record) elif src_record != dest_dict[key]: to_update.append(src_record) logger.info(f"Operaciones necesarias - Insertar: {len(to_insert)}, Actualizar: {len(to_update)}") dest_cursor = dest_conn.cursor() try: # Ejecutar inserciones if to_insert: logger.info(f"Insertando {len(to_insert)} registros...") columns_str = ', '.join([f"`{col}`" for col in config['columns']]) placeholders = ', '.join(['%s'] * len(config['columns'])) insert_sql = f"INSERT INTO `{table_name}` ({columns_str}) VALUES ({placeholders})" logger.info(f"SQL INSERT: {insert_sql}") logger.info(f"Columnas a insertar: {config['columns']}") for i in range(0, len(to_insert), BATCH_SIZE): batch = to_insert[i:i + BATCH_SIZE] if batch: logger.info(f"Insertando lote con {len(batch)} registros") logger.info(f"Primer registro del lote: {batch[0]}") dest_cursor.executemany(insert_sql, batch) dest_conn.commit() logger.info(f"Insertados {min(i + BATCH_SIZE, len(to_insert))}/{len(to_insert)} registros") # Ejecutar actualizaciones if to_update: logger.info(f"Actualizando {len(to_update)} registros...") set_clause = ', '.join([f"`{col}` = %s" for col in config['columns']]) for record in to_update: # Construir WHERE clause para claves primarias where_clause, where_values = build_where_clause(config['primary_key'], record, config['columns']) update_sql = f"UPDATE `{table_name}` SET {set_clause} WHERE {where_clause}" logger.info(f"SQL UPDATE: {update_sql}") logger.info(f"Valores UPDATE: {list(record) + where_values}") # Los valores para SET + los valores para WHERE update_values = list(record) + where_values dest_cursor.execute(update_sql, update_values) dest_conn.commit() logger.info(f"Actualizados {len(to_update)} registros") end_time = time.time() duration = end_time - start_time logger.info(f"=== Sincronización de '{table_name}' completada en {duration:.2f}s ===") return { 'inserted': len(to_insert), 'updated': len(to_update), 'duration': duration } except mysql.connector.Error as err: logger.error(f"Error sincronizando tabla '{table_name}': {err}") dest_conn.rollback() raise finally: dest_cursor.close() def main(): """Función principal que ejecuta la sincronización de todas las tablas.""" logger.info("=== INICIANDO SINCRONIZACIÓN DE INTEGRADORES ===") overall_start = time.time() try: # Conectar a origen (AWS RDS) logger.info("Conectando a origen (AWS RDS)...") try: src_conn = mysql.connector.connect(**SOURCE_CONFIG) logger.info("Conectado al origen exitosamente.") except mysql.connector.Error as err: logger.error(f"Error conectando al origen: {err}") if "Character set" in str(err): # Fallback sin charset específico source_config_fallback = SOURCE_CONFIG.copy() del source_config_fallback['charset'] del source_config_fallback['collation'] logger.info("Reintentando conexión al origen sin charset específico...") src_conn = mysql.connector.connect(**source_config_fallback) logger.info("Conectado al origen con configuración alternativa.") else: raise # Conectar a destino (MariaDB local) logger.info("Conectando a destino (MariaDB local)...") try: dest_conn = mysql.connector.connect(**DEST_CONFIG) logger.info("Conectado al destino exitosamente.") except mysql.connector.Error as err: logger.error(f"Error conectando al destino: {err}") if "Character set" in str(err): # Fallback sin charset específico para MariaDB dest_config_fallback = DEST_CONFIG.copy() del dest_config_fallback['charset'] del dest_config_fallback['collation'] if 'init_command' in dest_config_fallback: del dest_config_fallback['init_command'] logger.info("Reintentando conexión al destino sin charset específico...") dest_conn = mysql.connector.connect(**dest_config_fallback) logger.info("Conectado al destino con configuración alternativa.") else: raise # Resumen de operaciones total_stats = { 'inserted': 0, 'updated': 0, 'tables_processed': 0 } # Sincronizar cada tabla en orden (respetando integridad referencial) # Orden: clientes -> dispositivos -> moviles -> alertas -> control_alertas -> criterios -> destinatarios -> dispositivos_configuracion -> asignaciones -> integradores sync_order = [ 'clientes', # Base: no depende de otras 'integradores_empresas', # Base: configuración de empresas integradoras 'dispositivos', # Depende de clientes indirectamente 'moviles', # Depende de clientes 'destinatarios', # Depende de clientes 'criterios', # Depende de clientes 'control_alertas', # Depende de clientes 'dispositivos_configuracion', # Depende de dispositivos 'integradores_configuracion', # Depende de integradores_empresas 'alertas', # Depende de dispositivos/moviles 'asignacion_integracion', # Depende de moviles 'asignacion_servicios', # Depende de clientes 'autorizados_clientes', # Depende de clientes y moviles 'asig_estadistico_destinatario', # Depende de destinatarios 'asig_estadistico_movil', # Depende de moviles 'asig_movil_sensor', # Depende de moviles 'asignacion_controles', # Depende de control_alertas 'asignacion_destinatarios', # Depende de destinatarios y control_alertas 'asignacion_dispositivo', # Depende de moviles y dispositivos 'asignacion_movil_conductor', # Depende de moviles y conductores 'asignacion_moviles_usuario', # Depende de moviles y usuarios 'asignacion_emails_conductor', # Depende de conductores 'asignacion_emails_vencimientos' # Depende de datos moviles ] for table_name in sync_order: if table_name in TABLES_CONFIG: try: stats = sync_table(src_conn, dest_conn, table_name, TABLES_CONFIG[table_name]) total_stats['inserted'] += stats['inserted'] total_stats['updated'] += stats['updated'] total_stats['tables_processed'] += 1 except Exception as e: logger.error(f"Error procesando tabla '{table_name}': {e}") # Continúa con la siguiente tabla continue # Mostrar resumen final total_time = time.time() - overall_start logger.info(f"\n=== RESUMEN FINAL ===") logger.info(f"Tablas procesadas: {total_stats['tables_processed']}") logger.info(f"Total insertados: {total_stats['inserted']}") logger.info(f"Total actualizados: {total_stats['updated']}") logger.info(f"Tiempo total: {total_time:.2f}s") logger.info("=== SINCRONIZACIÓN COMPLETADA ===") except mysql.connector.Error as err: if err.errno == mysql.connector.errorcode.ER_ACCESS_DENIED_ERROR: logger.error("Error de autenticación: verifica usuario y contraseña.") elif err.errno == mysql.connector.errorcode.ER_BAD_DB_ERROR: logger.error("La base de datos no existe.") else: logger.error(f"Error de MySQL: {err}") sys.exit(1) except Exception as e: logger.error(f"Error inesperado: {e}") sys.exit(1) finally: # Cerrar conexiones if 'src_conn' in locals() and src_conn.is_connected(): src_conn.close() logger.info("Conexión de origen cerrada.") if 'dest_conn' in locals() and dest_conn.is_connected(): dest_conn.close() logger.info("Conexión de destino cerrada.") if __name__ == '__main__': main()