Actualizar backend/services.py

2026-10-09 13:24:57 -05:00
parent d8498dfd34
commit 01cefbecb6
1 changed files with 257 additions and 7 deletions
--- a/backend/services.py
+++ b/backend/services.py
@@ -166,6 +166,227 @@ def _clasif_filtro_programa(dm, dscp):
    return fdata.get("DEFAULT", "OTROS") if isinstance(fdata.get("DEFAULT"), str) else "OTROS"


+def _clasif_sede_fact(dm, dscp):
+    """Clasifica la sede del programa con sede.json (igual que Cobranza/Ocupabilidad)."""
+    up = str(dscp or "").upper()
+    if not up:
+        return ""
+    for s, data in dm.sede_data.get("clasificacion_sedes", {}).items():
+        if s == "DEFAULT":
+            continue
+        for pat in data.get("patrones", []):
+            if pat.upper() in up:
+                return s
+    _def = dm.sede_data.get("clasificacion_default", {})
+    return _def.get("sede", "LIMA") if isinstance(_def, dict) else "LIMA"
+
+
+def _fact_ym(fch):
+    """Devuelve (año, mes) de una fecha (datetime o str 'YYYY-MM-DD')."""
+    if fch is None:
+        return (None, None)
+    try:
+        return (fch.year, fch.month)
+    except Exception:
+        s = str(fch)
+        try:
+            return (int(s[0:4]), int(s[5:7]))
+        except Exception:
+            return (None, None)
+
+
+def facturacion(ano="TODOS", mes="TODOS", cuota="TODOS"):
+    """Matriz de Facturación: Ingresos (facturas) − Notas de Crédito = Neto,
+    agrupado Sede -> Tipo. Clasifica con los JSON (sede.json / REPLACE_META.json).
+    Filtros por fch_emision (año/mes) y num_cuota. La NC hereda programa y cuota
+    de su factura (enlace num_comprobante_BF = num_comprobante). Replica el Power BI."""
+    dm = get_dm()
+    _ano = None if str(ano).upper() == "TODOS" else int(ano)
+    # mes y cuota pueden venir como lista separada por comas (multi-selección).
+    # None = TODOS (sin filtro).
+    _mes = None
+    if str(mes).upper() != "TODOS":
+        _mes = {int(x) for x in str(mes).split(",") if str(x).strip().isdigit()}
+        if not _mes: _mes = None
+    def _ncuo(v):
+        # Normaliza num_cuota a string entero: 0 / 0.0 / "0.0" -> "0".
+        s = str(v if v is not None else "").strip()
+        if s == "": return ""
+        try: return str(int(float(s)))
+        except Exception: return s
+    _cuo = None
+    if str(cuota).upper() != "TODOS":
+        _cuo = {_ncuo(x) for x in str(cuota).split(",") if str(x).strip() != ""}
+        if not _cuo: _cuo = None
+
+    facturas = dm.ejecutar_consulta_facturas_cobranza("ALL") or []
+    notas = dm.ejecutar_consulta_notascredito() or []
+
+    def _num(v):
+        try: return float(v) if v is not None else 0.0
+        except Exception: return 0.0
+
+    # ── Índice de facturas por num_comprobante: programa, matrícula, min cuota ──
+    # (para enlazar NC -> factura). Guardamos el primer programa/matrícula visto
+    # y el MIN(num_cuota) del comprobante (num_cuota_NC del Power BI).
+    fact_idx = {}
+    # Cuántas veces aparece cada num_comprobante (regla IMP_SUBTOTAL_FINAL).
+    comp_rep = {}
+    for f in facturas:
+        if str(f.get("cod_estado") or "").strip().upper() == "ANU": continue
+        nc = str(f.get("num_comprobante") or "").strip()
+        if nc:
+            comp_rep[nc] = comp_rep.get(nc, 0) + 1
+            g = fact_idx.setdefault(nc, {"prog": f.get("dsc_det_programa") or "",
+                                         "glosa": f.get("dsc_glosa") or "",
+                                         "matricula": str(f.get("num_matricula") or "").strip(),
+                                         "min_cuota": None})
+            cu = f.get("num_cuota")
+            if cu is not None:
+                try:
+                    cu = int(cu)
+                    if g["min_cuota"] is None or cu < g["min_cuota"]:
+                        g["min_cuota"] = cu
+                except Exception:
+                    pass
+
+    # ── acumulador: {(sede,tipo,programa,comprobante): {"ing":x, "nc":y}} ──
+    from collections import defaultdict
+    acc = defaultdict(lambda: {"ing": 0.0, "nc": 0.0})
+
+    # ── INGRESOS ──
+    # 1) Filtrar por año/mes (SIN cuota). El conteo de repeticiones del comprobante
+    #    (regla IMP_SUBTOTAL_FINAL) se hace sobre ESTE conjunto, igual que el Power BI:
+    #    el slicer de cuota NO afecta cuántas cuotas tiene realmente la factura.
+    fact_am = []
+    for f in facturas:
+        if str(f.get("cod_estado") or "").strip().upper() == "ANU": continue  # excluir anuladas
+        a, m = _fact_ym(f.get("fch_emision"))
+        if _ano is not None and a != _ano: continue
+        if _mes is not None and m not in _mes: continue
+        fact_am.append(f)
+    # 2) Contar repeticiones de cada comprobante sobre año/mes (SIN filtro de cuota).
+    rep_filt = {}
+    for f in fact_am:
+        comp = str(f.get("num_comprobante") or "").strip()
+        if comp:
+            rep_filt[comp] = rep_filt.get(comp, 0) + 1
+    # 3) Aplicar el filtro de cuota solo para elegir QUÉ filas se suman.
+    fact_filt = fact_am if _cuo is None else [
+        f for f in fact_am if _ncuo(f.get("num_cuota")) in _cuo]
+    # 4) Calcular IMP_SUBTOTAL_FINAL por fila (de las cuotas que pasan el filtro).
+    for f in fact_filt:
+        prog = str(f.get("dsc_det_programa") or "").strip()
+        if not prog:
+            glosa = str(f.get("dsc_glosa") or "").strip() or "(SIN GLOSA)"
+            sede, tipo, pnom = "OTROS INGRESOS", "SIN CLASIFICAR", glosa
+        else:
+            sede = _clasif_sede_fact(dm, prog)
+            tipo = _clasif_filtro_programa(dm, prog)
+            pnom = prog
+        # único -> imp_subtotal; repetido -> imp_emitido
+        comp = str(f.get("num_comprobante") or "").strip()
+        base = _num(f.get("imp_subtotal")) if rep_filt.get(comp, 0) == 1 else _num(f.get("imp_emitido"))
+        if not base:
+            base = _num(f.get("imp_subtotal"))
+        if str(f.get("cod_moneda") or "").upper() != "SOL":
+            base *= _num(f.get("imp_tc"))
+        acc[(sede, tipo, pnom, comp or "(SIN COMPROBANTE)")]["ing"] += base
+
+    # ── NOTAS DE CRÉDITO ──
+    for n in notas:
+        a, m = _fact_ym(n.get("fch_emision"))
+        if _ano is not None and a != _ano: continue
+        if _mes is not None and m not in _mes: continue
+        comp_bf = str(n.get("num_comprobante_BF") or "").strip()
+        fi = fact_idx.get(comp_bf)
+        if not fi:
+            continue   # NC sin factura enlazada -> no clasifica (como BLANK del PBI)
+        # El enlace NC->factura por num_comprobante_BF ya identifica la factura;
+        # de ahí se hereda programa y cuota (como el LOOKUPVALUE del Power BI).
+        # Filtro cuota: usa la cuota de la factura enlazada (num_cuota_NC = MIN)
+        if _cuo is not None and _ncuo(fi["min_cuota"]) not in _cuo:
+            continue
+        prog = str(fi["prog"] or "").strip()
+        if not prog:
+            glosa = str(fi.get("glosa") or "").strip() or "(SIN GLOSA)"
+            sede, tipo, pnom = "OTROS INGRESOS", "SIN CLASIFICAR", glosa
+        else:
+            sede = _clasif_sede_fact(dm, prog)
+            tipo = _clasif_filtro_programa(dm, prog)
+            pnom = prog
+        base = _num(n.get("imp_subtotal"))
+        if str(n.get("cod_moneda") or "").upper() != "SOL":
+            base *= _num(n.get("imp_tc"))
+        acc[(sede, tipo, pnom, comp_bf or "(SIN COMPROBANTE)")]["nc"] += base
+
+    # ── armar jerarquía Sede -> Tipo -> Programa -> Comprobante ──
+    def _n0(): return {"ing": 0.0, "nc": 0.0}
+    def _nprog(): return {"ing": 0.0, "nc": 0.0, "comps": defaultdict(_n0)}
+    por_sede = defaultdict(lambda: {"ing": 0.0, "nc": 0.0,
+                 "tipos": defaultdict(lambda: {"ing": 0.0, "nc": 0.0, "progs": defaultdict(_nprog)})})
+    for (sede, tipo, pnom, comp), v in acc.items():
+        s = por_sede[sede]
+        s["ing"] += v["ing"]; s["nc"] += v["nc"]
+        t = s["tipos"][tipo]
+        t["ing"] += v["ing"]; t["nc"] += v["nc"]
+        p = t["progs"][pnom]
+        p["ing"] += v["ing"]; p["nc"] += v["nc"]
+        c = p["comps"][comp]
+        c["ing"] += v["ing"]; c["nc"] += v["nc"]
+
+    filas = []
+    for sede in sorted(por_sede.keys()):
+        s = por_sede[sede]
+        hijos = []
+        for tipo in sorted(s["tipos"].keys()):
+            t = s["tipos"][tipo]
+            progs = []
+            for pnom in sorted(t["progs"].keys()):
+                p = t["progs"][pnom]
+                comps = []
+                for comp in sorted(p["comps"].keys()):
+                    cc = p["comps"][comp]
+                    comps.append({"comprobante": comp, "ingresos": round(cc["ing"], 2),
+                                  "notas_credito": round(cc["nc"], 2),
+                                  "neto": round(cc["ing"] - cc["nc"], 2)})
+                progs.append({"programa": pnom, "ingresos": round(p["ing"], 2),
+                              "notas_credito": round(p["nc"], 2),
+                              "neto": round(p["ing"] - p["nc"], 2), "comprobantes": comps})
+            hijos.append({"tipo": tipo, "ingresos": round(t["ing"], 2),
+                          "notas_credito": round(t["nc"], 2),
+                          "neto": round(t["ing"] - t["nc"], 2), "programas": progs})
+        filas.append({"sede": sede, "ingresos": round(s["ing"], 2),
+                      "notas_credito": round(s["nc"], 2),
+                      "neto": round(s["ing"] - s["nc"], 2), "tipos": hijos})
+
+    tot = {"ingresos": round(sum(f["ingresos"] for f in filas), 2),
+           "notas_credito": round(sum(f["notas_credito"] for f in filas), 2),
+           "neto": round(sum(f["neto"] for f in filas), 2)}
+    return {"filas": filas, "total": tot}
+
+
+def facturacion_cuotas():
+    """Lista de num_cuota disponibles (para el filtro CUOTA)."""
+    dm = get_dm()
+    facturas = dm.ejecutar_consulta_facturas_cobranza("ALL") or []
+    def _ncuo(v):
+        s = str(v if v is not None else "").strip()
+        if s == "": return ""
+        try: return str(int(float(s)))
+        except Exception: return s
+    cuotas = set()
+    for f in facturas:
+        c = _ncuo(f.get("num_cuota"))
+        if c != "":
+            cuotas.add(c)
+    try:
+        orden = sorted(cuotas, key=lambda x: int(x))
+    except Exception:
+        orden = sorted(cuotas)
+    return {"cuotas": orden}
+
+
 def ventas_detalle(vendedor, ano, mes, tipo_lista, sede="TODOS", programa="TODOS", excluir_ret=False):
    def _load():
        dm = get_dm()
