Estructura inicial del backend
This commit is contained in:
Panchito
0
modules/ocupabilidad/__init__.py
Normal file
0
modules/ocupabilidad/__init__.py
Normal file
237
modules/ocupabilidad/logic.py
Normal file
237
modules/ocupabilidad/logic.py
Normal file
@@ -0,0 +1,237 @@
|
||||
# modules/ocupabilidad/logic.py
|
||||
import pandas as pd
|
||||
from .processor import CursoProcessor
|
||||
|
||||
class AnalizadorCursos:
|
||||
"""Analizador de cursos - Lógica de cálculo y estadísticas"""
|
||||
|
||||
def __init__(self, data_manager):
|
||||
self.data_manager = data_manager
|
||||
self.curso_processor = CursoProcessor(self.data_manager)
|
||||
self.columnas = self.curso_processor.columnas
|
||||
|
||||
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 get_current_month(self):
|
||||
return self.data_manager.get_current_month()
|
||||
|
||||
def get_current_year(self):
|
||||
return self.data_manager.get_current_year()
|
||||
|
||||
def actualizar_configuracion(self):
|
||||
print("🔄 Actualización automática de configuración...")
|
||||
self.data_manager.cargar_toda_configuracion()
|
||||
|
||||
# --- AGREGADO EL PARÁMETRO "TIPO" ---
|
||||
def obtener_datos_procesados(self, ano, mes, sede="TODOS", filtro_prog="TODOS"):
|
||||
# Sincronizamos el estado del toggle UI con el procesador
|
||||
self.curso_processor.mostrar_reprogramados = getattr(self, 'mostrar_reprogramados', True)
|
||||
|
||||
datos = self.curso_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', '')))
|
||||
# Inyectamos variables para el filtro UI
|
||||
d['Sede'] = self.identificar_sede(prog)
|
||||
d['Filtro_Programa'] = self.identificar_filtro_programa(prog)
|
||||
|
||||
if sede == "TODOS" or d['Sede'] == sede:
|
||||
programas_disponibles.add(d['Filtro_Programa'])
|
||||
|
||||
# Ordenar lista dinámica
|
||||
orden_deseado = ["SEMINARIOS", "OTROS", "TEAC", "TERC"]
|
||||
self.programas_disponibles_filtro = [p for p in orden_deseado if p in programas_disponibles]
|
||||
|
||||
# Auto-corrección si el programa no existe en la sede actual
|
||||
if filtro_prog != "TODOS" and filtro_prog not in programas_disponibles:
|
||||
filtro_prog = "TODOS"
|
||||
self.filtro_corregido = "TODOS"
|
||||
else:
|
||||
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)
|
||||
return datos_filtrados
|
||||
|
||||
return datos
|
||||
|
||||
def formatear_datos_para_tabla(self, datos):
|
||||
if not datos:
|
||||
return [], {}
|
||||
|
||||
sheet_data = []
|
||||
totales = {
|
||||
'mes': 0, 'retirados': 0, 'totales': 0,
|
||||
'activos': 0, 'pc': 0, 'continuidad': 0, 'meta': 0,
|
||||
'descuento': 0
|
||||
}
|
||||
|
||||
for registro in datos:
|
||||
fila = []
|
||||
for col in self.columnas:
|
||||
if col != 'Avance_Inscritos':
|
||||
valor = registro.get(col, '')
|
||||
fila.append(str(valor))
|
||||
self._acumular_totales(totales, col, registro)
|
||||
else:
|
||||
total_insc = registro.get('Inscritos_Totales', 0)
|
||||
meta = registro.get('Meta_Curso', 0)
|
||||
|
||||
if total_insc == "-" or meta == "-":
|
||||
fila.append("-")
|
||||
else:
|
||||
porcentaje = self.calcular_porcentaje_avance(total_insc, meta)
|
||||
fila.append(self.crear_barra_texto_color(porcentaje))
|
||||
|
||||
sheet_data.append(fila)
|
||||
|
||||
fila_totales = self.crear_fila_totales(totales)
|
||||
sheet_data.append(fila_totales)
|
||||
|
||||
return sheet_data, totales
|
||||
|
||||
def _acumular_totales(self, totales, col, registro):
|
||||
val = registro.get(col, 0)
|
||||
if val == "-": return # Ignoramos los guiones para que no rompa la suma
|
||||
|
||||
if col == 'Descuento':
|
||||
totales['descuento'] += val
|
||||
elif col == 'Inscritos_Mes':
|
||||
totales['mes'] += val
|
||||
elif col == 'Retirados':
|
||||
totales['retirados'] += val
|
||||
elif col == 'Inscritos_Totales':
|
||||
totales['totales'] += val
|
||||
elif col == 'Inscritos_Activos':
|
||||
totales['activos'] += val
|
||||
elif col == 'Inscritos_PC':
|
||||
totales['pc'] += val
|
||||
elif col == 'Inscritos_Continuidad':
|
||||
totales['continuidad'] += val
|
||||
elif col == 'Meta_Curso':
|
||||
totales['meta'] += val
|
||||
|
||||
def crear_fila_totales(self, totales):
|
||||
porcentaje = self.calcular_porcentaje_avance(totales['totales'], totales['meta'])
|
||||
# REORGANIZADO SEGÚN NUEVO ORDEN DE COLUMNAS SOLICITADO:
|
||||
# [0]Prog, [1]Fch, [2]Dias, [3]Mes, [4]Total, [5]Retirados, [6]Activos, [7]PC, [8]Refriperu, [9]Cont, [10]Meta, [11]Avance
|
||||
return [
|
||||
"TOTAL GENERAL", # PROGRAMA
|
||||
"", # FECHA INICIO
|
||||
"", # DIAS PARA INICIO
|
||||
str(totales['mes']), # INSCRITOS MES
|
||||
str(totales['totales']), # TOTAL INSCRITOS
|
||||
str(totales['retirados']), # RETIRADOS
|
||||
str(totales['activos']), # INSCRITOS EN CURSO
|
||||
str(totales['pc']), # INSCRITOS P.C
|
||||
str(totales['descuento']), # INSCRITOS REFRIPERU (Descuento)
|
||||
str(totales['continuidad']), # INSCRITOS CONTINUIDAD
|
||||
str(totales['meta']), # META
|
||||
self.crear_barra_texto_color(porcentaje) # AVANCE INSCRITOS
|
||||
]
|
||||
|
||||
def calcular_porcentaje_avance(self, activos, meta):
|
||||
return (activos / meta * 100) if meta > 0 else 0
|
||||
|
||||
def crear_barra_texto_color(self, porcentaje):
|
||||
longitud = 10
|
||||
llenas = int((min(porcentaje, 100) / 100) * longitud)
|
||||
vacias = longitud - llenas
|
||||
barra = '█' * llenas + '░' * vacias
|
||||
porc_text = f"{porcentaje:.1f}%"
|
||||
espacios = " " * (6 - len(porc_text))
|
||||
return f"{porc_text}{espacios}{barra}"
|
||||
|
||||
def get_intervalo_actualizacion(self):
|
||||
return self.data_manager.config_data.get(
|
||||
'config_general', {}
|
||||
).get('auto_update_minutos', 5) * 60000
|
||||
|
||||
def exportar_a_excel(self, datos, ano, mes):
|
||||
df = pd.DataFrame(datos)
|
||||
df_exportar = df[self.columnas]
|
||||
archivo = f"cursos_{ano}_{mes}.xlsx"
|
||||
df_exportar.to_excel(archivo, index=False)
|
||||
return archivo
|
||||
|
||||
def calcular_metricas_generales(self, datos):
|
||||
if not datos: return {}
|
||||
# Filtramos las filas fantasmas para no romper la matemática de Pandas
|
||||
datos_validos = [d for d in datos if d.get('dias_para_inicio') != 'REPROGRAMADO']
|
||||
if not datos_validos: return {}
|
||||
|
||||
df = pd.DataFrame(datos_validos)
|
||||
metricas = {
|
||||
'total_cursos': len(datos), # Mostramos el total real de filas (incluyendo fantasmas)
|
||||
'total_inscritos': df['Inscritos_Activos'].sum(),
|
||||
'total_meta': df['Meta_Curso'].sum(),
|
||||
'porcentaje_avance_general': (df['Inscritos_Activos'].sum() / df['Meta_Curso'].sum() * 100) if df['Meta_Curso'].sum() > 0 else 0,
|
||||
'cursos_sobre_meta': len(df[df['Inscritos_Activos'] >= df['Meta_Curso']]),
|
||||
'cursos_bajo_meta': len(df[df['Inscritos_Activos'] < df['Meta_Curso']]),
|
||||
'total_inscritos_pc': df['Inscritos_PC'].sum()
|
||||
}
|
||||
return metricas
|
||||
|
||||
def obtener_top_programas(self, datos, top_n=5):
|
||||
if not datos: return []
|
||||
# Filtramos las filas fantasmas
|
||||
datos_validos = [d for d in datos if d.get('dias_para_inicio') != 'REPROGRAMADO']
|
||||
if not datos_validos: return []
|
||||
|
||||
df = pd.DataFrame(datos_validos)
|
||||
df['Porcentaje_Avance'] = df.apply(
|
||||
lambda x: self.calcular_porcentaje_avance(x['Inscritos_Activos'], x['Meta_Curso']),
|
||||
axis=1
|
||||
)
|
||||
top_programas = df.nlargest(top_n, 'Porcentaje_Avance')[
|
||||
['programa_frecuencia', 'Inscritos_Activos', 'Meta_Curso', 'Porcentaje_Avance']
|
||||
].to_dict('records')
|
||||
return top_programas
|
||||
|
||||
def calcular_tendencias_mensuales(self, ano):
|
||||
tendencias = {}
|
||||
for mes in range(1, 13):
|
||||
try:
|
||||
datos_mes = self.obtener_datos_procesados(str(ano), str(mes))
|
||||
if datos_mes:
|
||||
metricas = self.calcular_metricas_generales(datos_mes)
|
||||
tendencias[mes] = metricas
|
||||
except Exception as e:
|
||||
print(f"Error procesando mes {mes}: {e}")
|
||||
continue
|
||||
return tendencias
|
||||
560
modules/ocupabilidad/processor.py
Normal file
560
modules/ocupabilidad/processor.py
Normal file
@@ -0,0 +1,560 @@
|
||||
# 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
|
||||
Reference in New Issue
Block a user