streamlit_app_uuid_fifo.py
Ruta original: docs/proyectos/consumo-interno-fifo/scripts/streamlit_app_uuid_fifo.py
"""
Movimientos en un periodo · V2 (con UUID de compra vía FIFO) - reporte Streamlit
Extiende "Movimientos en un periodo" (streamlit_app_movimientos.py, V1, que se conserva
sin cambios): cada SALIDA se enlaza con el UUID del CFDI de compra que la respalda, vía
FIFO "on the fly" (reusa fifo_lib.py, sibling en este mismo scripts/). Ver
docs/schema/inventario.md (raiz de este repo) y el index.md de este proyecto.
Corre standalone con:
streamlit run streamlit_app_uuid_fifo.py --server.port 8505
"""
import io
import os
import sys
from datetime import date
import pandas as pd
import streamlit as st
SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, SCRIPT_DIR)
from movimientos_uuid_lib import reporte_completo_uuid_fifo # noqa: E402
VERSION = "v2"
def _is_standalone():
return not hasattr(st, "_layout_gastos_embedded")
def _mes_anterior_completo():
hoy = date.today()
primero_mes_actual = hoy.replace(day=1)
fin = primero_mes_actual - pd.Timedelta(days=1)
ini = fin.replace(day=1)
return ini, fin
@st.cache_data(ttl=1800, show_spinner="Consultando TRIVASADB3 (movimientos + capas de compra + FIFO)...")
def cargar_reporte(fecha_ini: str, fecha_fin: str):
return reporte_completo_uuid_fifo(fecha_ini, fecha_fin)
def fmt_num(x):
if x is None or pd.isna(x):
return "-"
return f"{x:,.2f}"
def render():
st.title("Movimientos en un periodo · V2 (UUID de compra)")
st.caption(f"Version {VERSION} · Inventario · cada salida enlazada a su UUID de compra vía FIFO")
st.markdown(
"Extiende el reporte de [Movimientos en un periodo](movimientos-periodo): cada "
"**salida** se descuenta de la capa de compra más antigua disponible (FIFO), "
"resolviendo su UUID de factura. Si una salida cruza más de una capa, aparece "
"partida en varias filas -- una por capa tocada."
)
with st.sidebar:
st.subheader("Periodo")
default_ini, default_fin = _mes_anterior_completo()
fecha_ini = st.date_input("Desde", value=default_ini, key="uf_ini")
fecha_fin = st.date_input("Hasta", value=default_fin, key="uf_fin")
if fecha_ini > fecha_fin:
st.error("La fecha 'Desde' debe ser anterior o igual a 'Hasta'.")
return
resultado, salidas, mov, chequeo, descuadres = cargar_reporte(str(fecha_ini), str(fecha_fin))
if resultado.empty:
st.info("No hay salidas para este periodo.")
return
with st.sidebar:
st.subheader("Filtros")
categorias = sorted(resultado["categoria_movimiento"].dropna().unique().tolist())
categoria_sel = st.multiselect("Categoría de movimiento", categorias, default=[])
tipos_doc = sorted(resultado["tipo_documento_origen"].dropna().unique().tolist())
tipo_doc_sel = st.multiselect("Tipo de documento origen", tipos_doc, default=[])
sucursales = sorted(resultado["Sc_Cve_Sucursal"].dropna().unique().tolist())
sucursal_sel = st.multiselect("Sucursal", sucursales, default=[])
cobertura_sel = st.radio(
"Cobertura de compra", ["Todos", "Con capa de compra", "Sin capa de compra"], index=0,
)
solo_partidos = st.checkbox(
"Solo movimientos partidos en 2+ registros",
value=False,
help="Movimientos donde la salida cruzó más de una capa de compra.",
)
producto_buscar = st.text_input("Buscar producto (clave o descripción contiene)")
df = resultado.copy()
if categoria_sel:
df = df[df["categoria_movimiento"].isin(categoria_sel)]
if tipo_doc_sel:
df = df[df["tipo_documento_origen"].isin(tipo_doc_sel)]
if sucursal_sel:
df = df[df["Sc_Cve_Sucursal"].isin(sucursal_sel)]
if cobertura_sel == "Con capa de compra":
df = df[df["con_capa_de_compra"]]
elif cobertura_sel == "Sin capa de compra":
df = df[~df["con_capa_de_compra"]]
if solo_partidos:
df = df[df["total_partes_del_movimiento"] > 1]
if producto_buscar:
needle = producto_buscar.strip().upper()
df = df[
df["Pr_Cve_Producto"].str.upper().str.contains(needle, na=False)
| df["Pr_Descripcion"].str.upper().str.contains(needle, na=False)
]
cantidad_con_capa = resultado.loc[resultado["con_capa_de_compra"], "cantidad_asignada"].sum()
cantidad_total = resultado["cantidad_asignada"].sum()
pct_cobertura = (cantidad_con_capa / cantidad_total * 100) if cantidad_total else 0
n_partidos = resultado.loc[resultado["total_partes_del_movimiento"] > 1,
["Pr_Cve_Producto", "Mv_Folio", "Mv_ID"]].drop_duplicates().shape[0]
folios_usados = resultado["Co_Folio_origen"].dropna().nunique()
folios_con_uuid = resultado.loc[resultado["con_uuid"], "Co_Folio_origen"].nunique()
k1, k2, k3, k4, k5 = st.columns(5)
k1.metric("Filas (filtrado)", f"{len(df):,}")
k2.metric("Cobertura de compra", f"{pct_cobertura:.1f}%", help="% de la cantidad total con UUID de compra resuelto")
k3.metric("Movimientos partidos", f"{n_partidos:,}", help="Cruzaron 2+ capas de compra")
k4.metric("Folios con UUID", f"{folios_con_uuid} / {folios_usados}")
k5.metric(
"Cobertura FIFO exacta",
f"{len(chequeo) - len(descuadres)} / {len(chequeo)}",
help="Movimientos donde la suma de cantidad_asignada iguala la magnitud original",
)
tab_det, tab_cobertura, tab_partidos, tab_docs = st.tabs(
["Detalle", "Cobertura por producto", "Movimientos partidos", "Documentación"]
)
columnas_tabla = [
"Mv_Fecha", "Pr_Cve_Producto", "Pr_Descripcion", "categoria_movimiento",
"cantidad_asignada", "Co_Folio_origen", "fecha_compra", "UUID",
"n_asignaciones_del_movimiento", "total_partes_del_movimiento",
"tipo_documento_origen", "documento_origen", "Mv_Folio",
]
with tab_det:
st.dataframe(
df[columnas_tabla].sort_values("Mv_Fecha", ascending=False),
width="stretch", height=520, hide_index=True,
)
st.caption(f"{len(df):,} filas de {len(resultado):,} totales en el periodo")
buf = io.BytesIO()
with pd.ExcelWriter(buf, engine="openpyxl") as writer:
df[columnas_tabla].to_excel(writer, index=False, sheet_name="Movimientos_UUID")
st.download_button(
"Descargar Excel (filtrado)",
data=buf.getvalue(),
file_name=f"movimientos_uuid_fifo_{fecha_ini}_a_{fecha_fin}.xlsx",
mime="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
)
with tab_cobertura:
st.markdown("Por producto: cuánta de la salida total tiene respaldo de compra (UUID) vs. sin respaldo (producto fabricado, ajustes, etc.).")
por_prod = resultado.groupby(["Pr_Cve_Producto", "Pr_Descripcion"], as_index=False).apply(
lambda g: pd.Series({
"salida_total": g["cantidad_asignada"].sum(),
"con_capa_de_compra": g.loc[g["con_capa_de_compra"], "cantidad_asignada"].sum(),
}),
include_groups=False,
).reset_index()
por_prod["pct_cubierto"] = (por_prod["con_capa_de_compra"] / por_prod["salida_total"] * 100).round(1)
st.dataframe(
por_prod.sort_values("salida_total", ascending=False),
width="stretch", height=460, hide_index=True,
)
with tab_partidos:
st.markdown(
f"**{n_partidos} movimientos** se partieron en 2 o más registros porque la salida "
"cruzó más de una capa de compra."
)
partidos = resultado[resultado["total_partes_del_movimiento"] > 1]
if partidos.empty:
st.info("Ningún movimiento se partió en este periodo.")
else:
resumen_cat = (
partidos.drop_duplicates(["Pr_Cve_Producto", "Mv_Folio", "Mv_ID"])
.groupby("categoria_movimiento").size().sort_values(ascending=False)
.rename("movimientos_partidos").reset_index()
)
c1, c2 = st.columns(2)
c1.markdown("**Por categoría**")
c1.dataframe(resumen_cat, hide_index=True, width="stretch")
resumen_partes = (
partidos.drop_duplicates(["Pr_Cve_Producto", "Mv_Folio", "Mv_ID"])
.groupby("total_partes_del_movimiento").size()
.rename("movimientos").reset_index()
)
c2.markdown("**Por número de partes**")
c2.dataframe(resumen_partes, hide_index=True, width="stretch")
st.markdown("**Detalle de movimientos partidos**")
st.dataframe(
partidos.sort_values(
["Pr_Cve_Producto", "Mv_Folio", "Mv_ID", "n_asignaciones_del_movimiento"]
)[columnas_tabla],
width="stretch", height=420, hide_index=True,
)
with tab_docs:
st.markdown(
"Metodologia completa (FIFO on-the-fly, resolucion de UUID, cobertura por "
"producto) en `docs/schema/inventario.md` y "
"`docs/proyectos/consumo-interno-fifo/index.md` de este repo (trivasa-context)."
)
if _is_standalone():
st.set_page_config(page_title="Movimientos + UUID (FIFO) V2", layout="wide")
render()