@@ -1222,8 +1443,9 @@ def roles_modulos_disponibles():
    para armar las casillas al crear un rol. Ordenados alfabéticamente."""
    sb = _sb()
    res = sb.table("permisos").select("modulo").execute()
-    mods = sorted({str(r.get("modulo", "")).strip() for r in (res.data or []) if r.get("modulo")})
-    return {"modulos": mods}
+    mods = {str(r.get("modulo", "")).strip() for r in (res.data or []) if r.get("modulo")}
+    mods.add("facturacion")   # módulo nuevo: siempre asignable aunque no esté en la tabla
+    return {"modulos": sorted(mods)}

 def roles_listar():
    """Devuelve todos los roles con la lista de módulos que puede ver cada uno.
@@ -1763,7 +1985,7 @@ def comisiones_export_excel(ano, mes):
        cd.font = total_font; cd.fill = total_fill; cd.border = border
        ce = ws.cell(row=rtot, column=5, value=round(inv1 + inv2, 2))
        ce.font = total_font; ce.fill = total_fill; ce.border = border
-        return rtot + 1, (bool(c1) or bool(c2))
+        return rtot + 1, (bool(c1) or bool(c2)), round(inv1 + inv2, 2)

    for vend in vendedores:
        try:
@@ -1777,10 +1999,13 @@ def comisiones_export_excel(ano, mes):
        filas_pc = _aplicar_overrides_comisiones(list(filas_pc))
        filas_pend = _aplicar_overrides_comisiones(list(filas_pend))

