# modules/cobranza/processor.py from datetime import datetime import requests import unicodedata class CobranzaProcessor: def __init__(self, data_manager): self.data_manager = data_manager self.config_sedes = self._cargar_json_sedes() # ── Caché de consultas SQL ──────────────────────────────────────── self._cache_cronograma = None self._cache_facturas = None self._cache_ano = None def _obtener_datos_cached(self, ano): """Devuelve cronograma y facturas usando caché si el año coincide.""" if self._cache_ano == ano and self._cache_cronograma is not None: print(f"[CACHE] Reutilizando datos en caché para año {ano}") return self._cache_cronograma, self._cache_facturas print(f"[CACHE] Consultando BD para año {ano}...") cronograma = self.data_manager.ejecutar_consulta_cronograma_cobranza(ano) facturas = self.data_manager.ejecutar_consulta_facturas_cobranza(ano) self._cache_cronograma = cronograma self._cache_facturas = facturas self._cache_ano = ano return cronograma, facturas def invalidar_cache(self): """Fuerza recarga desde BD en la siguiente consulta.""" self._cache_cronograma = None self._cache_facturas = None self._cache_ano = None print("🧹 [CACHE] Memoria limpiada por actualización automática. La próxima consulta irá directo a la BD.") # ========================================================================= # CARGA DE CONFIGURACIÓN (sede.json) # ========================================================================= def _cargar_json_sedes(self): datos = None try: url = getattr(self.data_manager, 'github_sede_url', None) if url: response = requests.get(url) response.raise_for_status() datos = response.json() except Exception as e: pass if datos is None: datos = { "clasificacion_sedes": { "AREQUIPA": {"patrones": ["AREQUIPA"]}, "PIURA": {"patrones": ["PIURA"]}, "TRUJILLO": {"patrones": ["TRUJILLO"]} }, "clasificacion_default": {"sede": "LIMA"} } self._reemplazos_nombre = datos.get("reemplazos_nombre", []) return datos # ========================================================================= # REEMPLAZOS DE NOMBRE DE PROGRAMA # ========================================================================= @staticmethod def _normalizar(texto): texto = unicodedata.normalize('NFD', str(texto)) return texto.encode('ascii', 'ignore').decode('utf-8') def _aplicar_reemplazos_nombre(self, texto): if not texto or not self._reemplazos_nombre: return texto resultado = texto for entrada in self._reemplazos_nombre: buscar = str(entrada.get("buscar", "")).strip() reemplazar = str(entrada.get("reemplazar", "")).strip() if not buscar: continue buscar_norm = self._normalizar(buscar).upper() idx = self._normalizar(resultado).upper().find(buscar_norm) while idx != -1: resultado = resultado[:idx] + reemplazar + resultado[idx + len(buscar):] idx = self._normalizar(resultado).upper().find(buscar_norm, idx + len(reemplazar)) return resultado # ========================================================================= # CLASIFICACIÓN DE SEDE # ========================================================================= def clasificar_sede(self, dsc_programa): if not dsc_programa: return self.config_sedes.get("clasificacion_default", {}).get("sede", "LIMA") dsc_prog_upper = str(dsc_programa).upper() sedes = self.config_sedes.get("clasificacion_sedes", {}) for nombre_sede, datos in sedes.items(): for patron in datos.get("patrones", []): if patron.upper() in dsc_prog_upper: return nombre_sede return self.config_sedes.get("clasificacion_default", {}).get("sede", "LIMA") # ========================================================================= # UTILIDADES # ========================================================================= 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 def obtener_fechas_filtro(self, ano, mes_numero): ano_int = int(ano) mes_int = int(mes_numero) fecha_filtro = datetime(ano_int, mes_int, 1) if mes_int == 12: fecha_filtro_siguiente = datetime(ano_int + 1, 1, 1) else: fecha_filtro_siguiente = datetime(ano_int, mes_int + 1, 1) return fecha_filtro, fecha_filtro_siguiente def calcular_importe_final(self, fila): try: imp_total = float(fila.get('imp_total', 0) or 0) except: imp_total = 0.0 try: imp_dscto = float(fila.get('imp_dscto', 0) or 0) except: imp_dscto = 0.0 return imp_total - imp_dscto def evaluar_ctaxcob_suma(self, fila, importe_final, fecha_filtro, fecha_filtro_siguiente, hoy, debug_stats): sectorista = str(fila.get('dsc_sectorista', '')) if sectorista == " , " or sectorista.strip() == ",": return 0.0, 0.0 try: num_cuota = int(fila.get('num_cuota', 0)) except: num_cuota = 0 if not (num_cuota > 1): debug_stats['falla_cuota'] += 1 return 0.0, 0.0 fch_vencimiento = self.parse_fecha(fila.get('fch_vencimiento')) if not fch_vencimiento: debug_stats['falla_vencimiento'] += 1 return 0.0, 0.0 es_anterior = fch_vencimiento < fecha_filtro es_curso = (fch_vencimiento >= fecha_filtro) and (fch_vencimiento < fecha_filtro_siguiente) if not (es_anterior or es_curso): debug_stats['falla_vencimiento'] += 1 return 0.0, 0.0 fch_inicio = self.parse_fecha(fila.get('fch_inicio')) if not fch_inicio or not (fch_inicio <= hoy): debug_stats['falla_inicio'] += 1 return 0.0, 0.0 estado = str(fila.get('estado_matricula', '')).strip().upper() if estado in ["ALU", "PRE"]: pass elif estado == "RET": fch_retiro = self.parse_fecha(fila.get('fch_retiro')) if not fch_retiro or not (fch_retiro >= fecha_filtro_siguiente): debug_stats['falla_estado'] += 1 return 0.0, 0.0 else: debug_stats['falla_estado'] += 1 return 0.0, 0.0 importe_convertido = importe_final moneda = str(fila.get('cod_moneda', '')).strip().upper() if moneda == 'DOL': try: tc = float(fila.get('imp_tc', 1.0) or 1.0) except: tc = 1.0 importe_convertido *= tc debug_stats['pasa_todo'] += 1 if es_anterior: return importe_convertido, 0.0 if es_curso: return 0.0, importe_convertido return 0.0, 0.0 def evaluar_ctaxcob_resta(self, fila_factura, fecha_filtro, fecha_filtro_siguiente, hoy, debug_stats, tc_correcto): sectorista = str(fila_factura.get('dsc_sectorista', '')) if sectorista == " , " or sectorista.strip() == ",": return 0.0, 0.0, 0.0, 0.0 try: num_cuota = int(fila_factura.get('num_cuota', 0)) except: num_cuota = 0 if not (num_cuota > 1): debug_stats['falla_cuota'] += 1 return 0.0, 0.0, 0.0, 0.0 cod_estado = str(fila_factura.get('cod_estado', '')).strip().upper() if cod_estado != "CAN": debug_stats['falla_estado_can'] += 1 return 0.0, 0.0, 0.0, 0.0 flg_nc = str(fila_factura.get('flg_nc', '')).strip().upper() if flg_nc == "SI": debug_stats['falla_nc'] += 1 return 0.0, 0.0, 0.0, 0.0 fch_venc_cuota = self.parse_fecha(fila_factura.get('fch_vencimiento_cuota')) if not fch_venc_cuota: debug_stats['falla_venc_cuota'] += 1 return 0.0, 0.0, 0.0, 0.0 es_anterior = fch_venc_cuota < fecha_filtro es_curso = (fch_venc_cuota >= fecha_filtro) and (fch_venc_cuota < fecha_filtro_siguiente) if not (es_anterior or es_curso): debug_stats['falla_venc_cuota'] += 1 return 0.0, 0.0, 0.0, 0.0 fch_cancelacion = self.parse_fecha(fila_factura.get('fch_cancelacion')) if not fch_cancelacion: debug_stats['falla_cancelacion'] += 1 return 0.0, 0.0, 0.0, 0.0 pasa_anterior = es_anterior and (fch_cancelacion < fecha_filtro) pasa_curso = es_curso and (fch_cancelacion < fecha_filtro) pasa_cobrado_ant = es_anterior and (fch_cancelacion >= fecha_filtro) and (fch_cancelacion < fecha_filtro_siguiente) pasa_cobrado_curso = es_curso and (fch_cancelacion >= fecha_filtro) and (fch_cancelacion < fecha_filtro_siguiente) if not (pasa_anterior or pasa_curso or pasa_cobrado_ant or pasa_cobrado_curso): debug_stats['falla_cancelacion'] += 1 return 0.0, 0.0, 0.0, 0.0 fch_inicio = self.parse_fecha(fila_factura.get('fch_inicio')) if not fch_inicio or not (fch_inicio <= hoy): debug_stats['falla_inicio'] += 1 return 0.0, 0.0, 0.0, 0.0 estado_mat = str(fila_factura.get('estado_matricula', '')).strip().upper() if estado_mat in ["ALU", "PRE"]: pass elif estado_mat == "RET": fch_retiro = self.parse_fecha(fila_factura.get('fch_retiro')) if not fch_retiro or not (fch_retiro >= fecha_filtro_siguiente): debug_stats['falla_estado_mat'] += 1 return 0.0, 0.0, 0.0, 0.0 else: debug_stats['falla_estado_mat'] += 1 return 0.0, 0.0, 0.0, 0.0 try: imp_emitido = float(fila_factura.get('imp_emitido', 0) or 0) except: imp_emitido = 0.0 imp_emitido_convertido = imp_emitido moneda = str(fila_factura.get('cod_moneda', '')).strip().upper() if moneda == 'DOL': imp_emitido_convertido *= tc_correcto debug_stats['pasa_todo'] += 1 val_resta_ant = imp_emitido_convertido if pasa_anterior else 0.0 val_resta_cur = imp_emitido_convertido if pasa_curso else 0.0 val_cobrado_ant = imp_emitido_convertido if pasa_cobrado_ant else 0.0 val_cobrado_curso = imp_emitido_convertido if pasa_cobrado_curso else 0.0 return val_resta_ant, val_resta_cur, val_cobrado_ant, val_cobrado_curso # ========================================================================= # LISTA DE SECTORISTAS # ========================================================================= def obtener_lista_sectoristas(self, ano, mes): fecha_filtro, fecha_filtro_siguiente = self.obtener_fechas_filtro(ano, mes) hoy = datetime.now() try: cronograma_raw = self.data_manager.ejecutar_consulta_cronograma_cobranza(ano) facturas_raw = self.data_manager.ejecutar_consulta_facturas_cobranza(ano) except Exception: return ["TODOS"] sectoristas_neto = {} stats_suma = {'falla_cuota': 0, 'falla_vencimiento': 0, 'falla_inicio': 0, 'falla_estado': 0, 'pasa_todo': 0} stats_resta = {'falla_cuota': 0, 'falla_estado_can': 0, 'falla_nc': 0, 'falla_cancelacion': 0, 'falla_venc_cuota': 0, 'falla_inicio': 0, 'falla_estado_mat': 0, 'falla_treatas': 0, 'pasa_todo': 0} indices_cronograma = set() mapa_tc = {} mapa_sectorista = {} for fila in cronograma_raw: sec_str = str(fila.get('dsc_sectorista', '')) sec_clean = sec_str.strip() if sec_clean in ["", ",", "None"] or sec_str == " , ": continue idx = str(fila.get('num_indice', '')).strip() if idx: indices_cronograma.add(idx) mapa_sectorista[idx] = sec_clean mat = str(fila.get('num_matricula', '')).strip() if mat and mat not in mapa_tc: try: tc = float(fila.get('imp_tc', 1.0) or 1.0) except: tc = 1.0 mapa_tc[mat] = tc for fila in cronograma_raw: idx = str(fila.get('num_indice', '')).strip() if idx not in indices_cronograma: continue sec_clean = mapa_sectorista[idx] importe_final = self.calcular_importe_final(fila) suma_ant, suma_cur = self.evaluar_ctaxcob_suma(fila, importe_final, fecha_filtro, fecha_filtro_siguiente, hoy, stats_suma) if sec_clean not in sectoristas_neto: sectoristas_neto[sec_clean] = 0.0 sectoristas_neto[sec_clean] += (suma_ant + suma_cur) for fila in facturas_raw: idx_factura = str(fila.get('num_indice', '')).strip() if idx_factura not in indices_cronograma: continue sec_clean = mapa_sectorista.get(idx_factura, "") if not sec_clean: continue mat_factura = str(fila.get('num_matricula', '')).strip() tc_correcto = mapa_tc.get(mat_factura, 1.0) resta_ant, resta_cur, _, _ = self.evaluar_ctaxcob_resta( fila, fecha_filtro, fecha_filtro_siguiente, hoy, stats_resta, tc_correcto ) if sec_clean in sectoristas_neto: sectoristas_neto[sec_clean] -= (resta_ant + resta_cur) lista_final = [] for sec, total_cta_x_cob in sectoristas_neto.items(): if round(total_cta_x_cob, 2) > 0: lista_final.append(sec) lista_final = sorted(lista_final) lista_final.insert(0, "TODOS") return lista_final # ========================================================================= # DATOS PROCESADOS (TABLA PRINCIPAL CON VISTA DE ASESORES) # ========================================================================= def obtener_datos_procesados(self, ano, mes, sectorista_filtro="TODOS", agrupacion="SEDE"): fecha_filtro, fecha_filtro_siguiente = self.obtener_fechas_filtro(ano, mes) hoy = datetime.now() datos_agrupados = {} frecuencias_cuota = {} frecuencias_fecha_venc = {} mapa_frecuencia = {} alumnos_tracking = {} try: cronograma_raw, facturas_raw = self._obtener_datos_cached(ano) except Exception as e: print(f"Error crítico: {e}") return [] stats_suma = {'falla_cuota': 0, 'falla_vencimiento': 0, 'falla_inicio': 0, 'falla_estado': 0, 'pasa_todo': 0} stats_resta = {'falla_cuota': 0, 'falla_estado_can': 0, 'falla_nc': 0, 'falla_cancelacion': 0, 'falla_venc_cuota': 0, 'falla_inicio': 0, 'falla_estado_mat': 0, 'falla_treatas': 0, 'pasa_todo': 0} indices_cronograma = set() mapa_tc = {} mapa_programa = {} mapa_sectorista = {} # <-- NUEVO MAPA PARA CORREGIR ASESORES for fila in cronograma_raw: sectorista_str = str(fila.get('dsc_sectorista', '')) sectorista_clean = sectorista_str.strip() idx = str(fila.get('num_indice', '')).strip() if idx: # Guardamos el asesor real del cronograma para usarlo en las facturas mapa_sectorista[idx] = sectorista_clean if sectorista_clean and sectorista_clean not in [",", "None"] else "SIN ASESOR" if sectorista_clean in [",", "", "None"] or sectorista_str == " , ": continue if sectorista_filtro != "TODOS" and sectorista_clean != sectorista_filtro: continue if idx: indices_cronograma.add(idx) mat = str(fila.get('num_matricula', '')).strip() if mat: if mat not in mapa_tc: try: tc = float(fila.get('imp_tc', 1.0) or 1.0) except: tc = 1.0 mapa_tc[mat] = tc if mat not in mapa_programa: prog = str(fila.get('dsc_det_programa', '')).strip() mapa_programa[mat] = prog if (prog and prog != "None") else "SIN PROGRAMA" for fila in cronograma_raw: idx = str(fila.get('num_indice', '')).strip() if idx not in indices_cronograma: continue mat = str(fila.get('num_matricula', '')).strip() if agrupacion == "SEDE": llave = self.clasificar_sede(fila.get('dsc_programa', '')) elif agrupacion == "ASESOR": llave = mapa_sectorista.get(idx, "SIN ASESOR") else: llave = str(fila.get('dsc_det_programa', '')).strip() if not llave or llave == "None": llave = "SIN PROGRAMA" if llave not in datos_agrupados: datos_agrupados[llave] = {"Suma_Ant": 0.0, "Suma_Cur": 0.0, "Resta_Ant": 0.0, "Resta_Cur": 0.0, "Cob_Ant": 0.0, "Cob_Cur": 0.0} importe_final = self.calcular_importe_final(fila) suma_ant, suma_cur = self.evaluar_ctaxcob_suma(fila, importe_final, fecha_filtro, fecha_filtro_siguiente, hoy, stats_suma) datos_agrupados[llave]["Suma_Ant"] += suma_ant datos_agrupados[llave]["Suma_Cur"] += suma_cur clave_mat = (mat, llave) if clave_mat not in alumnos_tracking: alumnos_tracking[clave_mat] = {"suma": 0.0, "resta": 0.0, "cobrado": 0.0, "cuota_final": None, "fecha_final": None} alumnos_tracking[clave_mat]["suma"] += (suma_ant + suma_cur) if agrupacion == "PROGRAMA": if llave not in mapa_frecuencia: frec_val = str(fila.get('cod_frecuencia', '-')).strip() mapa_frecuencia[llave] = frec_val if frec_val and frec_val != 'None' and frec_val != '' else '-' fch_venc_raw = fila.get('fch_vencimiento') fch_venc_dt = self.parse_fecha(fch_venc_raw) if fch_venc_dt and (fecha_filtro <= fch_venc_dt < fecha_filtro_siguiente): try: n_cuota = int(fila.get('num_cuota', -1)) if n_cuota >= 0: alumnos_tracking[clave_mat]["cuota_final"] = n_cuota except: pass try: alumnos_tracking[clave_mat]["fecha_final"] = fch_venc_dt.strftime('%d/%m/%Y') except: pass for fila in facturas_raw: idx_factura = str(fila.get('num_indice', '')).strip() if idx_factura not in indices_cronograma: continue mat_factura = str(fila.get('num_matricula', '')).strip() tc_correcto = mapa_tc.get(mat_factura, 1.0) if agrupacion == "SEDE": llave = self.clasificar_sede(fila.get('dsc_programa', '')) elif agrupacion == "ASESOR": # AQUÍ ESTÁ LA MAGIA: usa el asesor guardado desde el cronograma llave = mapa_sectorista.get(idx_factura, "SIN ASESOR") else: llave = mapa_programa.get(mat_factura, "") if not llave: llave = str(fila.get('dsc_det_programa', '')).strip() if not llave or llave == "None": llave = "SIN PROGRAMA" if llave not in datos_agrupados: datos_agrupados[llave] = {"Suma_Ant": 0.0, "Suma_Cur": 0.0, "Resta_Ant": 0.0, "Resta_Cur": 0.0, "Cob_Ant": 0.0, "Cob_Cur": 0.0} resta_ant, resta_cur, cobrado_ant, cobrado_curso = self.evaluar_ctaxcob_resta( fila, fecha_filtro, fecha_filtro_siguiente, hoy, stats_resta, tc_correcto ) datos_agrupados[llave]["Resta_Ant"] += resta_ant datos_agrupados[llave]["Resta_Cur"] += resta_cur datos_agrupados[llave]["Cob_Ant"] += cobrado_ant datos_agrupados[llave]["Cob_Cur"] += cobrado_curso clave_mat = (mat_factura, llave) if clave_mat in alumnos_tracking: alumnos_tracking[clave_mat]["resta"] += (resta_ant + resta_cur) alumnos_tracking[clave_mat]["cobrado"] += (cobrado_ant + cobrado_curso) for (mat, llave), track in alumnos_tracking.items(): saldo_alumno = (track["suma"] - track["resta"]) - track["cobrado"] if round(saldo_alumno, 2) > 0: c = track["cuota_final"] f = track["fecha_final"] if c is not None: if llave not in frecuencias_cuota: frecuencias_cuota[llave] = {} frecuencias_cuota[llave][c] = frecuencias_cuota[llave].get(c, 0) + 1 if f is not None: if llave not in frecuencias_fecha_venc: frecuencias_fecha_venc[llave] = {} frecuencias_fecha_venc[llave][f] = frecuencias_fecha_venc[llave].get(f, 0) + 1 datos_limpios = [] llaves_ordenadas = sorted(datos_agrupados.keys()) asesores_validos = [] if agrupacion == "ASESOR": asesores_validos = [s for s in self.obtener_lista_sectoristas(ano, mes) if s != "TODOS"] if agrupacion == "SEDE": orden_sedes = ["LIMA", "AREQUIPA", "TRUJILLO", "PIURA"] llaves_ordenadas = [s for s in orden_sedes if s in datos_agrupados] + [s for s in llaves_ordenadas if s not in orden_sedes] for llave in llaves_ordenadas: if agrupacion == "ASESOR" and llave not in asesores_validos: continue totales = datos_agrupados[llave] cta_cob_ant = totales["Suma_Ant"] - totales["Resta_Ant"] cta_cob_curso = totales["Suma_Cur"] - totales["Resta_Cur"] total_cta = cta_cob_ant + cta_cob_curso cobrado_meses_ant = totales["Cob_Ant"] cobrado_mes_curso = totales["Cob_Cur"] total_cobrado_sede = cobrado_meses_ant + cobrado_mes_curso if round(total_cta, 2) <= 0 and round(total_cobrado_sede, 2) <= 0: continue ratio_1 = (cobrado_meses_ant / cta_cob_ant) * 100 if cta_cob_ant > 0 else 0.0 ratio_2 = (cobrado_mes_curso / cta_cob_curso) * 100 if cta_cob_curso > 0 else 0.0 ratio_3 = (total_cobrado_sede / total_cta) * 100 if total_cta > 0 else 0.0 saldo_sede = total_cta - total_cobrado_sede frec_str = "-" moda_cuota = "-" moda_fecha = "-" if agrupacion == "PROGRAMA": frec_str = mapa_frecuencia.get(llave, "-") if llave in frecuencias_cuota and frecuencias_cuota[llave]: moda_cuota = str(max(frecuencias_cuota[llave], key=frecuencias_cuota[llave].get)) if llave in frecuencias_fecha_venc and frecuencias_fecha_venc[llave]: moda_fecha = str(max(frecuencias_fecha_venc[llave], key=frecuencias_fecha_venc[llave].get)) if agrupacion == "PROGRAMA": grupo_display = self._aplicar_reemplazos_nombre(llave) else: grupo_display = llave datos_limpios.append({ 'GRUPO': grupo_display, 'GRUPO_ORIGINAL': llave, 'FRECUENCIA': frec_str, 'NUM_CUOTA': moda_cuota, 'FCH_VENC_MODA': moda_fecha, 'CTA_COB_ANT': cta_cob_ant, 'CTA_COB_MES_CURSO': cta_cob_curso, 'TOTAL_CTA_COB': total_cta, 'COB_ANT': cobrado_meses_ant, 'RATIO_1': ratio_1, 'COB_MES_CURSO': cobrado_mes_curso, 'RATIO_2': ratio_2, 'TOTAL_COBRADO': total_cobrado_sede, 'RATIO_3': ratio_3, 'SALDO': saldo_sede, 'OPCIONES': " ≡ ▼ " }) if agrupacion == "PROGRAMA": def orden_personalizado_programa(x): fecha_str = x['FCH_VENC_MODA'] if fecha_str == "-": return (0, datetime.min, x['GRUPO']) else: try: dt = datetime.strptime(fecha_str, '%d/%m/%Y') return (1, dt, x['GRUPO']) except: return (1, datetime.max, x['GRUPO']) return sorted(datos_limpios, key=orden_personalizado_programa) return datos_limpios # ========================================================================= # DESGLOSE POR ALUMNO/MATRÍCULA # ========================================================================= def obtener_detalle_programa(self, ano, mes, sectorista_filtro, programa_nombre): fecha_filtro, fecha_filtro_siguiente = self.obtener_fechas_filtro(ano, mes) hoy = datetime.now() try: cronograma_raw, facturas_raw = self._obtener_datos_cached(ano) except Exception: return [] stats_suma = {'falla_cuota': 0, 'falla_vencimiento': 0, 'falla_inicio': 0, 'falla_estado': 0, 'pasa_todo': 0} stats_resta = {'falla_cuota': 0, 'falla_estado_can': 0, 'falla_nc': 0, 'falla_cancelacion': 0, 'falla_venc_cuota': 0, 'falla_inicio': 0, 'falla_estado_mat': 0, 'falla_treatas': 0, 'pasa_todo': 0} indices_cronograma = set() mat_validas = set() mapa_tc = {} mapa_programa_frec = {} for fila in cronograma_raw: sectorista_str = str(fila.get('dsc_sectorista', '')).strip() if sectorista_str in [",", ""] or fila.get('dsc_sectorista') == " , ": continue if sectorista_filtro != "TODOS" and sectorista_str != sectorista_filtro: continue prog = str(fila.get('dsc_det_programa', '')).strip() if not prog or prog == "None": prog = "SIN PROGRAMA" if programa_nombre != "TODOS": if self._aplicar_reemplazos_nombre(prog) != programa_nombre: continue idx = str(fila.get('num_indice', '')).strip() if idx: indices_cronograma.add(idx) mat = str(fila.get('num_matricula', '')).strip() if mat: mat_validas.add(mat) if mat not in mapa_tc: try: tc = float(fila.get('imp_tc', 1.0) or 1.0) except: tc = 1.0 mapa_tc[mat] = tc if mat not in mapa_programa_frec: prog_bd = str(fila.get('dsc_det_programa', '')).strip() frec_bd = str(fila.get('cod_frecuencia', '')).strip() mapa_programa_frec[mat] = { "prog": prog_bd if prog_bd and prog_bd != "None" else "-", "frec": frec_bd if frec_bd and frec_bd != "None" else "-" } mapa_alumnos = {} if mat_validas: try: conn = self.data_manager.get_connection() cursor = conn.cursor() lista_mats = list(mat_validas) for i in range(0, len(lista_mats), 1000): chunk = lista_mats[i:i+1000] chunk_str = ",".join([f"'{m}'" for m in chunk]) sql = f""" SELECT sgeca_matricula.num_matricula, sgema_alumno.dsc_apellido_paterno + ' ' + sgema_alumno.dsc_apellido_materno + ', ' + sgema_alumno.dsc_nombres AS dsc_alumno FROM sgeca_matricula INNER JOIN sgema_alumno ON sgeca_matricula.cod_alumno = sgema_alumno.cod_alumno WHERE sgeca_matricula.num_matricula IN ({chunk_str}) """ cursor.execute(sql) for row in cursor.fetchall(): num_mat_bd = str(row[0]).strip() nombre_bd = str(row[1]).strip() mapa_alumnos[num_mat_bd] = nombre_bd conn.close() except Exception as e: print(f"Error consultando nombres de alumnos en BD: {e}") datos_agrupados = {} for fila in cronograma_raw: idx = str(fila.get('num_indice', '')).strip() if idx not in indices_cronograma: continue mat = str(fila.get('num_matricula', '')).strip() if mat not in datos_agrupados: prog_info = mapa_programa_frec.get(mat, {"prog": "-", "frec": "-"}) datos_agrupados[mat] = { "Alumno": mapa_alumnos.get(mat, "SIN NOMBRE"), "Programa": prog_info["prog"], "Frecuencia": prog_info["frec"], "Suma_Ant": 0.0, "Suma_Cur": 0.0, "Resta_Ant": 0.0, "Resta_Cur": 0.0, "Cob_Ant": 0.0, "Cob_Cur": 0.0, "Cuota_Mes": "-", "Fch_Venc_Mes": "-" } fch_venc_raw = fila.get('fch_vencimiento') fch_venc_dt = self.parse_fecha(fch_venc_raw) if fch_venc_dt and (fecha_filtro <= fch_venc_dt < fecha_filtro_siguiente): try: datos_agrupados[mat]["Cuota_Mes"] = str(int(fila.get('num_cuota', -1))) except: pass datos_agrupados[mat]["Fch_Venc_Mes"] = fch_venc_dt.strftime('%d/%m/%Y') importe_final = self.calcular_importe_final(fila) suma_ant, suma_cur = self.evaluar_ctaxcob_suma(fila, importe_final, fecha_filtro, fecha_filtro_siguiente, hoy, stats_suma) datos_agrupados[mat]["Suma_Ant"] += suma_ant datos_agrupados[mat]["Suma_Cur"] += suma_cur for fila in facturas_raw: idx_factura = str(fila.get('num_indice', '')).strip() if idx_factura not in indices_cronograma: continue mat_factura = str(fila.get('num_matricula', '')).strip() tc_correcto = mapa_tc.get(mat_factura, 1.0) if mat_factura not in datos_agrupados: prog_info = mapa_programa_frec.get(mat_factura, {"prog": "-", "frec": "-"}) datos_agrupados[mat_factura] = { "Alumno": mapa_alumnos.get(mat_factura, "SIN NOMBRE"), "Programa": prog_info["prog"], "Frecuencia": prog_info["frec"], "Suma_Ant": 0.0, "Suma_Cur": 0.0, "Resta_Ant": 0.0, "Resta_Cur": 0.0, "Cob_Ant": 0.0, "Cob_Cur": 0.0, "Cuota_Mes": "-", "Fch_Venc_Mes": "-" } resta_ant, resta_cur, cobrado_ant, cobrado_curso = self.evaluar_ctaxcob_resta( fila, fecha_filtro, fecha_filtro_siguiente, hoy, stats_resta, tc_correcto ) datos_agrupados[mat_factura]["Resta_Ant"] += resta_ant datos_agrupados[mat_factura]["Resta_Cur"] += resta_cur datos_agrupados[mat_factura]["Cob_Ant"] += cobrado_ant datos_agrupados[mat_factura]["Cob_Cur"] += cobrado_curso datos_limpios = [] for mat, totales in datos_agrupados.items(): cta_cob_ant = totales["Suma_Ant"] - totales["Resta_Ant"] cta_cob_curso = totales["Suma_Cur"] - totales["Resta_Cur"] total_cta = cta_cob_ant + cta_cob_curso cob_ant = totales["Cob_Ant"] cob_cur = totales["Cob_Cur"] total_cob = cob_ant + cob_cur saldo = total_cta - total_cob if round(total_cta, 2) <= 0 and round(total_cob, 2) <= 0: continue cuota_final = totales["Cuota_Mes"] fecha_final = totales["Fch_Venc_Mes"] if round(saldo, 2) <= 0: cuota_final = "-" fecha_final = "-" datos_limpios.append({ 'MATRICULA': mat, 'ALUMNO': totales["Alumno"], 'PROGRAMA': totales["Programa"], 'FRECUENCIA': totales["Frecuencia"], 'NUM_CUOTA': cuota_final, 'FCH_VENC': fecha_final, 'CTA_COB_ANT': cta_cob_ant, 'CTA_COB_MES_CURSO': cta_cob_curso, 'TOTAL_CTA_COB': total_cta, 'COB_ANT': cob_ant, 'COB_MES_CURSO': cob_cur, 'TOTAL_COBRADO': total_cob, 'SALDO': saldo }) def orden_personalizado(x): fecha_str = x['FCH_VENC'] if fecha_str == "-": return (0, datetime.min, x['ALUMNO']) else: try: dt = datetime.strptime(fecha_str, '%d/%m/%Y') return (1, dt, x['ALUMNO']) except: return (1, datetime.max, x['ALUMNO']) return sorted(datos_limpios, key=orden_personalizado)