01_conteo_polizas_por_origen.py

Ruta original: docs/proyectos/poliza-explor/scripts/01_conteo_polizas_por_origen.py

"""
Censo de polizas contables (Poliza) por tipo de documento de origen de mpro,
via el campo polimorfico Poliza_Control.Pc_Tabla (patron Xx_Tabla/Xx_Documento,
ver docs/schema/convenciones.md de trivasa-context).

Contexto: Poliza.Pl_Tabla / Pl_Documento existen como columnas pero vienen
vacias en la practica (confirmado con muestra real, enero 2026) -- el origen
real de cada poliza se resuelve solo via Poliza_Control (Pl_Folio -> Pc_Tabla
+ Pc_Documento), tabla puente 1:N (una poliza consolidada agrupa muchos
documentos de origen del mismo tipo, ej. muchos folios de Gasto_Registro en
una sola poliza diaria). Confirmado que la particion por Pc_Tabla es limpia:
sum(polizas) por grupo == total de folios distintos de Poliza en el rango,
sin folios repartidos entre mas de un Pc_Tabla.

Uso:
    python3 01_conteo_polizas_por_origen.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 = """
SELECT DISTINCT
    pl.Pl_Folio,
    pl.Es_Cve_Estado,
    ISNULL(NULLIF(plc.Pc_Tabla, ''), '(sin Poliza_Control)') AS ORIGEN
FROM Poliza pl
LEFT JOIN Poliza_Control plc ON plc.Pl_Folio = pl.Pl_Folio
WHERE pl.Pl_Fecha >= :fecha_ini AND pl.Pl_Fecha < :fecha_fin
"""


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

    df = pd.read_sql(
        text(QUERY), engine,
        params={"fecha_ini": args.fecha_ini, "fecha_fin": args.fecha_fin},
    )

    out = f"01_conteo_polizas_por_origen_{args.fecha_ini}_{args.fecha_fin}.csv"
    df.to_csv(out, index=False)

    por_origen = df.groupby("ORIGEN").agg(
        polizas=("Pl_Folio", "nunique"),
        activas=("Es_Cve_Estado", lambda s: (s == "AC").sum()),
        canceladas=("Es_Cve_Estado", lambda s: (s == "CA").sum()),
    ).sort_values("polizas", ascending=False)
    mostrar_tabla(por_origen.reset_index(), f"Polizas por origen (Pc_Tabla) {args.fecha_ini} a {args.fecha_fin}")

    resumen(
        polizas_totales=df["Pl_Folio"].nunique(),
        tipos_de_origen=df["ORIGEN"].nunique(),
    )
    console.print(f"\n[green]Guardado: {out}[/green]")
    console.print(
        "[dim]Nota: 'polizas' cuenta Pl_Folio distintos (la poliza consolidada), no "
        "documentos de origen -- una sola poliza de GASTO_REGISTRO puede agrupar miles "
        "de folios. Para contar documentos de origen (Pc_Documento) hace falta un query "
        "aparte por tipo, ya que Pc_Documento colisiona entre tipos distintos.[/dim]"
    )


if __name__ == "__main__":
    main()