Ir al contenido

Mejores Prácticas para Procesos RPA

Diseñar procesos RPA efectivos es un arte que combina principios de ingeniería de software, conocimiento del negocio y experiencia práctica. Esta guía detalla las mejores prácticas para crear automatizaciones robustas, mantenibles y eficientes con Heptora, aprovechando sus características únicas para facilitar la implementación de cada práctica.

Un proceso RPA debe ser comprensible por cualquier persona del equipo, no solo por quien lo creó.

Beneficios de la claridad:

  • Reducción del tiempo de onboarding de nuevos miembros
  • Facilita el mantenimiento y las actualizaciones
  • Disminuye la probabilidad de errores
  • Mejora la colaboración entre equipos técnicos y de negocio

Cómo Heptora facilita la claridad:

  • Interfaz visual intuitiva para diseñar flujos
  • Nomenclatura descriptiva en bloques y acciones
  • Documentación integrada en cada bloque
  • Vista de flujo que muestra el proceso completo

2. Automatiza lo Repetitivo, No lo Excepcional

Sección titulada «2. Automatiza lo Repetitivo, No lo Excepcional»

No intentes automatizar cada caso extremo desde el principio. Enfócate en el 80% de casos comunes.

Estrategia recomendada:

  1. Identifica el flujo principal (happy path)
  2. Automatiza ese flujo primero
  3. Mide resultados y detecta casos especiales
  4. Añade manejo de excepciones progresivamente
  5. Deja casos muy raros para procesamiento manual

Ejemplo práctico:

❌ Mal enfoque:
Intentar automatizar todos los formatos de facturas posibles
(PDF, XML, TXT, escaneadas, escritas a mano, etc.)
✅ Buen enfoque:
Automatizar facturas XML (80% del volumen)
Detectar otros formatos y enviarlos a revisión manual
Añadir soporte para PDF después si el volumen lo justifica

Es mejor que un proceso falle inmediatamente con un mensaje claro que continuar con datos incorrectos.

Principios:

  • Valida datos de entrada al inicio del proceso
  • Usa mensajes de error descriptivos
  • Registra el contexto del fallo
  • No ocultes errores esperando que se resuelvan solos

Cómo Heptora ayuda:

  • Sistema de resultados OK/KO/ERROR claramente diferenciados
  • Logs detallados automáticos
  • Notificaciones con contexto completo
  • Panel de ejecuciones con trazabilidad

Los bloques simples son más fáciles de:

  • Entender: Leer el código y comprender su propósito
  • Probar: Verificar que funcionen correctamente
  • Depurar: Identificar y corregir problemas
  • Reutilizar: Usar en múltiples procesos
  • Mantener: Actualizar sin romper otros componentes

Cada bloque debe hacer una cosa y hacerla bien.

❌ Ejemplo de Bloque Complejo (Anti-patrón)

Sección titulada «❌ Ejemplo de Bloque Complejo (Anti-patrón)»
# Bloque: "Procesar Cliente Completo"
def procesar_cliente_completo():
# Lee datos del Excel
clientes = heptora.excel.load("clientes.xlsx", "Hoja1", ["Nombre", "Email", "NIF"], True)
# Valida los datos
for cliente in clientes:
if not validar_email(cliente["Email"]):
heptora.log.force_ko(f"Email inválido: {cliente['Email']}")
# Se conecta a la web
page = heptora.browser.connect()
page.goto("https://portal.empresa.com")
# Hace login
page.locator("#user").fill("admin")
page.locator("#pass").fill("password123")
page.locator("#login").click()
# Busca el cliente
page.locator("#search").fill(cliente["NIF"])
page.locator("#btn_search").click()
# Actualiza los datos
page.locator("#nombre").fill(cliente["Nombre"])
page.locator("#email").fill(cliente["Email"])
page.locator("#guardar").click()
# Descarga el comprobante
page.locator("#download").click()
# Envía un email
enviar_email(cliente["Email"], "Cliente actualizado")
# Registra en base de datos
guardar_en_db(cliente)
heptora.log.result("ok", f"Cliente {cliente['Nombre']} procesado")

Problemas de este bloque:

  • Hace demasiadas cosas (leer, validar, login, buscar, actualizar, descargar, enviar email, guardar)
  • Difícil de probar individualmente
  • Si falla una parte, es difícil saber dónde
  • No se puede reutilizar el login o la búsqueda por separado
  • Credenciales hardcodeadas (inseguro)
  • Mezcla lógica de negocio con automatización web

✅ Ejemplo de Bloques Simples (Mejor Práctica)

Sección titulada «✅ Ejemplo de Bloques Simples (Mejor Práctica)»

Bloque 1: Cargar Clientes

# Bloque: "Cargar Clientes desde Excel"
def cargar_clientes():
"""Carga datos de clientes desde el archivo Excel."""
columnas = ["Nombre", "Email", "NIF"]
clientes = heptora.excel.load("clientes.xlsx", "Hoja1", columnas, True)
heptora.log.info(f"Cargados {len(clientes)} clientes")
heptora.data.set("clientes", clientes)

Bloque 2: Validar Cliente

# Bloque: "Validar Datos de Cliente"
def validar_cliente(cliente):
"""Valida que los datos del cliente sean correctos."""
errores = []
if not cliente.get("Nombre"):
errores.append("Falta nombre")
if not cliente.get("Email") or "@" not in cliente["Email"]:
errores.append("Email inválido")
if not cliente.get("NIF") or len(cliente["NIF"]) != 9:
errores.append("NIF inválido")
if errores:
return False, ", ".join(errores)
return True, "OK"

Bloque 3: Login en Portal

# Bloque: "Login en Portal Empresa"
def login_portal():
"""Realiza login en el portal web."""
# Obtener credenciales de forma segura
usuario = heptora.data.get("portal_usuario")
password = heptora.data.get("portal_password")
if not usuario or not password:
heptora.log.force_ko("Faltan credenciales. Configúralas en Secretos.")
return False
page = heptora.browser.connect()
page.goto("https://portal.empresa.com")
page.locator("#user").fill(usuario)
page.locator("#pass").fill(password)
page.locator("#login").click()
# Verificar login exitoso
try:
page.wait_for_selector(".dashboard", timeout=5000)
heptora.log.info("Login exitoso")
return True
except:
heptora.log.force_ko("Error en login")
return False

Bloque 4: Buscar Cliente

# Bloque: "Buscar Cliente por NIF"
def buscar_cliente(nif):
"""Busca un cliente en el portal por su NIF."""
page = heptora.browser.connect()
page.locator("#search").fill(nif)
page.locator("#btn_search").click()
# Esperar resultados
try:
page.wait_for_selector(".cliente-detalle", timeout=5000)
heptora.log.info(f"Cliente {nif} encontrado")
return True
except:
heptora.log.info(f"Cliente {nif} no encontrado")
return False

Bloque 5: Actualizar Datos Cliente

# Bloque: "Actualizar Datos de Cliente"
def actualizar_cliente(cliente):
"""Actualiza los datos del cliente en el portal."""
page = heptora.browser.connect()
page.locator("#nombre").fill(cliente["Nombre"])
page.locator("#email").fill(cliente["Email"])
page.locator("#guardar").click()
# Verificar guardado
try:
page.wait_for_selector(".mensaje-exito", timeout=3000)
heptora.log.info(f"Cliente {cliente['Nombre']} actualizado")
return True
except:
heptora.log.info(f"Error al actualizar cliente {cliente['Nombre']}")
return False

Proceso Principal: Orquestación

# Proceso: "Actualizar Clientes en Portal"
def main():
"""Proceso principal que orquesta los bloques."""
# 1. Cargar datos
cargar_clientes()
clientes = heptora.data.get("clientes")
# 2. Login una sola vez
if not login_portal():
heptora.log.force_ko("No se pudo iniciar sesión")
return
# 3. Procesar cada cliente
for cliente in clientes:
# Validar
es_valido, mensaje = validar_cliente(cliente)
if not es_valido:
heptora.log.result("ko", f"Cliente inválido: {mensaje}")
continue
# Buscar
if not buscar_cliente(cliente["NIF"]):
heptora.log.result("ko", f"Cliente {cliente['NIF']} no existe en el portal")
continue
# Actualizar
if actualizar_cliente(cliente):
heptora.log.result("ok", f"Cliente {cliente['Nombre']} procesado correctamente")
else:
heptora.log.result("error", f"Error al actualizar cliente {cliente['Nombre']}")
# Pausa entre clientes
heptora.time.pause(1)
# Ejecutar
main()
AspectoBloque ComplejoBloques Simples
ComprensiónDifícil, requiere leer todoFácil, cada bloque es autoexplicativo
PruebasDifícil probar componentes individualesFácil probar cada bloque por separado
DepuraciónDifícil localizar el problemaFácil identificar qué bloque falla
ReutilizaciónImposible reutilizar partesBloques se pueden usar en otros procesos
MantenimientoCambios arriesgadosCambios localizados y seguros
ColaboraciónDifícil trabajar en paraleloVarios desarrolladores pueden trabajar simultáneamente

1. Bloques de Automatización Predefinidos

Heptora proporciona bloques listos para usar que ya implementan tareas comunes de forma simple y eficiente:

  • Bloques de lectura/escritura de Excel
  • Bloques de navegación web
  • Bloques de integración con APIs
  • Bloques de gestión de archivos
  • Bloques de envío de emails

2. Validaciones Automáticas

El sistema valida automáticamente:

  • Parámetros requeridos
  • Tipos de datos correctos
  • Conexiones válidas
  • Permisos necesarios

3. Interfaz Visual

La vista de flujo visual ayuda a:

  • Identificar bloques que hacen demasiado
  • Visualizar dependencias entre bloques
  • Detectar flujos complejos que necesitan simplificación
  • Documentar visualmente el proceso

4. Límites Naturales

Heptora establece límites naturales que fomentan la simplicidad:

  • Timeouts razonables
  • Límites de recursos
  • Separación entre bloques de Python y lenguaje Heptora
  • Estructura clara de entrada/salida de datos

La reutilización es uno de los principios más importantes en RPA:

  • Reduce tiempo de desarrollo: No reinventar la rueda
  • Mejora la calidad: Componentes probados y validados
  • Facilita mantenimiento: Una corrección beneficia a todos los procesos
  • Estandariza procesos: Consistencia en toda la organización
  • Acelera escalabilidad: Más fácil crear nuevos procesos

