""" CULTIVA IA — Pipeline de procesamiento de leads

Microservicio Python que valida, normaliza e ingesta leads desde múltiples fuentes (formularios, LinkedIn, CSV) hacia el CRM.

Patrones de manejo de errores aplicados (skill: manejo-errores-python):

1. Validación temprana (fail fast) con Pydantic v2
2. Excepciones custom con código HTTP y contexto estructurado
3. Gestión de fallos parciales en operaciones batch
4. Encadenamiento de excepciones (raise ... from e)
5. Logging estructurado con IDs y contadores
6. Conversión a tipos de dominio en la frontera del sistema """

from future import annotations

import logging import re from dataclasses import dataclass, field from enum import Enum from typing import Any

from pydantic import BaseModel, Field, field_validator, model_validator

---------------------------------------------------------------------------

Logging estructurado

---------------------------------------------------------------------------

logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(name)s — %(message)s", datefmt="%Y-%m-%dT%H:%M:%S", ) logger = logging.getLogger("cultiva.leads")

---------------------------------------------------------------------------

Tipos de dominio (Pattern 2 — Convert to domain types early)

---------------------------------------------------------------------------

class OrigenLead(str, Enum): """Canales de captación de leads soportados.""" LINKEDIN = "linkedin" FORMULARIO = "formulario" CSV = "csv" WEBHOOK = "webhook"

def parse_origen(valor: str) -> OrigenLead: """Convierte un string de entrada en OrigenLead.

Args:
    valor: Cadena de texto proveniente del cliente.

Returns:
    Miembro válido de OrigenLead.

Raises:
    ValueError: Si el origen no está soportado.
"""
try:
    return OrigenLead(valor.lower().strip())
except ValueError:
    validos = [o.value for o in OrigenLead]
    raise ValueError(
        f"Origen '{valor}' no reconocido. "
        f"Opciones válidas: {', '.join(validos)}"
    )

---------------------------------------------------------------------------

Excepciones custom (Pattern 5 — Custom exceptions with context)

---------------------------------------------------------------------------

class LeadError(Exception): """Excepción base para errores del pipeline de leads."""

def __init__(
    self,
    mensaje: str,
    codigo_http: int = 400,
    campo: str | None = None,
    lead_email: str | None = None,
) -> None:
    self.codigo_http = codigo_http
    self.campo = campo
    self.lead_email = lead_email
    super().__init__(mensaje)

def to_dict(self) -> dict[str, Any]:
    return {
        "error": str(self),
        "codigo_http": self.codigo_http,
        "campo": self.campo,
        "lead_email": self.lead_email,
    }

class ValidacionError(LeadError): """Los datos del lead no superan la validación.""" def init(self, mensaje: str, campo: str, lead_email: str | None = None) -> None: super().init(mensaje, codigo_http=422, campo=campo, lead_email=lead_email)

class CRMError(LeadError): """Fallo al escribir en el CRM externo.""" def init(self, mensaje: str, lead_email: str | None = None) -> None: super().init(mensaje, codigo_http=502, lead_email=lead_email)

class RateLimitCRMError(CRMError): """El CRM ha devuelto 429 — demasiadas peticiones.""" def init(self, retry_after: int = 60) -> None: self.retry_after = retry_after super().init( f"CRM rate limit alcanzado. Reintentar en {retry_after}s" )

---------------------------------------------------------------------------

Modelo de entrada con Pydantic (Pattern 3 — Pydantic for complex validation)

---------------------------------------------------------------------------

_EMAIL_RE = re.compile(r"^[^@\s]+@[^@\s]+.[^@\s]+$")

class LeadInput(BaseModel): """Datos crudos de un lead recibidos en la API."""

email:   str = Field(..., min_length=5,  max_length=254, description="Email del contacto")
nombre:  str = Field(..., min_length=1,  max_length=120, description="Nombre completo")
empresa: str = Field(..., min_length=1,  max_length=200, description="Nombre de la empresa")
cargo:   str = Field("",  max_length=100,                description="Cargo o puesto")
origen:  str = Field(...,                                 description="Canal de captación")

