# modules/ocupabilidad/processor.py from datetime import datetime import unicodedata class CursoProcessor: """Procesador de datos de cursos - Aplica transformaciones y cálculos""" def __init__(self, data_manager): self.data_manager = data_manager # --- COLUMNAS REORDENADAS Y REORGANIZADAS --- self.columnas = [ 'programa_frecuencia', 'fch_inicio', 'dias_para_inicio', 'Inscritos_Mes', 'Inscritos_Totales', 'Retirados', 'Inscritos_Activos', 'Inscritos_PC', 'Descuento', # Esto es 'INSCRITOS REFRIPERU' 'Inscritos_Continuidad', 'Meta_Curso', 'Avance_Inscritos' ] def obtener_datos_procesados(self, ano, mes): try: # 1. Cargar datos básicos datos_matriculas = self.data_manager.cargar_datos_matriculas(ano, mes) datos_raw_matriculas = self.data_manager.ejecutar_consulta_matriculados_detalle(ano, mes) datos_originales = self.data_manager.ejecutar_consulta_cursos(ano, mes) # --- NUEVO: EL GUARDIÁN DE FECHAS (SUPABASE) --- fechas_historicas = self.data_manager.obtener_fechas_originales() indices_en_sql = set() for fila in datos_originales: indice = str(fila.get('num_indice', '')).strip() fecha_actual = fila.get('fch_inicio') programa = fila.get('dsc_programa', '') if indice and fecha_actual: indices_en_sql.add(indice) # Registramos que sí vino en SQL este mes if indice not in fechas_historicas: fecha_str = fecha_actual.strftime('%d/%m/%Y') if isinstance(fecha_actual, datetime) else str(fecha_actual) exito = self.data_manager.guardar_fecha_original(indice, programa, fecha_str) if exito: fechas_historicas[indice] = {'fecha': fecha_str, 'programa': programa} # ----------------------------------------------- # 2. CONTINUIDAD lista_dnis = [] if datos_raw_matriculas: for alumno in datos_raw_matriculas: dni = alumno.get('dsc_documento') if dni: lista_dnis.append(dni) datos_historial = self.data_manager.consultar_historial_continuidad(lista_dnis) # 3. Procesar todo datos_procesados = self.aplicar_personalizaciones( datos_originales, datos_matriculas, datos_raw_matriculas, datos_historial, ano, mes ) # --- INYECTAR CURSOS REPROGRAMADOS (FILAS FANTASMAS) --- if getattr(self, 'mostrar_reprogramados', True): for idx, info in fechas_historicas.items(): if idx not in indices_en_sql: # Si el curso NO vino este mes desde SQL... # Aseguramos compatibilidad si guardaste data vieja como texto o la nueva como diccionario if isinstance(info, dict): fecha_orig = info.get('fecha', '') prog_orig = info.get('programa', 'CURSO REPROGRAMADO') else: fecha_orig = str(info) prog_orig = 'CURSO REPROGRAMADO' # Verificamos si la fecha original correspondía a ESTE mes Y AÑO filtrado mes_orig = "" ano_orig = "" if "/" in fecha_orig: partes = fecha_orig.split("/") if len(partes) >= 2: mes_orig = str(int(partes[1])) if len(partes) >= 3: ano_orig = str(int(partes[2])) elif "-" in fecha_orig: partes = fecha_orig.split("-") if len(partes) >= 2: mes_orig = str(int(partes[1])) if len(partes) >= 3: ano_orig = str(int(partes[2])) # Mes debe coincidir; y si la fecha tiene año, el año también debe coincidir. # Si la fecha NO tiene año (formato viejo dd/mm), no se inyecta para evitar # mostrarla en años equivocados. coincide_mes = (mes_orig == str(int(mes))) coincide_ano = (ano_orig == str(int(ano))) if ano_orig else False if coincide_mes and coincide_ano: # Inyectamos la fila falsa para alertar en pantalla datos_procesados.append({ 'num_indice': idx, 'programa_frecuencia': prog_orig, 'dsc_programa': prog_orig, 'fch_inicio': fecha_orig, 'dias_para_inicio': 'REPROGRAMADO', 'Inscritos_Mes': "-", 'Inscritos_Totales': "-", 'Retirados': "-", 'Inscritos_Activos': "-", 'Inscritos_PC': "-", 'Descuento': "-", 'Inscritos_Continuidad': "-", 'Meta_Curso': "-", 'Avance_Inscritos': "-" }) # ----------------------------------------------- ## --- ORDENAR: NORMALES POR FECHA, REPROGRAMADOS POR SEDE --- def parse_fecha_orden(fecha_str): if not fecha_str or fecha_str == "-": return datetime.max s = str(fecha_str).replace("/", "-").strip() try: if len(s) == 5: return datetime.strptime(f"{s}-{ano}", "%d-%m-%Y") if len(s) >= 10: return datetime.strptime(s[:10], "%d-%m-%Y") except: pass return datetime.max def obtener_orden_sede(programa): prog_upper = str(programa).upper() if 'LIMA' in prog_upper: return 1 if 'PIURA' in prog_upper: return 2 if 'TRUJILLO' in prog_upper: return 3 if 'AREQUIPA' in prog_upper: return 4 return 5 def logica_ordenamiento(x): if x.get('dias_para_inicio') == 'REPROGRAMADO': # Reprogramados: Van al final (1), ordenados por sede, luego alfabéticamente return (1, obtener_orden_sede(x.get('dsc_programa', '')), x.get('dsc_programa', '')) else: # Normales: Van al inicio (0), ordenados por fecha return (0, 0, parse_fecha_orden(x.get('fch_inicio', ''))) # Aplicamos el ordenamiento inteligente datos_procesados.sort(key=logica_ordenamiento) return datos_procesados except Exception as e: print(f"❌ Error obteniendo datos procesados: {e}") return [] def normalizar_texto(self, texto): if not texto: return "" texto = str(texto).upper().strip() texto = unicodedata.normalize('NFD', texto) texto = texto.encode('ascii', 'ignore').decode("utf-8") return texto def obtener_categoria_programa(self, nombre_programa): try: nombre_limpio = self.normalizar_texto(nombre_programa) clasificaciones = self.data_manager.meta_data.get('clasificacion_programas', {}) for categoria, config in clasificaciones.items(): patrones = config.get('patrones', []) for patron in patrones: patron_limpio = self.normalizar_texto(patron) if patron_limpio in nombre_limpio: return categoria return "OTROS" except: return "OTROS" def aplicar_personalizaciones(self, datos_originales, datos_matriculas, datos_raw_matriculas, datos_historial, ano_filtro, mes_filtro): print("🎯 Aplicando personalizaciones...") cursos_personalizados = self.data_manager.config_data.get('cursos_personalizados', {}) datos_filtrados = [] try: ano_target = int(ano_filtro) mes_target = int(mes_filtro) except ValueError: ano_target = 0 mes_target = 0 # 1. FILTRADO for curso in datos_originales: num_indice = str(curso.get('num_indice', '')) if num_indice in cursos_personalizados: personalizacion = cursos_personalizados[num_indice] if 'fch_inicio' in personalizacion: curso['fch_inicio'] = personalizacion['fch_inicio'] if 'flg_activo' in personalizacion: curso['flg_activo'] = personalizacion['flg_activo'] if 'dsc_programa' in personalizacion: curso['dsc_programa'] = personalizacion['dsc_programa'] if 'dsc_det_programa' in personalizacion: curso['dsc_det_programa'] = personalizacion['dsc_det_programa'] if curso.get('flg_activo', '') == 'NO': continue fch_valida = True try: raw_fecha = curso.get('fch_inicio') if raw_fecha: f_obj = None s_fecha = str(raw_fecha).strip() if len(s_fecha) == 10 and s_fecha[2] == '-' and s_fecha[5] == '-': f_obj = datetime.strptime(s_fecha, '%d-%m-%Y') elif '-' in s_fecha: if ' ' in s_fecha: f_obj = datetime.strptime(s_fecha.split('.')[0], '%Y-%m-%d %H:%M:%S') else: f_obj = datetime.strptime(s_fecha, '%Y-%m-%d') if f_obj: if f_obj.year != ano_target or f_obj.month != mes_target: fch_valida = False except Exception: pass if fch_valida: datos_filtrados.append(curso) # 2. TRANSFORMACIONES datos_filtrados = self.aplicar_reemplazos_programas(datos_filtrados) datos_filtrados = self.concatenar_programa_frecuencia(datos_filtrados) for curso in datos_filtrados: if 'fch_inicio' in curso: curso['fch_inicio'] = self.formatear_fecha(curso['fch_inicio']) # 3. DATOS CALCULADOS CON PRIORIDAD (REFRIPERU > CONTINUIDAD) datos_filtrados, alumnos_refriperu = self.agregar_descuento(datos_filtrados, datos_raw_matriculas) datos_filtrados = self.agregar_continuidad(datos_filtrados, datos_raw_matriculas, datos_historial, alumnos_refriperu) datos_filtrados = self.calcular_dias_para_inicio_cursos(datos_filtrados) datos_filtrados = self.agregar_inscritos_mes(datos_filtrados, datos_matriculas) datos_filtrados = self.agregar_inscritos_pc(datos_filtrados, datos_raw_matriculas) datos_filtrados = self.agregar_meta_curso(datos_filtrados) datos_filtrados = self.agregar_avance_inscritos(datos_filtrados) return datos_filtrados def aplicar_reemplazos_programas(self, datos): if 'reemplazos_programas' in self.data_manager.replace_data: reemplazos = self.data_manager.replace_data['reemplazos_programas'] for curso in datos: programa_original = curso.get('dsc_programa', '') if programa_original in reemplazos: curso['dsc_programa'] = reemplazos[programa_original] return datos def concatenar_programa_frecuencia(self, datos): for curso in datos: programa = curso.get('dsc_programa', '') frecuencia = curso.get('cod_frecuencia', '') if programa and frecuencia: curso['programa_frecuencia'] = f"{programa} - {frecuencia}" elif programa: curso['programa_frecuencia'] = programa elif frecuencia: curso['programa_frecuencia'] = frecuencia else: curso['programa_frecuencia'] = "" return datos def formatear_fecha(self, fecha_str): try: if not fecha_str: return "" if isinstance(fecha_str, str) and len(fecha_str) == 10 and fecha_str[2] == '-': return fecha_str if '.' in str(fecha_str): return datetime.strptime(str(fecha_str), '%Y-%m-%d %H:%M:%S.%f').strftime('%d-%m-%Y') return datetime.strptime(str(fecha_str), '%Y-%m-%d %H:%M:%S').strftime('%d-%m-%Y') except: return str(fecha_str) def identificar_linea_carrera(self, nombre_programa): categoria = self.obtener_categoria_programa(nombre_programa) # --- AQUÍ ESTÁ EL CAMBIO --- # Si es cualquiera de estas 5, las unimos bajo una misma "Línea Universal" de compatibilidad if categoria in ["AREQUIPA", "TRUJILLO", "PIURA", "TEAC", "TERC"]: return "CARRERA_COMPATIBLE" # Si es GESTION, MASTERCLASS, SEMINARIOS, etc., retorna None (las sigue ignorando) return None def parse_fecha(self, fecha_raw): try: if isinstance(fecha_raw, datetime): return fecha_raw if not fecha_raw: return None s_fecha = str(fecha_raw).strip() if len(s_fecha) == 10 and s_fecha[2] == '-': return datetime.strptime(s_fecha, '%d-%m-%Y') if ' ' in s_fecha: return datetime.strptime(s_fecha.split('.')[0], '%Y-%m-%d %H:%M:%S') return datetime.strptime(s_fecha, '%Y-%m-%d') except: return None # ========================================================================= # ❄️ REFRIPERU / DESCUENTO (ALTA PRIORIDAD) # ========================================================================= def agregar_descuento(self, datos, datos_raw_matriculas): print("\n💰 Calculando REFRIPERU (Descuentos)...") print("="*80) print(f"{'DNI':<12} | {'NOMBRE ALUMNO':<35} | {'PROGRAMA':<20} | {'PAGO':<10} | {'ZONA/MOTIVO'}") print("-" * 80) conteo_por_curso = {} alumnos_refriperu = set() correcciones_descuento = self.data_manager.config_data.get('correcciones_descuento', {}) if datos_raw_matriculas: for alumno in datos_raw_matriculas: try: estado = str(alumno.get('estado_matricula', '')).strip().upper() if estado not in ['ALU', 'PRE']: continue num_indice = str(alumno.get('num_indice', '')) dni = str(alumno.get('dsc_documento', 'SD')).strip() nombre_alumno = str(alumno.get('nombre_alumno', 'SIN NOMBRE')).strip() mat_id = str(alumno.get('num_matricula', '')).strip() if mat_id.endswith('.0'): mat_id = mat_id[:-2] if mat_id in correcciones_descuento: accion = str(correcciones_descuento[mat_id]).upper().strip() if accion == "SI": conteo_por_curso[num_indice] = conteo_por_curso.get(num_indice, 0) + 1 alumnos_refriperu.add(mat_id) print(f"{dni:<12} | {nombre_alumno[:35]:<35} | {'---':<20} | {'---':<10} | MANUAL (JSON)") continue elif accion == "NO": continue nombre_programa = str(alumno.get('dsc_programa', '')) cod_moneda = str(alumno.get('cod_moneda', 'SOL')).strip().upper() try: inv_neta = float(alumno.get('INV_NETA', 0) or 0) imp_tc = float(alumno.get('imp_tc', 1) or 1) if imp_tc <= 0: imp_tc = 1.0 except: inv_neta = 0.0; imp_tc = 1.0 if cod_moneda == 'DOL': inv_neta_final = inv_neta * imp_tc else: inv_neta_final = inv_neta inv_neta_final = round(inv_neta_final, 2) categoria_json = self.obtener_categoria_programa(nombre_programa) es_descuento = False motivo_debug = "" if categoria_json in ["AREQUIPA", "TRUJILLO", "PIURA"]: if 1200 <= inv_neta_final <= 1700: es_descuento = True motivo_debug = f"{categoria_json} (Rango Prov)" elif categoria_json in ["TEAC", "TERC"]: if 1400 <= inv_neta_final <= 1900: es_descuento = True motivo_debug = f"{categoria_json} (Rango Lima)" if es_descuento: conteo_por_curso[num_indice] = conteo_por_curso.get(num_indice, 0) + 1 alumnos_refriperu.add(mat_id) print(f"{dni:<12} | {nombre_alumno[:35]:<35} | {nombre_programa[:20]:<20} | S/{inv_neta_final:<8} | {motivo_debug}") except Exception: continue for curso in datos: idx = str(curso.get('num_indice', '')) curso['Descuento'] = conteo_por_curso.get(idx, 0) return datos, alumnos_refriperu # ========================================================================= # 🔄 CONTINUIDAD (BAJA PRIORIDAD) # ========================================================================= def agregar_continuidad(self, datos, datos_raw_matriculas, datos_historial, alumnos_refriperu=None): if alumnos_refriperu is None: alumnos_refriperu = set() print("🔄 Calculando CONTINUIDAD...") print("="*80) print(f"{'DNI':<12} | {'NOMBRE ALUMNO':<35} | {'PROGRAMA ACTUAL':<30} | {'MOTIVO/ORIGEN'}") print("-" * 80) conteo_por_curso = {} correcciones_continuidad = self.data_manager.config_data.get('correcciones_continuidad', {}) MIN_DIAS_DIFERENCIA = 60 if datos_raw_matriculas and datos_historial: alumnos_por_curso = {} for alu in datos_raw_matriculas: idx = str(alu.get('num_indice', '')) if idx not in alumnos_por_curso: alumnos_por_curso[idx] = [] alumnos_por_curso[idx].append(alu) for curso in datos: num_indice = str(curso.get('num_indice', '')) nombre_curso_actual = str(curso.get('dsc_programa', '')) fecha_inicio_actual = self.parse_fecha(curso.get('fch_inicio')) # Identifica si el curso actual pertenece a la "Línea Universal" de 5 categorías linea_actual = self.identificar_linea_carrera(nombre_curso_actual) if not linea_actual or not fecha_inicio_actual: continue lista_alumnos = alumnos_por_curso.get(num_indice, []) contador_fieles = 0 for alumno in lista_alumnos: estado = str(alumno.get('estado_matricula', '')).strip().upper() if estado not in ['ALU', 'PRE']: continue mat_id = str(alumno.get('num_matricula', '')).strip() if mat_id.endswith('.0'): mat_id = mat_id[:-2] if mat_id in alumnos_refriperu: continue dni = str(alumno.get('dsc_documento', 'SD')).strip() nombre_alumno = str(alumno.get('nombre_alumno', 'SIN NOMBRE')).strip() es_manual_si = False if mat_id in correcciones_continuidad: accion = str(correcciones_continuidad[mat_id]).upper().strip() if accion == "SI": es_manual_si = True elif accion == "NO": continue es_fiel = False motivo = "" if es_manual_si: es_fiel = True motivo = "MANUAL (JSON)" elif dni and dni in datos_historial: historial_alumno = datos_historial[dni] for antecedente in historial_alumno: nombre_pasado = str(antecedente.get('programa', '')) fecha_pasada = antecedente.get('fecha') if not fecha_pasada: continue if not (fecha_pasada < fecha_inicio_actual): continue dias_diff = (fecha_inicio_actual - fecha_pasada).days if dias_diff < MIN_DIAS_DIFERENCIA: continue # Revisa si el curso pasado también es de esa misma línea universal linea_pasada = self.identificar_linea_carrera(nombre_pasado) # Si ambos devuelven "CARRERA_COMPATIBLE", entonces hacen match if linea_pasada == linea_actual: es_fiel = True motivo = f"Viene de: {nombre_pasado}" break if es_fiel: contador_fieles += 1 print(f"{dni:<12} | {nombre_alumno[:35]:<35} | {nombre_curso_actual[:30]:<30} | {motivo}") conteo_por_curso[num_indice] = contador_fieles for curso in datos: idx = str(curso.get('num_indice', '')) curso['Inscritos_Continuidad'] = conteo_por_curso.get(idx, 0) return datos def calcular_dias_para_inicio(self, fecha_str, cod_estado): try: if cod_estado == 'SUS': return "SUSPENDIDO" if not fecha_str: return "" if isinstance(fecha_str, str) and len(fecha_str) == 10 and fecha_str[2] == '-': fecha_curso = datetime.strptime(fecha_str, '%d-%m-%Y') else: fecha_curso = datetime.strptime(str(fecha_str), '%Y-%m-%d %H:%M:%S') hoy = datetime.now().replace(hour=0, minute=0, second=0, microsecond=0) fecha_curso = fecha_curso.replace(hour=0, minute=0, second=0, microsecond=0) diferencia = (fecha_curso - hoy).days if diferencia < 0: return "INICIADO" return str(diferencia) if diferencia != 0 else "0" except: return "" def calcular_dias_para_inicio_cursos(self, datos): for curso in datos: curso['dias_para_inicio'] = self.calcular_dias_para_inicio(curso.get('fch_inicio'), curso.get('cod_estado')) return datos def agregar_inscritos_mes(self, datos, datos_matriculas): dict_norm = {str(k): v for k, v in datos_matriculas.items()} for curso in datos: num_indice = str(curso.get('num_indice', '')) curso['Inscritos_Mes'] = dict_norm.get(num_indice, 0) return datos def agregar_inscritos_pc(self, datos, datos_raw_matriculas): conteo_pc = {} if datos_raw_matriculas: for m in datos_raw_matriculas: try: est = str(m.get('estado_matricula', '')).strip() nid = str(m.get('num_indice', '')) sm = float(m.get('imp_saldo_matricula', 0) or 0) sc1 = float(m.get('imp_saldo_cuota1', 0) or 0) if est in ['ALU', 'PRE'] and sm < 1 and sc1 < 1: conteo_pc[nid] = conteo_pc.get(nid, 0) + 1 except: continue for curso in datos: curso['Inscritos_PC'] = conteo_pc.get(str(curso.get('num_indice', '')), 0) curso['Retirados'] = int(curso.get('Inscritos_Retirados', 0)) # <--- NUEVA EXTRACCIÓN return datos def calcular_meta_curso(self, nombre_programa, cod_estado): try: if cod_estado == 'SUS': return 0 categoria = self.obtener_categoria_programa(nombre_programa) clasificaciones = self.data_manager.meta_data.get('clasificacion_programas', {}) if categoria in clasificaciones: return clasificaciones[categoria].get('valor', 0) return self.data_manager.meta_data.get('clasificacion_default', {}).get('valor', 15) except: return 0 def agregar_meta_curso(self, datos): print("🎯 Calculando META_CURSO...") for curso in datos: curso['Meta_Curso'] = self.calcular_meta_curso(curso.get('dsc_programa', ''), curso.get('cod_estado', '')) return datos def calcular_avance_inscritos(self, inscritos_activos, meta_curso): try: if meta_curso > 0: porcentaje = (inscritos_activos / meta_curso) * 100 return f"{porcentaje:.1f}%" else: return "0%" except: return "0%" def agregar_avance_inscritos(self, datos): print("📊 Calculando AVANCE_INSCRITOS...") for curso in datos: curso['Avance_Inscritos'] = self.calcular_avance_inscritos(curso.get('Inscritos_Totales', 0), curso.get('Meta_Curso', 0)) return datos