Los bloques de automatización son componentes reutilizables que encapsulan funcionalidad específica.

Ejemplos de bloques reutilizables:

  • Login en sistema específico (SAP, Salesforce, portal web)
  • Validación de datos (NIF, email, IBAN)
  • Envío de notificaciones
  • Lectura/escritura de bases de datos
  • Transformación de formatos (CSV a JSON, etc.)
  • Descarga de archivos
  • Generación de reportes

Bloque reutilizable: Validar NIF Español

# Bloque: "Validar NIF Español"
def validar_nif(nif):
"""
Valida que un NIF español tenga formato correcto.
Args:
nif (str): NIF a validar
Returns:
Tuple[bool, str]: (es_valido, mensaje)
"""
if not nif or not isinstance(nif, str):
return False, "NIF vacío o no es texto"
# Eliminar espacios
nif = nif.strip().upper()
# Validar longitud
if len(nif) != 9:
return False, f"NIF debe tener 9 caracteres, tiene {len(nif)}"
# Validar formato: 8 dígitos + 1 letra
if not nif[:8].isdigit() or not nif[8].isalpha():
return False, "Formato incorrecto. Debe ser 8 dígitos + 1 letra"
# Validar letra
letras = "TRWAGMYFPDXBNJZSQVHLCKE"
numero = int(nif[:8])
letra_calculada = letras[numero % 23]
if nif[8] != letra_calculada:
return False, f"Letra incorrecta. Debería ser {letra_calculada}"
return True, "NIF válido"
# Usar el bloque reutilizable
nif_cliente = "12345678Z"
es_valido, mensaje = validar_nif(nif_cliente)
if not es_valido:
heptora.log.force_ko(f"NIF inválido: {mensaje}")

Bloque reutilizable: Login Genérico

# Bloque: "Login Web Genérico"
def login_web(url, selector_user, selector_pass, selector_submit, verificador_exito):
"""
Realiza login en un portal web genérico.
Args:
url (str): URL del login
selector_user (str): Selector CSS del campo usuario
selector_pass (str): Selector CSS del campo contraseña
selector_submit (str): Selector CSS del botón submit
verificador_exito (str): Selector que confirma login exitoso
Returns:
bool: True si login exitoso
"""
# Obtener credenciales de forma segura
usuario = heptora.data.get("login_usuario")
password = heptora.data.get("login_password")
if not usuario or not password:
heptora.log.force_ko("Faltan credenciales")
return False
page = heptora.browser.connect()
page.goto(url)
try:
# Esperar formulario
page.wait_for_selector(selector_user, timeout=10000)
# Rellenar y enviar
page.locator(selector_user).fill(usuario)
page.locator(selector_pass).fill(password)
page.locator(selector_submit).click()
# Verificar éxito
page.wait_for_selector(verificador_exito, timeout=10000)
heptora.log.info("Login exitoso")
return True
except Exception as e:
heptora.log.info(f"Error en login: {str(e)}")
return False
# Usar en diferentes portales
# Portal 1: Sistema interno
login_web(
"https://sistema.empresa.com/login",
"#username",
"#password",
"button[type='submit']",
".dashboard"
)
# Portal 2: Proveedor externo
login_web(
"https://proveedor.com/acceso",
"input[name='user']",
"input[name='pass']",
"#btnLogin",
".home-page"
)

Las plantillas de procesos son estructuras completas que se pueden adaptar a diferentes casos de uso.

Cómo Heptora facilita las plantillas:

Heptora ofrece Plantillas de Procesos predefinidas para casos comunes:

  • Facturación electrónica (FACe)
  • Gestión de facturas con mutuas
  • Presentación de documentos
  • Consultas en portales web
  • Integración con ERPs

Ejemplo: Plantilla de Procesamiento de Documentos

Plantilla: "Procesamiento Masivo de Documentos"
Estructura:
1. [Bloque] Cargar documentos desde carpeta
2. [Bloque] Validar formato y contenido
3. [Bloque] Clasificar documentos por tipo
4. [Loop] Para cada documento válido:
4.1. [Bloque] Extraer información clave
4.2. [Bloque] Procesar según tipo
4.3. [Bloque] Guardar resultado
5. [Bloque] Generar reporte de procesamiento
6. [Bloque] Mover archivos a carpetas finales
7. [Bloque] Enviar notificación con resumen
Parámetros configurables:
- Carpeta de entrada
- Tipos de documentos soportados
- Reglas de validación
- Reglas de clasificación
- Destinatarios de notificaciones

Funciones de utilidad que se usan frecuentemente en muchos procesos.

Biblioteca de Funciones Reutilizables:

utils/validaciones.py
"""Funciones de validación reutilizables."""
def validar_email(email):
"""Valida formato de email."""
import re
patron = r'^[\w\.-]+@[\w\.-]+\.\w+$'
return re.match(patron, email) is not None
def validar_iban(iban):
"""Valida formato de IBAN español."""
if not iban or len(iban) != 24:
return False
if not iban.startswith("ES"):
return False
return iban[2:].replace(" ", "").isalnum()
def validar_fecha(fecha_str, formato="%Y-%m-%d"):
"""Valida que una cadena sea una fecha válida."""
from datetime import datetime
try:
datetime.strptime(fecha_str, formato)
return True
except ValueError:
return False
def validar_importe(importe_str):
"""Valida y normaliza un importe."""
try:
# Eliminar espacios, cambiar comas por puntos
importe_limpio = importe_str.strip().replace(",", ".")
importe = float(importe_limpio)
return importe > 0, importe
except:
return False, 0
utils/formateo.py
"""Funciones de formateo reutilizables."""
def normalizar_nif(nif):
"""Normaliza un NIF eliminando espacios y guiones."""
return nif.upper().replace(" ", "").replace("-", "")
def formatear_fecha(fecha_str, formato_entrada, formato_salida):
"""Convierte fecha de un formato a otro."""
from datetime import datetime
fecha = datetime.strptime(fecha_str, formato_entrada)
return fecha.strftime(formato_salida)
def formatear_importe(importe, decimales=2, separador_miles="."):
"""Formatea un importe para visualización."""
formato = f"{{:,.{decimales}f}}"
resultado = formato.format(importe)
# Cambiar separadores si es necesario
if separador_miles == ".":
resultado = resultado.replace(",", "X").replace(".", ",").replace("X", ".")
return resultado
def limpiar_texto(texto):
"""Elimina espacios extra y normaliza texto."""
return " ".join(texto.split())
utils/archivos.py
"""Funciones de gestión de archivos reutilizables."""
import os
from datetime import datetime
def crear_nombre_archivo_con_fecha(prefijo, extension):
"""Crea nombre de archivo con timestamp."""
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
return f"{prefijo}_{timestamp}.{extension}"
def asegurar_carpeta_existe(ruta_carpeta):
"""Crea carpeta si no existe."""
if not os.path.exists(ruta_carpeta):
os.makedirs(ruta_carpeta)
return True
return False
def mover_archivo_a_procesados(archivo_origen, carpeta_destino):
"""Mueve archivo a carpeta de procesados."""
import shutil
asegurar_carpeta_existe(carpeta_destino)
nombre_archivo = os.path.basename(archivo_origen)
ruta_destino = os.path.join(carpeta_destino, nombre_archivo)
shutil.move(archivo_origen, ruta_destino)
return ruta_destino
def listar_archivos_por_extension(carpeta, extension):
"""Lista todos los archivos de una extensión en una carpeta."""
archivos = []
for archivo in os.listdir(carpeta):
if archivo.endswith(f".{extension}"):
ruta_completa = os.path.join(carpeta, archivo)
archivos.append(ruta_completa)
return archivos

Usar las funciones reutilizables:

# En tu proceso
from utils.validaciones import validar_email, validar_nif, validar_importe
from utils.formateo import formatear_fecha, formatear_importe
from utils.archivos import crear_nombre_archivo_con_fecha
# Proceso principal
clientes = heptora.data.get("clientes")
for cliente in clientes:
# Usar validaciones reutilizables
if not validar_email(cliente["Email"]):
heptora.log.force_ko(f"Email inválido: {cliente['Email']}")
continue
if not validar_nif(cliente["NIF"]):
heptora.log.force_ko(f"NIF inválido: {cliente['NIF']}")
continue
# Usar formateo reutilizable
fecha_formateada = formatear_fecha(
cliente["Fecha"],
"%d/%m/%Y", # formato entrada
"%Y-%m-%d" # formato salida
)
# Procesar cliente...
heptora.log.result("ok", f"Cliente {cliente['Nombre']} procesado")
# Generar reporte con nombre único
nombre_reporte = crear_nombre_archivo_con_fecha("reporte_clientes", "xlsx")
heptora.excel.write(resultados, nombre_reporte, "Resultados")

Estructura de carpetas recomendada:

mi_organizacion_rpa/
├── bloques/
│ ├── validaciones/
│ │ ├── validar_nif.py
│ │ ├── validar_email.py
│ │ └── validar_iban.py
│ ├── login/
│ │ ├── login_sap.py
│ │ ├── login_salesforce.py
│ │ └── login_portal_interno.py
│ └── reportes/
│ ├── generar_reporte_excel.py
│ └── enviar_email_resumen.py
├── plantillas/
│ ├── procesamiento_facturas/
│ ├── actualizacion_inventario/
│ └── consulta_estado_pedidos/
├── utils/
│ ├── validaciones.py
│ ├── formateo.py
│ ├── archivos.py
│ └── fechas.py
└── procesos/
├── proceso_facturas_proveedores.py
├── proceso_pedidos_clientes.py
└── proceso_inventario_mensual.py

Cada componente reutilizable debe incluir:

  1. Descripción clara: Qué hace el componente
  2. Parámetros: Qué entradas necesita
  3. Retorno: Qué devuelve
  4. Ejemplos de uso: Cómo usarlo en diferentes contextos
  5. Dependencias: Qué otros componentes o librerías necesita
  6. Autor y fecha: Quién lo creó y cuándo
  7. Historial de cambios: Versiones y modificaciones

Ejemplo de documentación:

