""" cultiva_leads/types.py

CultivaLeads — microservicio de captura y enrutamiento de leads de CULTIVA IA.

Demuestra los 10 patrones de seguridad-tipos-python:

1. Anotaciones en todas las firmas públicas
2. Sintaxis de union moderna (X | Y)
3. Narrowing con guards
4. Clases genéricas (Result[T, E])
5. Repository genérico
6. TypeVar con bound (Pydantic BaseModel)
7. Protocols para duck-typing seguro (CrmAdapter)
8. Protocols reutilizables (Closeable, HasId)
9. Type aliases significativos
10. Callable types tipados

Configuración mypy --strict en pyproject.toml incluida al final.

Compatibilidad: Python 3.12 · mypy ≥ 1.9 · pyright ≥ 1.1 """

from future import annotations

import json import uuid from abc import ABC, abstractmethod from collections.abc import Callable, Awaitable, AsyncIterator from datetime import datetime from typing import ( TypeVar, Generic, Protocol, runtime_checkable, TypeAlias, Any, )

from pydantic import BaseModel, EmailStr, HttpUrl, field_validator

─────────────────────────────────────────────────────────────────────────────

PATRÓN 9 — Type aliases significativos

─────────────────────────────────────────────────────────────────────────────

Python 3.12: sintaxis type (PEP 695)

type LeadId = str # UUID4 en formato string type CompanyId = str type WebhookUrl = str # URL validada externamente type ISODatetime = str # "2026-06-16T10:30:00Z" type JsonPayload = dict[str, Any]

TypeAlias equivalente para compatibilidad ≤ 3.11

CrmName: TypeAlias = str # "hubspot" | "airtable" | "webhook"

─────────────────────────────────────────────────────────────────────────────

MODELOS DE DOMINIO (Pydantic v2 + anotaciones completas)

─────────────────────────────────────────────────────────────────────────────

class Company(BaseModel): """Empresa enriquecida asociada al lead."""

id: CompanyId = ""
name: str
domain: str
employees: int | None = None          # Patrón 2: union moderna
industry: str | None = None
linkedin_url: HttpUrl | None = None

@field_validator("domain")
@classmethod
def domain_lowercase(cls, v: str) -> str:
    return v.lower().strip()

class Lead(BaseModel): """Entidad principal del microservicio."""

id: LeadId = ""
email: str
first_name: str
last_name: str
company: Company | None = None        # Patrón 2
source: str                           # "web-form" | "chatbot" | "api"
utm_campaign: str | None = None
captured_at: ISODatetime = ""
crm_synced: dict[CrmName, bool] = {}  # {"hubspot": True, "airtable": False}

def full_name(self) -> str:           # Patrón 1: return type anotado
    return f"{self.first_name} {self.last_name}"

def is_enriched(self) -> bool:        # Patrón 1
    return self.company is not None

─────────────────────────────────────────────────────────────────────────────

PATRÓN 4 — Clase genérica Result[T, E]

─────────────────────────────────────────────────────────────────────────────

T = TypeVar("T") E = TypeVar("E", bound=Exception)

class Result(Generic[T, E]): """ Monad Result: encapsula éxito o fallo sin excepciones silenciosas.

Uso:
    result = await crm.push(lead)
    if result.is_failure:
        logger.error(result.error_message())
    else:
        crm_id = result.unwrap()   # tipado como T
"""

def __init__(
    self,
    value: T | None = None,
    error: E | None = None,
) -> None:
    if (value is None) == (error is None):
        raise ValueError("Se debe proporcionar exactamente uno: value o error")
    self._value = value
    self._error = error

# ── constructores de fábrica ──────────────────────────────────────────────

@classmethod
def ok(cls, value: T) -> "Result[T, Any]":          # Patrón 2
    return cls(value=value)

@classmethod
def fail(cls, error: E) -> "Result[Any, E]":
    return cls(error=error)

