Spaces:
Sleeping
Sleeping
File size: 5,813 Bytes
9c6a172 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 | """
WhatsApp Business API Integration
===================================
Handles inbound webhook messages and outbound replies via the
Meta WhatsApp Cloud API (v18+).
In the POC demo, all send/receive calls are stubbed and logged.
In production, set:
WHATSAPP_TOKEN = Bearer token from Meta Business Portal
WHATSAPP_PHONE_ID = Phone Number ID
WHATSAPP_VERIFY_TOKEN = Webhook verification token
"""
import os
import json
import logging
import hashlib
from datetime import datetime
from typing import Optional
logger = logging.getLogger(__name__)
WHATSAPP_API_URL = "https://graph.facebook.com/v18.0/{phone_id}/messages"
class WhatsAppClient:
def __init__(self):
self.token = os.getenv("WHATSAPP_TOKEN", "DEMO_TOKEN")
self.phone_id = os.getenv("WHATSAPP_PHONE_ID", "DEMO_PHONE_ID")
self.verify = os.getenv("WHATSAPP_VERIFY_TOKEN", "plotweaver_verify")
self._demo = self.token == "DEMO_TOKEN"
# ── Outbound ─────────────────────────────────────────────────────────────
def send_text(self, to: str, body: str) -> dict:
"""Send a plain text message."""
payload = {
"messaging_product": "whatsapp",
"recipient_type": "individual",
"to": to,
"type": "text",
"text": {"preview_url": False, "body": body},
}
return self._post(payload)
def send_audio(self, to: str, audio_url: str) -> dict:
"""Send a voice note (ogg/opus recommended by Meta)."""
payload = {
"messaging_product": "whatsapp",
"recipient_type": "individual",
"to": to,
"type": "audio",
"audio": {"link": audio_url},
}
return self._post(payload)
def send_interactive_buttons(self, to: str, body: str,
buttons: list[dict]) -> dict:
"""
Quick-reply buttons for disambiguation.
buttons = [{"id": "yes", "title": "Yes ✓"}, {"id": "no", "title": "No ✗"}]
"""
payload = {
"messaging_product": "whatsapp",
"to": to,
"type": "interactive",
"interactive": {
"type": "button",
"body": {"text": body},
"action": {
"buttons": [
{"type": "reply", "reply": btn} for btn in buttons
]
},
},
}
return self._post(payload)
# ── Inbound webhook ──────────────────────────────────────────────────────
def verify_webhook(self, mode: str, token: str, challenge: str) -> Optional[str]:
"""GET verification handshake from Meta."""
if mode == "subscribe" and token == self.verify:
logger.info("WhatsApp webhook verified.")
return challenge
return None
def parse_inbound(self, raw_body: dict) -> Optional[dict]:
"""
Extract relevant fields from inbound webhook payload.
Returns normalized event dict or None if not a user message.
"""
try:
entry = raw_body["entry"][0]
changes = entry["changes"][0]["value"]
message = changes["messages"][0]
contact = changes["contacts"][0]
event = {
"from": message["from"],
"name": contact["profile"]["name"],
"timestamp": datetime.fromtimestamp(int(message["timestamp"])).isoformat(),
"msg_id": message["id"],
"type": message["type"],
}
if message["type"] == "text":
event["text"] = message["text"]["body"]
elif message["type"] == "audio":
event["audio_id"] = message["audio"]["id"]
event["mime"] = message["audio"].get("mime_type", "audio/ogg")
elif message["type"] == "interactive":
event["button_id"] = message["interactive"]["button_reply"]["id"]
return event
except (KeyError, IndexError) as e:
logger.warning(f"Could not parse WhatsApp payload: {e}")
return None
def download_media(self, media_id: str) -> Optional[bytes]:
"""Fetch audio bytes for voice messages."""
if self._demo:
logger.info(f"[DEMO] Would download media: {media_id}")
return b"" # Return empty bytes in demo mode
# Production: GET https://graph.facebook.com/v18.0/{media_id}
# then fetch the returned URL with Bearer auth
raise NotImplementedError("Set WHATSAPP_TOKEN in production.")
# ── Internals ─────────────────────────────────────────────────────────────
def _post(self, payload: dict) -> dict:
if self._demo:
logger.info(f"[DEMO] WhatsApp → {payload['to']}: "
f"{json.dumps(payload)[:120]} …")
return {"status": "demo_ok",
"message_id": "demo_" + hashlib.md5(
json.dumps(payload).encode()).hexdigest()[:8]}
import requests
url = WHATSAPP_API_URL.format(phone_id=self.phone_id)
headers = {"Authorization": f"Bearer {self.token}",
"Content-Type": "application/json"}
r = requests.post(url, headers=headers, json=payload, timeout=10)
r.raise_for_status()
return r.json()
|