-        # Si el asesor NO tiene datos en ninguno de los dos bloques, NO crear su hoja.
+        # Si el asesor NO tiene datos en ningún bloque (ventas ni deudores), NO crear hoja.
        c1_pc, c2_pc = _clasificar(filas_pc)
        c1_pp, c2_pp = _clasificar(filas_pend)
-        if not (c1_pc or c2_pc or c1_pp or c2_pp):
+        _dm_raw = det.get("Deudores Mes", []) if isinstance(det, dict) else []
+        _da_raw = det.get("Deudores Anterior", []) if isinstance(det, dict) else []
+        _tiene_deudores = any(f and str(f[0]).strip().upper() != "TOTAL GENERAL" for f in (_dm_raw + _da_raw))
+        if not (c1_pc or c2_pc or c1_pp or c2_pp or _tiene_deudores):
            continue

        # Acumular para el RESUMEN (MES EN CURSO = Venta P.C; MES ANTERIOR = Pendientes)
@@ -1800,9 +2025,34 @@ def comisiones_export_excel(ano, mes):
            ws.column_dimensions[openpyxl.utils.get_column_letter(i)].width = w

        # BLOQUE 1: MES EN CURSO (Venta P.C)
-        prox, _ = _escribir_bloque(ws, 1, "MES EN CURSO", filas_pc)
+        prox, _, tot_cur = _escribir_bloque(ws, 1, "MES EN CURSO", filas_pc)
        # BLOQUE 2: MESES PASADOS (Venta Pendientes) — 3 filas más abajo
