import mysql.connector from datetime import datetime # --- Configuración de la base de datos --- DB_CONFIG = { 'host': '10.2.12.223', 'user': 'gps', 'password': 'p0o9i8u7', 'database': 'gps_reportes', 'charset': 'utf8mb4', 'use_pure': True, } # El cotejamiento y conjunto de caracteres al que quieres convertir las columnas COLLATE_TO_USE = "utf8mb4_unicode_ci" CHARSET_TO_USE = "utf8mb4" def log_status(cursor, table_name, description): """ Inserta o actualiza un registro en la tabla arreglocolation. Si ya existe un registro para la tabla, lo actualiza. De lo contrario, lo inserta. """ now = datetime.now().strftime('%Y-%m-%d %H:%M:%S') # 1. Verificar si ya existe un registro para esta tabla check_query = "SELECT COUNT(*) FROM arreglocolation WHERE tabla = %s" cursor.execute(check_query, (table_name,)) exists = cursor.fetchone()[0] > 0 if exists: # 2. Si existe, actualizamos la fecha y la descripción update_query = "UPDATE arreglocolation SET fecha = %s, descripcion = %s WHERE tabla = %s" cursor.execute(update_query, (now, description, table_name)) print(f" -> Registro de la tabla '{table_name}' actualizado en el log.") else: # 3. Si no existe, insertamos un nuevo registro insert_query = "INSERT INTO arreglocolation (tabla, fecha, descripcion) VALUES (%s, %s, %s)" cursor.execute(insert_query, (table_name, now, description)) print(f" -> Nuevo registro de la tabla '{table_name}' insertado en el log.") def get_db_collation(cursor, db_name): """Obtiene el cotejamiento predeterminado de la base de datos.""" query = f"SELECT DEFAULT_COLLATION_NAME FROM INFORMATION_SCHEMA.SCHEMATA WHERE SCHEMA_NAME = '{db_name}'" cursor.execute(query) result = cursor.fetchone() if result: return result[0] return None def get_tables_to_fix(cursor, db_name): """Obtiene una lista de tablas que comienzan con .""" query = f"SELECT table_name FROM information_schema.tables WHERE table_schema = '{db_name}' and table_name like 'temporal_%' " cursor.execute(query) return [row[0] for row in cursor.fetchall()] def get_columns_to_fix(cursor, db_name, table_name, db_collation): """Obtiene una lista de columnas con cotejamientos incorrectos.""" query = f""" SELECT COLUMN_NAME, DATA_TYPE, CHARACTER_MAXIMUM_LENGTH FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_SCHEMA = '{db_name}' AND TABLE_NAME = '{table_name}' AND COLLATION_NAME IS NOT NULL AND COLLATION_NAME != '{db_collation}' AND DATA_TYPE IN ('char', 'varchar', 'text', 'tinytext', 'mediumtext', 'longtext'); """ cursor.execute(query) return cursor.fetchall() def fix_column_collation(cursor, table_name, column_name, data_type, max_length): """Genera y ejecuta el comando ALTER TABLE para una columna.""" # Los tipos de datos TEXT no tienen longitud máxima en el comando ALTER ''' alter_table_command = f"ALTER TABLE `{table_name}` CHARACTER SET = utf8mb4 COLLATE = utf8mb4_unicode_ci;" try: cursor.execute(alter_table_command) print(f" -> ¡Éxito! tabla '{table_name}' corregida.") return True except mysql.connector.Error as err: print(f" -> ¡Error! No se pudo corregir la tabla '{table_name}': {err}") return False ''' if data_type in ('text', 'tinytext', 'mediumtext', 'longtext'): alter_command = ( f"ALTER TABLE `{table_name}` MODIFY COLUMN `{column_name}` " f"{data_type} CHARACTER SET {CHARSET_TO_USE} COLLATE {COLLATE_TO_USE};" ) else: alter_command = ( f"ALTER TABLE `{table_name}` MODIFY COLUMN `{column_name}` " f"{data_type}({max_length}) CHARACTER SET {CHARSET_TO_USE} COLLATE {COLLATE_TO_USE};" ) print(f" Ejecutando: {alter_command}") try: cursor.execute(alter_command) print(f" -> ¡Éxito! Columna '{column_name}' corregida.") return True except mysql.connector.Error as err: print(f" -> ¡Error! No se pudo corregir la columna '{column_name}': {err}") return False def main(): """Función principal para ejecutar el proceso.""" try: print("Conectando a la base de datos...") conn = mysql.connector.connect(**DB_CONFIG) cursor = conn.cursor() db_name = DB_CONFIG['database'] db_collation = get_db_collation(cursor, db_name) if not db_collation: print("Error: No se pudo obtener el cotejamiento de la base de datos.") return print(f"Cotejamiento de la base de datos: {db_collation}") print("Buscando tablas que empiezan con 'LOG_'...") tables_to_process = get_tables_to_fix(cursor, db_name) all_log_tables = get_all_log_tables(cursor, db_name) # Obtener todas las tablas LOG_ if not all_log_tables: print("No se encontraron tablas que empiecen con 'LOG_'.") return for table_name in all_log_tables: print(f"\n-> Procesando tabla '{table_name}'...") if table_name not in tables_to_process: print(" -> La tabla ya está OK. Registrando en log...") log_status(cursor, table_name, "Tabla ya OK con colation") continue columns_to_fix = get_columns_to_fix(cursor, db_name, table_name, db_collation) if not columns_to_fix: print(" -> No se encontraron campos con cotejamientos incorrectos. Registrando en log...") log_status(cursor, table_name, "Tabla ya OK con colation") continue # Bucle sobre las columnas a corregir all_fixed = True for column_info in columns_to_fix: column_name, data_type, max_length = column_info if not fix_column_collation(cursor, table_name, column_name, data_type, max_length): all_fixed = False if all_fixed: log_status(cursor, table_name, "Arreglo colation") print(f"\n-> Tabla '{table_name}' corregida y registrada.") conn.commit() print("\nProceso finalizado. Se han guardado todos los cambios.") except mysql.connector.Error as err: print(f"Error de conexión o de la base de datos: {err}") finally: if 'conn' in locals() and conn.is_connected(): cursor.close() conn.close() print("Conexión a la base de datos cerrada.") def get_all_log_tables(cursor, db_name): """Obtiene todas las tablas que comienzan con 'LOG_'.""" query = f"SELECT table_name FROM information_schema.tables WHERE table_schema = '{db_name}' " cursor.execute(query) return [row[0] for row in cursor.fetchall()] if __name__ == "__main__": main()