04_diagnostico_compra_indirecto.py

Ruta original: docs/proyectos/poliza-explor/scripts/04_diagnostico_compra_indirecto.py

"""
Diagnostico Compra_Indirecto: CORRIGE la clasificacion de 01_conteo_polizas_por_origen.py.

Ci_Tabla viene siempre vacio (sin sub-origenes propios), lo que en el
analisis inicial sugeria "0 casos de negocio = simple", igual que Cheque y
Recibo_Pago. FALSO -- Compra_Indirecto comparte el mismo patron de "dos
polizas paralelas" ya documentado para CONSUMO_INTERNO en
docs/schema/calidad-de-datos.md de trivasa-context: cada Ci_Folio genera
DOS polizas activas bajo el mismo Pc_Documento:

1. Poliza "real" (Pl_Comentario NOT LIKE '%CUENTAS ORDEN%') -- Cargo a
   cuenta de almacen/costo (ej. 1140.001.001.003), UN renglon por Ci_Folio
   con el TOTAL sumado: Cargo = SUM(Ci_Precio_Descontado_Importe) de todos
   los Ci_ID de ese folio.
2. Poliza "cuentas de orden" (Pl_Comentario LIKE '%CUENTAS ORDEN%') --
   Cargo a cuenta memo raiz H (ej. 10500.012.011), un renglon POR CADA
   Ci_ID (no sumado), mismo importe (Ci_Precio_Descontado_Importe) pero a
   nivel linea.

Ademas hay pares cancelada->activa del mismo folio (mismo patron de
Cheque) que se resuelven con el filtro estandar Es_Cve_Estado <> 'CA'.

Validado: enero 2026, 5/5 documentos (100%) -- la poliza real cuadra
exacto contra SUM(Ci_Precio_Descontado_Importe) filtrando por
Pl_Comentario. Volumen bajo (5 documentos / 14 polizas activas+canceladas
en enero 2026) -- no se investigo Q1 completo por ser de impacto marginal,
a diferencia de Cheque y Recibo_Pago.

Pendiente: nunca se investigo el otro lado (Abono) de ninguna de las dos
polizas -- este script solo valida el Cargo de la poliza "real".

Uso:
    python3 04_diagnostico_compra_indirecto.py --fecha-ini 2026-01-01 --fecha-fin 2026-02-01
"""
import argparse

from connection_205_trivasadb3 import engine  # default de exploracion -- ver skill trivasa-sql-exploracion

import pandas as pd
from sqlalchemy import text
from helpers_output import console, mostrar_tabla, resumen

QUERY_REAL = """
SELECT plc.Pl_Folio, plc.Pc_Documento, pl.Pl_Comentario,
       SUM(CASE WHEN pd.Pd_Tipo = 1 THEN pd.Pd_Importe ELSE 0 END) AS CARGO
FROM Poliza pl
JOIN Poliza_Control plc ON plc.Pl_Folio = pl.Pl_Folio AND plc.Pc_Tabla = 'COMPRA_INDIRECTO'
JOIN Poliza_Detalle pd ON pd.Pl_Folio = plc.Pl_Folio AND pd.Pd_Referencia = plc.Pc_Documento
WHERE pl.Pl_Fecha >= :fecha_ini AND pl.Pl_Fecha < :fecha_fin
  AND pl.Es_Cve_Estado <> 'CA'
  AND UPPER(pl.Pl_Comentario) NOT LIKE '%CUENTAS ORDEN%'
GROUP BY plc.Pl_Folio, plc.Pc_Documento, pl.Pl_Comentario
"""

QUERY_ESPERADO = """
SELECT Ci_Folio, SUM(Ci_Precio_Descontado_Importe) AS ESPERADO
FROM Compra_Indirecto
GROUP BY Ci_Folio
"""


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--fecha-ini", required=True)
    ap.add_argument("--fecha-fin", required=True)
    args = ap.parse_args()

    real = pd.read_sql(
        text(QUERY_REAL), engine,
        params={"fecha_ini": args.fecha_ini, "fecha_fin": args.fecha_fin},
    )
    esperado = pd.read_sql(text(QUERY_ESPERADO), engine)

    comp = real.merge(esperado, left_on="Pc_Documento", right_on="Ci_Folio", how="left")
    comp["DIFERENCIA"] = (comp["CARGO"] - comp["ESPERADO"]).abs()
    comp["CUADRA"] = comp["DIFERENCIA"] <= 1

    out = f"04_diagnostico_compra_indirecto_{args.fecha_ini}_{args.fecha_fin}.csv"
    comp.to_csv(out, index=False)

    mostrar_tabla(comp, "Poliza real (Compra_Indirecto) vs SUM(Ci_Precio_Descontado_Importe)")
    n, c = len(comp), int(comp["CUADRA"].sum())
    resumen(documentos=n, cuadran=c, pct=(f"{100 * c / n:.2f}%" if n else "n/a"))
    console.print(f"\n[green]Guardado: {out}[/green]")
    console.print(
        "[dim]Nota: solo valida la poliza 'real' (sin CUENTAS ORDEN). Volumen bajo -- "
        "confirmar si vale la pena profundizar la poliza memo y el lado Abono.[/dim]"
    )


if __name__ == "__main__":
    main()