""" PropVoz AI — ElevenLabs Webhook Handler

Cliente: PropVoz AI (agente de voz IA para inmobiliarias) Skill: elevenlabs-webhooks | CULTIVA IA

Procesa eventos webhook de ElevenLabs Conversational AI:

• post_call_transcription → cualifica lead, actualiza CRM HubSpot, notifica Slack
• voice_removal_notice → alerta equipo en Slack
• voice_removal_notice_withdrawn → cancela alerta previa
• voice_removed → limpia referencias en BD

Stack: FastAPI + Python 3.11 + PostgreSQL + HubSpot API v3 + Slack webhooks """

import os import json import hmac import hashlib import time import re import logging from typing import Optional, Dict, Any from datetime import datetime, timezone

import httpx from fastapi import FastAPI, Request, HTTPException, Header from fastapi.responses import JSONResponse from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession from sqlalchemy.orm import sessionmaker from sqlalchemy import text

─── Logging estructurado ─────────────────────────────────────────────────────

logging.basicConfig( level=logging.INFO, format='%(asctime)s %(levelname)s [%(name)s] %(message)s' ) logger = logging.getLogger("propvoz.webhooks")

─── Config desde entorno ─────────────────────────────────────────────────────

ELEVENLABS_WEBHOOK_SECRET = os.environ["ELEVENLABS_WEBHOOK_SECRET"] HUBSPOT_API_KEY = os.environ["HUBSPOT_API_KEY"] SLACK_WEBHOOK_URL = os.environ["SLACK_WEBHOOK_URL"] DATABASE_URL = os.environ.get( "DATABASE_URL", "postgresql+asyncpg://propvoz:propvoz@localhost/propvoz" )

─── DB async engine ──────────────────────────────────────────────────────────

engine = create_async_engine(DATABASE_URL, echo=False) AsyncSessionLocal = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

app = FastAPI(title="PropVoz AI — Webhook Handler", version="1.0.0")

═══════════════════════════════════════════════════════════════════════════════

1. VERIFICACIÓN DE FIRMA ELEVENLABS (HMAC-SHA256)

═══════════════════════════════════════════════════════════════════════════════

def verify_elevenlabs_signature(raw_body: bytes, signature_header: str, secret: str) -> None: """ Verifica la firma HMAC-SHA256 del webhook de ElevenLabs. Formato cabecera: "t=,v0=" Tolerancia: 30 minutos. Lanza ValueError si la verificación falla. """ if not signature_header: raise ValueError("Cabecera ElevenLabs-Signature ausente")

parts: Dict[str, Any] = {}
for chunk in signature_header.split(","):
    if "=" in chunk:
        k, v = chunk.split("=", 1)
        parts[k] = v

timestamp = parts.get("t")
received_sig = parts.get("v0")

if not timestamp or not received_sig:
    raise ValueError(f"Formato de firma inválido: {signature_header!r}")

# Validar antigüedad (máximo 30 minutos)
age_seconds = abs(int(time.time()) - int(timestamp))
if age_seconds > 1800:
    raise ValueError(f"Timestamp del webhook caducado ({age_seconds}s > 1800s)")

# Calcular firma esperada: HMAC-SHA256("<timestamp>.<raw_body>")
signed_payload = f"{timestamp}.{raw_body.decode('utf-8')}"
expected_sig = hmac.new(
    secret.encode("utf-8"),
    signed_payload.encode("utf-8"),
    hashlib.sha256
).hexdigest()

if not hmac.compare_digest(received_sig, expected_sig):
    raise ValueError("Firma HMAC inválida — posible payload manipulado")

logger.debug("Firma ElevenLabs verificada correctamente (age=%ds)", age_seconds)

═══════════════════════════════════════════════════════════════════════════════

2. IDEMPOTENCIA — evitar doble procesado por call_id

═══════════════════════════════════════════════════════════════════════════════

async def is_already_processed(call_id: str) -> bool: """Comprueba si este call_id ya fue procesado (tabla processed_calls).""" async with AsyncSessionLocal() as session: result = await session.execute( text("SELECT 1 FROM processed_calls WHERE call_id = :cid LIMIT 1"), {"cid": call_id} ) return result.scalar() is not None

