| """WhatsApp Business Cloud API webhook.""" |
|
|
| from __future__ import annotations |
|
|
| import logging |
| from typing import Any |
|
|
| import httpx |
| from fastapi import APIRouter, Depends, HTTPException, Query, Request |
| from sqlalchemy import select |
| from sqlalchemy.ext.asyncio import AsyncSession |
|
|
| from app.config import get_settings |
| from app.db import get_db |
| from app.models import Conversation |
| from app.services.chat import handle_user_message, start_conversation |
|
|
| logger = logging.getLogger(__name__) |
| router = APIRouter(prefix="/api/whatsapp", tags=["whatsapp"]) |
|
|
|
|
| @router.get("/webhook") |
| async def verify_webhook( |
| hub_mode: str = Query(alias="hub.mode", default=""), |
| hub_verify_token: str = Query( |
| alias="hub.verify_token", default="" |
| ), |
| hub_challenge: str = Query( |
| alias="hub.challenge", default="" |
| ), |
| ) -> Any: |
| """Meta webhook verification handshake.""" |
| settings = get_settings() |
| if ( |
| hub_mode == "subscribe" |
| and hub_verify_token == settings.whatsapp_verify_token |
| ): |
| return int(hub_challenge) if hub_challenge.isdigit() else hub_challenge |
| raise HTTPException(status_code=403, detail="Verification failed") |
|
|
|
|
| @router.post("/webhook") |
| async def receive_webhook( |
| request: Request, |
| db: AsyncSession = Depends(get_db), |
| ) -> dict[str, str]: |
| """Receive inbound WhatsApp messages.""" |
| payload = await request.json() |
| settings = get_settings() |
|
|
| try: |
| entries = payload.get("entry", []) |
| for entry in entries: |
| for change in entry.get("changes", []): |
| value = change.get("value", {}) |
| messages = value.get("messages", []) |
| for msg in messages: |
| if msg.get("type") != "text": |
| continue |
| wa_from = msg.get("from", "") |
| text = msg.get("text", {}).get("body", "") |
| if not wa_from or not text: |
| continue |
| await _process_incoming( |
| db, |
| wa_from=wa_from, |
| text=text, |
| settings=settings, |
| ) |
| except Exception as exc: |
| logger.exception("WhatsApp webhook error: %s", exc) |
|
|
| return {"status": "ok"} |
|
|
|
|
| async def _process_incoming( |
| db: AsyncSession, |
| *, |
| wa_from: str, |
| text: str, |
| settings: Any, |
| ) -> None: |
| result = await db.execute( |
| select(Conversation).where( |
| Conversation.channel == "whatsapp", |
| Conversation.external_id == wa_from, |
| Conversation.status == "active", |
| ) |
| ) |
| conv = result.scalar_one_or_none() |
| if conv is None: |
| conv, _greeting = await start_conversation( |
| db, |
| channel="whatsapp", |
| ui_language="en", |
| external_id=wa_from, |
| ) |
|
|
| data = await handle_user_message( |
| db, |
| conversation_id=conv.id, |
| message=text, |
| ui_language=conv.language or "en", |
| ) |
| await send_whatsapp_text(wa_from, data["reply"]) |
|
|
|
|
| async def send_whatsapp_text(to: str, body: str) -> None: |
| """Send a WhatsApp text message if credentials exist.""" |
| settings = get_settings() |
| if not settings.has_whatsapp: |
| logger.info( |
| "WhatsApp demo send to=%s body=%s", to, body[:120] |
| ) |
| return |
|
|
| url = ( |
| f"https://graph.facebook.com/v20.0/" |
| f"{settings.whatsapp_phone_number_id}/messages" |
| ) |
| headers = { |
| "Authorization": f"Bearer {settings.whatsapp_token}", |
| "Content-Type": "application/json", |
| } |
| payload = { |
| "messaging_product": "whatsapp", |
| "to": to, |
| "type": "text", |
| "text": {"body": body[:4000]}, |
| } |
| async with httpx.AsyncClient(timeout=20) as client: |
| response = await client.post( |
| url, headers=headers, json=payload |
| ) |
| if response.status_code >= 400: |
| logger.error( |
| "WhatsApp send failed: %s %s", |
| response.status_code, |
| response.text, |
| ) |
|
|