@field_validator("email")
@classmethod
def validar_email(cls, v: str) -> str:
    v = v.strip().lower()
    if not _EMAIL_RE.match(v):
        raise ValueError(
            f"'{v}' no tiene formato de email válido (esperado: usuario@dominio.tld)"
        )
    return v

@field_validator("nombre", "empresa")
@classmethod
def normalizar_texto(cls, v: str) -> str:
    return v.strip().title()

@field_validator("cargo")
@classmethod
def normalizar_cargo(cls, v: str) -> str:
    return v.strip().title() if v else "Sin especificar"

@model_validator(mode="after")
def validar_origen(self) -> LeadInput:
    # Convertimos a tipo dominio para detectar errores en la frontera
    parse_origen(self.origen)   # lanza ValueError si inválido
    return self

class LeadNormalizado(BaseModel): """Lead ya validado y listo para insertar en el CRM.""" email: str nombre: str empresa: str cargo: str origen: OrigenLead

---------------------------------------------------------------------------

Servicio CRM (simulado) con manejo de excepciones encadenadas (Pattern 6)

---------------------------------------------------------------------------

class CRMService: """Cliente simulado del CRM Huly."""

# Emails que simularán fallos de red para demostrar el patrón
_SIMULAR_FALLO_RED = {"fallo@red.test"}

def insertar_lead(self, lead: LeadNormalizado) -> str:
    """Inserta un lead en el CRM y devuelve el ID generado.

    Args:
        lead: Lead validado y normalizado.

    Returns:
        ID asignado por el CRM.

    Raises:
        CRMError: Si el CRM devuelve un error de escritura.
    """
    if lead.email in self._SIMULAR_FALLO_RED:
        # Simula excepción de red de bajo nivel
        causa = ConnectionError("Timeout conectando a CRM (10s)")
        raise CRMError(
            f"No se pudo insertar '{lead.email}' en el CRM: error de red",
            lead_email=lead.email,
        ) from causa

    # Simula inserción exitosa — devuelve ID determinista
    crm_id = f"HLY-{abs(hash(lead.email)) % 100000:05d}"
    logger.info(
        "Lead insertado en CRM",
        extra={"email": lead.email, "crm_id": crm_id, "origen": lead.origen.value},
    )
    return crm_id

---------------------------------------------------------------------------

Resultado de batch (Pattern 7 — Partial failures)

---------------------------------------------------------------------------

@dataclass class ResultadoBatch: """Resultado de procesar un lote de leads.""" exitosos: dict[int, dict[str, str]] = field(default_factory=dict) # idx -> {email, crm_id} fallidos: dict[int, dict[str, Any]] = field(default_factory=dict) # idx -> {email, error}

@property
def total(self) -> int:
    return len(self.exitosos) + len(self.fallidos)

@property
def tasa_exito(self) -> float:
    return len(self.exitosos) / self.total if self.total else 0.0

def resumen(self) -> str:
    return (
        f"Batch finalizado — "
        f"{len(self.exitosos)}/{self.total} exitosos "
        f"({self.tasa_exito:.0%}), "
        f"{len(self.fallidos)} fallidos"
    )

---------------------------------------------------------------------------

Pipeline principal (Pattern 1 + 3 + 7 combinados)

---------------------------------------------------------------------------

def procesar_batch_leads( datos_crudos: list[dict[str, Any]], crm: CRMService | None = None, ) -> ResultadoBatch: """Procesa un lote de leads: valida, normaliza e ingesta en el CRM.

Fallos individuales NO abortan el batch; se registran y continúa.

Args:
    datos_crudos: Lista de dicts con datos de leads sin validar.
    crm:          Instancia del servicio CRM (inyectable para tests).

Returns:
    ResultadoBatch con exitosos y fallidos indexados.

Raises:
    ValueError: Si datos_crudos está vacío o es demasiado grande.
"""
# --- Validación temprana del batch entero (Pattern 1: Fail Fast) ---
if not datos_crudos:
    raise ValueError("'datos_crudos' no puede estar vacío")