"""
Bloque: Validar y Normalizar Cliente
Descripción:
Valida que los datos de un cliente cumplan con los requisitos
mínimos y normaliza los campos para procesamiento consistente.
Parámetros:
cliente (dict): Diccionario con datos del cliente
- Nombre (str): Nombre completo
- Email (str): Dirección de email
- NIF (str): NIF español
- Telefono (str, opcional): Teléfono de contacto
Retorna:
Tuple[bool, dict, str]:
- bool: True si el cliente es válido
- dict: Cliente con datos normalizados
- str: Mensaje de error si no es válido
Ejemplo de uso:
cliente = {
"Nombre": " Juan García ",
"Email": "juan@empresa.com",
"NIF": "12345678-Z",
"Telefono": "+34 600 123 456"
}
es_valido, cliente_limpio, error = validar_normalizar_cliente(cliente)
if es_valido:
# Usar cliente_limpio
procesar(cliente_limpio)
else:
heptora.log.force_ko(f"Cliente inválido: {error}")
Dependencias:
- utils.validaciones: validar_email, validar_nif
- utils.formateo: limpiar_texto, normalizar_nif
Autor: Equipo RPA
Fecha: 2024-01-15
Versión: 1.2
Historial:
v1.0 (2024-01-15): Versión inicial
v1.1 (2024-02-01): Añadida validación de teléfono
v1.2 (2024-03-10): Mejorada normalización de NIF
"""
def validar_normalizar_cliente(cliente):
from utils.validaciones import validar_email, validar_nif
from utils.formateo import limpiar_texto, normalizar_nif
# Validar campos requeridos
if not cliente.get("Nombre"):
return False, None, "Falta nombre del cliente"
if not cliente.get("Email"):
return False, None, "Falta email del cliente"
if not cliente.get("NIF"):
return False, None, "Falta NIF del cliente"
# Validar formato de email
if not validar_email(cliente["Email"]):
return False, None, f"Email inválido: {cliente['Email']}"
# Validar y normalizar NIF
nif_normalizado = normalizar_nif(cliente["NIF"])
if not validar_nif(nif_normalizado):
return False, None, f"NIF inválido: {cliente['NIF']}"
# Normalizar datos
cliente_limpio = {
"Nombre": limpiar_texto(cliente["Nombre"]),
"Email": cliente["Email"].lower().strip(),
"NIF": nif_normalizado,
"Telefono": cliente.get("Telefono", "").strip()
}
return True, cliente_limpio, ""

Principios de versionado:

  • Versionado semántico: MAJOR.MINOR.PATCH
    • MAJOR: Cambios incompatibles (rompe código existente)
    • MINOR: Nueva funcionalidad compatible
    • PATCH: Correcciones de bugs

Ejemplo:

2.1.3
# Bloque: Login SAP
"""
Historial de versiones:
- 1.0.0: Versión inicial con login básico
- 1.1.0: Añadido soporte para autenticación de dos factores
- 1.2.0: Añadida gestión de sesiones persistentes
- 2.0.0: BREAKING CHANGE - Cambio en estructura de parámetros
- 2.1.0: Añadida detección automática de idioma del sistema
- 2.1.1: Corrección de bug en timeout
- 2.1.2: Corrección de bug en manejo de certificados
- 2.1.3: Optimización de rendimiento
"""

Mantén un catálogo interno con:

  • Lista de todos los componentes reutilizables
  • Descripción breve de cada uno
  • Ejemplos de uso
  • Procesos que los utilizan
  • Responsable del mantenimiento

Ejemplo de catálogo (formato Markdown):

# Catálogo de Componentes Reutilizables - RPA
## Validaciones
### validar_nif(nif) → bool
Valida formato de NIF español.
- **Ubicación**: `bloques/validaciones/validar_nif.py`
- **Versión**: 1.3
- **Usado en**: 15 procesos
- **Responsable**: Equipo RPA Core
### validar_email(email) → bool
Valida formato de dirección de email.
- **Ubicación**: `bloques/validaciones/validar_email.py`
- **Versión**: 1.1
- **Usado en**: 23 procesos
- **Responsable**: Equipo RPA Core
## Login
### login_sap(usuario, password) → bool
Realiza login en SAP.
- **Ubicación**: `bloques/login/login_sap.py`
- **Versión**: 2.1.3
- **Usado en**: 8 procesos
- **Responsable**: María García
### login_salesforce() → bool
Realiza login en Salesforce usando credenciales del sistema de secretos.
- **Ubicación**: `bloques/login/login_salesforce.py`
- **Versión**: 1.5
- **Usado en**: 5 procesos
- **Responsable**: Carlos Martínez
## Reportes
### generar_reporte_excel(datos, plantilla) → str
Genera reporte Excel usando plantilla predefinida.
- **Ubicación**: `bloques/reportes/generar_reporte_excel.py`
- **Versión**: 2.0
- **Usado en**: 18 procesos
- **Responsable**: Ana López

La gestión de excepciones es crítica para procesos RPA robustos. Un buen manejo de errores diferencia entre procesos que fallan constantemente y procesos confiables que manejan problemas elegantemente.

El mejor error es el que nunca ocurre.

Jerarquía de estrategias:

  1. Prevenir: Validar datos de entrada antes de procesarlos
  2. Detectar: Identificar problemas rápidamente cuando ocurren
  3. Corregir: Recuperarse automáticamente cuando sea posible
  4. Reportar: Notificar problemas que requieren intervención humana

No ocultes errores esperando que desaparezcan. Detecta rápido, registra claramente y decide la mejor acción.

3. Distingue Errores Controlados de Inesperados

Sección titulada «3. Distingue Errores Controlados de Inesperados»

Error Controlado (KO):

  • Problema esperado según reglas de negocio
  • Datos incorrectos o incompletos
  • Registro ya procesado
  • Estado que no permite la acción

Error Inesperado (ERROR):

  • Problema técnico no anticipado
  • Fallo de conectividad
  • Sistema no disponible
  • Elemento no encontrado en interfaz

Cómo Heptora Facilita la Gestión de Excepciones

Sección titulada «Cómo Heptora Facilita la Gestión de Excepciones»

Heptora proporciona un sistema claro de resultados que facilita la gestión de excepciones:

# Resultado exitoso
heptora.log.result("ok", "Factura procesada correctamente")
# Error controlado (dato incorrecto)
heptora.log.result("ko", "Factura ya está pagada, no requiere procesamiento")
# Error inesperado (problema técnico)
heptora.log.result("error", "No se pudo conectar al servidor")

Ventajas del sistema de Heptora:

  • Clasificación automática de resultados
  • Reportes diferenciados por tipo de resultado
  • Métricas claras (tasa de OK vs KO vs ERROR)
  • Notificaciones configurables según tipo
  • Trazabilidad completa en el panel de ejecuciones

heptora.log.force_ko() detiene el procesamiento del registro actual y marca como KO:

# Validación al inicio
if not factura.get("Numero"):
heptora.log.force_ko("Falta número de factura")
# El resto del código no se ejecuta para este registro
# El proceso continúa con el siguiente registro automáticamente

Cuándo usar force_ko():

  • Datos de entrada inválidos
  • Validaciones de negocio no cumplidas
  • Registros que no deben procesarse
  • Situaciones esperadas que requieren revisión manual

Cuándo NO usar force_ko():

  • Errores técnicos inesperados (usar result("error", ...))
  • Cuando quieres intentar recuperación automática
  • En errores de configuración del proceso

Valida todos los datos de entrada al inicio del proceso, antes de realizar cualquier acción costosa.

Patrón de validación en cascada:

def procesar_factura(factura):
"""Procesa una factura con validaciones tempranas."""
# ========================================
# VALIDACIONES TEMPRANAS
# ========================================
# Validación 1: Campos requeridos
campos_requeridos = ["Numero", "Cliente", "Importe", "Fecha"]
for campo in campos_requeridos:
if not factura.get(campo):
heptora.log.force_ko(f"Falta campo requerido: {campo}")
return
# Validación 2: Formato de datos
try:
importe = float(factura["Importe"])
except (ValueError, TypeError):
heptora.log.force_ko(f"Importe inválido: {factura['Importe']}")
return
# Validación 3: Rango de valores
if importe <= 0:
heptora.log.force_ko(f"Importe debe ser mayor a 0, es: {importe}")
return
if importe > 1000000:
heptora.log.force_ko(f"Importe excede el máximo permitido: {importe}")
return
# Validación 4: Reglas de negocio
if factura.get("Estado") == "Pagada":
heptora.log.force_ko("La factura ya está pagada")
return
# ========================================
# PROCESAMIENTO (solo si pasó validaciones)
# ========================================
try:
# Código de procesamiento...
resultado = enviar_factura_a_portal(factura)
heptora.log.result("ok", f"Factura {factura['Numero']} procesada")
except Exception as e:
# Error técnico inesperado
heptora.log.result("error", f"Error al procesar factura: {str(e)}")

Usa bloques try-except específicos para diferentes tipos de errores.

❌ Evitar: Try-except genérico que oculta problemas

try:
# Todo el proceso aquí...
cargar_datos()
validar_datos()
procesar_datos()
guardar_resultados()
enviar_notificacion()
except:
heptora.log.result("error", "Algo falló") # ¡Muy vago!

✅ Mejor: Try-except granular con manejo específico

# Cargar datos
try:
facturas = heptora.excel.load("facturas.xlsx", "Hoja1", columnas, True)
except FileNotFoundError:
heptora.log.force_ko("No se encontró el archivo facturas.xlsx")
return
except Exception as e:
heptora.log.force_ko(f"Error al leer Excel: {str(e)}")
return
# Procesar cada factura
for factura in facturas:
try:
# Validaciones (force_ko si falla)
validar_factura(factura)
# Procesamiento con manejo específico de errores
try:
resultado = enviar_a_portal(factura)
heptora.log.result("ok", f"Factura {factura['Numero']} procesada")
except ConnectionError as e:
# Error de red: puede ser temporal, marcar para reintento
heptora.log.result("error", f"Error de conexión: {str(e)}")
agregar_a_reintentos(factura)
except ValueError as e:
# Error de datos: problema con la factura específica
heptora.log.result("ko", f"Datos incorrectos: {str(e)}")
except TimeoutError as e:
# Timeout: puede ser carga del servidor
heptora.log.result("error", f"Timeout al procesar: {str(e)}")
agregar_a_reintentos(factura)
except Exception as e:
# Error inesperado en este registro
heptora.log.result("error", f"Error inesperado: {str(e)}")
continue # Continuar con siguiente factura
# Generar reporte final
try:
generar_reporte_resumen()
except Exception as e:
heptora.log.info(f"Advertencia: No se pudo generar reporte: {str(e)}")
# No fallar todo el proceso por esto