async def mark_as_processed(call_id: str) -> None: """Marca el call_id como procesado.""" async with AsyncSessionLocal() as session: await session.execute( text( "INSERT INTO processed_calls (call_id, processed_at) " "VALUES (:cid, NOW()) ON CONFLICT DO NOTHING" ), {"cid": call_id} ) await session.commit()

═══════════════════════════════════════════════════════════════════════════════

3. EXTRACCIÓN DE LEAD DESDE TRANSCRIPCIÓN

═══════════════════════════════════════════════════════════════════════════════

def extract_lead_data(transcript: str, call_data: Dict[str, Any]) -> Dict[str, Any]: """ Extrae datos del lead desde la transcripción de llamada usando regex + heurísticas. En producción, sustituir por llamada a Claude API para extracción LLM. """ lead = { "call_id" : call_data.get("call_id", ""), "agent_id" : call_data.get("agent_id", ""), "duration_secs" : call_data.get("duration", 0), "called_at" : call_data.get("start_time_unix_secs", 0), "transcript" : transcript, "phone" : None, "email" : None, "nombre" : None, "zona_interes" : None, "presupuesto" : None, "tipo_inmueble" : None, "cualificado" : False, }

# Teléfono español
phone_match = re.search(r'\b([6789]\d{8})\b', transcript)
if phone_match:
    lead["phone"] = phone_match.group(1)

# Email
email_match = re.search(r'[\w.+-]+@[\w.-]+\.\w{2,}', transcript, re.IGNORECASE)
if email_match:
    lead["email"] = email_match.group(0).lower()

# Zona de interés (barrios Madrid)
zonas = ["Salamanca", "Chamberí", "Retiro", "Malasaña", "Lavapiés",
         "Moncloa", "Hortaleza", "Vallecas", "Carabanchel", "Pozuelo",
         "Las Rozas", "Majadahonda", "Alcobendas", "Getafe", "Alcorcón"]
for zona in zonas:
    if zona.lower() in transcript.lower():
        lead["zona_interes"] = zona
        break

# Presupuesto aproximado
budget_match = re.search(r'(\d{2,4})\s*(?:mil|k)\b', transcript, re.IGNORECASE)
if budget_match:
    lead["presupuesto"] = int(budget_match.group(1)) * 1000

# Tipo de inmueble
for tipo in ["piso", "casa", "chalet", "ático", "dúplex", "local", "estudio"]:
    if tipo in transcript.lower():
        lead["tipo_inmueble"] = tipo
        break

# Lead cualificado si tiene al menos teléfono o email + zona
lead["cualificado"] = bool(
    (lead["phone"] or lead["email"]) and lead["zona_interes"]
)

return lead

═══════════════════════════════════════════════════════════════════════════════

4. PUSH A HUBSPOT CRM

═══════════════════════════════════════════════════════════════════════════════

async def push_to_hubspot(lead: Dict[str, Any]) -> Optional[str]: """ Crea contacto + deal en HubSpot con los datos del lead. Devuelve el contact_id creado, o None si falla. """ headers = { "Authorization": f"Bearer {HUBSPOT_API_KEY}", "Content-Type": "application/json", }

# Construir propiedades del contacto
props: Dict[str, str] = {
    "hs_lead_status": "NEW",
    "lead_source"   : "elevenlabs_voice_agent",
    "call_id_elevenlabs": lead["call_id"],
}
if lead.get("phone"):
    props["phone"] = lead["phone"]
if lead.get("email"):
    props["email"] = lead["email"]
if lead.get("zona_interes"):
    props["city"] = lead["zona_interes"]
if lead.get("presupuesto"):
    props["budget_range"] = str(lead["presupuesto"])
if lead.get("tipo_inmueble"):
    props["property_type"] = lead["tipo_inmueble"]

contact_payload = {"properties": props}

