Automatización de Procesos de Negocio: Más Allá del Script Frágil

{
  "nodes": [
    {
      "name": "Trigger: Nueva factura en ERP",
      "type": "n8n-nodes-base.webhook",
      "parameters": {
        "path": "/erp/nueva-factura",
        "responseMode": "lastNode"
      }
    },
    {
      "name": "Validar datos de factura",
      "type": "n8n-nodes-base.code",
      "parameters": {
        "jsCode": "const { numero, monto, cliente_id } = $input.first().json;\nif (!numero || !monto || !cliente_id) {\n  throw new Error('Datos de factura incompletos');\n}\nreturn $input.all();"
      }
    },
    {
      "name": "Enviar a sistema de cobros",
      "type": "n8n-nodes-base.httpRequest",
      "parameters": {
        "url": "https://cobros.empresa.com/api/facturas",
        "method": "POST",
        "authentication": "headerAuth"
      }
    },
    {
      "name": "Notificar equipo de finanzas",
      "type": "n8n-nodes-base.slack",
      "parameters": {
        "channel": "#finanzas",
        "text": "Nueva factura #{{ $node.Trigger.json.numero }} procesada"
      }
    }
  ]
}

from datetime import datetime, timedelta
from airflow.decorators import dag, task
from airflow.providers.postgres.hooks.postgres import PostgresHook
from airflow.providers.slack.operators.slack_webhook import SlackWebhookOperator

@dag(
    schedule_interval='@daily',
    start_date=datetime(2025, 1, 1),
    catchup=False,
    default_args={
        'retries': 3,
        'retry_delay': timedelta(minutes=5),
        'retry_exponential_backoff': True,
    }
)
def reporte_diario_ventas():

    @task()
    def extraer_ventas_del_dia():
        hook = PostgresHook(postgres_conn_id='erp_production')
        records = hook.get_records(
            "SELECT producto_id, sum(total) FROM ventas WHERE fecha = CURRENT_DATE GROUP BY 1"
        )
        return [{'producto_id': r[0], 'total': float(r[1])} for r in records]

    @task()
    def calcular_metricas(ventas: list):
        total = sum(v['total'] for v in ventas)
        top_productos = sorted(ventas, key=lambda x: x['total'], reverse=True)[:5]
        return {'total_dia': total, 'top_5': top_productos}

    @task()
    def guardar_en_warehouse(metricas: dict):
        hook = PostgresHook(postgres_conn_id='analytics_warehouse')
        hook.run(
            "INSERT INTO reportes_diarios (fecha, total, metricas_json) VALUES (CURRENT_DATE, %s, %s)",
            parameters=(metricas['total_dia'], str(metricas))
        )

    ventas = extraer_ventas_del_dia()
    metricas = calcular_metricas(ventas)
    guardar_en_warehouse(metricas)

dag = reporte_diario_ventas()

# Patrón idempotente con tabla de control de ejecución
import hashlib

def procesar_orden(orden_id: str, items: list) -> None:
    # Generar clave de idempotencia basada en los datos de entrada
    payload_hash = hashlib.sha256(
        f"{orden_id}:{sorted(items)}".encode()
    ).hexdigest()

    # Verificar si ya fue procesada (tabla de control)
    cursor.execute(
        "SELECT id FROM ordenes_procesadas WHERE idempotency_key = %s",
        (payload_hash,)
    )
    if cursor.fetchone():
        logger.info(f"Orden {orden_id} ya procesada, saltando")
        return

    # Procesar la orden
    resultado = _aplicar_descuentos_y_facturar(orden_id, items)

    # Registrar como procesada
    cursor.execute(
        "INSERT INTO ordenes_procesadas (idempotency_key, orden_id, resultado, processed_at) VALUES (%s, %s, %s, NOW())",
        (payload_hash, orden_id, str(resultado))
    )