Para errores transitorios (conectividad, timeouts), implementa reintentos con retroceso exponencial.

Patrón de reintentos:

def ejecutar_con_reintentos(funcion, max_intentos=3, espera_base=2):
"""
Ejecuta una función con reintentos exponenciales.
Args:
funcion: Función a ejecutar
max_intentos: Número máximo de intentos
espera_base: Segundos base de espera (se duplica en cada intento)
Returns:
Resultado de la función o None si falla todos los intentos
"""
for intento in range(1, max_intentos + 1):
try:
heptora.log.info(f"Intento {intento}/{max_intentos}")
resultado = funcion()
return resultado
except (ConnectionError, TimeoutError) as e:
# Errores que justifican reintento
if intento < max_intentos:
espera = espera_base ** intento # Retroceso exponencial: 2, 4, 8 segundos...
heptora.log.info(f"Error transitorio: {str(e)}. Reintentando en {espera}s...")
heptora.time.pause(espera)
else:
heptora.log.info(f"Agotados {max_intentos} intentos. Error: {str(e)}")
return None
except Exception as e:
# Errores que no justifican reintento
heptora.log.info(f"Error no recuperable: {str(e)}")
return None
return None
# Usar el patrón
def enviar_factura():
"""Función que puede fallar temporalmente."""
page = heptora.browser.connect()
page.goto("https://portal.facturas.com")
# ... resto del procesamiento
return "OK"
# Ejecutar con reintentos automáticos
resultado = ejecutar_con_reintentos(enviar_factura, max_intentos=3, espera_base=2)
if resultado:
heptora.log.result("ok", "Factura enviada correctamente")
else:
heptora.log.result("error", "No se pudo enviar factura tras múltiples intentos")

Cuándo usar reintentos:

  • Errores de conectividad
  • Timeouts
  • Errores 5xx del servidor (500, 503)
  • Sistema temporalmente no disponible
  • Recursos momentáneamente bloqueados

Cuándo NO usar reintentos:

  • Errores de validación de datos
  • Errores 4xx (400, 404, 401, 403)
  • Errores de lógica del programa
  • Recursos que no existen
  • Permisos insuficientes

Si una funcionalidad secundaria falla, el proceso principal debe continuar.

Ejemplo:

def procesar_factura_completo(factura):
"""Procesa factura con funciones principales y secundarias."""
# ========================================
# FUNCIONALIDAD PRINCIPAL (crítica)
# ========================================
try:
# Enviar factura al portal (CRÍTICO)
enviar_a_portal(factura)
heptora.log.result("ok", f"Factura {factura['Numero']} procesada")
except Exception as e:
# Si falla lo principal, es ERROR y no continuamos
heptora.log.result("error", f"Error crítico: {str(e)}")
return # No continuar con funciones secundarias
# ========================================
# FUNCIONALIDADES SECUNDARIAS (opcionales)
# ========================================
# Generar PDF de comprobante (secundario)
try:
generar_pdf_comprobante(factura)
heptora.log.info("PDF generado correctamente")
except Exception as e:
heptora.log.info(f"Advertencia: No se pudo generar PDF: {str(e)}")
# Continuar aunque falle esto
# Enviar email de confirmación (secundario)
try:
enviar_email_confirmacion(factura["Email"])
heptora.log.info("Email enviado correctamente")
except Exception as e:
heptora.log.info(f"Advertencia: No se pudo enviar email: {str(e)}")
# Continuar aunque falle esto
# Actualizar sistema auxiliar (secundario)
try:
actualizar_sistema_analytics(factura)
heptora.log.info("Analytics actualizado")
except Exception as e:
heptora.log.info(f"Advertencia: No se pudo actualizar analytics: {str(e)}")
# Continuar aunque falle esto

Para sistemas externos que fallan repetidamente, implementa un circuit breaker que detenga intentos temporalmente.

Patrón Circuit Breaker:

class CircuitBreaker:
"""
Implementa el patrón Circuit Breaker para proteger de fallos en cascada.
Estados:
- CLOSED: Funcionamiento normal, permite llamadas
- OPEN: Muchos fallos detectados, bloquea llamadas
- HALF_OPEN: Período de prueba, permite algunas llamadas
"""
def __init__(self, max_fallos=5, timeout_reset=60):
self.max_fallos = max_fallos
self.timeout_reset = timeout_reset
self.fallos = 0
self.ultimo_fallo = None
self.estado = "CLOSED"
def llamar(self, funcion):
"""Ejecuta función con protección de circuit breaker."""
from datetime import datetime, timedelta
# Si está OPEN, verificar si ya pasó el timeout
if self.estado == "OPEN":
if datetime.now() - self.ultimo_fallo > timedelta(seconds=self.timeout_reset):
heptora.log.info("Circuit Breaker: Cambiando a HALF_OPEN")
self.estado = "HALF_OPEN"
self.fallos = 0
else:
heptora.log.info("Circuit Breaker OPEN: Bloqueando llamada")
return None
# Intentar llamar a la función
try:
resultado = funcion()
# Éxito: resetear contador si estaba en HALF_OPEN
if self.estado == "HALF_OPEN":
heptora.log.info("Circuit Breaker: Llamada exitosa, cambiando a CLOSED")
self.estado = "CLOSED"
self.fallos = 0
return resultado
except Exception as e:
# Fallo: incrementar contador
self.fallos += 1
self.ultimo_fallo = datetime.now()
heptora.log.info(f"Circuit Breaker: Fallo {self.fallos}/{self.max_fallos}")
# Si superamos máximo de fallos, abrir circuito
if self.fallos >= self.max_fallos:
self.estado = "OPEN"
heptora.log.info(
f"Circuit Breaker: ABIERTO por {self.timeout_reset}s "
f"debido a {self.fallos} fallos consecutivos"
)
raise e
# Usar el Circuit Breaker
cb_portal = CircuitBreaker(max_fallos=5, timeout_reset=120)
for factura in facturas:
def enviar():
return enviar_a_portal(factura)
try:
resultado = cb_portal.llamar(enviar)
if resultado:
heptora.log.result("ok", f"Factura {factura['Numero']} procesada")
else:
heptora.log.result(
"error",
f"Portal no disponible (Circuit Breaker OPEN). Reintentar más tarde."
)
except Exception as e:
heptora.log.result("error", f"Error al procesar: {str(e)}")

El logging efectivo es crucial para debugging y auditoría.

# Información general del flujo
heptora.log.info("Iniciando procesamiento de facturas")
heptora.log.info(f"Cargadas {len(facturas)} facturas")
# Detalles de procesamiento
heptora.log.info(f"Procesando factura {i}/{total}: {numero}")
# Advertencias (problemas no críticos)
heptora.log.info(f"Advertencia: Campo opcional vacío: {campo}")
# Resultados de procesamiento
heptora.log.result("ok", "Factura procesada correctamente")
heptora.log.result("ko", "Factura con datos incorrectos")
heptora.log.result("error", "Error técnico al procesar")

✅ SI loggear:

  • Inicio y fin del proceso
  • Número de registros a procesar
  • Progreso (cada N registros)
  • Decisiones importantes (validaciones, bifurcaciones)
  • Errores y advertencias
  • Resultados finales (exitosos, fallidos)
  • Tiempos de ejecución de operaciones largas

❌ NO loggear:

  • Datos sensibles (contraseñas, tokens, NIFs completos)
  • Información personal identificable excesiva
  • Cada operación trivial (exceso de ruido)
  • Datos en formato no legible

Ejemplo de logging balanceado:

def procesar_facturas():
"""Procesa facturas con logging apropiado."""
# Log de inicio
heptora.log.info("=" * 60)
heptora.log.info("INICIO: Proceso de Facturas")
heptora.log.info("=" * 60)
# Cargar datos
facturas = heptora.excel.load("facturas.xlsx", "Hoja1", columnas, True)
total = len(facturas)
heptora.log.info(f"Cargadas {total} facturas para procesar")
# Procesar
exitosas = 0
kos = 0
errores = 0
for i, factura in enumerate(facturas, 1):
numero = factura.get("Numero", "SIN_NUM")
# Log cada 10 facturas o última
if i % 10 == 0 or i == total:
heptora.log.info(f"Progreso: {i}/{total} facturas procesadas")
# Validar
es_valida, mensaje = validar_factura(factura)
if not es_valida:
heptora.log.result("ko", f"Factura {numero}: {mensaje}")
kos += 1
continue
# Procesar
try:
enviar_a_portal(factura)
heptora.log.result("ok", f"Factura {numero} procesada correctamente")
exitosas += 1
except Exception as e:
# Log del error con contexto (pero sin datos sensibles)
heptora.log.result(
"error",
f"Factura {numero}: Error al enviar - {type(e).__name__}: {str(e)}"
)
errores += 1
# Log de resumen final
heptora.log.info("=" * 60)
heptora.log.info("RESUMEN FINAL")
heptora.log.info(f"Total procesadas: {total}")
heptora.log.info(f" ✓ Exitosas: {exitosas} ({exitosas/total*100:.1f}%)")
heptora.log.info(f" ⚠ KO (datos): {kos} ({kos/total*100:.1f}%)")
heptora.log.info(f" ✗ Errores: {errores} ({errores/total*100:.1f}%)")
heptora.log.info("=" * 60)
heptora.log.info("FIN: Proceso completado")

No notifiques todo; notifica lo importante.

Configurar notificaciones por criticidad:

