Este documento describe el proceso para crear una funcionalidad estable de un proceso de validación de stock de producto en transacciones.
Herramientas
- Bproc de Evento onExit
- Script de Guest Code
- Finni
- Transacciones con productos
Objetivo
Consiste en indicar paso a paso cómo implementar el caso y mostrar pruebas de su funcionamiento.
Descripción del problema
En una transacción donde se asocian productos, se puede introducir una cantidad mayor al stock disponible, el sistema notificará de este error únicamente al querer guardar la factura una vez completada. Esto genera que el usuario tenga que modificar la cantidad del ítem asociado.
Solución propuesta
Nuestra solución consiste en utilizar un bproc de evento onExit conectado a un script de guest code que notifique en caso de que la cantidad de producto ingresada en el input supere a la cantidad de stock del producto.
Configuración
Guest Code
Para la creación del script de guest code se utilizó la IA Finni, indicando la necesidad en cuestión. Se pasó por varias iteraciones.
Se deja adjunto el script final
# Script para Finnegans GO Guest Code
# Validado con información del MCP
# ──────────────────────────────────────────────────────────────────────────────
# BPROC: Evento onexit en widget "Cantidad Venta" – Factura de Venta
# Propósito: Validar que la cantidad ingresada no supere el stock disponible
# ──────────────────────────────────────────────────────────────────────────────
import json
from datetime import datetime
from finnegans.scripting import request, HTTPResponse
from finnegans.api import get_reporte
# ──────────────────────────────────────────────────────────────────────────────
# ⚙ CONFIGURACIÓN – Ajustar según la implementación de tu instancia GO
# ──────────────────────────────────────────────────────────────────────────────
# Nombres de campo tal como llegan en el payload del bproc onexit
CAMPO_CANTIDAD = "CantidadWorkflow" # Widget que contiene la cantidad
CAMPO_PRODUCTO = "ProductoCodigo" # Widget que contiene el código del producto
CAMPO_EMPRESA = "sucursal" # Campo que contiene la empresa
CAMPO_DEPOSITO = "DepositoOrigen" # Widget que contiene el depósito origen
# Endpoint de reporte de stock
STOCK_REPORT_API = "stockEmpresa"
def obtener_stock_producto(producto_id: str, empresa_id: str, deposito_id: str = "", organizacion_id: str = "") -> dict:
"""
Consulta el stock actual de un producto usando el reporte stockEmpresa de Finnegans GO.
Args:
producto_id: Identificador del producto a consultar.
empresa_id: Identificador de la empresa.
deposito_id: Identificador del depósito (opcional).
organizacion_id: Identificador de la organización (opcional).
Returns:
Diccionario con información del stock del producto.
Raises:
Exception: Si la consulta falla o no se encuentran resultados.
"""
# Obtener fecha actual para el reporte
fecha_actual = datetime.now().strftime("%Y-%m-%d")
# Parámetros para el reporte de stock
params = {
"PARAMWEBREPORT_fecha": fecha_actual,
"PARAMWEBREPORT_producto": producto_id,
"PARAMWEBREPORT_MonedaID":"PES"
}
# LOG: Mostrar URL y parámetros antes de la consulta
# Construir URL completa para logging
param_strings = []
for key, value in params.items():
param_strings.append(f"{key}={value}")
query_string = "&".join(param_strings)
full_url = f"{STOCK_REPORT_API}?{query_string}"
print(f"🔍 LOG CONSULTA STOCK:")
print(f" URL: {full_url}")
print(f" Parámetros principales:")
print(f" - Producto: {producto_id}")
print(f" - Empresa: {empresa_id}")
print(f" - Depósito: {deposito_id}")
print(f" - Fecha: {fecha_actual}")
print(f" Todos los parámetros: {json.dumps(params, indent=2)}")
# Ejecutar reporte de stock
try:
resultado_reporte = get_reporte(STOCK_REPORT_API, params)
# LOG: Mostrar respuesta cruda antes de procesar
print(f"📥 LOG RESPUESTA CRUDA:")
print(f" Tipo de dato recibido: {type(resultado_reporte)}")
if isinstance(resultado_reporte, str):
print(f" Contenido (primeros 500 chars): {str(resultado_reporte)[:500]}...")
if len(str(resultado_reporte)) > 500:
print(f" [Texto truncado - longitud total: {len(str(resultado_reporte))} caracteres]")
else:
print(f" Contenido completo: {resultado_reporte}")
# Verificar si el resultado es un string que necesita ser parseado
if isinstance(resultado_reporte, str):
print("🔄 Parseando respuesta como JSON string...")
resultados = json.loads(resultado_reporte)
elif isinstance(resultado_reporte, list):
print("✅ Respuesta ya es una lista, usando directamente...")
resultados = resultado_reporte
else:
print("⚠️ Respuesta no es string ni lista, convirtiendo...")
# Convertir a lista si es otro tipo
resultados = [resultado_reporte] if resultado_reporte else []
# LOG: Mostrar datos procesados
print(f"📋 LOG DATOS PROCESADOS:")
print(f" Número de registros encontrados: {len(resultados) if resultados else 0}")
if resultados and len(resultados) > 0:
print(f" Primer registro: {json.dumps(resultados[0], indent=2)}")
if len(resultados) > 1:
print(f" [Total de {len(resultados)} registros]")
except json.JSONDecodeError as e:
print(f"❌ ERROR JSON: {str(e)}")
raise Exception(f"Error al parsear respuesta del reporte de stock: {str(e)}")
except Exception as e:
print(f"❌ ERROR GENERAL: {str(e)}")
raise Exception(f"Error al ejecutar reporte de stock: {str(e)}")
if not resultados:
print("❌ No se encontraron resultados para el producto")
raise Exception("No se encontraron resultados para el producto especificado")
# Buscar el producto específico en los resultados y consolidar el stock
stock_total = 0.0
info_producto = None
registros_encontrados = 0
print(f"🔍 Buscando producto '{producto_id}' en los resultados...")
for item in resultados:
# Asegurar que item es un diccionario
if not isinstance(item, dict):
print(f"⚠️ Saltando item que no es diccionario: {type(item)}")
continue
# Verificar si es el producto buscado
codigo_producto = str(item.get("PRODUCTOCODIGO", ""))
nombre_producto = str(item.get("PRODUCTO", ""))
if (codigo_producto == str(producto_id) or nombre_producto == str(producto_id)):
registros_encontrados += 1
print(f"✅ Producto encontrado (registro #{registros_encontrados}):")
print(f" Código: {codigo_producto}")
print(f" Nombre: {nombre_producto}")
# Obtener la cantidad disponible (CANTIDAD1 es la cantidad en unidad principal)
cantidad_disponible = float(item.get("CANTIDAD1", 0) or 0)
stock_total += cantidad_disponible
print(f" Cantidad en este registro: {cantidad_disponible}")
print(f" Stock acumulado: {stock_total}")
# Guardar información adicional del producto (tomar del primer registro)
if info_producto is None:
info_producto = {
"codigo": item.get("PRODUCTOCODIGO", ""),
"nombre": item.get("PRODUCTO", ""),
"unidad1": item.get("UNIDAD1", ""),
"unidad2": item.get("UNIDAD2", ""),
"deposito": item.get("DEPOSITO", ""),
"empresa": item.get("EMPRESA", ""),
"organizacion": item.get("ORGANIZACION", ""),
"punto_reposicion": float(item.get("PUNTOREPOSICION", 0) or 0),
"cantidad_a_reponer": float(item.get("CANTIDADSTOCKAREPONER", 0) or 0),
"precio_unidad": float(item.get("PRECIOUNIDADSTOCK1", 0) or 0),
"moneda": item.get("MONEDA", ""),
"partida": item.get("PARTIDA", "")
}
print(f"📊 RESUMEN FINAL:")
print(f" Registros del producto encontrados: {registros_encontrados}")
print(f" Stock total consolidado: {stock_total}")
if info_producto is None:
print(f"❌ No se encontró información para el producto {producto_id}")
raise Exception(f"No se encontró información para el producto {producto_id}")
info_producto["stock_disponible"] = stock_total
print(f"✅ Información del producto consolidada correctamente")
return info_producto
def construir_respuesta_error(cantidad: float, info_stock: dict) -> dict:
"""
Arma el cuerpo del error a mostrarle al usuario con la disponibilidad real.
Args:
cantidad: Cantidad ingresada por el usuario.
info_stock: Información completa del stock del producto.
Returns:
Diccionario con el mensaje de error y detalles del stock.
"""
return {
"error": (
f"⚠ No hay stock disponible suficiente para el producto {info_stock['codigo']}. "
f"La cantidad solicitada ({cantidad:g} {info_stock['unidad1']}) supera las existencias actuales. "
f"Stock disponible: {info_stock['stock_disponible']:g} {info_stock['unidad1']}."
),
"stockDisponible": info_stock['stock_disponible'],
"producto": {
"codigo": info_stock['codigo'],
"nombre": info_stock['nombre'],
"unidad": info_stock['unidad1'],
"deposito": info_stock['deposito'],
"puntoReposicion": info_stock['punto_reposicion'],
"cantidadAReponer": info_stock['cantidad_a_reponer']
}
}
def construir_respuesta_exitosa(cantidad: float, info_stock: dict) -> dict:
"""
Arma la respuesta exitosa con información detallada del producto.
Args:
cantidad: Cantidad validada exitosamente.
info_stock: Información completa del stock del producto.
Returns:
Diccionario con el mensaje de éxito y detalles del stock.
"""
stock_restante = info_stock['stock_disponible'] - cantidad
return {
"mensaje": f"✅ Validación exitosa. La cantidad solicitada ({cantidad:g} {info_stock['unidad1']}) está disponible.",
"stockAntes": info_stock['stock_disponible'],
"cantidadSolicitada": cantidad,
"stockDespues": stock_restante,
"producto": {
"codigo": info_stock['codigo'],
"nombre": info_stock['nombre'],
"unidad": info_stock['unidad1'],
"deposito": info_stock['deposito'],
"empresa": info_stock['empresa'],
"organizacion": info_stock['organizacion'],
"puntoReposicion": info_stock['punto_reposicion'],
"cantidadAReponer": info_stock['cantidad_a_reponer'],
"precioUnidad": info_stock['precio_unidad'],
"moneda": info_stock['moneda'],
"partida": info_stock['partida']
},
"alertas": {
"bajoPuntoReposicion": stock_restante <= info_stock['punto_reposicion'],
"requiereReposicion": info_stock['cantidad_a_reponer'] > 0
}
}
def main():
# ── 1. Parsear el body enviado por el bproc onexit ───────────────────────
try:
body = json.loads((request.get_body() or b"{}").decode())
except Exception:
return HTTPResponse(400, body=json.dumps({"error": "Cuerpo de la solicitud inválido. Se esperaba JSON."}))
# ── 2. Extraer datos desde la estructura widgets ────────────────────────
widgets = body.get("widgets", {})
# Extraer cantidad desde CantidadWorkflow.value
cantidad_widget = widgets.get(CAMPO_CANTIDAD, {})
cantidad_raw = cantidad_widget.get("value")
# Extraer producto desde ProductoCodigo.value
producto_widget = widgets.get(CAMPO_PRODUCTO, {})
producto_id = producto_widget.get("value")
# Extraer empresa desde sucursal (nivel superior)
empresa_id = body.get(CAMPO_EMPRESA)
# Extraer depósito desde DepositoOrigen.value.id (si existe)
deposito_widget = widgets.get(CAMPO_DEPOSITO, {})
deposito_value = deposito_widget.get("value", {})
deposito_id = deposito_value.get("id") if deposito_value else ""
# Organización no se incluye según las instrucciones
organizacion_id = ""
print(f"📥 DATOS EXTRAÍDOS DEL FORMULARIO:")
print(f" Cantidad: {cantidad_raw}")
print(f" Producto: {producto_id}")
print(f" Empresa: {empresa_id}")
print(f" Depósito: {deposito_id}")
# ── 3. Validar presencia de campos obligatorios ──────────────────────────
if cantidad_raw is None:
return HTTPResponse(
400,
body=json.dumps({"error": f"No se recibió el campo 'cantidadVenta' en la solicitud."})
)
if not producto_id:
return HTTPResponse(
400,
body=json.dumps({"error": f"No se recibió el código del producto."})
)
if not empresa_id:
return HTTPResponse(
400,
body=json.dumps({"error": f"No se recibió el identificador de la empresa."})
)
# ── 4. Convertir cantidad a número ───────────────────────────────────────
try:
cantidad = float(cantidad_raw)
except (ValueError, TypeError):
return HTTPResponse(
400,
body=json.dumps({"error": f"El valor '{cantidad_raw}' no es un número válido para la cantidad."})
)
if cantidad <= 0:
return HTTPResponse(
400,
body=json.dumps({"error": "La cantidad debe ser un valor mayor a cero."})
)
# ── 5. Consultar stock disponible del producto ───────────────────────────
try:
info_stock = obtener_stock_producto(
producto_id=str(producto_id),
empresa_id=str(empresa_id),
deposito_id=str(deposito_id) if deposito_id else "",
organizacion_id=str(organizacion_id)
)
except Exception as e:
return HTTPResponse(
500,
body=json.dumps({"error": f"Error al consultar el stock del producto: {str(e)}"})
)
# ── 6. Validar cantidad vs stock y notificar al usuario ──────────────────
if cantidad > info_stock['stock_disponible']:
return HTTPResponse(
400,
body=json.dumps(construir_respuesta_error(cantidad, info_stock))
)
# ── 7. Validación exitosa con información detallada ──────────────────────
return HTTPResponse(
200,
body=json.dumps(construir_respuesta_exitosa(cantidad, info_stock))
)
if __name__ == "__main__":
result = main()
print(result)
Una vez terminado el script, copiamos el endpoint del script y pasamos a crear el bproc
BProc de Evento onExit
Factura de venta
Pruebas
Con todo configurado, podemos probarlo con facilidad yendo a una Transaccion y agregando un producto (Item). Si la cantidad es superior al stock, se lanzará un mensaje notificando que no hay stock disponible acompañado del stock actual del producto.
