# modules/rentabilidad/logic.py import pandas as pd import requests import re from datetime import datetime from .processor import RentabilidadProcessor class RentabilidadLogic: """Controlador lógico de Rentabilidad - Estructura la tabla y gestiona variables de clasificación""" def __init__(self, data_manager): self.data_manager = data_manager self.processor = RentabilidadProcessor(self.data_manager) def obtener_categorias(self): """Devuelve la lista simplificada y agrupada para el filtro del Dashboard""" return ["AREQUIPA", "TRUJILLO", "PIURA", "PROGRAMAS", "SEMINARIOS", "OTROS"] def obtener_sedes(self): """Devuelve la lista de sedes para el nuevo filtro""" return ["LIMA", "AREQUIPA", "PIURA", "TRUJILLO"] def identificar_sede(self, dsc_programa): """Identifica la sede basada en el nombre del programa y el JSON de GitHub""" if not dsc_programa: return "LIMA" dsc_prog_upper = str(dsc_programa).upper() sedes_data = self.data_manager.meta_data.get("clasificacion_sedes", {}) for sede, data in sedes_data.items(): if sede == "DEFAULT": continue for patron in data.get("patrones", []): if patron.upper() in dsc_prog_upper: return sede return sedes_data.get("DEFAULT", "LIMA") def obtener_filtros_programa(self): """Devuelve la lista de programas para el nuevo filtro combinado""" return ["SEMINARIOS", "OTROS", "TEAC", "TERC"] def identificar_filtro_programa(self, dsc_programa): """Identifica la categoría del programa para el filtro de la UI""" if not dsc_programa: return "OTROS" dsc_prog_upper = str(dsc_programa).upper() filtro_data = self.data_manager.meta_data.get("clasificacion_filtro_programa", {}) for categoria, data in filtro_data.items(): if categoria == "DEFAULT": continue for patron in data.get("patrones", []): if patron.upper() in dsc_prog_upper: return categoria return filtro_data.get("DEFAULT", "OTROS") def clasificar_programa(self, dsc_programa): if not dsc_programa: return "OTROS" dsc_prog_upper = str(dsc_programa).upper() meta_data = self.data_manager.meta_data clasificaciones = meta_data.get("clasificacion_programas", {}) for clase, data in clasificaciones.items(): for patron in data.get("patrones", []): if patron.upper() in dsc_prog_upper: return clase return meta_data.get("clasificacion_default", {}).get("categoria", "SEMINARIOS") def identificar_linea_carrera(self, nombre_programa): """Identifica si el programa es de la familia de Carreras Técnicas""" categoria = self.clasificar_programa(nombre_programa) if categoria in ["AREQUIPA", "TRUJILLO", "PIURA", "TEAC", "TERC"]: return "CARRERA_COMPATIBLE" return None def parse_fecha(self, fecha_raw): """Convierte diferentes formatos de fecha a objeto datetime""" 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 def calcular_descuento(self, clasificacion, inv_neta): try: inv_neta = float(inv_neta) except: inv_neta = 0.0 if clasificacion in ["TEAC", "TERC"]: return 3499.0 - inv_neta elif clasificacion in ["AREQUIPA", "PIURA", "TRUJILLO"]: return 2839.0 - inv_neta elif clasificacion == "CARRERA": return 4299.0 - inv_neta else: return None def calcular_estado_descuento(self, clasificacion, descuento): if descuento is None: return "" es_provincia = clasificacion in ["TRUJILLO", "AREQUIPA", "PIURA"] tope = 1000.0 if es_provincia else 800.0 if descuento > tope or descuento < 0: return "NO" return "SI" # ======================================================================== # MOTOR DINÁMICO DE COSTOS - BÚSQUEDA INTELIGENTE # ======================================================================== # ======================================================================== # MOTOR DINÁMICO DE COSTOS - BÚSQUEDA INTELIGENTE # ======================================================================== def _obtener_costo_segun_categoria(self, diccionario_costo, categoria, dsc_programa): """Busca el costo inteligente. Si es 'otros', busca similitud de palabras.""" cat_lower = str(categoria).lower() # 1. Agrupamos los técnicos dentro de 'programas' if cat_lower in ["teac", "terc"]: cat_lower = "programas" # 2. EL EMBUDO: Si la categoría no existe en el diccionario (ej. seminarios, masterclass, etc.), # la forzamos a que caiga siempre en la bolsa de "otros" if cat_lower not in diccionario_costo: cat_lower = "otros" # 3. Buscamos el valor if cat_lower in diccionario_costo: valor = diccionario_costo[cat_lower] # Si el valor es un bloque de similitudes (Como sucede ahora con "otros") if isinstance(valor, dict): dsc_upper = str(dsc_programa).upper() # Buscamos coincidencias con el nombre completo del curso for patron, monto in valor.items(): if patron != "DEFAULT" and patron.upper() in dsc_upper: return float(monto) # Si lee todo el diccionario y no encuentra coincidencia, usamos el DEFAULT return float(valor.get("DEFAULT", 0)) else: # Si es un número directo (ej. piura, trujillo, carrera) return float(valor) return 0.0 # ======================================================================== def obtener_datos_procesados(self, ano, mes, sede="TODOS", filtro_prog="TODOS"): """Obtiene datos, inyecta Categoria_Programa, Sede, Filtro_Programa y calcula promedios.""" datos = self.processor.obtener_datos_procesados(ano, mes) programas_disponibles = set() if datos: for d in datos: prog = str(d.get('programa_frecuencia', d.get('dsc_programa', ''))) d['Categoria_Programa'] = self.clasificar_programa(prog) d['Sede'] = self.identificar_sede(prog) d['Filtro_Programa'] = self.identificar_filtro_programa(prog) # Recolectar qué programas existen REALMENTE en la Sede seleccionada if sede == "TODOS" or d['Sede'] == sede: programas_disponibles.add(d['Filtro_Programa']) # Ordenar para el Dropdown en la UI orden_deseado = ["SEMINARIOS", "OTROS", "TEAC", "TERC"] self.programas_disponibles_filtro = [p for p in orden_deseado if p in programas_disponibles] # NO auto-corregir: se respeta el filtro y la tabla queda vacia si no hay. self.filtro_corregido = None if datos and (sede != "TODOS" or filtro_prog != "TODOS"): datos_filtrados = [] for d in datos: cumple_sede = (sede == "TODOS" or d.get('Sede') == sede) cumple_prog = (filtro_prog == "TODOS" or d.get('Filtro_Programa') == filtro_prog) if cumple_sede and cumple_prog: datos_filtrados.append(d) datos = datos_filtrados if not datos: return [] sql_matriculas = self._get_sql_matriculas_modificado() sql_cuotas = self._get_sql_cuotas() matriculados_raw = [] cuotas_raw = [] if sql_matriculas and sql_cuotas: try: conn = self.data_manager.get_connection() cursor = conn.cursor() try: cursor.execute(sql_matriculas, ano, mes) cols_mat = [c[0] for c in cursor.description] matriculados_raw = [dict(zip(cols_mat, row)) for row in cursor.fetchall()] cursor.execute(sql_cuotas, ano, mes) cols_cuo = [c[0] for c in cursor.description] cuotas_raw = [dict(zip(cols_cuo, row)) for row in cursor.fetchall()] finally: conn.close() except Exception as e: print(f"❌ Error en consultas DAX: {e}") promedios_dax = self._calcular_promedios_dax(matriculados_raw, cuotas_raw) for d in datos: idx = str(d.get('num_indice', '')).strip() if idx.endswith('.0'): idx = idx[:-2] if idx in promedios_dax: d['Promedio_Cuota'] = promedios_dax[idx]['promedio'] d['Suma_Dax_Real'] = promedios_dax[idx]['suma_real'] d['Conteo_Dax_Real'] = promedios_dax[idx]['conteo_real'] d['Promedio_Desc_E'] = promedios_dax[idx]['promedio_desc_e'] d['Suma_Desc_E_Real'] = promedios_dax[idx]['suma_desc_e_real'] d['Conteo_Desc_E_Real'] = promedios_dax[idx]['conteo_desc_e_real'] d['Valor_Venta'] = promedios_dax[idx]['suma_valor_venta'] d['Valor_Venta_Actual'] = promedios_dax[idx]['suma_valor_venta_actual'] else: d['Promedio_Cuota'] = 0.0 d['Suma_Dax_Real'] = 0.0 d['Conteo_Dax_Real'] = 0 d['Promedio_Desc_E'] = 0.0 d['Suma_Desc_E_Real'] = 0.0 d['Conteo_Desc_E_Real'] = 0 d['Valor_Venta'] = 0.0 d['Valor_Venta_Actual'] = 0.0 return datos def _calcular_promedios_dax(self, matriculados, cuotas): alumnos_dict = {} agrupado_indices = {} lista_mirar_cuotas = ["AREQUIPA", "TRUJILLO", "PIURA", "TEAC", "TERC"] correcciones_mat = self.data_manager.config_data.get('correcciones_matriculas', {}) correcciones_inv_neta = self.data_manager.config_data.get('Actualizar_INV_NETA', {}) for m in matriculados: estado_mat = str(m.get('estado_matricula', '')).strip().upper() if estado_mat not in ['ALU', 'PRE', 'RET']: continue num_matricula = str(m.get('num_matricula', '')).strip() if num_matricula.endswith('.0'): num_matricula = num_matricula[:-2] num_indice = str(m.get('num_indice', '')).strip() if num_indice.endswith('.0'): num_indice = num_indice[:-2] prog = str(m.get('dsc_programa', '')) try: inv_neta = float(m.get('INV_NETA', 0.0)) except: inv_neta = 0.0 tc_override = None if num_matricula in correcciones_mat: if "imp_tc" in correcciones_mat[num_matricula]: tc_override = float(correcciones_mat[num_matricula]["imp_tc"]) # TC del COMPROBANTE — único TC para todo (INV_NETA + cuotas) if tc_override is not None: imp_tc_comp = tc_override else: try: imp_tc_comp = float(m.get('imp_tc', 0) or 0) except: imp_tc_comp = 0 if imp_tc_comp < 2: imp_tc_comp = 3.45 # TC por defecto # CORRECCIÓN MANUAL DE INV_NETA (siempre en SOL, sin multiplicar) if num_matricula in correcciones_inv_neta: try: inv_neta = float(correcciones_inv_neta[num_matricula]) except: pass else: cod_moneda_mat = str(m.get('cod_moneda', '')).strip().upper() if cod_moneda_mat == "DOL": inv_neta = inv_neta * imp_tc_comp clase = self.clasificar_programa(prog) descuento = self.calcular_descuento(clase, inv_neta) estado_desc = self.calcular_estado_descuento(clase, descuento) alumnos_dict[num_matricula] = { 'num_indice': num_indice, 'clasificacion': clase, 'programa': prog, 'estado_descuento': estado_desc, 'inv_neta': inv_neta, 'estado_mat': estado_mat, 'monto_real_cuotas': 0.0, 'imp_tc_comprobante': imp_tc_comp } if num_indice not in agrupado_indices: agrupado_indices[num_indice] = { 'suma': 0.0, 'conteo': 0, 'suma_desc_e': 0.0, 'conteo_desc_e': 0, 'suma_valor_venta': 0.0, 'suma_valor_venta_actual': 0.0 } agrupado_indices[num_indice]['suma_valor_venta'] += inv_neta # ALU y PRE: suma INV_NETA completa if estado_mat in ['ALU', 'PRE']: agrupado_indices[num_indice]['suma_valor_venta_actual'] += inv_neta # RET: suma solo lo que ya pagó elif estado_mat == 'RET': try: imp_pagado = float(m.get('imp_total_pagado', 0) or 0) except: imp_pagado = 0.0 agrupado_indices[num_indice]['suma_valor_venta_actual'] += imp_pagado if clase not in lista_mirar_cuotas: agrupado_indices[num_indice]['suma'] += inv_neta agrupado_indices[num_indice]['conteo'] += 1 for c in cuotas: num_matricula = str(c.get('num_matricula', '')).strip() if num_matricula.endswith('.0'): num_matricula = num_matricula[:-2] if num_matricula not in alumnos_dict: continue alumno = alumnos_dict[num_matricula] num_idx = alumno['num_indice'] if alumno['clasificacion'] in lista_mirar_cuotas: try: num_cuota = int(c.get('num_cuota', 0)) except: num_cuota = 0 t_val = c.get('imp_total') d_val = c.get('imp_dscto') try: t_monto = float(t_val) if t_val is not None else 0.0 except: t_monto = 0.0 try: d_monto = float(d_val) if d_val is not None else 0.0 except: d_monto = 0.0 imp_cuota = t_monto - d_monto cod_moneda_cuo = str(c.get('cod_moneda', '')).strip().upper() if cod_moneda_cuo == "DOL": # USAR el TC del COMPROBANTE (ya validado con fallback 3.45) tc_cuo = alumno['imp_tc_comprobante'] imp_cuota = imp_cuota * tc_cuo cumple_cuota_normal = (alumno['estado_descuento'] == "SI" and num_cuota > 0 and imp_cuota < 1000) cumple_desc_especial = (alumno['estado_descuento'] == "SI" and num_cuota > 0) if cumple_cuota_normal: agrupado_indices[num_idx]['suma'] += imp_cuota agrupado_indices[num_idx]['conteo'] += 1 if cumple_desc_especial: alumno['monto_real_cuotas'] += imp_cuota for mat_id, alumno in alumnos_dict.items(): idx = alumno['num_indice'] clase = alumno['clasificacion'] estado_desc = alumno['estado_descuento'] if clase in lista_mirar_cuotas: monto_real = alumno['monto_real_cuotas'] else: monto_real = alumno['inv_neta'] valor_final = 0.0 if monto_real >= 1900 and estado_desc == "SI": if clase in ["TEAC", "TERC"]: desc_calc = 3400 - monto_real elif clase in ["AREQUIPA", "PIURA", "TRUJILLO"]: desc_calc = 2740 - monto_real else: desc_calc = 0 resta = desc_calc - 200 valor_final = min(max(resta, 0.0), 200.0) agrupado_indices[idx]['suma_desc_e'] += valor_final agrupado_indices[idx]['conteo_desc_e'] += 1 promedios = {} for idx, totales in agrupado_indices.items(): suma_total = totales['suma'] conteo_total = totales['conteo'] suma_desc_e = totales['suma_desc_e'] conteo_desc_e = totales['conteo_desc_e'] promedios[idx] = { 'promedio': suma_total / conteo_total if conteo_total > 0 else 0.0, 'suma_real': suma_total, 'conteo_real': conteo_total, 'promedio_desc_e': suma_desc_e / conteo_desc_e if conteo_desc_e > 0 else 0.0, 'suma_desc_e_real': suma_desc_e, 'conteo_desc_e_real': conteo_desc_e, 'suma_valor_venta': totales['suma_valor_venta'], 'suma_valor_venta_actual': totales['suma_valor_venta_actual'] # <-- NUEVO } return promedios def formatear_datos_para_tabla(self, datos): """Estructura las filas finales para tksheet (14 columnas)""" if not datos: return [] cfg_costos = self.data_manager.costos_data if not cfg_costos: cfg_costos = { "epp": {"piura": 28, "trujillo": 28, "arequipa": 28, "programas": 28, "carrera": 28, "otros": 0}, "certificado": {"piura": 7, "trujillo": 7, "arequipa": 7, "programas": 7, "carrera": 7, "otros": 0}, "consumibles": {"piura": 676, "trujillo": 676, "arequipa": 676, "programas": 614, "carrera": 614, "otros": 0}, "marketing": {"piura": 2839, "trujillo": 2839, "arequipa": 2839, "programas": 3499, "carrera": 3499, "otros": {"VRV/VRF": 4500, "INDUSTRIAL POR AMONIACO (NH3)": 1000, "VIRTUAL SEMINARIO DISEÑO DE CHILLERS": 900, "DIPLOMADO INTERNACIONAL": 1300, "SEMINARIO VIRTUAL METRADO , COSTEO": 800, "VIRTUAL SEMINARIO DISEÑO DE CAMARAS": 1000, "DEFAULT": 0}}, "docente": {"piura": 5115, "trujillo": 5115, "arequipa": 6500, "programas": 4560, "carrera": 4560, "otros": {"VRV/VRF": 4500, "INDUSTRIAL POR AMONIACO (NH3)": 4000, "VIRTUAL SEMINARIO DISEÑO DE CHILLERS": 3000, "DIPLOMADO INTERNACIONAL": 7000, "SEMINARIO VIRTUAL METRADO , COSTEO": 2000, "VIRTUAL SEMINARIO DISEÑO DE CAMARAS": 4000, "DEFAULT": 0}} } # CARGA OVERRIDES DESDE SUPABASE (una sola vez por render) overrides_supabase = self.data_manager.cargar_overrides_costos() sheet_data = [] tot_total = tot_retirados = tot_curso = 0 tot_venta = 0.0 tot_venta_actual = 0.0 tot_costo = 0.0 tot_costo_actual = 0.0 bolsa_plata_global = 0.0 bolsa_recibos_global = 0 bolsa_desc_e_plata_global = 0.0 bolsa_desc_e_recibos_global = 0 for d in datos: try: total_inscritos = int(d.get('Inscritos_Totales', 0)) except: total_inscritos = 0 try: retirados = int(d.get('Retirados', 0)) except: retirados = 0 try: en_curso = int(d.get('Inscritos_En_Curso', d.get('Inscritos_Activos', 0))) except: en_curso = 0 p_cuota = float(d.get('Promedio_Cuota', 0.0)) p_desc = float(d.get('Promedio_Desc_E', 0.0)) venta = float(d.get('Valor_Venta', 0.0)) venta_actual = float(d.get('Valor_Venta_Actual', 0.0)) conteo_cuota_real = int(d.get('Conteo_Dax_Real', 0)) str_cuota_tabla = f"S/ {p_cuota:,.0f}" if conteo_cuota_real > 0 else "" conteo_desc_e = int(d.get('Conteo_Desc_E_Real', 0)) str_desc_e_tabla = f"S/ {p_desc:,.0f}" if conteo_desc_e > 0 else "" cat = d.get('Categoria_Programa', 'OTROS') nombre_prog_crudo = d.get('dsc_programa', '') # Identificador de curso para buscar override num_idx_curso = str(d.get('num_indice', '')).strip() if num_idx_curso.endswith('.0'): num_idx_curso = num_idx_curso[:-2] ov_curso = overrides_supabase.get(num_idx_curso, {}) ov_inicial = ov_curso.get('inicial', {}) ov_actual = ov_curso.get('actual', {}) # ============= COSTO INICIAL (con override si existe) ============= def _costo_inicial(key_supa, key_cfg, multiplicar): val = ov_inicial.get(key_supa) if val is not None: return float(val) # Supabase manda directo, sin multiplicar base = self._obtener_costo_segun_categoria(cfg_costos.get(key_cfg, {}), cat, nombre_prog_crudo) return total_inscritos * base if multiplicar else base costo_epp = _costo_inicial('epp', 'epp', True) costo_cert = _costo_inicial('certificado', 'certificado', True) costo_cons = _costo_inicial('consumibles', 'consumibles', True) costo_mkt = _costo_inicial('marketing', 'marketing', False) costo_doc = _costo_inicial('docente', 'docente', False) valor_costo_inicial = costo_epp + costo_cert + costo_cons + costo_mkt + costo_doc # ============= COSTO ACTUAL (con override si existe) ============= def _costo_actual(key_supa, key_cfg, multiplicador): val = ov_actual.get(key_supa) if val is not None: return float(val) # Supabase manda directo, sin multiplicar base = self._obtener_costo_segun_categoria(cfg_costos.get(key_cfg, {}), cat, nombre_prog_crudo) return multiplicador * base if multiplicador is not None else base costo_epp_actual = _costo_actual('epp', 'epp', total_inscritos) costo_cert_actual = _costo_actual('certificado', 'certificado', en_curso) costo_cons_actual = _costo_actual('consumibles', 'consumibles', total_inscritos) costo_mkt_actual = _costo_actual('marketing', 'marketing', None) costo_doc_actual = _costo_actual('docente', 'docente', None) costo_actual = costo_epp_actual + costo_cert_actual + costo_cons_actual + costo_mkt_actual + costo_doc_actual # MARGEN BRUTO % if venta > 0: margen_bruto = 1 - (valor_costo_inicial / venta) str_margen = f"{margen_bruto * 100:,.1f}%" else: str_margen = "" # MARGEN BRUTO ACTUAL % if venta_actual > 0: margen_bruto_actual = 1 - (costo_actual / venta_actual) str_margen_actual = f"{margen_bruto_actual * 100:,.1f}%" else: str_margen_actual = "" tot_total += total_inscritos tot_retirados += retirados tot_curso += en_curso tot_venta += venta tot_venta_actual += venta_actual tot_costo += valor_costo_inicial tot_costo_actual += costo_actual bolsa_plata_global += float(d.get('Suma_Dax_Real', 0.0)) bolsa_recibos_global += int(d.get('Conteo_Dax_Real', 0)) bolsa_desc_e_plata_global += float(d.get('Suma_Desc_E_Real', 0.0)) bolsa_desc_e_recibos_global += int(d.get('Conteo_Desc_E_Real', 0)) nombre_prog = d.get('programa_frecuencia', d.get('dsc_programa', '')) fila = [ nombre_prog, d.get('fch_inicio', ''), total_inscritos, retirados, en_curso, str_cuota_tabla, str_desc_e_tabla, f"S/ {venta:,.0f}", f"S/ {valor_costo_inicial:,.0f}", str_margen, f"S/ {venta_actual:,.0f}", f"S/ {costo_actual:,.0f}", str_margen_actual, " ≡ ▼ " ] sheet_data.append(fila) promedio_total_final = (bolsa_plata_global / bolsa_recibos_global) if bolsa_recibos_global > 0 else 0.0 promedio_desc_e_final = (bolsa_desc_e_plata_global / bolsa_desc_e_recibos_global) if bolsa_desc_e_recibos_global > 0 else 0.0 str_total_cuota = f"S/ {promedio_total_final:,.0f}" if bolsa_recibos_global > 0 else "" str_total_desc_e = f"S/ {promedio_desc_e_final:,.0f}" if bolsa_desc_e_recibos_global > 0 else "" margen_total = 1 - (tot_costo / tot_venta) if tot_venta > 0 else 0 margen_total_actual = 1 - (tot_costo_actual / tot_venta_actual) if tot_venta_actual > 0 else 0 fila_total = [ "TOTAL GENERAL", "", tot_total, tot_retirados, tot_curso, str_total_cuota, str_total_desc_e, f"S/ {tot_venta:,.0f}", f"S/ {tot_costo:,.0f}", f"{margen_total * 100:,.1f}%", f"S/ {tot_venta_actual:,.0f}", f"S/ {tot_costo_actual:,.0f}", f"{margen_total_actual * 100:,.1f}%", "" ] sheet_data.append(fila_total) return sheet_data def exportar_detalle_alumnos_excel(self, programa, headers, datos): if not datos: raise ValueError("No hay datos para exportar") prog_limpio = "".join([c if c.isalnum() else "_" for c in str(programa)])[:40] archivo = f"Detalle_Alumnos_{prog_limpio}.xlsx" df = pd.DataFrame(datos, columns=headers) try: with pd.ExcelWriter(archivo, engine='openpyxl') as writer: df.to_excel(writer, index=False, sheet_name='Detalle_Alumnos') worksheet = writer.sheets['Detalle_Alumnos'] try: from openpyxl.styles import PatternFill, Font, Border, Side, Alignment fill_header = PatternFill(start_color="12345F", end_color="12345F", fill_type="solid") font_header = Font(color="FFFFFF", bold=True) thin_border = Border( left=Side(style='thin', color="DDDDDD"), right=Side(style='thin', color="DDDDDD"), top=Side(style='thin', color="DDDDDD"), bottom=Side(style='thin', color="DDDDDD") ) for cell in worksheet[1]: cell.fill = fill_header cell.font = font_header cell.border = thin_border cell.alignment = Alignment(horizontal="center", vertical="center", wrap_text=True) for row in worksheet.iter_rows(min_row=2): for cell in row: cell.border = thin_border cell.alignment = Alignment(vertical="center") for col in worksheet.columns: max_len = 0 col_letter = col[0].column_letter for cell in col: if cell.value: max_len = max(max_len, len(str(cell.value))) worksheet.column_dimensions[col_letter].width = min(max_len + 3, 50) except Exception: pass except Exception as e: df.to_excel(archivo, index=False) return archivo def obtener_detalle_programa(self, programa_target, ano, mes_numero): prog_t = str(programa_target).strip().upper() sql_matriculas = self._get_sql_matriculas_modificado() sql_cuotas = self._get_sql_cuotas() matriculados_raw = [] cuotas_raw = [] if sql_matriculas and sql_cuotas: try: conn = self.data_manager.get_connection() cursor = conn.cursor() try: cursor.execute(sql_matriculas, ano, mes_numero) cols_mat = [c[0] for c in cursor.description] matriculados_raw = [dict(zip(cols_mat, row)) for row in cursor.fetchall()] cursor.execute(sql_cuotas, ano, mes_numero) cols_cuo = [c[0] for c in cursor.description] cuotas_raw = [dict(zip(cols_cuo, row)) for row in cursor.fetchall()] finally: conn.close() except Exception as e: print(f"❌ Error en consultas detalle: {e}") cuotas_por_mat = {} for c in cuotas_raw: m_id = str(c.get('num_matricula', '')).strip() if m_id.endswith('.0'): m_id = m_id[:-2] if m_id not in cuotas_por_mat: cuotas_por_mat[m_id] = [] cuotas_por_mat[m_id].append(c) cursos = self.processor.obtener_datos_procesados(ano, mes_numero) indices_target = [] fecha_inicio_actual = None linea_actual = self.identificar_linea_carrera(prog_t) for c in cursos: nombre = str(c.get('programa_frecuencia', c.get('dsc_programa', ''))).strip().upper() if nombre == prog_t: idx = str(c.get('num_indice', '')).strip() if idx.endswith('.0'): idx = idx[:-2] indices_target.append(idx) if not fecha_inicio_actual: fecha_inicio_actual = self.parse_fecha(c.get('fch_inicio')) lista_dnis = [] for m in matriculados_raw: dni = m.get('dsc_documento') if dni: lista_dnis.append(str(dni).strip()) try: datos_historial = self.data_manager.consultar_historial_continuidad(lista_dnis) except: datos_historial = {} correcciones_desc = self.data_manager.config_data.get('correcciones_descuento', {}) correcciones_cont = self.data_manager.config_data.get('correcciones_continuidad', {}) correcciones_mat = self.data_manager.config_data.get('correcciones_matriculas', {}) correcciones_inv_neta = self.data_manager.config_data.get('Actualizar_INV_NETA', {}) lista_mirar_cuotas = ["AREQUIPA", "TRUJILLO", "PIURA", "TEAC", "TERC"] acum_inv_neta_global = 0.0 acum_suma_cuotas_global = 0.0 acum_cont_cuotas_global = 0 acum_suma_desc_e_global = 0.0 acum_cont_desc_e_global = 0 alumnos_lista = [] for m in matriculados_raw: estado_mat = str(m.get('estado_matricula', '')).strip().upper() if estado_mat not in ['ALU', 'PRE', 'RET']: continue idx_mat = str(m.get('num_indice', '')).strip() if idx_mat.endswith('.0'): idx_mat = idx_mat[:-2] nombre_prog_crudo = str(m.get('dsc_programa', '')).strip().upper() if (idx_mat in indices_target) or (prog_t in nombre_prog_crudo) or (nombre_prog_crudo in prog_t): vendedor = str(m.get('dsc_vendedor', 'SIN VENDEDOR')).strip() if vendedor == "None" or not vendedor: vendedor = "SIN VENDEDOR" alumno_nombre = str(m.get('nombre_alumno', m.get('dsc_alumno', 'SIN NOMBRE'))).strip() if alumno_nombre == "None" or not alumno_nombre: alumno_nombre = "SIN NOMBRE" mat_id = str(m.get('num_matricula', '')).strip() if mat_id.endswith('.0'): mat_id = mat_id[:-2] dni = str(m.get('dsc_documento', '')).strip() try: inv_neta_raw = float(m.get('INV_NETA', 0.0)) except: inv_neta_raw = 0.0 cod_moneda = str(m.get('cod_moneda', 'SOL')).strip().upper() try: tc = float(m.get('imp_tc', 0) or 0) except: tc = 0 if tc < 2: tc = 3.45 # TC por defecto si comprobante inválido fch_canc_raw = m.get('fch_cancelacion_cuota1', '') if mat_id in correcciones_mat: datos_corregidos = correcciones_mat[mat_id] if "imp_tc" in datos_corregidos: tc = float(datos_corregidos["imp_tc"]) if "fch_cancelacion_cuota1" in datos_corregidos: fch_canc_raw = datos_corregidos["fch_cancelacion_cuota1"] # CORRECCIÓN MANUAL DE INV_NETA (siempre en SOL, sin multiplicar) if mat_id in correcciones_inv_neta: try: inv_neta_soles = float(correcciones_inv_neta[mat_id]) except: inv_neta_soles = inv_neta_raw elif cod_moneda == "DOL": inv_neta_soles = inv_neta_raw * tc else: inv_neta_soles = inv_neta_raw inv_neta_soles = round(inv_neta_soles, 2) acum_inv_neta_global += inv_neta_soles fch_mat = m.get('fch_matricula', '') fecha_mat_limpia = "" if fch_mat: try: if hasattr(fch_mat, 'strftime'): fecha_mat_limpia = fch_mat.strftime('%d/%m/%Y') else: tmp = str(fch_mat)[:10] if len(tmp) == 10 and tmp[4] == '-': p = tmp.split('-'); fecha_mat_limpia = f"{p[2]}/{p[1]}/{p[0]}" else: fecha_mat_limpia = tmp except: fecha_mat_limpia = str(fch_mat)[:10] fch_canc_limpia = "" if fch_canc_raw and str(fch_canc_raw).strip() != "None": try: if hasattr(fch_canc_raw, 'strftime'): fch_canc_limpia = fch_canc_raw.strftime('%d/%m/%Y') else: tmp = str(fch_canc_raw)[:10] if len(tmp) == 10 and tmp[4] == '-': p = tmp.split('-'); fch_canc_limpia = f"{p[2]}/{p[1]}/{p[0]}" else: fch_canc_limpia = tmp except: fch_canc_limpia = str(fch_canc_raw)[:10] clase_alumno = self.clasificar_programa(nombre_prog_crudo) es_refriperu = False if mat_id in correcciones_desc: if str(correcciones_desc[mat_id]).upper() == "SI": es_refriperu = True elif str(correcciones_desc[mat_id]).upper() == "NO": es_refriperu = False else: if clase_alumno in ["AREQUIPA", "TRUJILLO", "PIURA"] and 1200 <= inv_neta_soles <= 1700: es_refriperu = True elif clase_alumno in ["TEAC", "TERC"] and 1400 <= inv_neta_soles <= 1900: es_refriperu = True tipo_final = "NUEVO" if es_refriperu: tipo_final = "REFRIPERU" else: es_cont = False if mat_id in correcciones_cont: if str(correcciones_cont[mat_id]).upper() == "SI": es_cont = True elif str(correcciones_cont[mat_id]).upper() == "NO": es_cont = False else: if fecha_inicio_actual and linea_actual and (dni in datos_historial): for antecedente in datos_historial[dni]: nombre_pasado = str(antecedente.get('programa', '')) fecha_pasada = antecedente.get('fecha') if not fecha_pasada: continue if not (fecha_pasada < fecha_inicio_actual): continue if (fecha_inicio_actual - fecha_pasada).days < 60: continue if self.identificar_linea_carrera(nombre_pasado) == linea_actual: es_cont = True; break if es_cont: tipo_final = "CONTINUIDAD" descuento = self.calcular_descuento(clase_alumno, inv_neta_soles) estado_desc = self.calcular_estado_descuento(clase_alumno, descuento) str_cuota_val = "" monto_eval = 0.0 if clase_alumno in lista_mirar_cuotas: suma_c = 0.0; cont_c = 0; monto_real_c = 0.0 for c_dict in cuotas_por_mat.get(mat_id, []): try: num_c = int(c_dict.get('num_cuota', 0)) except: num_c = 0 t_val = c_dict.get('imp_total'); d_val = c_dict.get('imp_dscto') try: t_monto = float(t_val) if t_val is not None else 0.0 except: t_monto = 0.0 try: d_monto = float(d_val) if d_val is not None else 0.0 except: d_monto = 0.0 imp_c = t_monto - d_monto tc_over = None if mat_id in correcciones_mat and "imp_tc" in correcciones_mat[mat_id]: tc_over = float(correcciones_mat[mat_id]["imp_tc"]) cod_m_c = str(c_dict.get('cod_moneda', '')).strip().upper() if cod_m_c == "DOL": # USAR TC del COMPROBANTE (no del cronograma) tc_c = tc # tc ya está validado arriba (línea 654) imp_c *= tc_c if estado_desc == "SI" and num_c > 0 and imp_c < 1000: suma_c += imp_c; cont_c += 1 if estado_desc == "SI" and num_c > 0: monto_real_c += imp_c if cont_c > 0: str_cuota_val = f"S/ {suma_c / cont_c:,.0f}" monto_eval = monto_real_c acum_suma_cuotas_global += suma_c acum_cont_cuotas_global += cont_c else: str_cuota_val = f"S/ {inv_neta_soles:,.0f}" monto_eval = inv_neta_soles acum_suma_cuotas_global += inv_neta_soles acum_cont_cuotas_global += 1 str_desc_e_val = "" if monto_eval >= 1900 and estado_desc == "SI": if clase_alumno in ["TEAC", "TERC"]: d_calc = 3400 - monto_eval elif clase_alumno in ["AREQUIPA", "PIURA", "TRUJILLO"]: d_calc = 2740 - monto_eval else: d_calc = 0 desc_e_val = min(max(d_calc - 200, 0.0), 200.0) str_desc_e_val = f"S/ {desc_e_val:,.0f}" acum_suma_desc_e_global += desc_e_val acum_cont_desc_e_global += 1 alumnos_lista.append([ vendedor, alumno_nombre, fecha_mat_limpia, fch_canc_limpia, tipo_final, str_cuota_val, str_desc_e_val, f"S/ {inv_neta_soles:,.0f}", estado_mat # columna oculta para detectar RET ]) alumnos_lista.sort(key=lambda x: x[1]) prom_gral_cuota = (acum_suma_cuotas_global / acum_cont_cuotas_global) if acum_cont_cuotas_global > 0 else 0.0 prom_gral_desc = (acum_suma_desc_e_global / acum_cont_desc_e_global) if acum_cont_desc_e_global > 0 else 0.0 str_gral_cuota = f"S/ {prom_gral_cuota:,.0f}" if acum_cont_cuotas_global > 0 else "" str_gral_desc = f"S/ {prom_gral_desc:,.0f}" if acum_cont_desc_e_global > 0 else "" alumnos_lista.append([ "TOTAL GENERAL", "", "", "", "", str_gral_cuota, str_gral_desc, f"S/ {acum_inv_neta_global:,.0f}", "" # columna oculta extra para TOTAL ]) return alumnos_lista def exportar_a_excel(self, datos, ano, mes): if not datos: raise ValueError("No hay datos para exportar") df = pd.DataFrame(datos) archivo = f"reporte_rentabilidad_{ano}_{mes}.xlsx" df.to_excel(archivo, index=False) return archivo def _get_sql_matriculas_modificado(self): if not hasattr(self, '_sql_matriculas_cache') or not self._sql_matriculas_cache: sql = self.data_manager.query_matriculas_sql if not sql: return "" sql = re.sub( r'YEAR\(\s*sgeca_matricula\.fch_matricula\s*\)', 'YEAR(sgede_RP_programa.fch_inicio)', sql, flags=re.IGNORECASE ) sql = re.sub( r'MONTH\(\s*sgeca_matricula\.fch_matricula\s*\)', 'MONTH(sgede_RP_programa.fch_inicio)', sql, flags=re.IGNORECASE ) self._sql_matriculas_cache = sql return self._sql_matriculas_cache def _get_sql_cuotas(self): if not hasattr(self, '_sql_cuotas_cache') or not self._sql_cuotas_cache: try: res = requests.get(self.data_manager.github_cuota_url) self._sql_cuotas_cache = res.text except: self._sql_cuotas_cache = "" return self._sql_cuotas_cache def obtener_datos_costos_programa(self, programa, ano, mes): """Devuelve el desglose de costos e ingresos para el modal Editar""" datos = self.obtener_datos_procesados(ano, mes) if not datos: return None curso = None for d in datos: nombre = d.get('programa_frecuencia', d.get('dsc_programa', '')) if str(nombre).strip() == str(programa).strip(): curso = d; break if not curso: return None cfg_costos = self.data_manager.costos_data if not cfg_costos: return None # OVERRIDES desde Supabase num_idx = str(curso.get('num_indice', '')).strip() if num_idx.endswith('.0'): num_idx = num_idx[:-2] overrides = self.data_manager.cargar_overrides_costos() ov_curso = overrides.get(num_idx, {}) ov_inicial = ov_curso.get('inicial', {}) ov_actual = ov_curso.get('actual', {}) cat = curso.get('Categoria_Programa', 'OTROS') prog_crudo = curso.get('dsc_programa', '') try: total_inscritos = int(curso.get('Inscritos_Totales', 0)) except: total_inscritos = 0 try: en_curso = int(curso.get('Inscritos_En_Curso', curso.get('Inscritos_Activos', 0))) except: en_curso = 0 epp_u = self._obtener_costo_segun_categoria(cfg_costos.get("epp", {}), cat, prog_crudo) cert_u = self._obtener_costo_segun_categoria(cfg_costos.get("certificado", {}), cat, prog_crudo) cons_u = self._obtener_costo_segun_categoria(cfg_costos.get("consumibles", {}), cat, prog_crudo) mkt = self._obtener_costo_segun_categoria(cfg_costos.get("marketing", {}), cat, prog_crudo) doc = self._obtener_costo_segun_categoria(cfg_costos.get("docente", {}), cat, prog_crudo) # INICIAL epp_i = float(ov_inicial['epp']) if ov_inicial.get('epp') is not None else total_inscritos * epp_u cert_i = float(ov_inicial['certificado']) if ov_inicial.get('certificado') is not None else total_inscritos * cert_u cons_i = float(ov_inicial['consumibles']) if ov_inicial.get('consumibles') is not None else total_inscritos * cons_u mkt_i = float(ov_inicial['marketing']) if ov_inicial.get('marketing') is not None else mkt doc_i = float(ov_inicial['docente']) if ov_inicial.get('docente') is not None else doc costo_i = epp_i + cert_i + cons_i + mkt_i + doc_i # ACTUAL epp_a = float(ov_actual['epp']) if ov_actual.get('epp') is not None else total_inscritos * epp_u cert_a = float(ov_actual['certificado']) if ov_actual.get('certificado') is not None else en_curso * cert_u cons_a = float(ov_actual['consumibles']) if ov_actual.get('consumibles') is not None else total_inscritos * cons_u mkt_a = float(ov_actual['marketing']) if ov_actual.get('marketing') is not None else mkt doc_a = float(ov_actual['docente']) if ov_actual.get('docente') is not None else doc costo_a = epp_a + cert_a + cons_a + mkt_a + doc_a vi = float(curso.get('Valor_Venta', 0.0)) va = float(curso.get('Valor_Venta_Actual', 0.0)) mbi = (1 - costo_i / vi) * 100 if vi > 0 else 0.0 mba = (1 - costo_a / va) * 100 if va > 0 else 0.0 return { 'num_indice': num_idx, 'venta_inicial': vi, 'costo_inicial': costo_i, 'mb_inicial': mbi, 'venta_actual': va, 'costo_actual': costo_a, 'mb_actual': mba, 'epp_inicial': epp_i, 'cert_inicial': cert_i, 'cons_inicial': cons_i, 'mkt_inicial': mkt_i, 'doc_inicial': doc_i, 'epp_actual': epp_a, 'cert_actual': cert_a, 'cons_actual': cons_a, 'mkt_actual': mkt_a, 'doc_actual': doc_a, # Lo que ya estaba en Supabase, para detectar qué fue editado 'ov_inicial': ov_inicial, 'ov_actual': ov_actual, }