async with httpx.AsyncClient(timeout=10.0) as client:
    try:
        # 1. Crear contacto
        resp = await client.post(
            "https://api.hubapi.com/crm/v3/objects/contacts",
            headers=headers,
            json=contact_payload,
        )
        if resp.status_code not in (200, 201):
            logger.warning("HubSpot contacto error %s: %s", resp.status_code, resp.text)
            return None

        contact_id = resp.json()["id"]
        logger.info("HubSpot contacto creado: %s", contact_id)

        # 2. Crear deal asociado
        deal_payload = {
            "properties": {
                "dealname"  : f"Lead voz — {lead.get('zona_interes', 'Sin zona')} ({lead['call_id'][:8]})",
                "dealstage" : "appointmentscheduled",
                "pipeline"  : "default",
                "amount"    : str(lead.get("presupuesto") or 0),
            },
            "associations": [
                {
                    "to": {"id": contact_id},
                    "types": [{"associationCategory": "HUBSPOT_DEFINED", "associationTypeId": 3}]
                }
            ]
        }
        deal_resp = await client.post(
            "https://api.hubapi.com/crm/v3/objects/deals",
            headers=headers,
            json=deal_payload,
        )
        if deal_resp.status_code in (200, 201):
            logger.info("HubSpot deal creado: %s", deal_resp.json()["id"])

        return contact_id

    except httpx.RequestError as exc:
        logger.error("HubSpot request error: %s", exc)
        return None

═══════════════════════════════════════════════════════════════════════════════

5. NOTIFICACIONES SLACK

═══════════════════════════════════════════════════════════════════════════════

async def notify_slack(message: str, urgente: bool = False) -> None: """Envía notificación al canal de Slack de PropVoz AI.""" emoji = ":rotating_light:" if urgente else ":telephone_receiver:" payload = {"text": f"{emoji} PropVoz AI — {message}"} async with httpx.AsyncClient(timeout=5.0) as client: try: await client.post(SLACK_WEBHOOK_URL, json=payload) except httpx.RequestError as exc: logger.warning("Slack notify falló: %s", exc)

═══════════════════════════════════════════════════════════════════════════════

6. HANDLERS DE EVENTOS

═══════════════════════════════════════════════════════════════════════════════

async def handle_post_call_transcription(data: Dict[str, Any]) -> None: """ Evento: post_call_transcription Ejecuta: idempotencia → extracción de lead → HubSpot → Slack """ call_id = data.get("call_id", "unknown")

# Idempotencia
if await is_already_processed(call_id):
    logger.info("call_id %s ya procesado — saltando", call_id)
    return

# Obtener transcripción
transcript = ""
for turn in data.get("transcript", []):
    role = turn.get("role", "")
    msg  = turn.get("message", "")
    transcript += f"{role}: {msg}\n"

# Extraer datos del lead
lead = extract_lead_data(transcript, data)
logger.info(
    "Lead extraído — call_id=%s cualificado=%s zona=%s presupuesto=%s",
    call_id, lead["cualificado"], lead["zona_interes"], lead["presupuesto"]
)

# Push a HubSpot si está cualificado
contact_id = None
if lead["cualificado"]:
    contact_id = await push_to_hubspot(lead)

# Notificación Slack
duracion_min = lead["duration_secs"] // 60
if lead["cualificado"] and contact_id:
    await notify_slack(
        f"Nuevo lead cualificado ({duracion_min} min)\n"
        f"> Zona: {lead['zona_interes']} | "
        f"Presupuesto: {lead.get('presupuesto', 'n/d')}€ | "
        f"Tipo: {lead.get('tipo_inmueble', 'n/d')}\n"
        f"> HubSpot contact: `{contact_id}`"
    )
else:
    await notify_slack(
        f"Llamada NO cualificada ({duracion_min} min) — call_id: `{call_id[:8]}`"
    )

# Marcar como procesado
await mark_as_processed(call_id)

async def handle_voice_removal_notice(data: Dict[str, Any]) -> None: """Evento: voice_removal_notice — alerta urgente al equipo.""" voice_id = data.get("voice_id", "unknown") voice_name = data.get("voice_name", "Desconocida") removal_at = data.get("removal_date", "fecha desconocida")