def configurar_notificaciones(exitosas, kos, errores):
"""Decide qué notificaciones enviar según resultados."""
total = exitosas + kos + errores
tasa_error = errores / total if total > 0 else 0
tasa_ko = kos / total if total > 0 else 0
# Notificación siempre: resumen
enviar_email_resumen(exitosas, kos, errores)
# Notificación crítica: muchos errores técnicos
if tasa_error > 0.2: # Más del 20% errores
enviar_alerta_critica(
"ALERTA: Alta tasa de errores técnicos",
f"El proceso tiene {errores} errores de {total} registros ({tasa_error*100:.1f}%)"
)
# Notificación advertencia: muchos KOs
if tasa_ko > 0.3: # Más del 30% KOs
enviar_alerta_advertencia(
"Advertencia: Alta tasa de datos incorrectos",
f"El proceso tiene {kos} registros con problemas de datos de {total} ({tasa_ko*100:.1f}%)"
)
# Notificación éxito: proceso perfecto
if exitosas == total:
enviar_notificacion_exito(
f"Proceso completado sin errores",
f"Todas las {total} facturas se procesaron correctamente"
)

La navegación web es particularmente propensa a errores.

Estrategias específicas para web:

def navegar_con_manejo_robusto(page, url, selector_verificacion, timeout=10000):
"""Navega a URL con manejo robusto de errores."""
intentos_maximos = 3
for intento in range(1, intentos_maximos + 1):
try:
heptora.log.info(f"Navegando a {url} (intento {intento}/{intentos_maximos})")
# Navegar
page.goto(url, wait_until="domcontentloaded", timeout=timeout)
# Verificar que cargó correctamente
page.wait_for_selector(selector_verificacion, timeout=timeout)
heptora.log.info("Página cargada correctamente")
return True
except TimeoutError:
if intento < intentos_maximos:
heptora.log.info("Timeout al cargar página, reintentando...")
heptora.time.pause(2)
else:
heptora.log.info("Timeout: La página no cargó tras múltiples intentos")
return False
except Exception as e:
heptora.log.info(f"Error al navegar: {str(e)}")
return False
return False
def llenar_formulario_con_validacion(page, campo_selector, valor, nombre_campo):
"""Llena campo de formulario con validación."""
try:
# Esperar que el campo esté presente
page.wait_for_selector(campo_selector, timeout=5000)
# Llenar campo
page.locator(campo_selector).fill(valor)
# Verificar que se llenó correctamente
valor_actual = page.locator(campo_selector).input_value()
if valor_actual != valor:
raise ValueError(
f"El valor del campo {nombre_campo} no se guardó correctamente. "
f"Esperado: '{valor}', Actual: '{valor_actual}'"
)
heptora.log.info(f"Campo {nombre_campo} llenado correctamente")
return True
except TimeoutError:
heptora.log.info(f"Campo {nombre_campo} no encontrado en la página")
return False
except Exception as e:
heptora.log.info(f"Error al llenar campo {nombre_campo}: {str(e)}")
return False
def click_con_verificacion(page, boton_selector, nombre_boton, verificador_siguiente):
"""Hace click en botón y verifica que la acción se completó."""
try:
# Esperar botón
page.wait_for_selector(boton_selector, timeout=5000)
# Click
page.locator(boton_selector).click()
heptora.log.info(f"Click en {nombre_boton}")
# Verificar que se procesó la acción
page.wait_for_selector(verificador_siguiente, timeout=10000)
heptora.log.info(f"Acción de {nombre_boton} completada correctamente")
return True
except TimeoutError:
heptora.log.info(
f"Timeout: La acción de {nombre_boton} no se completó o "
f"el siguiente elemento no apareció"
)
return False
except Exception as e:
heptora.log.info(f"Error al hacer click en {nombre_boton}: {str(e)}")
return False
# Usar las funciones robustas
def procesar_factura_en_portal(factura):
"""Procesa factura en portal web con manejo robusto."""
page = heptora.browser.connect()
# Navegar con reintentos
if not navegar_con_manejo_robusto(
page,
"https://portal.facturas.com/nueva",
"#formulario-factura"
):
heptora.log.result("error", "No se pudo cargar el formulario")
return False
# Llenar campos con validación
if not llenar_formulario_con_validacion(
page, "#numero", factura["Numero"], "Número"
):
heptora.log.result("error", "No se pudo llenar número de factura")
return False
if not llenar_formulario_con_validacion(
page, "#importe", str(factura["Importe"]), "Importe"
):
heptora.log.result("error", "No se pudo llenar importe")
return False
# Enviar con verificación
if not click_con_verificacion(
page,
"button#enviar",
"Enviar Factura",
".mensaje-confirmacion"
):
heptora.log.result("error", "No se pudo enviar la factura")
return False
heptora.log.result("ok", f"Factura {factura['Numero']} enviada correctamente")
return True

Antes de desplegar un proceso, verifica:

  • ✅ Todos los datos de entrada se validan al inicio
  • ✅ Las validaciones usan force_ko() apropiadamente
  • ✅ Los errores técnicos usan result("error", ...)
  • ✅ Hay try-except granulares, no un solo try-except global
  • ✅ Los errores transitorios tienen reintentos implementados
  • ✅ Los logs incluyen contexto suficiente para debugging
  • ✅ No se loggean datos sensibles
  • ✅ Las funcionalidades secundarias no rompen el proceso principal
  • ✅ Hay notificaciones configuradas apropiadamente
  • ✅ Los tiempos de timeout son razonables
  • ✅ El proceso maneja elegantemente recursos no disponibles
  • ✅ Hay logs de inicio, progreso y fin
  • ✅ Los mensajes de error son claros y accionables

Procesa datos a través de etapas secuenciales de transformación.

Estructura:

Entrada → Validar → Transformar → Enriquecer → Procesar → Salida

Ejemplo: Procesamiento de Pedidos:

def pipeline_pedidos():
"""Pipeline de procesamiento de pedidos."""
# Etapa 1: Extraer
pedidos_raw = extraer_pedidos_del_sistema()
heptora.log.info(f"Extraídos {len(pedidos_raw)} pedidos")
# Etapa 2: Validar
pedidos_validos, pedidos_invalidos = validar_pedidos(pedidos_raw)
heptora.log.info(f"Válidos: {len(pedidos_validos)}, Inválidos: {len(pedidos_invalidos)}")
# Etapa 3: Normalizar
pedidos_normalizados = normalizar_formato(pedidos_validos)
# Etapa 4: Enriquecer
pedidos_enriquecidos = enriquecer_con_datos_cliente(pedidos_normalizados)
# Etapa 5: Transformar
pedidos_transformados = aplicar_reglas_negocio(pedidos_enriquecidos)
# Etapa 6: Cargar
resultados = cargar_en_sistema_destino(pedidos_transformados)
# Etapa 7: Reportar
generar_reporte_procesamiento(resultados, pedidos_invalidos)

Distribuye trabajo según tipo o categoría.

Estructura:

def dispatcher_documentos(documento):
"""Despacha documentos según su tipo."""
tipo = clasificar_documento(documento)
if tipo == "factura":
return procesar_factura(documento)
elif tipo == "pedido":
return procesar_pedido(documento)
elif tipo == "albaran":
return procesar_albaran(documento)
else:
heptora.log.result("ko", f"Tipo de documento no soportado: {tipo}")
return None
# Procesar todos los documentos
for documento in documentos:
dispatcher_documentos(documento)

Procesa registros maestros y sus detalles relacionados.

Ejemplo: Facturas con Líneas:

def procesar_factura_completa(factura):
"""Procesa factura maestra y todas sus líneas de detalle."""
# Procesar cabecera de factura
try:
crear_factura_en_sistema(factura)
heptora.log.info(f"Factura {factura['Numero']} creada")
except Exception as e:
heptora.log.result("error", f"Error al crear factura: {str(e)}")
return # Si falla la cabecera, no procesar líneas
# Procesar líneas de detalle
lineas = factura.get("Lineas", [])
lineas_ok = 0
lineas_error = 0
for i, linea in enumerate(lineas, 1):
try:
agregar_linea_factura(factura["Numero"], linea)
lineas_ok += 1
except Exception as e:
heptora.log.info(f"Error en línea {i}: {str(e)}")
lineas_error += 1
# Resultado final
if lineas_error == 0:
heptora.log.result(
"ok",
f"Factura {factura['Numero']}: {lineas_ok} líneas procesadas"
)
elif lineas_ok > 0:
heptora.log.result(
"ko",
f"Factura {factura['Numero']}: {lineas_ok} OK, {lineas_error} errores"
)
else:
heptora.log.result(
"error",
f"Factura {factura['Numero']}: Todas las líneas fallaron"
)

Coordina operaciones en múltiples sistemas con capacidad de rollback.

Ejemplo: Creación de Cuenta Cliente:

def crear_cuenta_cliente_saga(cliente):
"""
Crea cuenta de cliente en múltiples sistemas con rollback.
Pasos:
1. Crear en CRM
2. Crear en ERP
3. Crear en sistema de facturación
4. Enviar email de bienvenida
Si cualquier paso falla, revertir los anteriores.
"""
rollback_actions = []
try:
# Paso 1: CRM
heptora.log.info("Creando cliente en CRM...")
id_crm = crear_en_crm(cliente)
rollback_actions.append(lambda: eliminar_de_crm(id_crm))
heptora.log.info(f"Cliente creado en CRM con ID: {id_crm}")
# Paso 2: ERP
heptora.log.info("Creando cliente en ERP...")
id_erp = crear_en_erp(cliente, id_crm)
rollback_actions.append(lambda: eliminar_de_erp(id_erp))
heptora.log.info(f"Cliente creado en ERP con ID: {id_erp}")
# Paso 3: Sistema de facturación
heptora.log.info("Creando cliente en sistema de facturación...")
id_facturacion = crear_en_facturacion(cliente, id_crm, id_erp)
rollback_actions.append(lambda: eliminar_de_facturacion(id_facturacion))
heptora.log.info(f"Cliente creado en facturación con ID: {id_facturacion}")
# Paso 4: Email (este no requiere rollback)
heptora.log.info("Enviando email de bienvenida...")
enviar_email_bienvenida(cliente["Email"])
heptora.log.result("ok", f"Cliente {cliente['Nombre']} creado en todos los sistemas")
return True
except Exception as e:
# Si algo falla, ejecutar rollback de lo que se hizo
heptora.log.info(f"Error en saga: {str(e)}. Iniciando rollback...")
for i, accion_rollback in enumerate(reversed(rollback_actions), 1):
try:
heptora.log.info(f"Rollback {i}/{len(rollback_actions)}...")
accion_rollback()
except Exception as e_rollback:
heptora.log.info(f"Error en rollback: {str(e_rollback)}")
heptora.log.result("error", f"Cliente {cliente['Nombre']}: Error y rollback completado")
return False