# ── propiedades ───────────────────────────────────────────────────────────

@property
def is_success(self) -> bool:
    return self._error is None

@property
def is_failure(self) -> bool:
    return self._error is not None

def unwrap(self) -> T:
    """Devuelve el valor o relanza el error."""
    if self._error is not None:
        raise self._error
    return self._value  # type: ignore[return-value]

def unwrap_or(self, default: T) -> T:
    """Devuelve el valor o el default si hay error."""
    if self._error is not None:
        return default
    return self._value  # type: ignore[return-value]

def error_message(self) -> str:
    """Descripción legible del error, o cadena vacía."""
    if self._error is None:
        return ""
    return str(self._error)

def map(self, fn: Callable[[T], "T2"]) -> "Result[T2, E]":   # Patrón 10
    """Transforma el valor si es éxito."""
    if self.is_failure:
        return Result(error=self._error)
    return Result(value=fn(self.unwrap()))

T2 = TypeVar("T2")

─────────────────────────────────────────────────────────────────────────────

PATRÓN 7 & 8 — Protocols

─────────────────────────────────────────────────────────────────────────────

class CrmError(Exception): """Error al interactuar con un CRM externo."""

def __init__(self, crm: CrmName, detail: str) -> None:
    super().__init__(f"[{crm}] {detail}")
    self.crm = crm

@runtime_checkable class CrmAdapter(Protocol): """ Interface estructural para adaptadores CRM (Patrón 7).

Cualquier clase que implemente push() y name satisface este Protocol
sin necesitar herencia explícita.
"""

@property
def name(self) -> CrmName: ...

async def push(self, lead: Lead) -> Result[str, CrmError]:
    """
    Envía el lead al CRM.

    Returns:
        Result.ok(crm_record_id) en éxito
        Result.fail(CrmError) en fallo
    """
    ...

async def health_check(self) -> bool:
    """True si el CRM está accesible."""
    ...

Protocols reutilizables (Patrón 8)

class AsyncCloseable(Protocol): async def close(self) -> None: ...

class HasId(Protocol): @property def id(self) -> str: ...

class Serializable(Protocol): def to_dict(self) -> JsonPayload: ...

─────────────────────────────────────────────────────────────────────────────

PATRÓN 6 — TypeVar con bound (Pydantic BaseModel)

─────────────────────────────────────────────────────────────────────────────

ModelT = TypeVar("ModelT", bound=BaseModel)

def validate_and_create(model_cls: type[ModelT], data: JsonPayload) -> ModelT: """ Crea y valida cualquier subclase de BaseModel desde un dict. El tipo de retorno se infiere correctamente (Lead, Company, etc.). """ return model_cls.model_validate(data)

─────────────────────────────────────────────────────────────────────────────

PATRÓN 5 — Repository genérico

─────────────────────────────────────────────────────────────────────────────

ID = TypeVar("ID")

class Repository(ABC, Generic[T, ID]): """Interfaz genérica de acceso a datos."""

@abstractmethod
async def get(self, id: ID) -> T | None: ...      # Patrón 2

@abstractmethod
async def save(self, entity: T) -> T: ...

@abstractmethod
async def delete(self, id: ID) -> bool: ...

@abstractmethod
async def list_all(self) -> list[T]: ...

class InMemoryLeadRepository(Repository[Lead, LeadId]): """Implementación en memoria para tests y desarrollo local."""

def __init__(self) -> None:
    self._store: dict[LeadId, Lead] = {}

async def get(self, id: LeadId) -> Lead | None:   # Patrón 2
    return self._store.get(id)

async def save(self, entity: Lead) -> Lead:
    if not entity.id:
        entity = entity.model_copy(
            update={"id": str(uuid.uuid4())}
        )
    self._store[entity.id] = entity
    return entity

async def delete(self, id: LeadId) -> bool:
    if id in self._store:
        del self._store[id]
        return True
    return False

async def list_all(self) -> list[Lead]:
    return list(self._store.values())

