# 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 # 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()] # 🔥 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_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") # MES PASADO: matrícula anterior + pagó 1° cuota en el mes del filtro if matricula_mes_pasado and saldos_ok and pago_en_fecha: 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