Distribuye trabajo en paralelo y luego consolida resultados.

Ejemplo: Consultar Múltiples Proveedores:

def consultar_precio_mejor_proveedor(producto):
"""
Consulta precio en múltiples proveedores y devuelve el mejor.
Fan-Out: Consulta a varios proveedores simultáneamente
Fan-In: Consolida respuestas y elige mejor precio
"""
proveedores = ["ProveedorA", "ProveedorB", "ProveedorC"]
# Fan-Out: Consultar todos los proveedores
heptora.log.info(f"Consultando precio de {producto} en {len(proveedores)} proveedores...")
resultados = []
for proveedor in proveedores:
try:
precio = consultar_precio_proveedor(proveedor, producto)
resultados.append({
"proveedor": proveedor,
"precio": precio,
"disponible": True
})
heptora.log.info(f"{proveedor}: {precio}€")
except Exception as e:
heptora.log.info(f"{proveedor}: No disponible ({str(e)})")
resultados.append({
"proveedor": proveedor,
"precio": None,
"disponible": False
})
# Fan-In: Consolidar y elegir mejor opción
disponibles = [r for r in resultados if r["disponible"]]
if not disponibles:
heptora.log.result("ko", f"Producto {producto} no disponible en ningún proveedor")
return None
mejor = min(disponibles, key=lambda x: x["precio"])
heptora.log.result(
"ok",
f"Mejor precio para {producto}: {mejor['proveedor']} - {mejor['precio']}€"
)
return mejor

Cada módulo debe tener un propósito único y bien definido.

✅ Alta cohesión:

validaciones_cliente.py
"""Todas las validaciones relacionadas con clientes."""
def validar_email_cliente(email):
"""Valida email de cliente."""
pass
def validar_nif_cliente(nif):
"""Valida NIF de cliente."""
pass
def validar_telefono_cliente(telefono):
"""Valida teléfono de cliente."""
pass

❌ Baja cohesión:

utilidades_varias.py
"""Funciones varias mezcladas."""
def validar_email(email):
pass
def generar_pdf(datos):
pass
def enviar_email(destinatario):
pass
def calcular_descuento(precio):
pass

Los módulos deben ser lo más independientes posible.

✅ Bajo acoplamiento:

# Módulo A: No depende de B
def procesar_datos(datos):
resultado = transformar(datos)
return resultado
# Módulo B: No depende de A
def guardar_resultado(resultado):
escribir_en_db(resultado)
# Orquestador: Usa ambos pero ellos no se conocen
def flujo_completo(datos):
resultado = procesar_datos(datos) # Módulo A
guardar_resultado(resultado) # Módulo B

❌ Alto acoplamiento:

# Módulo A: Depende directamente de B
def procesar_datos(datos):
resultado = transformar(datos)
modulo_b.guardar_en_formato_especifico(resultado) # Acoplado!
return resultado

Organización recomendada:

proyecto_rpa/
├── README.md # Documentación principal
├── requirements.txt # Dependencias Python
├── config/
│ ├── configuracion.json # Configuración del proceso
│ └── mappings.json # Mapeos de datos
├── src/
│ ├── bloques/ # Bloques reutilizables
│ │ ├── __init__.py
│ │ ├── validaciones.py
│ │ ├── login.py
│ │ └── reportes.py
│ ├── utils/ # Utilidades generales
│ │ ├── __init__.py
│ │ ├── archivos.py
│ │ ├── fechas.py
│ │ └── formateo.py
│ ├── procesos/ # Procesos principales
│ │ ├── __init__.py
│ │ ├── proceso_facturas.py
│ │ └── proceso_pedidos.py
│ └── main.py # Punto de entrada
├── tests/ # Tests
│ ├── test_validaciones.py
│ └── test_procesos.py
├── data/ # Datos de entrada
│ ├── input/
│ ├── processed/
│ └── errors/
└── logs/ # Logs del proceso

Cada archivo debe incluir:

"""
Módulo: Validaciones de Cliente
================================
Descripción:
Contiene todas las funciones de validación relacionadas con
datos de clientes, incluyendo NIF, email, teléfono y dirección.
Dependencias:
- re: Para expresiones regulares
- utils.formateo: Para normalización de datos
Autor: Equipo RPA
Fecha: 2024-01-15
Versión: 1.2
Ejemplo de uso:
>>> from bloques.validaciones import validar_email_cliente
>>> validar_email_cliente("juan@empresa.com")
True
>>> validar_email_cliente("correo_invalido")
False
"""
import re
from utils.formateo import normalizar_texto
def validar_email_cliente(email):
"""
Valida que un email tenga formato correcto.
Args:
email (str): Dirección de email a validar
Returns:
bool: True si el email es válido, False en caso contrario
Examples:
>>> validar_email_cliente("usuario@dominio.com")
True
>>> validar_email_cliente("correo_sin_arroba.com")
False
Notes:
- El email debe contener @ y un dominio válido
- No se permiten espacios
- Se aceptan caracteres especiales estándar en emails
"""
if not email or not isinstance(email, str):
return False
patron = r'^[\w\.-]+@[\w\.-]+\.\w+$'
return re.match(patron, email.strip()) is not None

Convenciones de commits:

feat: Nueva funcionalidad
ejemplo: feat: Añadir validación de IBAN
fix: Corrección de bug
ejemplo: fix: Corregir formato de fecha en reporte
refactor: Refactorización sin cambio de funcionalidad
ejemplo: refactor: Simplificar lógica de validación NIF
docs: Cambios en documentación
ejemplo: docs: Actualizar README con ejemplos
test: Añadir o modificar tests
ejemplo: test: Añadir tests para validaciones de cliente
perf: Mejora de rendimiento
ejemplo: perf: Optimizar consulta a base de datos
chore: Tareas de mantenimiento
ejemplo: chore: Actualizar dependencias

Ejemplo de flujo Git:

Ventana de terminal
# Crear rama para nueva funcionalidad
git checkout -b feat/validacion-iban
# Hacer cambios y commit
git add src/bloques/validaciones.py
git commit -m "feat: Añadir validación de IBAN español"
# Añadir tests
git add tests/test_validaciones.py
git commit -m "test: Añadir tests para validación IBAN"
# Actualizar documentación
git add README.md
git commit -m "docs: Documentar función validar_iban"
# Merge a main
git checkout main
git merge feat/validacion-iban

Docstrings en funciones y clases (ya cubierto arriba).

Archivo README específico para cada proceso.

Ejemplo: README-proceso-facturas.md:

# Proceso: Facturación Automática FACe
## Descripción General
Este proceso automatiza el envío de facturas al portal FACe de
administraciones públicas españolas.
## Flujo del Proceso
1. Carga facturas desde Excel
2. Valida formato y contenido
3. Conecta al portal FACe
4. Envía cada factura
5. Descarga acuses de recibo
6. Genera reporte de resultados
## Requisitos Previos
- Robot Heptora instalado y configurado
- Certificado digital válido
- Usuario registrado en FACe
- Facturas en formato Facturae 3.2
## Configuración
### Secretos Necesarios
- `face_usuario`: Usuario del portal FACe
- `face_password`: Contraseña del portal
- `certificado_path`: Ruta al certificado digital
### Archivos de Entrada
- **Ubicación**: `data/input/facturas.xlsx`
- **Formato**: Excel con las siguientes columnas:
- NumeroFactura (texto)
- CIF (texto, 9 caracteres)
- Importe (número decimal)
- Fecha (formato YYYY-MM-DD)
### Archivos de Salida
- **Facturas procesadas**: `data/processed/facturas_YYYYMMDD_HHMMSS.xlsx`
- **Reporte de errores**: `data/errors/errores_YYYYMMDD_HHMMSS.xlsx`
- **Acuses de recibo**: `data/output/acuses/`
## Ejecución
### Manual
```bash
python src/procesos/proceso_facturas.py

Configurar en Heptora para ejecutar:

  • Lunes a Viernes
  • 09:00, 14:00 y 18:00
  • Solo si hay archivos en carpeta de entrada
  • OK: Factura enviada correctamente y acuse descargado
  • KO: Factura con datos incorrectos o ya enviada
  • ERROR: Error técnico (conectividad, portal no disponible, etc.)
  • Email resumen a: finanzas@empresa.com
  • Alerta crítica si > 20% errores
  • Alerta advertencia si > 30% KOs
  • Tiempo promedio por factura: 45 segundos
  • Tasa de éxito esperada: > 95%
  • SLA: Todas las facturas procesadas en < 2 horas

Causa: Certificado digital expirado o no instalado Solución: Renovar certificado e instalarlo en almacén de Windows

Error: “Usuario o contraseña incorrectos”

Sección titulada «Error: “Usuario o contraseña incorrectos”»

Causa: Credenciales incorrectas en secretos Solución: Verificar y actualizar credenciales en robot local

Causa: La factura ya existe en FACe Solución: Verificar manualmente, puede ser correcto

#### 3. Documentación de Arquitectura
Documento de alto nivel explicando la estructura general.
**Ejemplo: ARCHITECTURE.md**:
```markdown
# Arquitectura del Sistema RPA - Empresa XYZ
## Visión General
Sistema de automatización RPA basado en Heptora para procesos
de back-office: facturación, pedidos e inventario.
## Componentes Principales
### 1. Capa de Datos
- **Excel Files**: Fuente principal de datos
- **API REST**: Integración con ERP
- **Base de Datos Local**: Cache temporal
### 2. Capa de Procesamiento
- **Bloques de Validación**: Verifican datos de entrada
- **Bloques de Transformación**: Normalizan y enriquecen datos
- **Bloques de Integración**: Conectan con sistemas externos
### 3. Capa de Presentación
- **Reportes Excel**: Resultados de procesamiento
- **Emails**: Notificaciones automáticas
- **Dashboard Heptora**: Monitoreo en tiempo real
## Flujo de Datos

