734 lines
33 KiB
Python
734 lines
33 KiB
Python
# 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) |