767 lines
40 KiB
Python
767 lines
40 KiB
Python
# modules/ventas/logic.py
|
|
import pandas as pd
|
|
from datetime import datetime
|
|
import requests
|
|
import re
|
|
|
|
class VentasLogic:
|
|
"""Clase que maneja la lógica de negocio y procesamiento de datos para Ventas"""
|
|
|
|
def __init__(self, data_manager):
|
|
self.data_manager = data_manager
|
|
self._config_sedes = None
|
|
|
|
def _clasificar_sede_comisiones(self, dsc_programa):
|
|
"""Clasifica la sede usando EXACTAMENTE la misma lógica de Cobranza (sede.json)."""
|
|
try:
|
|
if not hasattr(self, '_cob_processor') or self._cob_processor is None:
|
|
from modules.cobranza.processor import CobranzaProcessor
|
|
self._cob_processor = CobranzaProcessor(self.data_manager)
|
|
return self._cob_processor.clasificar_sede(dsc_programa)
|
|
except Exception:
|
|
prog_up = str(dsc_programa or "").upper()
|
|
for s in ("AREQUIPA", "PIURA", "TRUJILLO"):
|
|
if s in prog_up:
|
|
return s
|
|
return "LIMA"
|
|
|
|
def buscar_todos_matriculados(self):
|
|
"""Devuelve TODOS los matriculados (todos los años/meses) para el buscador.
|
|
Ejecuta la query de matrículas quitando el filtro de YEAR/MONTH de fch_matricula.
|
|
Retorna filas con el mismo formato que el detalle (para reusar columnas)."""
|
|
import re as _re
|
|
sql_base = self.data_manager.query_matriculas_sql
|
|
if not sql_base:
|
|
return []
|
|
# Quitar las condiciones de año y mes de fch_matricula
|
|
sql = _re.sub(r"AND\s+YEAR\(\s*sgeca_matricula\.fch_matricula\s*\)\s*=\s*\?", "", sql_base, flags=_re.IGNORECASE)
|
|
sql = _re.sub(r"AND\s+MONTH\(\s*sgeca_matricula\.fch_matricula\s*\)\s*=\s*\?", "", sql, flags=_re.IGNORECASE)
|
|
# Por si están sin AND (primera condición)
|
|
sql = _re.sub(r"YEAR\(\s*sgeca_matricula\.fch_matricula\s*\)\s*=\s*\?", "1=1", sql, flags=_re.IGNORECASE)
|
|
sql = _re.sub(r"MONTH\(\s*sgeca_matricula\.fch_matricula\s*\)\s*=\s*\?", "1=1", sql, flags=_re.IGNORECASE)
|
|
# El buscador debe traer TODOS los matriculados excepto ANU (incluye SUS, no pagados, etc.)
|
|
# El buscador muestra TODOS los matriculados, incluso ANU
|
|
sql = _re.sub(r"AND\s+sgeca_matricula\.cod_estado\s+NOT\s+IN\s*\([^)]*\)", "", sql, flags=_re.IGNORECASE)
|
|
sql = _re.sub(r"NOT\s+IN\s*\(\s*'ANU'\s*,\s*'SUS'\s*\)", "NOT IN ('ZZZ')", sql, flags=_re.IGNORECASE)
|
|
sql = _re.sub(r"NOT\s+IN\s*\(\s*'ANU'\s*,\s*'SUS'\s*,\s*'RET'\s*\)", "NOT IN ('ZZZ')", sql, flags=_re.IGNORECASE)
|
|
|
|
conn = self.data_manager.get_connection()
|
|
cursor = conn.cursor()
|
|
cursor.execute(sql) # sin parámetros de fecha
|
|
columns = [c[0] for c in cursor.description]
|
|
datos = [dict(zip(columns, row)) for row in cursor.fetchall()]
|
|
|
|
# Traer fecha de cancelación de la cuota 1 de TODAS las matrículas (cualquier refinanciamiento)
|
|
fechas_canc = {}
|
|
try:
|
|
cursor.execute("""
|
|
SELECT num_matricula, MAX(fch_cancelacion)
|
|
FROM sgede_cronograma_matricula
|
|
WHERE num_cuota = 1 AND fch_cancelacion IS NOT NULL
|
|
GROUP BY num_matricula
|
|
""")
|
|
for r in cursor.fetchall():
|
|
k = str(r[0]).strip()
|
|
if k.endswith('.0'): k = k[:-2]
|
|
fechas_canc[k] = r[1]
|
|
except Exception:
|
|
pass
|
|
conn.close()
|
|
|
|
filas = []
|
|
for d in datos:
|
|
raw_mat = d.get('num_matricula')
|
|
mat_id = ""
|
|
if raw_mat is not None:
|
|
v = str(raw_mat).strip()
|
|
if v.endswith('.0'): v = v[:-2]
|
|
mat_id = v
|
|
alumno = d.get('dsc_alumno', 'SIN NOMBRE')
|
|
# MOSTRAR: detallado (dsc_promocion = sgede_RP_programa.dsc_det_programa)
|
|
programa = (d.get('dsc_promocion') or d.get('dsc_det_programa') or d.get('dsc_programa') or '')
|
|
# CLASIFICAR sede: prioriza dsc_programa
|
|
prog_clasificacion = (d.get('dsc_programa') or d.get('dsc_det_programa') or d.get('dsc_promocion') or '')
|
|
vendedor = d.get('dsc_vendedor', 'SIN VENDEDOR') or 'SIN VENDEDOR'
|
|
inv = d.get('INV_NETA', 0) or 0
|
|
saldo_mat = d.get('imp_saldo_matricula', 0) or 0
|
|
saldo_c1 = d.get('imp_saldo_cuota1', 0) or 0
|
|
def _fdate(v):
|
|
if not v or str(v).strip() in ("","None"): return ""
|
|
if hasattr(v,'strftime'): return v.strftime('%d/%m/%Y')
|
|
s = str(v)[:10]
|
|
if len(s)==10 and s[4]=='-': p=s.split('-'); return f"{p[2]}/{p[1]}/{p[0]}"
|
|
return s
|
|
f_mat = _fdate(d.get('fch_matricula'))
|
|
f_ini = _fdate(d.get('fch_inicio'))
|
|
f_canc = _fdate(fechas_canc.get(mat_id))
|
|
sede = self._clasificar_sede_comisiones(prog_clasificacion)
|
|
# mismo layout que el detalle: 16 columnas
|
|
fila = [
|
|
vendedor, alumno, f_mat, f_canc, "", "", "",
|
|
f"S/ {float(inv):,.0f}", programa, f_ini,
|
|
f"S/ {float(saldo_mat):,.0f}", f"S/ {float(saldo_c1):,.0f}",
|
|
"", "-", sede, mat_id,
|
|
]
|
|
filas.append(fila)
|
|
return filas
|
|
|
|
def obtener_datos_brutos(self, ano, mes_numero):
|
|
datos = self.data_manager.ejecutar_consulta_ventas(str(ano), str(mes_numero))
|
|
return datos or []
|
|
|
|
def obtener_datos_brutos_filtrado(self, ano, mes_numero, sede="TODOS", programa="TODOS"):
|
|
datos = self.data_manager.ejecutar_consulta_ventas(str(ano), str(mes_numero), sede, programa)
|
|
return datos or []
|
|
|
|
# =========================================================================
|
|
# FUNCIONES AUXILIARES
|
|
# =========================================================================
|
|
def clasificar_programa(self, dsc_programa):
|
|
if not dsc_programa: return "OTROS"
|
|
dsc_prog_upper = str(dsc_programa).upper()
|
|
meta_data = self.data_manager.meta_data
|
|
clasificaciones = meta_data.get("clasificacion_programas", {})
|
|
|
|
for clase, data in clasificaciones.items():
|
|
for patron in data.get("patrones", []):
|
|
if patron.upper() in dsc_prog_upper:
|
|
return clase
|
|
return meta_data.get("clasificacion_default", {}).get("categoria", "SEMINARIOS")
|
|
|
|
def identificar_linea_carrera(self, nombre_programa):
|
|
categoria = self.clasificar_programa(nombre_programa)
|
|
if categoria in ["AREQUIPA", "TRUJILLO", "PIURA", "TEAC", "TERC", "CARRERA"]:
|
|
return "CARRERA_COMPATIBLE"
|
|
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
|
|
|
|
# =========================================================================
|
|
# OBTENCIÓN DE DETALLE POR VENDEDOR (LÓGICA PURA)
|
|
# =========================================================================
|
|
def obtener_detalle_vendedor(self, vendedor, ano, mes_numero):
|
|
try:
|
|
key_mes = f"{int(mes_numero):02d}-{ano}"
|
|
lista_raw = self.data_manager.historico_pendientes.get(key_mes, [])
|
|
set_vip_actual = set()
|
|
lista_vip_str = []
|
|
|
|
for item in lista_raw:
|
|
try:
|
|
val = int(item)
|
|
set_vip_actual.add(val)
|
|
lista_vip_str.append(str(val))
|
|
except: pass
|
|
# Lista PURA de BASE_PENDIENTES (sin overrides). Venta Pendientes usa SOLO esto.
|
|
set_lista_pendientes = set(set_vip_actual)
|
|
# Incluir matrículas con override de fecha (igual que ejecutar_consulta_ventas)
|
|
|
|
# Incluir matrículas con override de fecha (igual que ejecutar_consulta_ventas)
|
|
_fov_vip = getattr(self.data_manager, '_fecha_canc_overrides', None) or {}
|
|
for mk in _fov_vip.keys():
|
|
try:
|
|
val = int(float(str(mk)))
|
|
set_vip_actual.add(val)
|
|
if str(val) not in lista_vip_str:
|
|
lista_vip_str.append(str(val))
|
|
except: pass
|
|
|
|
correcciones_mat = self.data_manager.config_data.get("correcciones_matriculas", {})
|
|
correcciones_desc = self.data_manager.config_data.get('correcciones_descuento', {})
|
|
correcciones_cont = self.data_manager.config_data.get('correcciones_continuidad', {})
|
|
|
|
sql_base = self.data_manager.query_matriculas_sql
|
|
if not sql_base:
|
|
return {"Venta Inscritos": [], "Venta P.C": [], "Venta Pendientes": []}
|
|
|
|
sql_in_clause = "(-1)"
|
|
if lista_vip_str:
|
|
sql_in_clause = "(" + ",".join(lista_vip_str) + ")"
|
|
|
|
sql_modificado = re.sub(
|
|
r"YEAR\(\s*sgeca_matricula\.fch_matricula\s*\)\s*=\s*\?",
|
|
r"( (YEAR(sgeca_matricula.fch_matricula) = ?",
|
|
sql_base, flags=re.IGNORECASE
|
|
)
|
|
sql_modificado = re.sub(
|
|
r"MONTH\(\s*sgeca_matricula\.fch_matricula\s*\)\s*=\s*\?",
|
|
rf"""MONTH(sgeca_matricula.fch_matricula) = ?) OR sgeca_matricula.num_matricula IN {sql_in_clause}
|
|
OR EXISTS (SELECT 1 FROM sgede_cronograma_matricula crono
|
|
WHERE crono.cod_localidad = sgeca_matricula.cod_localidad
|
|
AND crono.num_matricula = sgeca_matricula.num_matricula
|
|
AND crono.num_cuota = 1 AND crono.fch_cancelacion IS NOT NULL
|
|
AND YEAR(crono.fch_cancelacion) = ? AND MONTH(crono.fch_cancelacion) = ?) )""",
|
|
sql_modificado, flags=re.IGNORECASE
|
|
)
|
|
|
|
# Permitir matrículas VIP/override aunque estén ANU/SUS (igual que la tabla principal)
|
|
if lista_vip_str:
|
|
sql_modificado = re.sub(
|
|
r"sgeca_matricula\.cod_estado\s+NOT\s+IN\s*\(\s*'ANU'\s*,\s*'SUS'\s*\)",
|
|
rf"(sgeca_matricula.cod_estado NOT IN ('ANU', 'SUS') OR sgeca_matricula.num_matricula IN {sql_in_clause})",
|
|
sql_modificado, flags=re.IGNORECASE
|
|
)
|
|
|
|
conn = self.data_manager.get_connection()
|
|
cursor = conn.cursor()
|
|
|
|
cursor.execute(sql_modificado, ano, mes_numero, ano, mes_numero)
|
|
columns = [column[0] for column in cursor.description]
|
|
datos_raw = [dict(zip(columns, row)) for row in cursor.fetchall()]
|
|
|
|
# DEDUP por num_matricula: el LEFT JOIN con RP_programa puede duplicar filas
|
|
# (una con fch_cancelacion y otra sin). Nos quedamos con UNA sola por matrícula,
|
|
# priorizando la que TENGA fecha de cancelación (verdad del pago).
|
|
_dedup = {}
|
|
for _d in datos_raw:
|
|
_m = str(_d.get('num_matricula', '')).strip()
|
|
if _m.endswith('.0'): _m = _m[:-2]
|
|
_prev = _dedup.get(_m)
|
|
_tiene_f = bool(_d.get('fch_cancelacion_cuota1'))
|
|
if _prev is None:
|
|
_dedup[_m] = _d
|
|
elif _tiene_f and not bool(_prev.get('fch_cancelacion_cuota1')):
|
|
_dedup[_m] = _d
|
|
datos_raw = list(_dedup.values())
|
|
|
|
# 🔥 DESCARGA DEL CRONOGRAMA COMPLETO (SIN IMPORTAR EL MES) 🔥
|
|
lista_mats = list(set([str(d.get('num_matricula')).replace('.0','').strip() for d in datos_raw if d.get('num_matricula')]))
|
|
cuotas_raw = []
|
|
|
|
if lista_mats:
|
|
mats_str_q = ",".join([f"'{m}'" for m in lista_mats])
|
|
try:
|
|
res_cuo = requests.get(self.data_manager.github_cuota_url)
|
|
sql_cuotas_base = res_cuo.text
|
|
|
|
# Convertimos las validaciones de Año y Mes en "1=1" para que no esconda los cronogramas de otros meses
|
|
sql_cuotas_base = re.sub(r"YEAR\s*\([^)]+\)\s*=\s*\?", "1=1", sql_cuotas_base, flags=re.IGNORECASE)
|
|
sql_cuotas_base = re.sub(r"MONTH\s*\([^)]+\)\s*=\s*\?", "1=1", sql_cuotas_base, flags=re.IGNORECASE)
|
|
sql_cuotas_base = re.sub(r"ORDER\s+BY\s+.*$", "", sql_cuotas_base, flags=re.IGNORECASE | re.DOTALL)
|
|
|
|
sql_final_cuotas = f"SELECT * FROM ({sql_cuotas_base}) AS sub_cuotas WHERE num_matricula IN ({mats_str_q})"
|
|
|
|
cursor.execute(sql_final_cuotas)
|
|
cols_cuo = [c[0] for c in cursor.description]
|
|
cuotas_raw = [dict(zip(cols_cuo, row)) for row in cursor.fetchall()]
|
|
except Exception as e:
|
|
print(f"Error DAX Cuotas: {e}")
|
|
|
|
try: cursos_raw = self.data_manager.ejecutar_consulta_cursos(ano, mes_numero)
|
|
except: cursos_raw = []
|
|
|
|
conn.close()
|
|
|
|
# Agrupar Cuotas
|
|
cuotas_por_mat = {}
|
|
for c in cuotas_raw:
|
|
m_id = str(c.get('num_matricula', '')).strip()
|
|
if m_id.endswith('.0'): m_id = m_id[:-2]
|
|
if m_id not in cuotas_por_mat: cuotas_por_mat[m_id] = []
|
|
cuotas_por_mat[m_id].append(c)
|
|
|
|
fecha_inicio_prog = {}
|
|
for c in cursos_raw:
|
|
prog = str(c.get('dsc_programa', '')).strip().upper()
|
|
if prog not in fecha_inicio_prog:
|
|
fecha_inicio_prog[prog] = self.parse_fecha(c.get('fch_inicio'))
|
|
|
|
lista_dnis = [str(d.get('dsc_documento', '')).strip() for d in datos_raw if d.get('dsc_documento')]
|
|
try: datos_historial = self.data_manager.consultar_historial_continuidad(lista_dnis)
|
|
except: datos_historial = {}
|
|
|
|
listas_datos = {
|
|
"Venta Inscritos": {"filas": [], "s_cuota": 0.0, "c_cuota": 0, "s_desc": 0.0, "c_desc": 0, "s_neta": 0.0},
|
|
"Venta P.C": {"filas": [], "s_cuota": 0.0, "c_cuota": 0, "s_desc": 0.0, "c_desc": 0, "s_neta": 0.0},
|
|
"Venta Pendientes":{"filas": [], "s_cuota": 0.0, "c_cuota": 0, "s_desc": 0.0, "c_desc": 0, "s_neta": 0.0}
|
|
}
|
|
|
|
vendedor_target = str(vendedor).upper().replace(".", "").replace(",", "")
|
|
vendedor_target = " ".join(vendedor_target.split())
|
|
traer_todos = (vendedor_target == "__TODOS__")
|
|
|
|
for d in datos_raw:
|
|
vend_name = str(d.get('dsc_vendedor', 'SIN VENDEDOR')).upper().replace(".", "").replace(",", "")
|
|
vend_name_clean = " ".join(vend_name.split())
|
|
|
|
if not traer_todos and vendedor_target not in vend_name_clean and vend_name_clean not in vendedor_target:
|
|
continue
|
|
|
|
estado = str(d.get('estado_matricula', '')).strip().upper()
|
|
# Permitir ANU/otros si tiene override de fecha; si no, solo ALU/PRE/RET
|
|
_rm = d.get('num_matricula')
|
|
_mk = ""
|
|
if _rm is not None:
|
|
_mk = str(_rm).strip()
|
|
if _mk.endswith('.0'): _mk = _mk[:-2]
|
|
_fov_chk = getattr(self.data_manager, '_fecha_canc_overrides', None) or {}
|
|
if estado not in ['ALU', 'PRE', 'RET'] and _mk not in _fov_chk: continue
|
|
|
|
raw_mat = d.get('num_matricula')
|
|
mat_id = ""
|
|
matricula_int = -1
|
|
if raw_mat is not None:
|
|
val_str = str(raw_mat).strip()
|
|
if val_str.endswith('.0'): val_str = val_str[:-2]
|
|
mat_id = val_str
|
|
try: matricula_int = int(float(val_str))
|
|
except: pass
|
|
|
|
dni = str(d.get('dsc_documento', '')).strip()
|
|
alumno_nombre = d.get('dsc_alumno', 'SIN NOMBRE')
|
|
nombre_prog_crudo = str(d.get('dsc_programa', '')).strip().upper()
|
|
|
|
inv_neta_raw = float(d.get('INV_NETA', 0.0))
|
|
cod_moneda = str(d.get('cod_moneda', 'SOL')).strip().upper()
|
|
|
|
# Buscar imp_tc desde el CRONOGRAMA (no del comprobante)
|
|
imp_tc = 0
|
|
cuotas_alumno = cuotas_por_mat.get(mat_id, [])
|
|
for c_aux in cuotas_alumno:
|
|
try:
|
|
tc_aux = float(c_aux.get('imp_tc', 0) or 0)
|
|
if tc_aux >= 2: # Tomar el primer TC válido del cronograma
|
|
imp_tc = tc_aux
|
|
break
|
|
except: continue
|
|
if imp_tc < 2: imp_tc = 3.45 # TC por defecto si no hay cronograma o es inválido
|
|
|
|
saldo_mat = float(d.get('imp_saldo_matricula', 0.0))
|
|
saldo_c1 = float(d.get('imp_saldo_cuota1', 0.0))
|
|
fch_canc_raw = d.get('fch_cancelacion_cuota1')
|
|
fch_canc_real_sql = d.get('fch_cancelacion_cuota1') # SQL puro, sin override
|
|
fch_mat = d.get('fch_matricula')
|
|
|
|
# Override de fecha cancelación 1 (Supabase) → prioridad para clasificación + cuenta como pagado
|
|
_fov = getattr(self.data_manager, '_fecha_canc_overrides', None) or {}
|
|
_mat_key = str(mat_id).strip()
|
|
if _mat_key.endswith('.0'): _mat_key = _mat_key[:-2]
|
|
tiene_override_fecha = _mat_key in _fov
|
|
fecha_vaciada_override = False # override "borrar" → tiene prioridad sobre todo
|
|
if tiene_override_fecha:
|
|
s = str(_fov[_mat_key]).strip()
|
|
if s == "__VACIO__":
|
|
# Fecha borrada a propósito → anula la del SQL (no cuenta como pagado)
|
|
fch_canc_raw = None
|
|
tiene_override_fecha = False
|
|
fecha_vaciada_override = True
|
|
else:
|
|
# Normalizar dd/mm/yyyy o d/m/yyyy → yyyy-mm-dd
|
|
if '/' in s:
|
|
p = s.split('/')
|
|
if len(p) == 3:
|
|
s = f"{p[2]}-{int(p[1]):02d}-{int(p[0]):02d}"
|
|
fch_canc_raw = s
|
|
|
|
if mat_id in correcciones_mat:
|
|
datos_corregidos = correcciones_mat[mat_id]
|
|
if "imp_tc" in datos_corregidos: imp_tc = float(datos_corregidos["imp_tc"])
|
|
if "imp_saldo_cuota1" in datos_corregidos: saldo_c1 = float(datos_corregidos["imp_saldo_cuota1"])
|
|
if "imp_saldo_matricula" in datos_corregidos: saldo_mat = float(datos_corregidos["imp_saldo_matricula"])
|
|
# Supabase (vaciado) tiene prioridad: la corrección manual NO restaura la fecha
|
|
if "fch_cancelacion_cuota1" in datos_corregidos and not fecha_vaciada_override:
|
|
fch_canc_raw = datos_corregidos["fch_cancelacion_cuota1"]
|
|
|
|
# Prioridad TC: Supabase (mes) > corrección > cronograma
|
|
_tc_ov = getattr(self.data_manager, '_tc_override_mes', None)
|
|
if _tc_ov: imp_tc = float(_tc_ov)
|
|
|
|
if cod_moneda == 'DOL': inv_neta_soles = round(inv_neta_raw * imp_tc, 2)
|
|
else: inv_neta_soles = round(inv_neta_raw, 2)
|
|
|
|
row_year, row_month = -1, -1
|
|
fecha_mat_limpia = ""
|
|
if fch_mat:
|
|
try:
|
|
if hasattr(fch_mat, 'strftime'):
|
|
row_year, row_month = fch_mat.year, fch_mat.month
|
|
fecha_mat_limpia = fch_mat.strftime('%d/%m/%Y')
|
|
else:
|
|
f_obj = datetime.strptime(str(fch_mat)[:10], '%Y-%m-%d')
|
|
row_year, row_month = f_obj.year, f_obj.month
|
|
fecha_mat_limpia = f_obj.strftime('%d/%m/%Y')
|
|
except: fecha_mat_limpia = str(fch_mat)[:10]
|
|
|
|
fch_canc_limpia = ""
|
|
if fch_canc_raw and str(fch_canc_raw).strip() != "None":
|
|
try:
|
|
if hasattr(fch_canc_raw, 'strftime'): fch_canc_limpia = fch_canc_raw.strftime('%d/%m/%Y')
|
|
else:
|
|
tmp = str(fch_canc_raw)[:10]
|
|
if len(tmp) == 10 and tmp[4] == '-': p = tmp.split('-'); fch_canc_limpia = f"{p[2]}/{p[1]}/{p[0]}"
|
|
else: fch_canc_limpia = tmp
|
|
except: fch_canc_limpia = str(fch_canc_raw)[:10]
|
|
|
|
clase_alumno = self.clasificar_programa(nombre_prog_crudo)
|
|
linea_actual = self.identificar_linea_carrera(nombre_prog_crudo)
|
|
fecha_inicio_actual = fecha_inicio_prog.get(nombre_prog_crudo)
|
|
|
|
es_refriperu = False
|
|
if mat_id in correcciones_desc:
|
|
if str(correcciones_desc[mat_id]).upper() == "SI": es_refriperu = True
|
|
elif str(correcciones_desc[mat_id]).upper() == "NO": es_refriperu = False
|
|
else:
|
|
if clase_alumno in ["AREQUIPA", "TRUJILLO", "PIURA"] and 1200 <= inv_neta_soles <= 1700: es_refriperu = True
|
|
elif clase_alumno in ["TEAC", "TERC", "CARRERA"] and 1400 <= inv_neta_soles <= 1900: es_refriperu = True
|
|
|
|
tipo_final = "NUEVO"
|
|
if es_refriperu: tipo_final = "REFRIPERU"
|
|
else:
|
|
es_cont = False
|
|
if mat_id in correcciones_cont:
|
|
if str(correcciones_cont[mat_id]).upper() == "SI": es_cont = True
|
|
elif str(correcciones_cont[mat_id]).upper() == "NO": es_cont = False
|
|
else:
|
|
if fecha_inicio_actual and linea_actual and (dni in datos_historial):
|
|
for antecedente in datos_historial[dni]:
|
|
nombre_pasado = str(antecedente.get('programa', ''))
|
|
fecha_pasada = antecedente.get('fecha')
|
|
if not fecha_pasada: continue
|
|
if not (fecha_pasada < fecha_inicio_actual): continue
|
|
dias_diff = (fecha_inicio_actual - fecha_pasada).days
|
|
if dias_diff < 60: continue
|
|
linea_pasada = self.identificar_linea_carrera(nombre_pasado)
|
|
if linea_pasada == linea_actual:
|
|
es_cont = True
|
|
break
|
|
if es_cont: tipo_final = "CONTINUIDAD"
|
|
|
|
# ===============================================================
|
|
# 🔥 MATEMÁTICA PURA EXACTAMENTE COMO LA PEDISTE 🔥
|
|
# ===============================================================
|
|
suma_cuotas = 0.0
|
|
mis_cuotas = cuotas_por_mat.get(mat_id, [])
|
|
|
|
for c_dict in mis_cuotas:
|
|
try: num_c = int(c_dict.get('num_cuota', 0))
|
|
except: num_c = 0
|
|
|
|
# REGLA: Si el número de cuota es mayor a 0, sumar (imp_total - imp_dscto)
|
|
if num_c > 0:
|
|
t_val = c_dict.get('imp_total', 0)
|
|
d_val = c_dict.get('imp_dscto', 0)
|
|
|
|
try: t_monto = float(t_val) if t_val is not None else 0.0
|
|
except: t_monto = 0.0
|
|
try: d_monto = float(d_val) if d_val is not None else 0.0
|
|
except: d_monto = 0.0
|
|
|
|
imp_c = t_monto - d_monto
|
|
|
|
# Conversión si es Dólares
|
|
tc_over = None
|
|
if mat_id in correcciones_mat and "imp_tc" in correcciones_mat[mat_id]:
|
|
tc_over = float(correcciones_mat[mat_id]["imp_tc"])
|
|
|
|
cod_m_c = str(c_dict.get('cod_moneda', 'SOL')).strip().upper()
|
|
if cod_m_c == "DOL":
|
|
if tc_over is not None: tc_c = tc_over
|
|
else:
|
|
try: tc_c = float(c_dict.get('imp_tc', 0) or 0)
|
|
except: tc_c = 0
|
|
if tc_c < 2: tc_c = 3.45 # TC por defecto si es inválido
|
|
imp_c *= tc_c
|
|
|
|
suma_cuotas += imp_c
|
|
|
|
# ===============================================================
|
|
# DIVISIÓN SEGÚN CLASIFICACIÓN
|
|
# ===============================================================
|
|
if suma_cuotas > 0:
|
|
if clase_alumno in ["TEAC", "TERC", "CARRERA"]:
|
|
valor_cuota_final = suma_cuotas / 5.0
|
|
elif clase_alumno in ["AREQUIPA", "TRUJILLO", "PIURA"]:
|
|
valor_cuota_final = suma_cuotas / 6.0
|
|
else:
|
|
valor_cuota_final = suma_cuotas
|
|
|
|
str_cuota_val = f"S/ {valor_cuota_final:,.0f}"
|
|
else:
|
|
valor_cuota_final = 0.0
|
|
# Si no tiene cuotas mayores a 0, se queda en blanco limpio
|
|
str_cuota_val = ""
|
|
|
|
# ===============================================================
|
|
# LÓGICA DESC. ESPECIAL (Sobre la suma pura)
|
|
# ===============================================================
|
|
desc_e_raw = 0.0
|
|
if suma_cuotas >= 1900:
|
|
if clase_alumno in ["TEAC", "TERC", "CARRERA"]: d_calc = 3400 - suma_cuotas
|
|
elif clase_alumno in ["AREQUIPA", "PIURA", "TRUJILLO"]: d_calc = 2740 - suma_cuotas
|
|
else: d_calc = 0
|
|
|
|
if d_calc > 0:
|
|
desc_e_raw = min(max(d_calc - 200, 0.0), 200.0)
|
|
|
|
str_desc_e_val = f"S/ {desc_e_raw:,.0f}" if desc_e_raw > 0 else ""
|
|
|
|
# ==========================================
|
|
# DISTRIBUCIÓN EN LAS 3 LISTAS
|
|
# ==========================================
|
|
vendedor_real = d.get('dsc_vendedor', 'SIN VENDEDOR')
|
|
if not vendedor_real: vendedor_real = 'SIN VENDEDOR'
|
|
|
|
# Campos extra para el popup de Comisiones (mismo alumno / num_indice)
|
|
# NOTA: la query trae el programa DETALLADO con alias 'dsc_promocion'
|
|
# (sgede_RP_programa.dsc_det_programa AS dsc_promocion).
|
|
# MOSTRAR: prioriza el detallado (dsc_promocion)
|
|
det_programa = (d.get('dsc_promocion') or d.get('dsc_det_programa')
|
|
or d.get('dsc_programa') or '')
|
|
# CLASIFICAR (sede/filtro): prioriza dsc_programa
|
|
prog_clasificacion = (d.get('dsc_programa') or d.get('dsc_det_programa')
|
|
or d.get('dsc_promocion') or '')
|
|
fch_ini_raw = d.get('fch_inicio', '')
|
|
fch_ini_limpia = ""
|
|
if fch_ini_raw and str(fch_ini_raw).strip() != "None":
|
|
try:
|
|
if hasattr(fch_ini_raw, 'strftime'): fch_ini_limpia = fch_ini_raw.strftime('%d/%m/%Y')
|
|
else:
|
|
t = str(fch_ini_raw)[:10]
|
|
if len(t) == 10 and t[4] == '-':
|
|
p = t.split('-'); fch_ini_limpia = f"{p[2]}/{p[1]}/{p[0]}"
|
|
else: fch_ini_limpia = t
|
|
except: fch_ini_limpia = str(fch_ini_raw)[:10]
|
|
str_saldo_mat = f"S/ {saldo_mat:,.0f}" if saldo_mat else "S/ 0"
|
|
str_saldo_c1 = f"S/ {saldo_c1:,.0f}" if saldo_c1 else "S/ 0"
|
|
|
|
# ── Cálculos extra para Comisiones ──────────────────────────
|
|
# DÍAS ANTICIPACIÓN = Fecha Inicio - Fecha Cancelación 1 (en días)
|
|
dias_anticipacion = "-"
|
|
try:
|
|
def _to_date(v):
|
|
if not v or str(v).strip() in ("", "None"): return None
|
|
if hasattr(v, 'year'): return datetime(v.year, v.month, v.day)
|
|
s = str(v)[:10]
|
|
if len(s) == 10 and s[4] == '-': return datetime.strptime(s, '%Y-%m-%d')
|
|
if len(s) == 10 and s[2] == '/':
|
|
p = s.split('/'); return datetime(int(p[2]), int(p[1]), int(p[0]))
|
|
return None
|
|
di = _to_date(fch_ini_raw)
|
|
dc = _to_date(fch_canc_raw)
|
|
if di and dc:
|
|
dias_anticipacion = str((di - dc).days)
|
|
except Exception:
|
|
dias_anticipacion = "-"
|
|
|
|
# TIPO PROGRAMA y sede (clasificación tipo Cobranza vía sede.json)
|
|
sede_alumno = self._clasificar_sede_comisiones(prog_clasificacion or nombre_prog_crudo)
|
|
tipo_programa = sede_alumno # según sede.json (LIMA/AREQUIPA/PIURA/TRUJILLO)
|
|
|
|
# VALOR CUOTA ADICIONAL = Promedio cuota - 640 (solo si sede = LIMA, mínimo 0)
|
|
if sede_alumno == "LIMA" and valor_cuota_final > 0:
|
|
valor_adicional = valor_cuota_final - 640
|
|
if valor_adicional < 0:
|
|
valor_adicional = 0
|
|
str_valor_adicional = f"S/ {valor_adicional:,.0f}"
|
|
else:
|
|
str_valor_adicional = "-"
|
|
|
|
fila_final = [
|
|
vendedor_real, alumno_nombre, fecha_mat_limpia, fch_canc_limpia,
|
|
tipo_final, str_cuota_val, str_desc_e_val, f"S/ {inv_neta_soles:,.0f}",
|
|
# Extra para Comisiones (índices 8+): programa, fch inicio, saldos
|
|
det_programa, fch_ini_limpia, str_saldo_mat, str_saldo_c1,
|
|
# Nuevos (índices 12+): días anticipación, valor cuota adicional, tipo programa
|
|
dias_anticipacion, str_valor_adicional, tipo_programa,
|
|
# Índice [15]: num_matricula (clave para overrides de Comisiones)
|
|
mat_id,
|
|
# Índice [16]: nombre para CLASIFICAR (prioriza dsc_programa). Oculto en UI.
|
|
prog_clasificacion
|
|
]
|
|
|
|
es_venta_del_mes = (str(row_year) == str(ano)) and (str(row_month) == str(mes_numero))
|
|
saldos_ok = (saldo_mat == 0 and saldo_c1 == 0)
|
|
if tiene_override_fecha:
|
|
saldos_ok = True
|
|
|
|
def agregar_a_lista(nombre_lista):
|
|
listas_datos[nombre_lista]["filas"].append(fila_final)
|
|
listas_datos[nombre_lista]["s_neta"] += inv_neta_soles
|
|
|
|
if valor_cuota_final > 0:
|
|
listas_datos[nombre_lista]["s_cuota"] += valor_cuota_final
|
|
listas_datos[nombre_lista]["c_cuota"] += 1
|
|
|
|
if desc_e_raw > 0:
|
|
listas_datos[nombre_lista]["s_desc"] += desc_e_raw
|
|
listas_datos[nombre_lista]["c_desc"] += 1
|
|
|
|
|
|
|
|
|
|
|
|
# Sacamos la validación de fecha de cancelación AFUERA para que sirva a ambas listas
|
|
# Sacamos la validación de fecha de cancelación AFUERA para que sirva a ambas listas
|
|
pago_en_fecha = False
|
|
if fch_canc_raw and str(fch_canc_raw).strip() != "None":
|
|
try:
|
|
if isinstance(fch_canc_raw, str):
|
|
f_obj = datetime.strptime(fch_canc_raw[:10], '%Y-%m-%d')
|
|
f_ano, f_mes = f_obj.year, f_obj.month
|
|
else:
|
|
f_ano, f_mes = fch_canc_raw.year, fch_canc_raw.month
|
|
|
|
if str(f_ano) == str(ano) and str(f_mes) == str(mes_numero):
|
|
pago_en_fecha = True
|
|
except: pass
|
|
|
|
# ¿Matrícula de un mes ANTERIOR al filtro?
|
|
matricula_mes_pasado = False
|
|
try:
|
|
ym_mat = int(row_year) * 100 + int(row_month)
|
|
ym_filtro = int(ano) * 100 + int(mes_numero)
|
|
matricula_mes_pasado = ym_mat < ym_filtro
|
|
except: pass
|
|
|
|
if es_venta_del_mes:
|
|
agregar_a_lista("Venta Inscritos")
|
|
if saldos_ok and pago_en_fecha:
|
|
agregar_a_lista("Venta P.C")
|
|
|
|
# VENTA PENDIENTES: SOLO de tu lista BASE_PENDIENTES y que pagaron su 1ª
|
|
# cuota EN EL MES/AÑO DEL FILTRO. Usa la FECHA REAL DEL SQL (sin override).
|
|
_pago_mes_real = False
|
|
if fch_canc_real_sql and str(fch_canc_real_sql).strip() != "None":
|
|
try:
|
|
if isinstance(fch_canc_real_sql, str):
|
|
_fo = datetime.strptime(fch_canc_real_sql[:10], '%Y-%m-%d')
|
|
_fa, _fm = _fo.year, _fo.month
|
|
else:
|
|
_fa, _fm = fch_canc_real_sql.year, fch_canc_real_sql.month
|
|
if int(_fa) == int(ano) and int(_fm) == int(mes_numero):
|
|
_pago_mes_real = True
|
|
except: pass
|
|
if matricula_int in set_lista_pendientes and _pago_mes_real:
|
|
agregar_a_lista("Venta Pendientes")
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# ==========================================
|
|
# ORDENAR Y ARMAR TOTAL GENERAL
|
|
# ==========================================
|
|
resultado_final = {}
|
|
for k, dict_data in listas_datos.items():
|
|
dict_data["filas"].sort(key=lambda x: x[1])
|
|
|
|
p_cuota = (dict_data["s_cuota"] / dict_data["c_cuota"]) if dict_data["c_cuota"] > 0 else 0.0
|
|
p_desc = (dict_data["s_desc"] / dict_data["c_desc"]) if dict_data["c_desc"] > 0 else 0.0
|
|
|
|
str_gral_cuota = f"S/ {p_cuota:,.0f}" if dict_data["c_cuota"] > 0 else ""
|
|
str_gral_desc = f"S/ {p_desc:,.0f}" if dict_data["c_desc"] > 0 else ""
|
|
|
|
dict_data["filas"].append([
|
|
"TOTAL GENERAL", "", "", "", "",
|
|
str_gral_cuota, str_gral_desc, f"S/ {dict_data['s_neta']:,.0f}"
|
|
])
|
|
resultado_final[k] = dict_data["filas"]
|
|
|
|
return resultado_final
|
|
|
|
except Exception as e:
|
|
print(f"Error procesando alumnos para modal Ventas: {e}")
|
|
return {"Venta Inscritos": [], "Venta P.C": [], "Venta Pendientes": []}
|
|
|
|
def formatear_datos_para_tabla(self, datos):
|
|
if not datos:
|
|
return []
|
|
|
|
sheet_data = []
|
|
total_monto = total_pc = total_pendientes = total_avance_pc_total = 0.0
|
|
total_cantidad = total_cantidad_pc = total_cantidad_pendientes = 0
|
|
|
|
for registro in datos:
|
|
vendedor = registro.get('VENDEDOR', 'SIN VENDEDOR')
|
|
monto = float(registro.get('MONTO', 0.0))
|
|
cantidad = int(registro.get('CANTIDAD', 0))
|
|
ventas_pc = float(registro.get('VENTAS_PC', 0.0))
|
|
cantidad_pc = int(registro.get('INSCRITOS_PC', 0))
|
|
pendientes = float(registro.get('PENDIENTES', 0.0))
|
|
cant_pendientes = int(registro.get('INSCRITOS_PENDIENTES', 0))
|
|
|
|
avance_pc_total = ventas_pc + pendientes
|
|
|
|
total_monto += monto; total_cantidad += cantidad; total_pc += ventas_pc
|
|
total_cantidad_pc += cantidad_pc; total_pendientes += pendientes
|
|
total_cantidad_pendientes += cant_pendientes; total_avance_pc_total += avance_pc_total
|
|
|
|
sheet_data.append([
|
|
vendedor, cantidad, f"S/ {monto:,.0f}", cantidad_pc, f"S/ {ventas_pc:,.0f}",
|
|
cant_pendientes, f"S/ {pendientes:,.0f}", f"S/ {avance_pc_total:,.0f}",
|
|
" ≡ ▼ "
|
|
])
|
|
|
|
sheet_data.append([
|
|
"TOTAL GENERAL", total_cantidad, f"S/ {total_monto:,.0f}",
|
|
total_cantidad_pc, f"S/ {total_pc:,.0f}", total_cantidad_pendientes,
|
|
f"S/ {total_pendientes:,.0f}", f"S/ {total_avance_pc_total:,.0f}",
|
|
""
|
|
])
|
|
|
|
return sheet_data
|
|
|
|
def exportar_detalle_alumnos_excel(self, vendedor, lista_nombre, headers, datos):
|
|
if not datos: raise ValueError("No hay datos para exportar")
|
|
vend_limpio = "".join([c if c.isalnum() else "_" for c in str(vendedor)])[:30]
|
|
lista_limpia = "".join([c if c.isalnum() else "_" for c in str(lista_nombre)])
|
|
archivo = f"Ventas_{vend_limpio}_{lista_limpia}.xlsx"
|
|
|
|
df = pd.DataFrame(datos, columns=headers)
|
|
try:
|
|
with pd.ExcelWriter(archivo, engine='openpyxl') as writer:
|
|
df.to_excel(writer, index=False, sheet_name='Detalle_Ventas')
|
|
worksheet = writer.sheets['Detalle_Ventas']
|
|
try:
|
|
from openpyxl.styles import PatternFill, Font, Border, Side, Alignment
|
|
fill_header = PatternFill(start_color="12345F", end_color="12345F", fill_type="solid")
|
|
font_header = Font(color="FFFFFF", bold=True)
|
|
thin_border = Border(left=Side(style='thin', color="DDDDDD"), right=Side(style='thin', color="DDDDDD"),
|
|
top=Side(style='thin', color="DDDDDD"), bottom=Side(style='thin', color="DDDDDD"))
|
|
|
|
for cell in worksheet[1]:
|
|
cell.fill = fill_header
|
|
cell.font = font_header
|
|
cell.border = thin_border
|
|
cell.alignment = Alignment(horizontal="center", vertical="center", wrap_text=True)
|
|
|
|
for row in worksheet.iter_rows(min_row=2):
|
|
for cell in row:
|
|
cell.border = thin_border
|
|
cell.alignment = Alignment(vertical="center")
|
|
|
|
for col in worksheet.columns:
|
|
max_len = 0
|
|
col_letter = col[0].column_letter
|
|
for cell in col:
|
|
if cell.value: max_len = max(max_len, len(str(cell.value)))
|
|
worksheet.column_dimensions[col_letter].width = min(max_len + 3, 50)
|
|
except: pass
|
|
except: df.to_excel(archivo, index=False)
|
|
return archivo
|
|
|
|
def exportar_a_excel(self, datos, ano, nombre_mes):
|
|
if not datos: raise ValueError("No hay datos para exportar")
|
|
datos_export = []
|
|
for d in datos:
|
|
nuevo_d = dict(d)
|
|
nuevo_d['AVANCE_PC_TOTAL'] = float(d.get('VENTAS_PC', 0)) + float(d.get('PENDIENTES', 0))
|
|
datos_export.append(nuevo_d)
|
|
|
|
df = pd.DataFrame(datos_export)
|
|
archivo = f"ventas_{ano}_{nombre_mes}.xlsx"
|
|
df.to_excel(archivo, index=False)
|
|
return archivo |