Uso de Guest Code con Bproc de transacción
Este documento muestra como utilizar la herramienta guest code y un bproc de tipo transacción para enviar un mail al cliente al superar cierto monto de deuda.
Descripción del caso
Automatizar el envío de una notificación por correo electrónico al cliente cuando el sistema registre una nueva factura de venta, siempre y cuando el monto de dicha factura provoque que el saldo total de su deuda supere el límite de crédito previamente establecido.
Configuraciones
Para que se envíe el mail al cliente se deben configurar:
1 - Guest code: Para realizar el script utilizamos la IA Fini. Se le indica la necesidad junto con el contexto necesario y luego de algunas iteraciones se obtiene el script deseado:
"""
Guest Code: webhook BProc → facturaVenta (identificación externa) → composicionSaldoCliente (cliente + empresa) → mail.
Modificación: el email del cliente se obtiene desde el endpoint organizacion con el código resuelto en facturaVenta.
"""
from __future__ import annotations
import html
import json
from datetime import datetime
from typing import Any
from finnegans.api import ApiException, obtener_entidad, get_reporte
from finnegans.mail import enviar_email
from finnegans.scripting import HTTPResponse, request
from finnegans.bproc import BProcWebhook
_TABLA_COLUMNAS: tuple[tuple[str, tuple[str, ...], str], ...] = (
("FECHA", ("FECHA",), "texto"),
("DOCUMENTO", ("DOCUMENTO", "IDENTIFICACIONEXTERNA"), "texto"),
("Importe moneda principal", ("IMPORTEMONPPAL",), "importe"),
("Importe moneda transacción", ("IMPORTEMONTRAN",), "importe"),
("MONEDA", ("MONEDA",), "texto"),
("FECHAVTO", ("FECHAVTO", "FECHACASHFLOW"), "texto"),
)
_NUM_COLS = len(_TABLA_COLUMNAS)
def _a_float(val: Any) -> float | None:
if val is None:
return None
if isinstance(val, bool):
return None
if isinstance(val, (int, float)):
return float(val)
s = str(val).strip().replace(" ", "")
if not s:
return None
if "," in s and "." not in s:
normalized = s.replace(",", ".")
elif "," in s and "." in s:
normalized = s.replace(".", "").replace(",", ".")
else:
normalized = s
try:
return float(normalized)
except ValueError:
return None
def _format_importe_numero(n: float) -> str:
sign = "-" if n < 0 else ""
n_abs = abs(float(n))
s = f"{n_abs:,.4f}"
whole, frac = s.split(".", 1)
whole = whole.replace(",", ".")
return f"{sign}{whole},{frac}"
def _sumatoria_importemonppal(filas: list[dict[str, Any]]) -> float:
total = 0.0
for row in filas:
n = _a_float(_valor_fila(row, ("IMPORTEMONPPAL",)))
if n is not None:
total += n
return total
def _organizacion_desde_factura_venta(fv: dict[str, Any], cliente_codigo: str) -> str:
for nk in ("OrganizacionCliente", "Cliente", "cliente"):
val = fv.get(nk)
if val is None:
continue
if isinstance(val, str):
s = val.strip()
if s:
return s
if isinstance(val, dict):
for k in ("Nombre", "nombre", "RazonSocial", "razonSocial", "NombreFantasia"):
x = val.get(k)
if x is not None and str(x).strip():
return str(x).strip()
return cliente_codigo
def _parece_fila_composicion(row: dict[str, Any]) -> bool:
u = {str(k).upper() for k in row}
if "IMPORTEMONTRAN" not in u and "IMPORTEMONTRANS" not in u:
return False
if "FECHA" not in u:
return False
return "DOCUMENTO" in u or "DOCUMENTOID" in u or "IDENTIFICACIONEXTERNA" in u
def _normalizar_filas_reporte(data: Any, depth: int = 0) -> list[dict[str, Any]]:
if data is None or depth > 8:
return []
if isinstance(data, list):
candidatos = [x for x in data if isinstance(x, dict)]
if candidatos and _parece_fila_composicion(candidatos[0]):
return candidatos
out: list[dict[str, Any]] = []
for x in data:
out.extend(_normalizar_filas_reporte(x, depth + 1))
return out
if isinstance(data, dict):
if _parece_fila_composicion(data):
return [data]
for k in (
"data", "items", "result", "results", "rows",
"records", "lista", "values", "report", "reportData",
):
inner = data.get(k)
nested = _normalizar_filas_reporte(inner, depth + 1)
if nested:
return nested
for _k, v in data.items():
nested = _normalizar_filas_reporte(v, depth + 1)
if nested:
return nested
return []
def _valor_fila(row: dict[str, Any], claves: tuple[str, ...]) -> Any:
upper = {str(k).upper(): v for k, v in row.items()}
for c in claves:
cu = c.upper()
if cu in upper and upper[cu] is not None:
return upper[cu]
return None
def _format_importe(val: Any) -> str:
if val is None:
return ""
if isinstance(val, bool):
return html.escape(str(val))
n = _a_float(val)
if n is not None:
return _format_importe_numero(n)
s = str(val).strip()
return html.escape(s) if s else ""
def _celda_texto(val: Any) -> str:
if val is None:
return ""
return html.escape(str(val).strip())
def _armar_html_tabla(filas: list[dict[str, Any]], organizacion: str, ident_ext: str) -> str:
th_base = (
"padding:10px 14px;border:1px solid #c9d1d9;background:#f6f8fa;"
"font-weight:600;font-size:13px;color:#24292f;font-family:Segoe UI,Roboto,Helvetica,Arial,sans-serif"
)
td_base = (
"padding:8px 14px;border:1px solid #d8dee4;font-size:13px;"
"font-family:Segoe UI,Roboto,Helvetica,Arial,sans-serif;color:#1f2328"
)
td_num = td_base + ";text-align:right;font-variant-numeric:tabular-nums;white-space:nowrap"
thead_cells: list[str] = []
for etiqueta, _claves, tipo in _TABLA_COLUMNAS:
align = "right" if tipo == "importe" else "left"
thead_cells.append(
f'<th style="{th_base};text-align:{align}">{html.escape(etiqueta)}</th>'
)
thead = "".join(thead_cells)
body_rows: list[str] = []
for idx, row in enumerate(filas):
row_bg = "#ffffff" if idx % 2 == 0 else "#fafbfc"
tds: list[str] = []
for _etiqueta, claves, tipo in _TABLA_COLUMNAS:
raw = _valor_fila(row, claves)
if tipo == "importe":
inner = _format_importe(raw)
tds.append(f'<td style="{td_num};background:{row_bg}">{inner}</td>')
else:
tds.append(
f'<td style="{td_base};text-align:left;background:{row_bg}">'
f"{_celda_texto(raw)}</td>"
)
body_rows.append("<tr>" + "".join(tds) + "</tr>")
empty_msg = (
f'<tr><td colspan="{_NUM_COLS}" style="{td_base};text-align:center;'
f'background:#fff;color:#57606a;">Sin registros</td></tr>'
)
tbody = "".join(body_rows) if body_rows else empty_msg
tfoot_html = ""
if filas:
idx_ppal = next(
i
for i, (_e, claves, _t) in enumerate(_TABLA_COLUMNAS)
if claves == ("IMPORTEMONPPAL",)
)
colspan_before = idx_ppal
colspan_after = _NUM_COLS - idx_ppal - 1
total_ppal = _sumatoria_importemonppal(filas)
total_str = _format_importe(total_ppal)
tf_bg = "#eef2f7"
tf_label_style = (
f"{td_base};font-weight:700;text-align:right;background:{tf_bg};"
"border-top:2px solid #c9d1d9;color:#24292f"
)
tf_num_style = (
f"{td_num};font-weight:700;background:{tf_bg};"
"border-top:2px solid #c9d1d9"
)
tf_empty_style = (
f"{td_base};background:{tf_bg};border-top:2px solid #c9d1d9"
)
parts = []
if colspan_before > 0:
parts.append(
f'<td colspan="{colspan_before}" style="{tf_label_style}">'
"Total moneda principal</td>"
)
parts.append(f'<td style="{tf_num_style}">{total_str}</td>')
if colspan_after > 0:
parts.append(
f'<td colspan="{colspan_after}" style="{tf_empty_style}"></td>'
)
tfoot_html = f"<tfoot><tr>{''.join(parts)}</tr></tfoot>"
table_wrap = (
"border-collapse:collapse;width:100%;max-width:920px;margin:12px 0;"
"box-shadow:0 1px 3px rgba(31,35,40,0.12);border-radius:6px;overflow:hidden"
)
intro = (
f'<p style="margin:0 0 10px;font-family:Segoe UI,Roboto,Helvetica,Arial,sans-serif;'
f'font-size:14px;color:#24292f">El cliente <b>{html.escape(organizacion)}</b> '
f"supera la tolerancia permitida con la carga del documento "
f"<b>{html.escape(ident_ext)}</b>.</p>"
)
return (
f"{intro}"
f'<table role="presentation" style="{table_wrap}">'
f"<thead><tr>{thead}</tr></thead><tbody>{tbody}</tbody>{tfoot_html}</table>"
)
def main():
try:
payload = json.loads(request.get_body())
bproc = BProcWebhook(payload)
tolerancia = _a_float(bproc.params.get("Tolerancia"))
email_raw = bproc.params.get("emailAlerta") #Se obtienen los parámetros configurados en el bproc
email_alerta = str(email_raw).strip() if email_raw else None
if tolerancia is None:
return HTTPResponse(
400,
body=json.dumps(
{"error": "Falta el parámetro Tolerancia en el body del BProc"},
ensure_ascii=False,
),
)
# Obtengo la identificacion externa obtenida del Bproc
ident_ext = bproc.object.code # Obtiene la Identificación externa de la transacción que activó el webhook
if not ident_ext:
return HTTPResponse(
400,
body=json.dumps(
{"error": "No se pudo resolver identificacionExterna del webhook"},
ensure_ascii=False,
),
)
# Hago un GET de facturaVenta con la identificacion externa
fv = obtener_entidad("facturaVenta", ident_ext) # Realiza un GET de la api de facturaVenta. De la respuesta se obtienen los datos del cliente y de la empresa
cliente = fv.get("Cliente")
empresa = fv.get("EmpresaCodigo")
if not cliente or not empresa:
return HTTPResponse(
400,
body=json.dumps(
{
"error": "facturaVenta sin cliente y/o empresa resolvibles",
"cliente": cliente,
"empresa": empresa,
},
ensure_ascii=False,
),
)
# ── NUEVO: email del cliente vía endpoint organizacion ───────────────
email_cliente = None
if cliente:
try:
org = obtener_entidad("organizacion", cliente) # Realiza un GET de la api de organización para obtener el mail del cliente.
email_cliente = org.get("Email")
except Exception:
return HTTPResponse(
400,
body=json.dumps(
{"error": "No se pudo resolver email del cliente desde organizacion"},
ensure_ascii=False,
),
)
# Lista de destinatarios: siempre email_cliente + email_alerta si tiene email configurado
destinatarios: list[str] = [email_cliente]
if email_alerta and email_alerta != email_cliente:
destinatarios.append(email_alerta)
# ────────────────────────────────────────────────────────────────────
raw_composicion = get_reporte("composicionSaldoCliente", {"PARAMWEBREPORT_organizacion": cliente, "PARAMWEBREPORT_fecha": datetime.now().strftime("%Y-%m-%d")}) #Realiza un GET de una api de tipo View/informe. Recibe como parámetro un mapping con los parámetros del informe
filas = _normalizar_filas_reporte(raw_composicion) #Parseo del response del informe de composicion de saldos
sum_pp = _sumatoria_importemonppal(filas)
debe_enviar_mail = sum_pp > tolerancia # Calculo de la sumatoria de importes y comparacion del saldo respecto a la Tolerancia
organizacion = " ".join(_organizacion_desde_factura_venta(fv, cliente).split())
asunto = f"Seguimiento de deuda: {organizacion}"
cuerpo_html = _armar_html_tabla(filas, organizacion, ident_ext) # Función que convierte en una tabla HTML la respuesta parseada de la composición para el armado del cuerpo del mail.
if debe_enviar_mail:
enviar_email(asunto, cuerpo_html, destinatarios, html=True) #Función interna que se encarga del envío de mails.
return HTTPResponse(
200,
body=json.dumps(
{
"status": "success",
"identificacionExterna": ident_ext,
"cliente": cliente,
"organizacion": organizacion,
"empresa": empresa,
"filas": len(filas),
"tolerancia": tolerancia,
"sumatoriaImportemonPpal": sum_pp,
"emailAlerta": email_alerta,
"emailCliente": email_cliente,
"emailEnviado": debe_enviar_mail,
"destinatarios": destinatarios if debe_enviar_mail else [],
},
ensure_ascii=False,
),
) # si todo se ejecutó correctamente, se devuelve un status response success
except ApiException as e:
return HTTPResponse(
502,
body=json.dumps({"status": "error", "error": str(e)}, ensure_ascii=False),
)
except Exception as e:
return HTTPResponse(
500,
body=json.dumps({"status": "error", "error": str(e)}, ensure_ascii=False),
) # En caso contrario, retorna la excepción capturada.
2 - Bproc: En este caso se utilizó un bproc de tipo transacción y modo webhook
Se configuró para que se ejecute el bproc cuando se cree o actualice una factura
En los parámetros se indica el límite de la deuda y un correo usado como copia interna cuando se envía el aviso de la deuda.
En el campo URL se pega la obtenida en el guest code
Resultado
Al guardar una nueva factura para un cliente que supera la tolerancia establecida, se envía un correo electrónico al cliente y una copia al correo indicado en el bproc