-        _escribir_bloque(ws, prox + 3, "MESES PASADOS", filas_pend)
+        prox2, _, tot_ant = _escribir_bloque(ws, prox + 3, "MESES PASADOS", filas_pend)
+
+        # TOTAL COMISIÓN = suma de los totales de los 2 bloques de arriba
+        rtc = prox2 + 1
+        c_tc = ws.cell(row=rtc, column=4, value="TOTAL COMISIÓN")
+        c_tc.font = Font(bold=True, size=12, color="1F4E79"); c_tc.fill = total_fill; c_tc.border = border
+        c_tcv = ws.cell(row=rtc, column=5, value=round(tot_cur + tot_ant, 2))
+        c_tcv.font = Font(bold=True, size=12); c_tcv.fill = total_fill; c_tcv.border = border
+
+        # Separador visual + título SALDO PENDIENTE antes de los deudores
+        r_sep = rtc + 3
+        c_sp = ws.cell(row=r_sep, column=1, value="SALDO PENDIENTE")
+        c_sp.font = Font(bold=True, size=13, color="FFFFFF")
+        c_sp.fill = PatternFill("solid", fgColor="992B2B")
+        for _c in range(1, 15):
+            cc = ws.cell(row=r_sep, column=_c)
+            if _c != 1:
+                cc.fill = PatternFill("solid", fgColor="992B2B")
+
+        # BLOQUES DE DEUDORES (debían al cierre del mes del filtro; solo ALU/PRE):
+        #   - "NO PAGADOS — MES"      : matriculados en el mes del filtro
+        #   - "NO PAGADOS — ANTERIOR" : matriculados antes del filtro
+        filas_dm = _aplicar_overrides_comisiones(list(det.get("Deudores Mes", []) if isinstance(det, dict) else []))
+        filas_da = _aplicar_overrides_comisiones(list(det.get("Deudores Anterior", []) if isinstance(det, dict) else []))
+        prox3, _, _ = _escribir_bloque(ws, r_sep + 2, "NO PAGADOS — MES", filas_dm)
+        _escribir_bloque(ws, prox3 + 3, "NO PAGADOS — MESES ANTERIORES", filas_da)

    # Agregar VENDEDORES MANUALES al final del RESUMEN: todo su monto va a
    # OTROS → MES EN CURSO. Agrupados por nombre de vendedor manual.