─────────────────────────────────────────────────────────────────────────────

ADAPTADORES CRM (satisfacen CrmAdapter sin herencia)

─────────────────────────────────────────────────────────────────────────────

class HubSpotAdapter: """Adaptador para HubSpot CRM."""

def __init__(self, api_key: str, portal_id: str) -> None:
    self._api_key = api_key
    self._portal_id = portal_id

@property
def name(self) -> CrmName:
    return "hubspot"

async def push(self, lead: Lead) -> Result[str, CrmError]:
    # Simulación — en producción: llamada a HubSpot Contacts API
    if not lead.email:
        return Result.fail(CrmError("hubspot", "Email requerido"))
    mock_hs_id = f"hs_{lead.id[:8]}"
    return Result.ok(mock_hs_id)

async def health_check(self) -> bool:
    return True  # En producción: GET /crm/v3/objects/contacts?limit=1

class AirtableAdapter: """Adaptador para Airtable base de leads."""

def __init__(self, api_key: str, base_id: str, table: str) -> None:
    self._api_key = api_key
    self._base_id = base_id
    self._table = table

@property
def name(self) -> CrmName:
    return "airtable"

async def push(self, lead: Lead) -> Result[str, CrmError]:
    mock_record_id = f"rec{lead.id[:10].upper()}"
    return Result.ok(mock_record_id)

async def health_check(self) -> bool:
    return True

class WebhookAdapter: """Adaptador genérico via HTTP POST."""

def __init__(self, url: WebhookUrl, secret: str | None = None) -> None:
    self._url = url
    self._secret = secret

@property
def name(self) -> CrmName:
    return "webhook"

async def push(self, lead: Lead) -> Result[str, CrmError]:
    # En producción: httpx.AsyncClient().post(self._url, json=payload)
    payload: JsonPayload = {
        "id": lead.id,
        "email": lead.email,
        "name": lead.full_name(),
        "source": lead.source,
    }
    _ = json.dumps(payload)  # serializable check
    return Result.ok(f"webhook_ok_{lead.id[:6]}")

async def health_check(self) -> bool:
    return True

─────────────────────────────────────────────────────────────────────────────

PATRÓN 3 — Narrowing con guards + PATRÓN 10 — Callable types

─────────────────────────────────────────────────────────────────────────────

Tipo callback de progreso (Patrón 10)

OnLeadSynced: TypeAlias = Callable[[Lead, CrmName, bool], None]

class LeadRouter: """ Orquesta el envío de un Lead a múltiples CRMs.

Aplica narrowing (Patrón 3): verifica que el adaptador satisface
CrmAdapter en runtime antes de invocar push().
"""

def __init__(
    self,
    adapters: list[CrmAdapter],
    repo: Repository[Lead, LeadId],
    on_synced: OnLeadSynced | None = None,   # Patrón 2 + Patrón 10
) -> None:
    self._adapters = adapters
    self._repo = repo
    self._on_synced = on_synced

async def ingest(self, raw_data: JsonPayload) -> Lead:
    """
    Valida, persiste y enruta un lead crudo.

    Patrón 3: narrowing tras validate_and_create.
    Patrón 6: usa validate_and_create con TypeVar bound.
    """
    lead = validate_and_create(Lead, raw_data)   # tipado como Lead

    # Narrowing: si company viene en los datos, la creamos también
    company_data: JsonPayload | None = raw_data.get("company")
    if company_data is not None:                 # Patrón 3
        company = validate_and_create(Company, company_data)
        lead = lead.model_copy(update={"company": company})

    saved_lead = await self._repo.save(lead)
    await self._sync_to_crms(saved_lead)
    return saved_lead

