movimientos_uuid_lib.py

Ruta original: docs/proyectos/consumo-interno-fifo/scripts/movimientos_uuid_lib.py

"""
Libreria reusable: enlaza cada SALIDA de movimientos_periodo con el UUID del CFDI de
compra que la respalda, via FIFO "on the fly" (reusa integramente fifo_lib.py, sibling
en este mismo scripts/). Bug de orden corregido aqui, sin tocar fifo_lib.py: fifo_lib
no ordena la cola por su cuenta -- ver el sort explicito en preparar_capas_y_eventos()
mas abajo, antes de invocar fifo_lineas_por_producto().

Usada por:
  - 06_movimientos_uuid_fifo.py (CLI, genera CSV)
  - streamlit_app_uuid_fifo.py (reporte interactivo V2)
"""
import pandas as pd

import fifo_lib as lib  # noqa: E402
from movimientos_lib import reporte_completo  # noqa: E402

TOLERANCIA = 0.5


def preparar_capas_y_eventos(fecha_ini: str, fecha_fin: str, salidas: pd.DataFrame):
    fecha_ini_excl_prev = (pd.Timestamp(fecha_ini) - pd.Timedelta(days=1)).strftime("%Y-%m-%d")
    fecha_fin_excl = (pd.Timestamp(fecha_fin) + pd.Timedelta(days=1)).strftime("%Y-%m-%d")

    claves = sorted(salidas["Pr_Cve_Producto"].unique().tolist())

    semilla = lib.reconstruir_semilla_on_the_fly(claves, fecha_ini_excl_prev)
    nuevas = lib.query_capas(claves, fecha_desde=fecha_ini, fecha_hasta=fecha_fin_excl)

    capas = pd.concat([semilla, nuevas], ignore_index=True)
    # Fix de orden (ver decision doc): fecha_compra ascendente = FIFO real. fifo_lib no
    # ordena por su cuenta -- se ordena aqui, sin tocar la libreria compartida.
    capas = capas.sort_values(["Pr_Cve_Producto", "fecha_compra", "Co_Folio_origen"])
    capas_por_producto = {p: g.copy() for p, g in capas.groupby("Pr_Cve_Producto")}

    eventos = salidas.rename(columns={"Mv_Cantidad_Control_1": "cantidad_neta"})
    eventos = eventos.sort_values(["Pr_Cve_Producto", "Mv_Fecha", "Mv_Folio", "Mv_ID"])
    eventos_por_producto = {p: g.copy() for p, g in eventos.groupby("Pr_Cve_Producto")}

    return capas_por_producto, eventos_por_producto


def correr_fifo(capas_por_producto: dict, eventos_por_producto: dict) -> pd.DataFrame:
    todas_filas = []
    for prod, eventos_prod in eventos_por_producto.items():
        capas_prod = capas_por_producto.get(prod)
        todas_filas.extend(lib.fifo_lineas_por_producto(capas_prod, eventos_prod))
    return pd.DataFrame(todas_filas)


def reporte_completo_uuid_fifo(fecha_ini: str, fecha_fin: str):
    """Todo en uno. Regresa (resultado, salidas, mov, chequeo, descuadres)."""
    mov, resumen_existencia, sin_cuadrar = reporte_completo(fecha_ini, fecha_fin)
    salidas = mov[mov["entrada_salida"] == "Salida"].copy()

    capas_por_producto, eventos_por_producto = preparar_capas_y_eventos(fecha_ini, fecha_fin, salidas)
    asignaciones = correr_fifo(capas_por_producto, eventos_por_producto)

    if asignaciones.empty:
        cols = ["Pr_Cve_Producto", "Mv_Folio", "Mv_ID", "cantidad_asignada",
                "Co_Folio_origen", "fecha_compra", "UUID"]
        return (pd.DataFrame(columns=cols), salidas, mov,
                pd.DataFrame(), pd.DataFrame())

    folios_usados = asignaciones["Co_Folio_origen"].dropna().unique().tolist()
    mapa_uuid = lib.query_uuids(folios_usados)
    asignaciones["UUID"] = asignaciones["Co_Folio_origen"].map(mapa_uuid)

    asignaciones = asignaciones.rename(columns={"Ci_Folio": "Mv_Folio", "Ci_ID": "Mv_ID"})
    salidas_idx = salidas.set_index(["Pr_Cve_Producto", "Mv_Folio", "Mv_ID"])
    asignaciones = asignaciones.set_index(["Pr_Cve_Producto", "Mv_Folio", "Mv_ID"])

    asignaciones["n_asignaciones_del_movimiento"] = asignaciones.groupby(level=[0, 1, 2]).cumcount() + 1
    total_partes = asignaciones.groupby(level=[0, 1, 2]).size()
    asignaciones["total_partes_del_movimiento"] = asignaciones.index.map(total_partes)

    columnas_movimiento = [c for c in salidas_idx.columns if c not in asignaciones.columns]
    resultado = asignaciones.join(salidas_idx[columnas_movimiento], how="left").reset_index()

    resultado["con_capa_de_compra"] = resultado["Co_Folio_origen"].notna()
    resultado["con_uuid"] = resultado["UUID"].notna()

    chequeo = resultado.groupby(["Pr_Cve_Producto", "Mv_Folio", "Mv_ID"], as_index=False).agg(
        cantidad_asignada_total=("cantidad_asignada", "sum"),
        Mv_Cantidad_Control_1=("Mv_Cantidad_Control_1", "first"),
    )
    chequeo["magnitud_original"] = chequeo["Mv_Cantidad_Control_1"].abs()
    chequeo["diferencia"] = chequeo["cantidad_asignada_total"] - chequeo["magnitud_original"]
    descuadres = chequeo[chequeo["diferencia"].abs() > TOLERANCIA]

    return resultado, salidas, mov, chequeo, descuadres