[Excel] → [Validar] → [Transformar] → [Procesar] → [Portal Web] ↓ [Reportes] ← [Consolidar] ← [Registrar] ← [Confirmar] ←

## Patrones Utilizados
- **Pipeline**: Para procesamiento secuencial de datos
- **Dispatcher**: Para routing según tipo de documento
- **Circuit Breaker**: Para protección contra fallos en sistemas externos
## Decisiones de Diseño
### ¿Por qué validación temprana?
Para fallar rápido y evitar procesamiento costoso de datos inválidos.
### ¿Por qué bloques pequeños?
Para reutilización, testing y mantenimiento más fácil.
### ¿Por qué almacenamiento local de secretos?
Para máxima seguridad y cumplimiento con GDPR.

Incluir diagramas visuales:

  • Diagrama de flujo: Representa el proceso paso a paso
  • Diagrama de componentes: Muestra módulos y sus relaciones
  • Diagrama de datos: Muestra flujo de datos entre componentes
  • Diagrama de despliegue: Muestra dónde se ejecuta cada componente

Herramientas recomendadas:

  • Mermaid (integrado en Markdown)
  • Draw.io / Lucidchart
  • PlantUML

Ejemplo con Mermaid:

## Diagrama de Flujo
```mermaid
flowchart TD
A[Inicio] --> B[Cargar Facturas]
B --> C{¿Facturas válidas?}
C -->|No| D[Registrar KO]
C -->|Sí| E[Login Portal]
E --> F{¿Login OK?}
F -->|No| G[Registrar ERROR]
F -->|Sí| H[Enviar Factura]
H --> I{¿Envío OK?}
I -->|No| J[Registrar ERROR]
I -->|Sí| K[Descargar Acuse]
K --> L[Registrar OK]
D --> M[Generar Reporte]
G --> M
J --> M
L --> M
M --> N[Fin]
---
## Estrategias de Testing
### Tipos de Tests
#### 1. Tests Unitarios
Prueban funciones individuales de forma aislada.
```python
# tests/test_validaciones.py
"""Tests unitarios para validaciones."""
import pytest
from src.bloques.validaciones import validar_nif, validar_email, validar_importe
class TestValidarNIF:
"""Tests para validación de NIF."""
def test_nif_valido(self):
"""Test con NIF válido."""
assert validar_nif("12345678Z") == (True, "NIF válido")
def test_nif_letra_incorrecta(self):
"""Test con letra incorrecta."""
valido, mensaje = validar_nif("12345678X")
assert not valido
assert "letra incorrecta" in mensaje.lower()
def test_nif_formato_incorrecto(self):
"""Test con formato incorrecto."""
valido, mensaje = validar_nif("1234567")
assert not valido
assert "formato" in mensaje.lower()
def test_nif_vacio(self):
"""Test con NIF vacío."""
valido, mensaje = validar_nif("")
assert not valido
def test_nif_none(self):
"""Test con NIF None."""
valido, mensaje = validar_nif(None)
assert not valido
def test_nif_con_espacios(self):
"""Test con espacios al inicio/fin."""
assert validar_nif(" 12345678Z ") == (True, "NIF válido")
class TestValidarEmail:
"""Tests para validación de email."""
def test_email_valido(self):
"""Test con email válido."""
assert validar_email("usuario@dominio.com") == True
def test_email_sin_arroba(self):
"""Test sin @."""
assert validar_email("usuariodominio.com") == False
def test_email_sin_dominio(self):
"""Test sin dominio."""
assert validar_email("usuario@") == False
def test_email_vacio(self):
"""Test con email vacío."""
assert validar_email("") == False
class TestValidarImporte:
"""Tests para validación de importe."""
def test_importe_valido(self):
"""Test con importe válido."""
valido, importe = validar_importe("123.45")
assert valido == True
assert importe == 123.45
def test_importe_con_coma(self):
"""Test con coma como separador decimal."""
valido, importe = validar_importe("123,45")
assert valido == True
assert importe == 123.45
def test_importe_negativo(self):
"""Test con importe negativo."""
valido, importe = validar_importe("-100")
assert valido == False
def test_importe_cero(self):
"""Test con importe cero."""
valido, importe = validar_importe("0")
assert valido == False
def test_importe_no_numerico(self):
"""Test con texto no numérico."""
valido, importe = validar_importe("abc")
assert valido == False
# Ejecutar tests
# pytest tests/test_validaciones.py -v

Prueban la interacción entre componentes.

tests/test_integracion.py
"""Tests de integración."""
import pytest
import os
from src.procesos.proceso_facturas import cargar_facturas, validar_facturas, procesar_facturas
class TestIntegracionFacturas:
"""Tests de integración del proceso de facturas."""
@pytest.fixture
def archivo_test(self):
"""Crea archivo Excel de prueba."""
# Crear archivo temporal con datos de test
# ...
yield "tests/data/facturas_test.xlsx"
# Limpiar después del test
def test_flujo_completo_facturas_validas(self, archivo_test):
"""Test del flujo completo con facturas válidas."""
# Cargar
facturas = cargar_facturas(archivo_test)
assert len(facturas) > 0
# Validar
validas, invalidas = validar_facturas(facturas)
assert len(validas) == len(facturas)
assert len(invalidas) == 0
# Procesar (en modo test, sin conexión real)
resultados = procesar_facturas(validas, modo_test=True)
assert all(r["resultado"] == "ok" for r in resultados)
def test_flujo_con_facturas_invalidas(self, archivo_test):
"""Test del flujo con facturas inválidas mezcladas."""
# Similar al anterior pero con datos inválidos
pass

Prueban el proceso completo en entorno real.

tests/test_e2e.py
"""Tests end-to-end."""
import pytest
from src.main import ejecutar_proceso
@pytest.mark.e2e
@pytest.mark.slow
class TestE2EFacturas:
"""Tests end-to-end del proceso de facturas."""
def test_proceso_completo_produccion(self):
"""
Test del proceso completo en entorno de prueba.
Nota: Este test requiere:
- Acceso al portal de prueba
- Credenciales configuradas
- Datos de prueba en la ubicación esperada
"""
resultado = ejecutar_proceso(
archivo="tests/data/facturas_e2e.xlsx",
entorno="test"
)
assert resultado["exitosas"] > 0
assert resultado["errores"] == 0

Pirámide de tests:

/\
/ \
/ E2E \ (Pocos tests, lentos, completos)
/--------\
/ \
/ Integración \ (Algunos tests, medios)
/--------------\
/ \
/ Unitarios \ (Muchos tests, rápidos, específicos)
/____________________\

Cobertura recomendada:

  • Tests unitarios: 80%+ del código
  • Tests de integración: Flujos principales
  • Tests E2E: Casos críticos de negocio

Crear datasets de prueba representativos:

tests/fixtures/datos_test.py
"""Datos de prueba para tests."""
FACTURAS_VALIDAS = [
{
"Numero": "FAC-2024-001",
"CIF": "A12345678",
"Importe": "1234.56",
"Fecha": "2024-01-15"
},
{
"Numero": "FAC-2024-002",
"CIF": "B87654321",
"Importe": "9876.54",
"Fecha": "2024-01-16"
}
]
FACTURAS_INVALIDAS = [
{
"Numero": "", # Número vacío
"CIF": "A12345678",
"Importe": "1000",
"Fecha": "2024-01-15"
},
{
"Numero": "FAC-2024-003",
"CIF": "INVALID", # CIF inválido
"Importe": "1000",
"Fecha": "2024-01-15"
},
{
"Numero": "FAC-2024-004",
"CIF": "A12345678",
"Importe": "-100", # Importe negativo
"Fecha": "2024-01-15"
}
]
CASOS_BORDE = [
{
"Numero": "A" * 100, # Número muy largo
"CIF": "A12345678",
"Importe": "0.01", # Importe mínimo
"Fecha": "2024-01-15"
},
{
"Numero": "FAC-2024-005",
"CIF": "A12345678",
"Importe": "999999.99", # Importe máximo
"Fecha": "2024-01-15"
}
]

Medir tiempos de ejecución:

import time
from functools import wraps
def medir_tiempo(func):
"""Decorator para medir tiempo de ejecución."""
@wraps(func)
def wrapper(*args, **kwargs):
inicio = time.time()
resultado = func(*args, **kwargs)
fin = time.time()
tiempo_ms = (fin - inicio) * 1000
heptora.log.info(f"{func.__name__} ejecutado en {tiempo_ms:.2f}ms")
return resultado
return wrapper
# Usar en funciones
@medir_tiempo
def cargar_facturas():
"""Carga facturas desde Excel."""
# ... código ...
@medir_tiempo
def validar_facturas(facturas):
"""Valida facturas."""
# ... código ...
@medir_tiempo
def procesar_facturas(facturas):
"""Procesa facturas."""
# ... código ...
# Resultado en logs:
# cargar_facturas ejecutado en 234.56ms
# validar_facturas ejecutado en 89.12ms
# procesar_facturas ejecutado en 45678.90ms <- Cuello de botella!

Agrupa operaciones similares.

# ❌ Ineficiente: Una consulta por registro
for factura in facturas:
cliente = buscar_cliente_en_db(factura["CIF"])
# ...
# ✅ Eficiente: Una consulta para todos
cifs = [f["CIF"] for f in facturas]
clientes_dict = buscar_clientes_en_db_lote(cifs) # Una sola consulta
for factura in facturas:
cliente = clientes_dict.get(factura["CIF"])
# ...

Evita consultas repetidas.

# Cache simple con diccionario
cache_clientes = {}
def obtener_cliente_con_cache(cif):
"""Obtiene cliente desde cache o DB."""
if cif not in cache_clientes:
cache_clientes[cif] = buscar_cliente_en_db(cif)
return cache_clientes[cif]
# Usar en el proceso
for factura in facturas:
cliente = obtener_cliente_con_cache(factura["CIF"])
# ...

Carga datos solo cuando sean necesarios.