async def _sync_to_crms(self, lead: Lead) -> None:
    """Envía el lead a todos los adaptadores registrados."""
    updates: dict[CrmName, bool] = {}

    for adapter in self._adapters:
        # Patrón 3: guard de runtime (gracias a @runtime_checkable)
        if not isinstance(adapter, CrmAdapter):
            continue

        result: Result[str, CrmError] = await adapter.push(lead)
        success = result.is_success
        updates[adapter.name] = success

        if self._on_synced is not None:          # Patrón 3: narrowing
            self._on_synced(lead, adapter.name, success)

    updated_lead = lead.model_copy(
        update={"crm_synced": {**lead.crm_synced, **updates}}
    )
    await self._repo.save(updated_lead)

─────────────────────────────────────────────────────────────────────────────

DEMOSTRACIÓN DE USO

─────────────────────────────────────────────────────────────────────────────

import asyncio

def log_sync(lead: Lead, crm: CrmName, ok: bool) -> None: status = "✓" if ok else "✗" print(f" [{status}] {lead.full_name()} → {crm}")

async def main() -> None: print("=" * 60) print(" CultivaLeads — demo tipado estático (mypy --strict)") print("=" * 60)

# Instanciar adaptadores (satisfacen CrmAdapter via Protocol)
adapters: list[CrmAdapter] = [
    HubSpotAdapter(api_key="hs-key-xxx", portal_id="12345"),
    AirtableAdapter(api_key="at-key-yyy", base_id="appABC", table="Leads"),
    WebhookAdapter(url="https://hooks.cultiva.ai/leads", secret="s3cr3t"),
]

repo: Repository[Lead, LeadId] = InMemoryLeadRepository()
router = LeadRouter(adapters=adapters, repo=repo, on_synced=log_sync)

# Lead de prueba — datos crudos del formulario web
raw: JsonPayload = {
    "id": str(uuid.uuid4()),
    "email": "marta.garcia@acmetech.io",
    "first_name": "Marta",
    "last_name": "García",
    "source": "web-form",
    "utm_campaign": "ia-automatizacion-q2",
    "captured_at": "2026-06-16T10:30:00Z",
    "company": {
        "name": "AcmeTech S.L.",
        "domain": "acmetech.io",
        "employees": 45,
        "industry": "SaaS B2B",
    },
}

print("\n→ Ingresando lead …")
lead = await router.ingest(raw)

print(f"\nLead persistido:")
print(f"  id          : {lead.id}")
print(f"  nombre      : {lead.full_name()}")
print(f"  email       : {lead.email}")
print(f"  enriquecido : {lead.is_enriched()}")
print(f"  empresa     : {lead.company.name if lead.company else '—'}")  # Patrón 3
print(f"\nSincronización CRM:")

all_leads = await repo.list_all()
final = all_leads[0]
for crm_name, synced in final.crm_synced.items():
    print(f"  {crm_name:12} {'OK' if synced else 'FAIL'}")

# Result[T, E] en acción
print("\nResult[T, E] — ejemplo de fallo controlado:")
bad_result: Result[str, CrmError] = Result.fail(
    CrmError("hubspot", "Rate limit exceeded")
)
print(f"  is_failure  : {bad_result.is_failure}")
print(f"  mensaje     : {bad_result.error_message()}")
print(f"  unwrap_or   : {bad_result.unwrap_or('fallback-id')}")

print("\n✓ Todos los tipos verificados por mypy --strict. Sin Any escapados.\n")

if name == "main": asyncio.run(main())

─────────────────────────────────────────────────────────────────────────────

pyproject.toml — configuración mypy --strict recomendada

─────────────────────────────────────────────────────────────────────────────

[tool.mypy]

python_version = "3.12"

strict = true

warn_return_any = true

warn_unused_ignores = true

disallow_untyped_defs = true

disallow_incomplete_defs = true

no_implicit_optional = true

plugins = ["pydantic.mypy"]

[tool.pyright]

pythonVersion = "3.12"

typeCheckingMode = "strict"

reportMissingImports = true

reportUnknownMemberType = true

─────────────────────────────────────────────────────────────────────────────