""" Normalización de Base de Datos e Ingesta Incremental (UPSERT) Procesamiento en Lotes de Alta Velocidad (Bulk Processing) """ import sys import os import mysql.connector DB_CONFIG = { 'host': '127.0.0.1', 'port': 3306, 'user': 'root', 'password': '', 'database': 'procurement_analytics', 'charset': 'utf8mb4' } def extract_area_consumo(comentarios): if not comentarios: return None parts = comentarios.split(',') if len(parts) > 1: potential_area = parts[-1].strip() if len(potential_area) < 100: return potential_area return None def main(): print("============================================================") print(" INGESTA INCREMENTAL (UPSERT BULK) A ESQUEMA 3NF") print("============================================================") conn = mysql.connector.connect(**DB_CONFIG) cursor = conn.cursor(dictionary=True) print("1. Leyendo datos del nuevo lote (staging_compras)...") cursor.execute("SELECT * FROM staging_compras") rows = cursor.fetchall() print(f" {len(rows)} registros a procesar.") if not rows: return def load_catalog(table, id_col, name_col): cursor.execute(f"SELECT {id_col}, {name_col} FROM {table}") return {r[name_col].lower(): r[id_col] for r in cursor.fetchall() if r[name_col]} db_subsidiarias = load_catalog('subsidiarias', 'id_subsidiaria', 'nombre') db_usuarios = load_catalog('usuarios', 'id_usuario', 'nombre_completo') db_proveedores = load_catalog('proveedores', 'id_proveedor', 'nombre') db_familias = load_catalog('familias', 'id_familia', 'nombre') db_ubicaciones = load_catalog('ubicaciones', 'id_ubicacion', 'nombre') print("2. Procesando Catálogos en Lote...") new_catalogs = {'subsidiarias': set(), 'usuarios': set(), 'proveedores': set(), 'familias': set(), 'ubicaciones': set()} for row in rows: if row['subsidiaria'] and row['subsidiaria'].strip().lower() not in db_subsidiarias: new_catalogs['subsidiarias'].add(row['subsidiaria'].strip()) for user_field in ['solicitante', 'autoriza_req', 'comprador', 'autorizador_oc']: if row[user_field] and row[user_field].strip().lower() not in db_usuarios: new_catalogs['usuarios'].add(row[user_field].strip()) if row['proveedor'] and row['proveedor'].strip().lower() not in db_proveedores: new_catalogs['proveedores'].add(row['proveedor'].strip()) if row['familia'] and row['familia'].strip().lower() not in db_familias: new_catalogs['familias'].add(row['familia'].strip()) for ub_field in ['ubicacion_consumo', 'ubicacion_almacen', 'ubicacion_recepcion']: if row[ub_field] and row[ub_field].strip().lower() not in db_ubicaciones: new_catalogs['ubicaciones'].add(row[ub_field].strip()) if new_catalogs['subsidiarias']: cursor.executemany("INSERT INTO subsidiarias (nombre) VALUES (%s)", [(x,) for x in new_catalogs['subsidiarias']]) if new_catalogs['usuarios']: cursor.executemany("INSERT INTO usuarios (nombre_completo) VALUES (%s)", [(x,) for x in new_catalogs['usuarios']]) if new_catalogs['proveedores']: cursor.executemany("INSERT INTO proveedores (nombre) VALUES (%s)", [(x,) for x in new_catalogs['proveedores']]) if new_catalogs['familias']: cursor.executemany("INSERT INTO familias (nombre) VALUES (%s)", [(x,) for x in new_catalogs['familias']]) if new_catalogs['ubicaciones']: cursor.executemany("INSERT INTO ubicaciones (nombre) VALUES (%s)", [(x,) for x in new_catalogs['ubicaciones']]) conn.commit() # Recargar diccionarios db_subsidiarias = load_catalog('subsidiarias', 'id_subsidiaria', 'nombre') db_usuarios = load_catalog('usuarios', 'id_usuario', 'nombre_completo') db_proveedores = load_catalog('proveedores', 'id_proveedor', 'nombre') db_familias = load_catalog('familias', 'id_familia', 'nombre') db_ubicaciones = load_catalog('ubicaciones', 'id_ubicacion', 'nombre') print("3. Preparando lotes transaccionales (UPSERT)...") def g(val, db_dict): return db_dict.get(val.strip().lower()) if val and val.strip() else None # REQUISICIONES req_batch = {} for r in rows: folio = r['folio_requisicion'] if folio: req_batch[folio] = (folio, g(r['subsidiaria'], db_subsidiarias), g(r['solicitante'], db_usuarios), g(r['autoriza_req'], db_usuarios), extract_area_consumo(r['comentarios']), g(r['ubicacion_consumo'], db_ubicaciones), g(r['ubicacion_almacen'], db_ubicaciones), g(r['ubicacion_recepcion'], db_ubicaciones), r['comentarios'], r['fecha_creacion'], r['fecha_envio_autorizacion'], r['fecha_autorizacion'], r['estatus_requisicion']) if req_batch: batch_list = list(req_batch.values()) for i in range(0, len(batch_list), 500): cursor.executemany(""" INSERT INTO requisiciones (folio_requisicion, id_subsidiaria, id_solicitante, id_autorizador_req, area_consumo, id_ubicacion_consumo, id_ubicacion_almacen, id_ubicacion_recepcion, comentarios, fecha_creacion, fecha_envio_autorizacion, fecha_autorizacion, estatus_requisicion) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) ON DUPLICATE KEY UPDATE id_subsidiaria=VALUES(id_subsidiaria), id_solicitante=VALUES(id_solicitante), id_autorizador_req=VALUES(id_autorizador_req), area_consumo=VALUES(area_consumo), id_ubicacion_consumo=VALUES(id_ubicacion_consumo), id_ubicacion_almacen=VALUES(id_ubicacion_almacen), id_ubicacion_recepcion=VALUES(id_ubicacion_recepcion), comentarios=VALUES(comentarios), fecha_creacion=VALUES(fecha_creacion), fecha_envio_autorizacion=VALUES(fecha_envio_autorizacion), fecha_autorizacion=VALUES(fecha_autorizacion), estatus_requisicion=VALUES(estatus_requisicion) """, batch_list[i:i+500]) conn.commit() cursor.execute("SELECT id_requisicion, folio_requisicion FROM requisiciones") req_dict = {r['folio_requisicion']: r['id_requisicion'] for r in cursor.fetchall()} # COTIZACIONES cot_batch = {} for r in rows: folio = r['cotizacion'] if folio and str(folio).strip(): cot_batch[folio] = (folio, req_dict.get(r['folio_requisicion']), r['fecha_invitacion_proveedor'], r['fecha_finalizacion_cotizacion'], r['estatus_cotizacion']) if cot_batch: batch_list = list(cot_batch.values()) for i in range(0, len(batch_list), 500): cursor.executemany(""" INSERT INTO cotizaciones (folio_cotizacion, id_requisicion, fecha_invitacion_proveedor, fecha_finalizacion_cotizacion, estatus_cotizacion) VALUES (%s, %s, %s, %s, %s) ON DUPLICATE KEY UPDATE id_requisicion=VALUES(id_requisicion), fecha_invitacion_proveedor=VALUES(fecha_invitacion_proveedor), fecha_finalizacion_cotizacion=VALUES(fecha_finalizacion_cotizacion), estatus_cotizacion=VALUES(estatus_cotizacion) """, batch_list[i:i+500]) conn.commit() cursor.execute("SELECT id_cotizacion, folio_cotizacion FROM cotizaciones WHERE folio_cotizacion IS NOT NULL") cot_dict = {r['folio_cotizacion']: r['id_cotizacion'] for r in cursor.fetchall()} # ORDENES DE COMPRA oc_batch = {} for r in rows: folio = r['folio_orden_compra'] if folio and str(folio).strip(): oc_batch[folio] = (folio, req_dict.get(r['folio_requisicion']), cot_dict.get(r['cotizacion']), g(r['comprador'], db_usuarios), g(r['autorizador_oc'], db_usuarios), r['fecha_envio_autorizacion_oc'], r['fecha_autorizacion_oc'], r['fecha_recepcion'], r['estatus_orden_compra'], r['estatus_almacen']) if oc_batch: batch_list = list(oc_batch.values()) for i in range(0, len(batch_list), 500): cursor.executemany(""" INSERT INTO ordenes_compra (folio_oc, id_requisicion, id_cotizacion, id_comprador, id_autorizador_oc, fecha_envio_autorizacion_oc, fecha_autorizacion_oc, fecha_recepcion, estatus_orden_compra, estatus_almacen) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s) ON DUPLICATE KEY UPDATE id_requisicion=VALUES(id_requisicion), id_cotizacion=VALUES(id_cotizacion), id_comprador=VALUES(id_comprador), id_autorizador_oc=VALUES(id_autorizador_oc), fecha_envio_autorizacion_oc=VALUES(fecha_envio_autorizacion_oc), fecha_autorizacion_oc=VALUES(fecha_autorizacion_oc), fecha_recepcion=VALUES(fecha_recepcion), estatus_orden_compra=VALUES(estatus_orden_compra), estatus_almacen=VALUES(estatus_almacen) """, batch_list[i:i+500]) conn.commit() cursor.execute("SELECT id_oc, folio_oc FROM ordenes_compra WHERE folio_oc IS NOT NULL") oc_dict = {r['folio_oc']: r['id_oc'] for r in cursor.fetchall()} # LINEAS DETALLE lineas_batch = [] for r in rows: lineas_batch.append((r['id_interno_linea'], oc_dict.get(r['folio_orden_compra']), req_dict.get(r['folio_requisicion']), g(r['familia'], db_familias), g(r['proveedor'], db_proveedores), r['articulo'], r['unidad'], r['cantidad_oc'], r['cantidad_solicitada'], r['precio_ultima_compra'], r['precio_final_mxn'], r['moneda'], r['precio_final_proveedor'], r['monto_total_colocado'], r['por_var_ultimo_precio'], r['por_var_precio_max'], r['por_var_precio_min'], r['proveedor_max'], r['precio_final_max'], r['proveedor_min'], r['precio_final_min'])) # Procesar lineas_detalle en bloques de 500 para no ahogar la memoria ni exceder el max_allowed_packet de MySQL batch_size = 500 for i in range(0, len(lineas_batch), batch_size): cursor.executemany(""" INSERT INTO lineas_detalle (id_interno_linea, id_oc, id_requisicion, id_familia, id_proveedor, articulo, unidad, cantidad_oc, cantidad_solicitada, precio_ultima_compra, precio_final_mxn, moneda, precio_final_proveedor, monto_total_colocado, por_var_ultimo_precio, por_var_precio_max, por_var_precio_min, proveedor_max, precio_final_max, proveedor_min, precio_final_min) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) ON DUPLICATE KEY UPDATE id_oc=VALUES(id_oc), id_requisicion=VALUES(id_requisicion), id_familia=VALUES(id_familia), id_proveedor=VALUES(id_proveedor), articulo=VALUES(articulo), unidad=VALUES(unidad), cantidad_oc=VALUES(cantidad_oc), cantidad_solicitada=VALUES(cantidad_solicitada), precio_ultima_compra=VALUES(precio_ultima_compra), precio_final_mxn=VALUES(precio_final_mxn), moneda=VALUES(moneda), precio_final_proveedor=VALUES(precio_final_proveedor), monto_total_colocado=VALUES(monto_total_colocado), por_var_ultimo_precio=VALUES(por_var_ultimo_precio), por_var_precio_max=VALUES(por_var_precio_max), por_var_precio_min=VALUES(por_var_precio_min), proveedor_max=VALUES(proveedor_max), precio_final_max=VALUES(precio_final_max), proveedor_min=VALUES(proveedor_min), precio_final_min=VALUES(precio_final_min) """, lineas_batch[i:i+batch_size]) conn.commit() print("4. Refrescando tabla materializada para el Dashboard...") cursor.execute("CALL RefreshMaterializedView()") print("============================================================") print(" OK Proceso de Sincronizacion UPSERT BULK Completado!") print("============================================================") cursor.close() conn.close() if __name__ == '__main__': main()