Solucionado submodulos y subiendo codigo real
This commit is contained in:
Panchito
237
backend/modules/ocupabilidad/logic.py
Normal file
237
backend/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
|
||||
Reference in New Issue
Block a user