if len(datos_crudos) > 500:
    raise ValueError(
        f"El batch supera el límite de 500 leads (recibidos: {len(datos_crudos)})"
    )

if crm is None:
    crm = CRMService()

resultado = ResultadoBatch()
total = len(datos_crudos)

logger.info("Iniciando procesamiento de batch", extra={"total_leads": total})

for idx, raw in enumerate(datos_crudos):
    email_hint = raw.get("email", f"lead[{idx}]")

    try:
        # 1. Validación y normalización con Pydantic
        lead_input = LeadInput(**raw)

        # 2. Conversión a tipo de dominio
        lead = LeadNormalizado(
            email=lead_input.email,
            nombre=lead_input.nombre,
            empresa=lead_input.empresa,
            cargo=lead_input.cargo,
            origen=parse_origen(lead_input.origen),
        )

        # 3. Inserción en CRM
        crm_id = crm.insertar_lead(lead)
        resultado.exitosos[idx] = {"email": lead.email, "crm_id": crm_id}

    except Exception as e:  # noqa: BLE001  — captura intencional para batch
        resultado.fallidos[idx] = {
            "email": email_hint,
            "error": str(e),
            "tipo": type(e).__name__,
        }
        logger.warning(
            "Lead fallido en batch",
            extra={
                "idx": idx,
                "email": email_hint,
                "error": str(e),
                "tipo": type(e).__name__,
            },
        )

logger.info(resultado.resumen(), extra={"exitosos": len(resultado.exitosos), "fallidos": len(resultado.fallidos)})
return resultado

---------------------------------------------------------------------------

Demo / ejecución directa

---------------------------------------------------------------------------

LEADS_EJEMPLO = [ {"email": "ana.garcia@startuptech.es", "nombre": "Ana García", "empresa": "StartupTech SL", "cargo": "CEO", "origen": "linkedin"}, {"email": "invalid-email", "nombre": "Carlos Pérez", "empresa": "", "cargo": "CTO", "origen": "formulario"}, {"email": "pedro.ruiz@marketingpro.com","nombre": "Pedro Ruiz", "empresa": "MarketingPro", "cargo": "Director", "origen": "csv"}, {"email": "maria@", "nombre": "María López", "empresa": "Consultoría Digital","cargo": "", "origen": "linkedin"}, {"email": "javier.m@agencia360.es", "nombre": "Javier Martínez","empresa": "Agencia 360", "cargo": "Fundador", "origen": "formulario"}, {"email": "laura.s@ecommerce.io", "nombre": "Laura Sánchez", "empresa": "EcommerceIO", "cargo": "CMO", "origen": "csv"}, ]

def main() -> None: print("=" * 62) print(" CULTIVA IA — Pipeline de Leads con manejo robusto de errores") print("=" * 62)

resultado = procesar_batch_leads(LEADS_EJEMPLO)

print(f"\n{resultado.resumen()}\n")

print("  EXITOSOS:")
for idx, item in resultado.exitosos.items():
    print(f"    [{idx}] ✓ {item['email']:<38} → {item['crm_id']}")

print("\n  FALLIDOS:")
for idx, item in resultado.fallidos.items():
    print(f"    [{idx}] ✗ {item['email']:<30} | {item['tipo']}: {item['error'][:60]}")

# Demostrar fallo temprano en batch vacío
print("\n  TEST fail-fast (batch vacío):")
try:
    procesar_batch_leads([])
except ValueError as e:
    print(f"    ValueError capturado correctamente: {e}")

# Demostrar fallo temprano en batch demasiado grande
print("\n  TEST fail-fast (batch > 500):")
try:
    procesar_batch_leads([{"x": i} for i in range(501)])
except ValueError as e:
    print(f"    ValueError capturado correctamente: {e}")

print("\n" + "=" * 62)

if name == "main": main()