# ❌ Carga todo al inicio (puede ser innecesario)
todos_los_productos = cargar_todos_los_productos() # 10,000 productos
for pedido in pedidos:
if pedido["estado"] == "pendiente":
producto = todos_los_productos[pedido["producto_id"]]
# ...
# ✅ Carga solo lo necesario
for pedido in pedidos:
if pedido["estado"] == "pendiente":
producto = cargar_producto(pedido["producto_id"]) # Solo si es necesario
# ...
# ❌ Leer archivo múltiples veces
for factura in facturas:
plantilla = leer_plantilla_pdf() # Lee archivo cada vez
pdf = generar_pdf(factura, plantilla)
# ...
# ✅ Leer archivo una vez
plantilla = leer_plantilla_pdf() # Una sola vez
for factura in facturas:
pdf = generar_pdf(factura, plantilla)
# ...

Definir KPIs de rendimiento:

  • Tiempo por registro: Cuánto tarda procesar un registro
  • Throughput: Registros procesados por hora
  • Tasa de utilización: % del tiempo ejecutando vs esperando
  • Tiempo total: Duración completa del proceso

Ejemplo de reporte de rendimiento:

def generar_reporte_rendimiento(estadisticas):
"""Genera reporte de métricas de rendimiento."""
total_registros = estadisticas["total"]
tiempo_total = estadisticas["tiempo_total_segundos"]
tiempo_procesamiento = estadisticas["tiempo_procesamiento_segundos"]
tiempo_espera = estadisticas["tiempo_espera_segundos"]
tiempo_por_registro = tiempo_total / total_registros
throughput = (total_registros / tiempo_total) * 3600 # registros/hora
utilizacion = (tiempo_procesamiento / tiempo_total) * 100
reporte = f"""
=== MÉTRICAS DE RENDIMIENTO ===
Total registros: {total_registros}
Tiempo total: {tiempo_total:.2f}s ({tiempo_total/60:.2f}min)
Tiempo por registro: {tiempo_por_registro:.2f}s
Throughput: {throughput:.0f} registros/hora
Tiempo procesando: {tiempo_procesamiento:.2f}s ({utilizacion:.1f}%)
Tiempo esperando: {tiempo_espera:.2f}s ({100-utilizacion:.1f}%)
Tiempos por operación:
- Cargar datos: {estadisticas["t_carga"]:.2f}s
- Validar: {estadisticas["t_validacion"]:.2f}s
- Procesar: {estadisticas["t_procesamiento"]:.2f}s
- Generar reporte: {estadisticas["t_reporte"]:.2f}s
"""
heptora.log.info(reporte)
return reporte

❌ Nunca hacer esto:

# ¡NUNCA!
usuario = "admin@empresa.com"
password = "MiPassword123"
api_key = "sk-1234567890abcdef"

✅ Siempre hacer esto:

# Usar sistema de secretos de Heptora
usuario = heptora.data.get("portal_usuario")
password = heptora.data.get("portal_password")
api_key = heptora.data.get("api_key")
# Verificar que existen
if not usuario or not password:
heptora.log.force_ko("Faltan credenciales. Configúralas en Secretos.")

❌ Evitar loggear datos sensibles completos:

heptora.log.info(f"Procesando cliente: {cliente['Nombre']} - NIF: {cliente['NIF']}")
heptora.log.info(f"Tarjeta de crédito: {cliente['NumeroTarjeta']}")

✅ Loggear datos ofuscados:

# Ofuscar NIF (mostrar solo primeros y últimos caracteres)
nif_ofuscado = cliente['NIF'][:2] + "***" + cliente['NIF'][-1:]
heptora.log.info(f"Procesando cliente: {cliente['Nombre']} - NIF: {nif_ofuscado}")
# No loggear tarjetas de crédito completas
heptora.log.info("Tarjeta de crédito validada correctamente")
def sanitizar_entrada(texto):
"""Sanitiza entrada del usuario eliminando caracteres peligrosos."""
import re
# Eliminar caracteres especiales peligrosos
texto_limpio = re.sub(r'[<>"\';]', '', texto)
# Limitar longitud
texto_limpio = texto_limpio[:255]
return texto_limpio.strip()
# Usar en procesamiento
nombre_usuario = sanitizar_entrada(datos_entrada["nombre"])
  • Usa cuentas con permisos mínimos necesarios
  • No uses cuentas de administrador para procesos RPA
  • Crea usuarios específicos para automatización

Registra todas las operaciones críticas:

def registrar_auditoria(operacion, usuario, datos_relevantes):
"""Registra operación para auditoría."""
from datetime import datetime
registro = {
"timestamp": datetime.now().isoformat(),
"operacion": operacion,
"usuario": usuario,
"datos": datos_relevantes
}
# Guardar en log de auditoría
with open("logs/auditoria.log", "a") as f:
f.write(json.dumps(registro) + "\n")
# Usar en operaciones críticas
registrar_auditoria(
operacion="crear_factura",
usuario=heptora.data.get("usuario_actual"),
datos_relevantes={
"numero_factura": factura["Numero"],
"importe": factura["Importe"]
}
)

1. Gestión de Secretos Local:

  • Los secretos se almacenan solo en tu equipo
  • Nunca se transmiten a la nube
  • Encriptación de nivel empresarial

2. Ejecución Local:

  • El proceso se ejecuta en tu infraestructura
  • Control total sobre datos sensibles
  • Cumplimiento con regulaciones de protección de datos

3. Auditoría Automática:

  • Registro completo de ejecuciones
  • Trazabilidad de todas las acciones
  • Logs detallados para investigación

Desarrollador RPA:

  • Crea y mantiene procesos
  • Implementa nuevas funcionalidades
  • Optimiza rendimiento

Analista de Negocio:

  • Define requisitos
  • Valida resultados
  • Propone mejoras

Administrador de Sistemas:

  • Gestiona infraestructura
  • Configura permisos
  • Monitorea salud del sistema

Usuario Final:

  • Usa los procesos
  • Reporta problemas
  • Proporciona feedback

1. Planificación:

Usuario Final → Identifica proceso a automatizar
Analista de Negocio → Define requisitos y casos de uso
Desarrollador RPA → Evalúa viabilidad técnica
Equipo → Aprueba y prioriza

2. Desarrollo:

Desarrollador → Desarrolla en rama de feature
Desarrollador → Tests unitarios y de integración
Analista → Revisa y valida contra requisitos
Desarrollador → Ajusta según feedback

3. Despliegue:

Desarrollador → Merge a rama main
Admin Sistemas → Despliega en entorno de prueba
Usuario Final → Pruebas de aceptación
Admin Sistemas → Despliega en producción

4. Operación:

Robot → Ejecuta proceso automáticamente
Usuario Final → Revisa resultados
Analista → Monitorea métricas
Desarrollador → Ajusta según necesidad

Definir y documentar:

  • Convenciones de nomenclatura
  • Estructura de proyecto
  • Estilo de código
  • Proceso de revisión de código
  • Criterios de aceptación

Ejemplo: STANDARDS.md:

# Estándares de Desarrollo RPA - Empresa XYZ
## Nomenclatura
### Archivos
- Procesos: `proceso_[nombre].py`
- Bloques: `bloque_[funcion].py`
- Utils: `util_[categoria].py`
- Tests: `test_[modulo].py`
### Variables
- Snake case: `nombre_variable`
- Constantes: `NOMBRE_CONSTANTE`
- Privadas: `_nombre_privado`
### Funciones
- Snake case: `nombre_funcion()`
- Verbos descriptivos: `validar_`, `procesar_`, `generar_`
- Docstrings obligatorios
## Estructura de Proyecto
Seguir estructura definida en docs/architecture.md
## Revisión de Código
- Toda rama de feature requiere pull request
- Mínimo 1 aprobación antes de merge
- Tests deben pasar antes de merge
- Cobertura de tests > 80%
## Commits
- Formato: `tipo: descripción`
- Tipos: feat, fix, docs, refactor, test, perf, chore
- Descripción en español, imperativo
- Ejemplo: `feat: Añadir validación de IBAN`

Checklist para revisor:

  • ¿El código sigue los estándares del equipo?
  • ¿Las funciones tienen docstrings?
  • ¿Hay tests para la nueva funcionalidad?
  • ¿El código es fácil de entender?
  • ¿Hay manejo apropiado de errores?
  • ¿Se evita código duplicado?
  • ¿Las validaciones son robustas?
  • ¿Los logs son apropiados (ni demasiados ni muy pocos)?
  • ¿No hay credenciales hardcodeadas?
  • ¿La documentación está actualizada?

  • El proceso tiene un objetivo claro y bien definido
  • El 80% de casos comunes están cubiertos
  • Los casos excepcionales se manejan apropiadamente
  • El flujo es fácil de entender visualmente
  • Cada bloque tiene una responsabilidad única
  • Los bloques son fáciles de explicar en una frase
  • No hay bloques de más de 100 líneas
  • La lógica compleja está separada en funciones auxiliares
  • Código común está en componentes reutilizables
  • Los componentes tienen documentación clara
  • Existe un catálogo de componentes disponibles
  • Se usan plantillas de Heptora cuando es posible
  • Todos los datos de entrada se validan
  • Se distingue claramente entre KO y ERROR
  • Hay reintentos para errores transitorios
  • Los logs proporcionan contexto suficiente
  • Las notificaciones son apropiadas
  • El código está organizado en módulos cohesivos
  • Los módulos tienen bajo acoplamiento
  • Existe una estructura de carpetas clara
  • El código es fácil de navegar
  • Existe README del proceso
  • Las funciones tienen docstrings
  • Hay ejemplos de uso
  • La documentación está actualizada
  • Hay tests unitarios para funciones críticas
  • Hay tests de integración para flujos principales
  • Se han probado casos extremos
  • Hay datos de prueba representativos
  • No hay credenciales hardcodeadas
  • Se usan secretos de Heptora
  • Los logs no exponen datos sensibles
  • Se validan y sanitizan entradas
  • Se han identificado cuellos de botella
  • Hay métricas de rendimiento
  • Las operaciones costosas están optimizadas
  • Se usa cache cuando es apropiado
  • El código sigue estándares del equipo
  • Hay revisión de código antes de merge
  • Los commits son descriptivos
  • La rama está actualizada con main

Si esta guía no resolvió tu problema o encontraste algún error en la documentación:

  • Soporte técnico: help@heptora.com
  • Describe claramente el problema que encontraste
  • Incluye capturas de pantalla si es posible
  • Indica qué pasos de la documentación seguiste

Nuestro equipo de soporte te ayudará a resolver cualquier problema.