Skip to content

05_huecos_ceco_configuracion.py

Ruta original: docs/proyectos/poliza-explor/scripts/05_huecos_ceco_configuracion.py

"""
Detector de huecos de centro de costo en las configuraciones de poliza.

QUE DETECTA
-----------
Los renglones de cargo de una configuracion eligen su cuenta contable con
listas IN de CeCos cableadas a mano en Pcd_Condicion (ver
configuracion-polizas.md). Un gasto cuyo CeCo no cae en NINGUNA de esas
listas no genera renglon -- pero sus contrapartes (abono a acreedores,
impuestos) si se generan. Resultado: la poliza descuadra.

POR QUE ES PREVENTIVO
---------------------
El descuadre casi nunca llega a la tabla Poliza: el motor no genera la
poliza si no cuadra (solo 4 de ~40,000 polizas de 2026 estan descuadradas,
3 canceladas). Auditar el resultado no sirve -- hay que auditar la
configuracion contra los datos que la alimentan.

COMO
----
Reconstruye el query que genera el boton "Generar script" de la pantalla
CT001, a partir de las tres piezas que lo componen:

    FROM   Poliza_Configuracion.Pc_Relacion          (joins de cabecera)
           Poliza_Configuracion_Detalle.Pcd_Relacion (joins del renglon)
    WHERE  Poliza_Configuracion_Condicion.Pcc_Valor  (condiciones globales)
           AND NOT (cond_renglon_1 OR cond_renglon_2 OR ...)

Las filas que sobreviven ese NOT(...) son exactamente los gastos que
entran a la poliza pero no encuentran renglon de cargo. Cero filas = la
configuracion cubre todo su universo.

Esto SUSTITUYE un intento previo por regex (extraer los CeCos citados en
las condiciones y restarlos del catalogo Centro_Costo). Aquel enfoque daba
69 "candidatos" sobre configs en uso; casi todos falsos positivos, porque
una config filtra ademas por grupo de tipo de gasto, proveedor y fecha, y
un CeCo ausente de sus listas simplemente nunca le llega. La version
rigurosa deja 2 configuraciones con huecos reales.

RESULTADO H1 2026 (empresa 0001, 80 configs de Gasto_Registro en uso)
---------------------------------------------------------------------
- 0394 COMPROBACION DE GASTOS NACIONAL: CeCo 000501 PLANTA CONCRETERA.
  Causa confirmada del unico descuadre vivo de 2026 (poliza 0000478288).
- 0371 RECLASIFICACION DE GASTO: 20 CeCos. NO descuadra -- sus renglones
  son simetricos (12 cargo / 12 abono), asi que un CeCo descubierto pierde
  los dos lados a la vez. Falla silenciosa: la poliza cuadra pero la
  reclasificacion no se hace.

Uso:  python3 05_huecos_ceco_configuracion.py [--empresa 0001]
                                              [--desde 2026-01-01]
                                              [--hasta 2026-06-30]
                                              [CONFIG ...]
"""
import argparse
import sys

import pandas as pd

from connection_205_trivasadb3 import q
from helpers_output import console_err, mostrar_tabla_err, resumen_err


def piezas_configuracion(cve, empresa):
    """Las tres piezas del query generado, o None si la config no aplica."""
    cab = q(f"""
        SELECT CAST(Pc_Relacion AS varchar(max)) rel, Pc_Descripcion descr
        FROM Poliza_Configuracion
        WHERE Pc_Cve_Poliza_Configuracion = '{cve}'
    """)
    if cab.empty or not (cab.rel[0] or "").strip():
        return None

    globales = q(f"""
        SELECT CAST(Pcc_Valor AS varchar(max)) v
        FROM Poliza_Configuracion_Condicion
        WHERE Pc_Cve_Poliza_Configuracion = '{cve}'
        ORDER BY Pcc_ID
    """).v.dropna()
    globales = [g.replace("<EMPRESA>", empresa) for g in globales if g.strip()]

    det = q(f"""
        SELECT CAST(Pcd_Condicion AS varchar(max)) cond,
               CAST(Pcd_Relacion AS varchar(max)) rel,
               CAST(Pcd_Cargo AS varchar(max)) cargo
        FROM Poliza_Configuracion_Detalle
        WHERE Pc_Cve_Poliza_Configuracion = '{cve}'
    """)
    # solo renglones de cargo que resuelven cuenta por centro de costo
    cargos = det[
        (det.cargo.fillna("").str.strip() != "")
        & det.rel.fillna("").str.contains("Centro_Costo", na=False)
    ]
    condiciones = [c.strip() for c in cargos.cond.fillna("") if c.strip()]
    if cargos.empty or not condiciones:
        return None

    return cab.descr[0], cab.rel[0], cargos.rel.iloc[0], globales, condiciones


