fifo_lib.py
Ruta original: docs/proyectos/consumo-interno-fifo/scripts/fifo_lib.py
"""
Funciones compartidas de FIFO on-the-fly: existencia a una fecha via Metodo C,
semilla reconstruida hacia atras, FIFO sobre la cola de capas de compra. Usadas por
movimientos_uuid_lib.py (asignacion de UUID a nivel movimiento) y por las
funciones de nivel producto/linea de mas abajo (exploracion original de
consumo interno, se conservan por si se retoma ese angulo).
"""
from collections import deque
import pandas as pd
import connection_205_trivasadb3 as c
PR_CVE_LEN = 10
TIPOS_CASO_SIMPLE = {"050", "051", "800", "801", "060", "061"}
# ---------------------------------------------------------------------------
# Nivel producto (agregado) -- usado por streamlit_app.py / _normalizado.py
# ---------------------------------------------------------------------------
def query_consumo_reportado(fecha_ini, fecha_fin, clave_producto=None):
fecha_ini_sql = fecha_ini.replace("-", "")
fecha_fin_sql = fecha_fin.replace("-", "")
filtro_producto = ""
if clave_producto:
filtro_producto = f" AND Consumo_Interno.Pr_Cve_Producto = '{clave_producto}'"
df = c.q(f"""
SELECT Producto.Pr_Cve_Producto, Producto.Pr_Descripcion,
SUM(Consumo_Interno.Ci_Cantidad_Control_1) AS consumo_reportado,
SUM(Consumo_Interno.Ci_Costo_Importe) AS importe_reportado
FROM Consumo_Interno
INNER JOIN Sucursal ON Sucursal.Sc_Cve_Sucursal = Consumo_Interno.Sc_Cve_Sucursal
INNER JOIN Producto ON Producto.Pr_Cve_Producto = Consumo_Interno.Pr_Cve_Producto
WHERE Sucursal.Em_Cve_Empresa = '0001'
AND Consumo_Interno.Es_Cve_Estado <> 'CA'
AND Consumo_Interno.Ci_Fecha BETWEEN '{fecha_ini_sql}' AND '{fecha_fin_sql}'
{filtro_producto}
GROUP BY Producto.Pr_Cve_Producto, Producto.Pr_Descripcion
ORDER BY Producto.Pr_Cve_Producto
""")
df["Pr_Cve_Producto"] = df["Pr_Cve_Producto"].astype(str).str.zfill(PR_CVE_LEN)
return df
def query_existencia_a_fecha_metodo_c(claves, fecha):
if not claves:
return pd.DataFrame(columns=["Pr_Cve_Producto", "existencia_ctrl1"])
fecha_sql = fecha.replace("-", "")
claves_sql = ",".join(f"'{c_}'" for c_ in claves)
df = c.q(f"""
SELECT Pr_Cve_Producto, SUM(Mv_Cantidad_Control_1) AS existencia_ctrl1
FROM Movimiento
WHERE Pr_Cve_Producto IN ({claves_sql}) AND Mv_Fecha <= '{fecha_sql}'
GROUP BY Pr_Cve_Producto
HAVING (SUM(Mv_Cantidad_Control_1) <> 0 OR SUM(Mv_Cantidad_Control_2) <> 0)
""")
df["Pr_Cve_Producto"] = df["Pr_Cve_Producto"].astype(str).str.zfill(PR_CVE_LEN)
return df
def query_capas(claves, fecha_desde=None, fecha_hasta=None):
if not claves:
return pd.DataFrame(columns=["Pr_Cve_Producto", "Co_Folio_origen", "fecha_compra", "cantidad_disponible"])
claves_sql = ",".join(f"'{c_}'" for c_ in claves)
filtro_fecha = ""
if fecha_desde:
filtro_fecha += f" AND Mv_Fecha >= '{fecha_desde}'"
if fecha_hasta:
filtro_fecha += f" AND Mv_Fecha < '{fecha_hasta}'"
df050 = c.q(f"""
SELECT Mv_Folio, Mv_Fecha, Mv_Documento AS Co_Folio, Pr_Cve_Producto, Mv_Cantidad_1
FROM Movimiento
WHERE Tm_Cve_Tipo_Movimiento = '050' AND Mv_Tabla = 'Compra'
AND Pr_Cve_Producto IN ({claves_sql}) {filtro_fecha}
""")
df051 = c.q(f"""
SELECT Mv_Documento AS Mv_Folio, Pr_Cve_Producto, Mv_Cantidad_1
FROM Movimiento
WHERE Tm_Cve_Tipo_Movimiento = '051' AND Pr_Cve_Producto IN ({claves_sql})
""")
capas050 = (df050.groupby(["Pr_Cve_Producto", "Mv_Folio", "Co_Folio", "Mv_Fecha"], as_index=False)
["Mv_Cantidad_1"].sum().rename(columns={"Mv_Cantidad_1": "cantidad_bruta"}))
anuladas = (df051.groupby(["Pr_Cve_Producto", "Mv_Folio"], as_index=False)["Mv_Cantidad_1"]
.sum().rename(columns={"Mv_Cantidad_1": "cantidad_anulada"}))
capas = capas050.merge(anuladas, on=["Pr_Cve_Producto", "Mv_Folio"], how="left")
capas["cantidad_anulada"] = capas["cantidad_anulada"].fillna(0)
capas["cantidad_neta"] = capas["cantidad_bruta"] + capas["cantidad_anulada"]
capas = capas[capas["cantidad_neta"] > 0.0001].copy()
capas["Pr_Cve_Producto"] = capas["Pr_Cve_Producto"].astype(str).str.zfill(PR_CVE_LEN)
return capas.rename(columns={"Mv_Fecha": "fecha_compra", "Co_Folio": "Co_Folio_origen",
"cantidad_neta": "cantidad_disponible"})[
["Pr_Cve_Producto", "Co_Folio_origen", "fecha_compra", "cantidad_disponible"]]
def reconstruir_semilla_on_the_fly(claves, fecha_corte):
objetivo_df = query_existencia_a_fecha_metodo_c(claves, fecha_corte)
capas_hist = query_capas(claves, fecha_hasta=fecha_corte)
capas_por_producto = {p: g.sort_values("fecha_compra", ascending=False)
for p, g in capas_hist.groupby("Pr_Cve_Producto")}
filas_semilla = []
for _, row in objetivo_df.iterrows():
prod, objetivo = row["Pr_Cve_Producto"], row["existencia_ctrl1"]
if objetivo <= 0:
continue
capas_prod = capas_por_producto.get(prod)
if capas_prod is None or capas_prod.empty:
continue
restante = objetivo
for _, capa in capas_prod.iterrows():
if restante <= 0.0001:
break
usar = min(restante, capa["cantidad_disponible"])
filas_semilla.append({
"Pr_Cve_Producto": prod, "Co_Folio_origen": capa["Co_Folio_origen"],
"fecha_compra": capa["fecha_compra"], "cantidad_disponible": usar,
})
restante -= usar
return pd.DataFrame(filas_semilla) if filas_semilla else pd.DataFrame(
columns=["Pr_Cve_Producto", "Co_Folio_origen", "fecha_compra", "cantidad_disponible"])
def fifo_total_por_producto(capas_prod, consumo_reportado):
"""Version agregada: 1 resultado por producto (usada por streamlit_app.py)."""
if capas_prod is None or capas_prod.empty:
return 0.0, consumo_reportado, None, 0.0, []
capas_ordenadas = capas_prod.sort_values(["fecha_compra", "Co_Folio_origen"])
restante = consumo_reportado
cubierto = 0.0
folios_usados = []
fecha_mas_antigua = None
for _, capa in capas_ordenadas.iterrows():
if restante <= 0.0001:
break
usar = min(restante, capa["cantidad_disponible"])
if usar > 0.0001:
cubierto += usar
restante -= usar
folios_usados.append(capa["Co_Folio_origen"])
if fecha_mas_antigua is None:
fecha_mas_antigua = capa["fecha_compra"]
total_capas = capas_ordenadas["cantidad_disponible"].sum()
return cubierto, max(restante, 0.0), fecha_mas_antigua, total_capas, folios_usados
def fifo_total_por_producto_detallado(capas_prod, consumo_reportado):
"""Version 'tipo BD': una fila por cada capa (folio) realmente usada, en vez
de un solo total agregado. Devuelve lista de dicts:
{Co_Folio_origen, fecha_compra, cantidad_asignada}
mas el remanente sin_capa_disponible (0 si se cubrio completo)."""
filas = []
if capas_prod is None or capas_prod.empty:
return filas, consumo_reportado
capas_ordenadas = capas_prod.sort_values(["fecha_compra", "Co_Folio_origen"])
restante = consumo_reportado
for _, capa in capas_ordenadas.iterrows():
if restante <= 0.0001:
break
usar = min(restante, capa["cantidad_disponible"])
if usar > 0.0001:
filas.append({
"Co_Folio_origen": capa["Co_Folio_origen"],
"fecha_compra": capa["fecha_compra"],
"cantidad_asignada": usar,
})
restante -= usar
return filas, max(restante, 0.0)
def query_uuids(folios):
"""Mismo mecanismo validado en scripts/20_compra_uuid_map.py."""
folios = sorted(set(f for f in folios if f))
if not folios:
return {}
folios_sql = ",".join(f"'{f}'" for f in folios)
df = c.q(f"""
SELECT LEFT(Cd_Documento, 10) AS Co_Folio, Cd_Timbre_UUID AS UUID, Cd_Timbre_Fecha
FROM Comprobante_Digital
WHERE Cd_Tabla = 'COMPRA' AND LEFT(Cd_Documento, 10) IN ({folios_sql})
AND Cd_Timbre_UUID IS NOT NULL AND Cd_Timbre_UUID <> ''
""")
if df.empty:
return {}
vigente = df.sort_values("Cd_Timbre_Fecha").drop_duplicates("Co_Folio", keep="last")
return dict(zip(vigente["Co_Folio"], vigente["UUID"]))
# ---------------------------------------------------------------------------
# Nivel linea (PK Consumo_Interno = Ci_Folio+Ci_ID) -- usado por _lineas.py y
# _caso_simple.py
# ---------------------------------------------------------------------------
def query_consumo_lineas(fecha_ini, fecha_fin, clave_producto=None):
"""Una fila por (Ci_Folio, Ci_ID) -- el PK real de Consumo_Interno."""
fecha_ini_sql = fecha_ini.replace("-", "")
fecha_fin_sql = fecha_fin.replace("-", "")
filtro_producto = ""
if clave_producto:
filtro_producto = f" AND Consumo_Interno.Pr_Cve_Producto = '{clave_producto}'"
df = c.q(f"""
SELECT Consumo_Interno.Ci_Folio, Consumo_Interno.Ci_ID, Consumo_Interno.Ci_Fecha,
Producto.Pr_Cve_Producto, Producto.Pr_Descripcion,
Consumo_Interno.Ci_Cantidad_Control_1, Consumo_Interno.Ci_Unidad_Control_1,
Consumo_Interno.Ci_Costo_Importe
FROM Consumo_Interno
INNER JOIN Sucursal ON Sucursal.Sc_Cve_Sucursal = Consumo_Interno.Sc_Cve_Sucursal
INNER JOIN Producto ON Producto.Pr_Cve_Producto = Consumo_Interno.Pr_Cve_Producto
WHERE Sucursal.Em_Cve_Empresa = '0001'
AND Consumo_Interno.Es_Cve_Estado <> 'CA'
AND Consumo_Interno.Ci_Fecha BETWEEN '{fecha_ini_sql}' AND '{fecha_fin_sql}'
{filtro_producto}
ORDER BY Producto.Pr_Cve_Producto, Consumo_Interno.Ci_Fecha, Consumo_Interno.Ci_ID
""")
df["Pr_Cve_Producto"] = df["Pr_Cve_Producto"].astype(str).str.zfill(PR_CVE_LEN)
df["Ci_Folio"] = df["Ci_Folio"].astype(str)
df["Ci_ID"] = df["Ci_ID"].astype(str)
return df
def query_eventos_netos_movimiento(claves, fecha_ini, fecha_fin_excl):
"""Eventos de consumo (Tm 800/801/060/061), neteados de anulacion via el join
de 2 saltos ya validado, a nivel Mv_Folio/Mv_ID (= Ci_Folio/Ci_ID por
convencion del proyecto). Identico a scripts/27_fifo_consumo_interno_2025_01.py."""
if not claves:
return pd.DataFrame(columns=["Pr_Cve_Producto", "Mv_Folio", "Mv_ID", "Mv_Fecha", "cantidad_neta"])
claves_sql = ",".join(f"'{c_}'" for c_ in claves)
bruta = c.q(f"""
SELECT Mv_Folio, Mv_ID, Mv_Fecha, Pr_Cve_Producto, Mv_Cantidad_1
FROM Movimiento
WHERE Tm_Cve_Tipo_Movimiento IN ('800','060')
AND Mv_Fecha >= '{fecha_ini}' AND Mv_Fecha < '{fecha_fin_excl}'
AND Pr_Cve_Producto IN ({claves_sql})
""")
anulaciones = c.q(f"""
SELECT Mv_Documento AS Mv_Folio, Pr_Cve_Producto, Mv_Cantidad_1
FROM Movimiento
WHERE Tm_Cve_Tipo_Movimiento IN ('801','061') AND Pr_Cve_Producto IN ({claves_sql})
""")
anuladas_agg = (anulaciones.groupby(["Pr_Cve_Producto", "Mv_Folio"], as_index=False)["Mv_Cantidad_1"]
.sum().rename(columns={"Mv_Cantidad_1": "cantidad_anulada"}))
eventos = bruta.merge(anuladas_agg, on=["Pr_Cve_Producto", "Mv_Folio"], how="left")
eventos["cantidad_anulada"] = eventos["cantidad_anulada"].fillna(0)
eventos["cantidad_neta"] = eventos["Mv_Cantidad_1"] + eventos["cantidad_anulada"]
eventos = eventos[eventos["cantidad_neta"].abs() > 0.0001].copy()
eventos["Pr_Cve_Producto"] = eventos["Pr_Cve_Producto"].astype(str).str.zfill(PR_CVE_LEN)
eventos["Mv_Folio"] = eventos["Mv_Folio"].astype(str)
eventos["Mv_ID"] = eventos["Mv_ID"].astype(str)
return eventos.sort_values(["Pr_Cve_Producto", "Mv_Fecha", "Mv_Folio", "Mv_ID"])
def fifo_lineas_por_producto(capas_prod, eventos_prod):
"""FIFO linea a linea (cola real, deque) -- identico a la logica de
scripts/27, on the fly. Devuelve 1+ filas por (Ci_Folio, Ci_ID) si el
consumo de esa linea se reparte entre varias capas de compra."""
cola = deque(
{"Co_Folio_origen": r["Co_Folio_origen"], "fecha_compra": r["fecha_compra"],
"cantidad_disponible": r["cantidad_disponible"]}
for _, r in capas_prod.iterrows()
) if capas_prod is not None and not capas_prod.empty else deque()
filas = []
for _, ev in eventos_prod.iterrows():
cantidad_neta = ev["cantidad_neta"]
if cantidad_neta >= 0:
continue
restante = -cantidad_neta
while restante > 0.0001 and cola:
capa = cola[0]
usar = min(restante, capa["cantidad_disponible"])
capa["cantidad_disponible"] -= usar
restante -= usar
filas.append({
"Pr_Cve_Producto": ev["Pr_Cve_Producto"], "Ci_Folio": ev["Mv_Folio"], "Ci_ID": ev["Mv_ID"],
"cantidad_asignada": usar, "Co_Folio_origen": capa["Co_Folio_origen"],
"fecha_compra": capa["fecha_compra"],
})
if capa["cantidad_disponible"] <= 0.0001:
cola.popleft()
if restante > 0.0001:
filas.append({
"Pr_Cve_Producto": ev["Pr_Cve_Producto"], "Ci_Folio": ev["Mv_Folio"], "Ci_ID": ev["Mv_ID"],
"cantidad_asignada": restante, "Co_Folio_origen": None, "fecha_compra": None,
})
return filas
# ---------------------------------------------------------------------------
# Caso simple: producto cuyo historial COMPLETO de Movimiento solo tiene
# compra (050/051) y consumo interno (800/801/060/061) -- ningun otro tipo de
# salida (venta, produccion, ajuste, traspaso, etc.)
# ---------------------------------------------------------------------------
def query_tipos_movimiento_por_producto(claves):
if not claves:
return {}
claves_sql = ",".join(f"'{c_}'" for c_ in claves)
df = c.q(f"""
SELECT DISTINCT Pr_Cve_Producto, Tm_Cve_Tipo_Movimiento
FROM Movimiento
WHERE Pr_Cve_Producto IN ({claves_sql})
""")
df["Pr_Cve_Producto"] = df["Pr_Cve_Producto"].astype(str).str.zfill(PR_CVE_LEN)
tipos_por_producto = {}
for prod, grupo in df.groupby("Pr_Cve_Producto"):
tipos_por_producto[prod] = set(grupo["Tm_Cve_Tipo_Movimiento"].astype(str))
return tipos_por_producto
def es_caso_simple(tipos_producto):
"""True si el producto SOLO tiene compra+consumo en TODO su historico.
(Version lifetime -- se conserva por si se necesita, pero el reporte
Caso simple usa la version period-scoped de abajo)."""
if not tipos_producto:
return False
return tipos_producto.issubset(TIPOS_CASO_SIMPLE)
def query_productos_con_otras_salidas_en_periodo(claves, fecha_ini, fecha_fin_excl):
"""Caso simple, on the fly, escopado al PERIODO (no al historico completo):
un producto deja de ser caso simple si, DENTRO del rango de fechas
seleccionado, tuvo alguna salida real (Mv_Cantidad_1 < 0) de un tipo de
movimiento que NO sea compra/anulacion-compra (050/051) ni consumo interno
(800/801/060/061) -- venta, produccion, ajuste, traspaso con salida neta,
devolucion, etc. Devuelve el SET de claves que SI tuvieron alguna otra
salida (es decir, las que NO son caso simple por este motivo)."""
if not claves:
return set()
claves_sql = ",".join(f"'{c_}'" for c_ in claves)
df = c.q(f"""
SELECT DISTINCT Pr_Cve_Producto
FROM Movimiento
WHERE Pr_Cve_Producto IN ({claves_sql})
AND Mv_Fecha >= '{fecha_ini}' AND Mv_Fecha < '{fecha_fin_excl}'
AND Tm_Cve_Tipo_Movimiento NOT IN ('050','051','800','801','060','061')
AND Mv_Cantidad_1 < 0
""")
return set(df["Pr_Cve_Producto"].astype(str).str.zfill(PR_CVE_LEN))