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.
Principios de Diseño de Procesos RPA
Sección titulada «Principios de Diseño de Procesos RPA»1. Claridad Sobre Complejidad
Sección titulada «1. Claridad Sobre Complejidad»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:
- Identifica el flujo principal (happy path)
- Automatiza ese flujo primero
- Mide resultados y detecta casos especiales
- Añade manejo de excepciones progresivamente
- 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 justifica3. Falla Rápido y Falla Claro
Sección titulada «3. Falla Rápido y Falla Claro»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
Mantener Bloques Simples
Sección titulada «Mantener Bloques Simples»Por Qué la Simplicidad es Crítica
Sección titulada «Por Qué la Simplicidad es Crítica»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
Principio de Responsabilidad Única
Sección titulada «Principio de Responsabilidad Única»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 FalseBloque 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 FalseBloque 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 FalseProceso 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)
# Ejecutarmain()Ventajas de los Bloques Simples
Sección titulada «Ventajas de los Bloques Simples»| Aspecto | Bloque Complejo | Bloques Simples |
|---|---|---|
| Comprensión | Difícil, requiere leer todo | Fácil, cada bloque es autoexplicativo |
| Pruebas | Difícil probar componentes individuales | Fácil probar cada bloque por separado |
| Depuración | Difícil localizar el problema | Fácil identificar qué bloque falla |
| Reutilización | Imposible reutilizar partes | Bloques se pueden usar en otros procesos |
| Mantenimiento | Cambios arriesgados | Cambios localizados y seguros |
| Colaboración | Difícil trabajar en paralelo | Varios desarrolladores pueden trabajar simultáneamente |
Cómo Heptora Facilita Bloques Simples
Sección titulada «Cómo Heptora Facilita Bloques Simples»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
Reutilización de Componentes
Sección titulada «Reutilización de Componentes»Por Qué Reutilizar es Fundamental
Sección titulada «Por Qué Reutilizar es Fundamental»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
Tipos de Componentes Reutilizables
Sección titulada «Tipos de Componentes Reutilizables»1. Bloques de Automatización
Sección titulada «1. Bloques de Automatización»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 reutilizablenif_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 internologin_web( "https://sistema.empresa.com/login", "#username", "#password", "button[type='submit']", ".dashboard")
# Portal 2: Proveedor externologin_web( "https://proveedor.com/acceso", "input[name='user']", "input[name='pass']", "#btnLogin", ".home-page")2. Plantillas de Procesos
Sección titulada «2. Plantillas de Procesos»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 carpeta2. [Bloque] Validar formato y contenido3. [Bloque] Clasificar documentos por tipo4. [Loop] Para cada documento válido: 4.1. [Bloque] Extraer información clave 4.2. [Bloque] Procesar según tipo 4.3. [Bloque] Guardar resultado5. [Bloque] Generar reporte de procesamiento6. [Bloque] Mover archivos a carpetas finales7. [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 notificaciones3. Funciones Auxiliares
Sección titulada «3. Funciones Auxiliares»Funciones de utilidad que se usan frecuentemente en muchos procesos.
Biblioteca de Funciones Reutilizables:
"""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"""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())"""Funciones de gestión de archivos reutilizables."""
import osfrom 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 archivosUsar las funciones reutilizables:
# En tu procesofrom utils.validaciones import validar_email, validar_nif, validar_importefrom utils.formateo import formatear_fecha, formatear_importefrom utils.archivos import crear_nombre_archivo_con_fecha
# Proceso principalclientes = 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 úniconombre_reporte = crear_nombre_archivo_con_fecha("reporte_clientes", "xlsx")heptora.excel.write(resultados, nombre_reporte, "Resultados")Organización de Componentes Reutilizables
Sección titulada «Organización de Componentes Reutilizables»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.pyDocumentación de Componentes Reutilizables
Sección titulada «Documentación de Componentes Reutilizables»Cada componente reutilizable debe incluir:
- Descripción clara: Qué hace el componente
- Parámetros: Qué entradas necesita
- Retorno: Qué devuelve
- Ejemplos de uso: Cómo usarlo en diferentes contextos
- Dependencias: Qué otros componentes o librerías necesita
- Autor y fecha: Quién lo creó y cuándo
- 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 RPAFecha: 2024-01-15Versió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, ""Versionado de Componentes
Sección titulada «Versionado de Componentes»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:
# 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"""Catálogo de Componentes
Sección titulada «Catálogo de Componentes»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) → boolValida 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) → boolValida 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) → boolRealiza 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() → boolRealiza 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) → strGenera reporte Excel usando plantilla predefinida.- **Ubicación**: `bloques/reportes/generar_reporte_excel.py`- **Versión**: 2.0- **Usado en**: 18 procesos- **Responsable**: Ana LópezGestión de Excepciones
Sección titulada «Gestión de Excepciones»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.
Principios de Gestión de Errores
Sección titulada «Principios de Gestión de Errores»1. Prevención > Detección > Corrección
Sección titulada «1. Prevención > Detección > Corrección»El mejor error es el que nunca ocurre.
Jerarquía de estrategias:
- Prevenir: Validar datos de entrada antes de procesarlos
- Detectar: Identificar problemas rápidamente cuando ocurren
- Corregir: Recuperarse automáticamente cuando sea posible
- Reportar: Notificar problemas que requieren intervención humana
2. Falla Rápido, Recupera Inteligentemente
Sección titulada «2. Falla Rápido, Recupera Inteligentemente»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»Sistema de Resultados OK/KO/ERROR
Sección titulada «Sistema de Resultados OK/KO/ERROR»Heptora proporciona un sistema claro de resultados que facilita la gestión de excepciones:
# Resultado exitosoheptora.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
Force KO para Errores Controlados
Sección titulada «Force KO para Errores Controlados»heptora.log.force_ko() detiene el procesamiento del registro actual y marca como KO:
# Validación al inicioif 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áticamenteCuá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
Estrategias de Manejo de Errores
Sección titulada «Estrategias de Manejo de Errores»1. Validación Temprana
Sección titulada «1. Validación Temprana»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)}")2. Try-Except Granular
Sección titulada «2. Try-Except Granular»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 datostry: facturas = heptora.excel.load("facturas.xlsx", "Hoja1", columnas, True)except FileNotFoundError: heptora.log.force_ko("No se encontró el archivo facturas.xlsx") returnexcept Exception as e: heptora.log.force_ko(f"Error al leer Excel: {str(e)}") return
# Procesar cada facturafor 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 finaltry: 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 esto3. Reintentos Inteligentes
Sección titulada «3. Reintentos Inteligentes»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óndef 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áticosresultado = 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
4. Degradación Elegante
Sección titulada «4. Degradación Elegante»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 esto5. Circuit Breaker
Sección titulada «5. Circuit Breaker»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 Breakercb_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)}")Estrategias de Logging
Sección titulada «Estrategias de Logging»El logging efectivo es crucial para debugging y auditoría.
Niveles de Logging
Sección titulada «Niveles de Logging»# Información general del flujoheptora.log.info("Iniciando procesamiento de facturas")heptora.log.info(f"Cargadas {len(facturas)} facturas")
# Detalles de procesamientoheptora.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 procesamientoheptora.log.result("ok", "Factura procesada correctamente")heptora.log.result("ko", "Factura con datos incorrectos")heptora.log.result("error", "Error técnico al procesar")Qué Loggear
Sección titulada «Qué Loggear»✅ 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")Notificaciones Inteligentes
Sección titulada «Notificaciones Inteligentes»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" )Manejo de Errores en Navegación Web
Sección titulada «Manejo de Errores en Navegación Web»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 robustasdef 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 TrueChecklist de Gestión de Excepciones
Sección titulada «Checklist de Gestión de Excepciones»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
Patrones de Diseño de Procesos
Sección titulada «Patrones de Diseño de Procesos»Patrón 1: Pipeline de Transformación
Sección titulada «Patrón 1: Pipeline de Transformación»Procesa datos a través de etapas secuenciales de transformación.
Estructura:
Entrada → Validar → Transformar → Enriquecer → Procesar → SalidaEjemplo: 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)Patrón 2: Dispatcher (Despachador)
Sección titulada «Patrón 2: Dispatcher (Despachador)»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 documentosfor documento in documentos: dispatcher_documentos(documento)Patrón 3: Master-Detail
Sección titulada «Patrón 3: Master-Detail»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" )Patrón 4: Saga (Transacción Distribuida)
Sección titulada «Patrón 4: Saga (Transacción Distribuida)»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 FalsePatrón 5: Fan-Out / Fan-In
Sección titulada «Patrón 5: Fan-Out / Fan-In»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 mejorModularidad y Mantenibilidad
Sección titulada «Modularidad y Mantenibilidad»Principios de Modularidad
Sección titulada «Principios de Modularidad»1. Alta Cohesión
Sección titulada «1. Alta Cohesión»Cada módulo debe tener un propósito único y bien definido.
✅ Alta cohesión:
"""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:
"""Funciones varias mezcladas."""
def validar_email(email): pass
def generar_pdf(datos): pass
def enviar_email(destinatario): pass
def calcular_descuento(precio): pass2. Bajo Acoplamiento
Sección titulada «2. Bajo Acoplamiento»Los módulos deben ser lo más independientes posible.
✅ Bajo acoplamiento:
# Módulo A: No depende de Bdef procesar_datos(datos): resultado = transformar(datos) return resultado
# Módulo B: No depende de Adef guardar_resultado(resultado): escribir_en_db(resultado)
# Orquestador: Usa ambos pero ellos no se conocendef flujo_completo(datos): resultado = procesar_datos(datos) # Módulo A guardar_resultado(resultado) # Módulo B❌ Alto acoplamiento:
# Módulo A: Depende directamente de Bdef procesar_datos(datos): resultado = transformar(datos) modulo_b.guardar_en_formato_especifico(resultado) # Acoplado! return resultadoEstructura de Proyecto Mantenible
Sección titulada «Estructura de Proyecto Mantenible»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 procesoDocumentación en Código
Sección titulada «Documentación en Código»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 RPAFecha: 2024-01-15Versió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 refrom 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 NoneControl de Versiones
Sección titulada «Control de Versiones»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 dependenciasEjemplo de flujo Git:
# Crear rama para nueva funcionalidadgit checkout -b feat/validacion-iban
# Hacer cambios y commitgit add src/bloques/validaciones.pygit commit -m "feat: Añadir validación de IBAN español"
# Añadir testsgit add tests/test_validaciones.pygit commit -m "test: Añadir tests para validación IBAN"
# Actualizar documentacióngit add README.mdgit commit -m "docs: Documentar función validar_iban"
# Merge a maingit checkout maingit merge feat/validacion-ibanDocumentación de Procesos
Sección titulada «Documentación de Procesos»Niveles de Documentación
Sección titulada «Niveles de Documentación»1. Documentación de Código
Sección titulada «1. Documentación de Código»Docstrings en funciones y clases (ya cubierto arriba).
2. Documentación de Proceso
Sección titulada «2. Documentación de Proceso»Archivo README específico para cada proceso.
Ejemplo: README-proceso-facturas.md:
# Proceso: Facturación Automática FACe
## Descripción GeneralEste proceso automatiza el envío de facturas al portal FACe deadministraciones públicas españolas.
## Flujo del Proceso1. Carga facturas desde Excel2. Valida formato y contenido3. Conecta al portal FACe4. Envía cada factura5. Descarga acuses de recibo6. 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```bashpython src/procesos/proceso_facturas.pyProgramada
Sección titulada «Programada»Configurar en Heptora para ejecutar:
- Lunes a Viernes
- 09:00, 14:00 y 18:00
- Solo si hay archivos en carpeta de entrada
Resultados
Sección titulada «Resultados»Tipos de Resultado
Sección titulada «Tipos de Resultado»- OK: Factura enviada correctamente y acuse descargado
- KO: Factura con datos incorrectos o ya enviada
- ERROR: Error técnico (conectividad, portal no disponible, etc.)
Notificaciones
Sección titulada «Notificaciones»- Email resumen a: finanzas@empresa.com
- Alerta crítica si > 20% errores
- Alerta advertencia si > 30% KOs
Métricas
Sección titulada «Métricas»- Tiempo promedio por factura: 45 segundos
- Tasa de éxito esperada: > 95%
- SLA: Todas las facturas procesadas en < 2 horas
Resolución de Problemas
Sección titulada «Resolución de Problemas»Error: “Certificado no válido”
Sección titulada «Error: “Certificado no válido”»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
KO: “Factura ya enviada”
Sección titulada «KO: “Factura ya enviada”»Causa: La factura ya existe en FACe Solución: Verificar manualmente, puede ser correcto
Contacto
Sección titulada «Contacto»- Responsable: Equipo de Finanzas
- Email: finanzas@empresa.com
- Soporte técnico: help@heptora.com
#### 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 GeneralSistema de automatización RPA basado en Heptora para procesosde 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.Diagramas
Sección titulada «Diagramas»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
```mermaidflowchart 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 pytestfrom 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 -v2. Tests de Integración
Sección titulada «2. Tests de Integración»Prueban la interacción entre componentes.
"""Tests de integración."""
import pytestimport osfrom 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 pass3. Tests End-to-End
Sección titulada «3. Tests End-to-End»Prueban el proceso completo en entorno real.
"""Tests end-to-end."""
import pytestfrom src.main import ejecutar_proceso
@pytest.mark.e2e@pytest.mark.slowclass 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"] == 0Estrategia de Testing
Sección titulada «Estrategia de Testing»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
Datos de Prueba
Sección titulada «Datos de Prueba»Crear datasets de prueba representativos:
"""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" }]Optimización de Rendimiento
Sección titulada «Optimización de Rendimiento»Identificar Cuellos de Botella
Sección titulada «Identificar Cuellos de Botella»Medir tiempos de ejecución:
import timefrom 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_tiempodef cargar_facturas(): """Carga facturas desde Excel.""" # ... código ...
@medir_tiempodef validar_facturas(facturas): """Valida facturas.""" # ... código ...
@medir_tiempodef 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!Técnicas de Optimización
Sección titulada «Técnicas de Optimización»1. Procesamiento por Lotes
Sección titulada «1. Procesamiento por Lotes»Agrupa operaciones similares.
# ❌ Ineficiente: Una consulta por registrofor factura in facturas: cliente = buscar_cliente_en_db(factura["CIF"]) # ...
# ✅ Eficiente: Una consulta para todoscifs = [f["CIF"] for f in facturas]clientes_dict = buscar_clientes_en_db_lote(cifs) # Una sola consultafor factura in facturas: cliente = clientes_dict.get(factura["CIF"]) # ...2. Cache de Datos
Sección titulada «2. Cache de Datos»Evita consultas repetidas.
# Cache simple con diccionariocache_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 procesofor factura in facturas: cliente = obtener_cliente_con_cache(factura["CIF"]) # ...3. Lazy Loading
Sección titulada «3. Lazy Loading»Carga datos solo cuando sean necesarios.
# ❌ Carga todo al inicio (puede ser innecesario)todos_los_productos = cargar_todos_los_productos() # 10,000 productosfor pedido in pedidos: if pedido["estado"] == "pendiente": producto = todos_los_productos[pedido["producto_id"]] # ...
# ✅ Carga solo lo necesariofor pedido in pedidos: if pedido["estado"] == "pendiente": producto = cargar_producto(pedido["producto_id"]) # Solo si es necesario # ...4. Optimizar Operaciones de Archivo
Sección titulada «4. Optimizar Operaciones de Archivo»# ❌ Leer archivo múltiples vecesfor factura in facturas: plantilla = leer_plantilla_pdf() # Lee archivo cada vez pdf = generar_pdf(factura, plantilla) # ...
# ✅ Leer archivo una vezplantilla = leer_plantilla_pdf() # Una sola vezfor factura in facturas: pdf = generar_pdf(factura, plantilla) # ...Métricas de Rendimiento
Sección titulada «Métricas de Rendimiento»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 reporteSeguridad
Sección titulada «Seguridad»Principios de Seguridad
Sección titulada «Principios de Seguridad»1. Nunca Exponer Credenciales
Sección titulada «1. Nunca Exponer Credenciales»❌ Nunca hacer esto:
# ¡NUNCA!usuario = "admin@empresa.com"password = "MiPassword123"api_key = "sk-1234567890abcdef"✅ Siempre hacer esto:
# Usar sistema de secretos de Heptorausuario = heptora.data.get("portal_usuario")password = heptora.data.get("portal_password")api_key = heptora.data.get("api_key")
# Verificar que existenif not usuario or not password: heptora.log.force_ko("Faltan credenciales. Configúralas en Secretos.")2. Minimizar Logs de Datos Sensibles
Sección titulada «2. Minimizar Logs de Datos Sensibles»❌ 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 completasheptora.log.info("Tarjeta de crédito validada correctamente")3. Validar y Sanitizar Entradas
Sección titulada «3. Validar y Sanitizar Entradas»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 procesamientonombre_usuario = sanitizar_entrada(datos_entrada["nombre"])4. Principio de Mínimo Privilegio
Sección titulada «4. Principio de Mínimo Privilegio»- Usa cuentas con permisos mínimos necesarios
- No uses cuentas de administrador para procesos RPA
- Crea usuarios específicos para automatización
5. Auditoría
Sección titulada «5. Auditoría»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íticasregistrar_auditoria( operacion="crear_factura", usuario=heptora.data.get("usuario_actual"), datos_relevantes={ "numero_factura": factura["Numero"], "importe": factura["Importe"] })Cómo Heptora Ayuda con la Seguridad
Sección titulada «Cómo Heptora Ayuda con la Seguridad»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
Colaboración en Equipo
Sección titulada «Colaboración en Equipo»Roles en Proyectos RPA
Sección titulada «Roles en Proyectos RPA»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
Flujo de Trabajo en Equipo
Sección titulada «Flujo de Trabajo en Equipo»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 prioriza2. Desarrollo:
Desarrollador → Desarrolla en rama de feature ↓Desarrollador → Tests unitarios y de integración ↓Analista → Revisa y valida contra requisitos ↓Desarrollador → Ajusta según feedback3. Despliegue:
Desarrollador → Merge a rama main ↓Admin Sistemas → Despliega en entorno de prueba ↓Usuario Final → Pruebas de aceptación ↓Admin Sistemas → Despliega en producción4. Operación:
Robot → Ejecuta proceso automáticamente ↓Usuario Final → Revisa resultados ↓Analista → Monitorea métricas ↓Desarrollador → Ajusta según necesidadEstándares de Equipo
Sección titulada «Estándares de Equipo»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 ProyectoSeguir 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`Revisión de Código
Sección titulada «Revisión de Código»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?
Checklist General de Mejores Prácticas
Sección titulada «Checklist General de Mejores Prácticas»- 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
Simplicidad de Bloques
Sección titulada «Simplicidad de Bloques»- 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
Reutilización
Sección titulada «Reutilización»- 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
Gestión de Excepciones
Sección titulada «Gestión de Excepciones»- 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
Modularidad
Sección titulada «Modularidad»- 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
Documentación
Sección titulada «Documentación»- Existe README del proceso
- Las funciones tienen docstrings
- Hay ejemplos de uso
- La documentación está actualizada
Testing
Sección titulada «Testing»- 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
Seguridad
Sección titulada «Seguridad»- No hay credenciales hardcodeadas
- Se usan secretos de Heptora
- Los logs no exponen datos sensibles
- Se validan y sanitizan entradas
Rendimiento
Sección titulada «Rendimiento»- Se han identificado cuellos de botella
- Hay métricas de rendimiento
- Las operaciones costosas están optimizadas
- Se usa cache cuando es apropiado
Colaboración
Sección titulada «Colaboración»- 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
Recursos Relacionados
Sección titulada «Recursos Relacionados»- Selección de Procesos para Automatizar - Cómo elegir qué automatizar
- Bloques de Automatización - Componentes reutilizables de Heptora
- Plantillas de Procesos - Procesos predefinidos listos para usar
- Gestión de Secretos - Protección de credenciales
- Python en Heptora - Referencia completa de Python
- Resultados de Ejecución - Sistema OK/KO/ERROR
- Grabador de Procesos - Herramienta para desarrollo rápido
¿Necesitas más ayuda?
Sección titulada «¿Necesitas más ayuda?»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.