def huecos(cve, empresa, desde, hasta):
    piezas = piezas_configuracion(cve, empresa)
    if piezas is None:
        return None, pd.DataFrame()
    descr, rel_cab, rel_linea, globales, condiciones = piezas

    where = " AND ".join(f"({g})" for g in globales) if globales else "1=1"
    no_cae = " OR ".join(f"({c})" for c in condiciones)

    # sin SUM de importe: la columna de monto cambia segun el modulo
    # (Grc_Importe en gastos directos, otra en consumo interno), y el
    # conteo de renglones basta para priorizar.
    sql = f"""
        SELECT Centro_Costo.Cc_Cve_Centro_Costo ceco,
               MAX(Centro_Costo.Cc_Descripcion) descripcion,
               COUNT(*) renglones
        FROM {rel_cab}
        {rel_linea}
        WHERE {where}
          AND Gasto_Registro.Gr_Fecha BETWEEN '{desde}' AND '{hasta}'
          AND NOT ({no_cae})
        GROUP BY Centro_Costo.Cc_Cve_Centro_Costo
    """
    return descr, q(sql)


def configs_en_uso(desde, hasta):
    return q(f"""
        SELECT DISTINCT c.Pc_Cve_Poliza_Configuracion cve
        FROM Poliza_Configuracion c
        JOIN Poliza p ON p.Pl_Configuracion = c.Pc_Cve_Poliza_Configuracion
        WHERE c.Es_Cve_Estado = 'AC'
          AND c.Pc_Tabla = 'Gasto_Registro'
          AND p.Es_Cve_Estado <> 'CA'
          AND p.Pl_Fecha BETWEEN '{desde}' AND '{hasta}'
        ORDER BY cve
    """).cve.tolist()


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("configs", nargs="*")
    ap.add_argument("--empresa", default="0001")
    ap.add_argument("--desde", default="2026-01-01")
    ap.add_argument("--hasta", default="2026-06-30")
    a = ap.parse_args()

    objetivo = a.configs or configs_en_uso(a.desde, a.hasta)
    console_err.print(
        f"[bold]Auditando {len(objetivo)} configuraciones[/bold] "
        f"({a.desde} a {a.hasta}, empresa {a.empresa})"
    )

    filas, revisadas, errores = [], 0, []
    for cve in objetivo:
        try:
            descr, r = huecos(cve, a.empresa, a.desde, a.hasta)
        except Exception as e:
            errores.append((cve, str(e)[:120]))
            continue
        if descr is None:
            continue
        revisadas += 1
        for _, x in r.iterrows():
            filas.append(
                dict(config=cve, descripcion_config=descr, ceco=x.ceco,
                     descripcion_ceco=x.descripcion, renglones=int(x.renglones))
            )

    df = pd.DataFrame(filas)
    if df.empty:
        console_err.print("[green]Sin huecos: toda config revisada cubre su universo.[/green]")
    else:
        mostrar_tabla_err(
            df.sort_values("renglones", ascending=False),
            "Huecos de CeCo confirmados (gasto entra a la poliza, no halla renglon de cargo)",
            max_filas=50,
        )
    for cve, err in errores:
        console_err.print(f"[yellow]{cve}: {err}[/yellow]")
    resumen_err(revisadas=revisadas, con_huecos=df.config.nunique() if not df.empty else 0,
                pares=len(df), errores=len(errores))

    print("config,descripcion_config,ceco,descripcion_ceco,renglones")
    for _, x in df.iterrows():
        print(f"{x.config},{x.descripcion_config},{x.ceco},{x.descripcion_ceco},{x.renglones}")


if __name__ == "__main__":
    main()