logger.warning("ALERTA: voz '%s' (%s) será eliminada el %s", voice_name, voice_id, removal_at)
await notify_slack(
    f"ALERTA: La voz *{voice_name}* (`{voice_id}`) será eliminada el `{removal_at}`.\n"
    f"Acción requerida: exportar/respaldar antes de esa fecha.",
    urgente=True
)

async def handle_voice_removal_notice_withdrawn(data: Dict[str, Any]) -> None: """Evento: voice_removal_notice_withdrawn — cancelar alerta previa.""" voice_id = data.get("voice_id", "unknown") voice_name = data.get("voice_name", "Desconocida")

logger.info("Aviso de eliminación retirado para voz '%s' (%s)", voice_name, voice_id)
await notify_slack(
    f"Aviso retirado: la voz *{voice_name}* (`{voice_id}`) ya no será eliminada. Todo OK."
)

async def handle_voice_removed(data: Dict[str, Any]) -> None: """Evento: voice_removed — limpiar referencias en BD.""" voice_id = data.get("voice_id", "unknown") voice_name = data.get("voice_name", "Desconocida")

logger.warning("Voz eliminada: %s (%s)", voice_name, voice_id)

# Limpiar referencias en BD
async with AsyncSessionLocal() as session:
    await session.execute(
        text("UPDATE agents SET voice_id = NULL WHERE voice_id = :vid"),
        {"vid": voice_id}
    )
    await session.commit()

await notify_slack(
    f"Voz *{voice_name}* (`{voice_id}`) eliminada por ElevenLabs.\n"
    f"Referencias limpiadas en BD. Asignar nueva voz a los agentes afectados.",
    urgente=True
)

═══════════════════════════════════════════════════════════════════════════════

7. ENDPOINT PRINCIPAL

═══════════════════════════════════════════════════════════════════════════════

@app.get("/health") async def health(): return {"status": "ok", "service": "propvoz-webhook-handler", "version": "1.0.0"}

@app.post("/webhooks/elevenlabs") async def elevenlabs_webhook( request: Request, elevenlabs_signature: Optional[str] = Header(None, alias="ElevenLabs-Signature"), ): """ Endpoint webhook ElevenLabs para PropVoz AI. 1. Verifica firma HMAC → 401 si inválida 2. Parsea evento 3. Despacha al handler correspondiente 4. Devuelve 200 para ACK """ raw_body = await request.body()

# Intentar cabecera en minúsculas si no viene en camel-case
sig = elevenlabs_signature or request.headers.get("elevenlabs-signature")

# 1. Verificar firma
try:
    verify_elevenlabs_signature(raw_body, sig or "", ELEVENLABS_WEBHOOK_SECRET)
except ValueError as exc:
    logger.warning("Firma rechazada: %s | ip=%s", exc, request.client.host)
    raise HTTPException(status_code=401, detail=str(exc))

# 2. Parsear evento
try:
    event = json.loads(raw_body)
except json.JSONDecodeError:
    raise HTTPException(status_code=400, detail="Payload no es JSON válido")

event_type = event.get("type", "unknown")
event_data = event.get("data", {})

logger.info("Evento recibido: type=%s", event_type)

# 3. Despachar
try:
    if event_type == "post_call_transcription":
        await handle_post_call_transcription(event_data)
    elif event_type == "voice_removal_notice":
        await handle_voice_removal_notice(event_data)
    elif event_type == "voice_removal_notice_withdrawn":
        await handle_voice_removal_notice_withdrawn(event_data)
    elif event_type == "voice_removed":
        await handle_voice_removed(event_data)
    else:
        logger.info("Tipo de evento no gestionado: %s", event_type)
except Exception as exc:
    # Loguear pero devolver 200 para que ElevenLabs no reintente indefinidamente
    # Usar un sistema de dead-letter queue en producción.
    logger.exception("Error al procesar evento %s: %s", event_type, exc)

# 4. ACK siempre 200
return JSONResponse(status_code=200, content={"received": True, "type": event_type})

─── Inicio local ──────────────────────────────────────────────────────────────

if name == "main": import uvicorn uvicorn.run(app, host="0.0.0.0", port=3000, reload=False)