File size: 3,550 Bytes
cafdb78 | 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 | import os, json
from fastapi import FastAPI, Request, HTTPException, Query
from fastapi.responses import PlainTextResponse, JSONResponse
import httpx
import base64
app = FastAPI()
VERIFY_TOKEN = os.getenv("VERIFY_TOKEN", "") # set in HF Secrets
PHONE_NUMBER_ID = os.getenv("PHONE_NUMBER_ID", "")
WHATSAPP_API_TOKEN = os.getenv("WHATSAPP_API_TOKEN", "")
@app.get("/webhook") # verification
def verify(
hub_mode: str = Query(None, alias="hub.mode"),
hub_token: str = Query(None, alias="hub.verify_token"),
hub_challenge: str = Query(None, alias="hub.challenge")
):
if hub_mode == "subscribe" and hub_token == VERIFY_TOKEN:
return PlainTextResponse(hub_challenge)
raise HTTPException(status_code=403, detail="Invalid token")
def send_whatsapp_message(wa_id, reply):
url = f"https://graph.facebook.com/v23.0/{PHONE_NUMBER_ID}/messages"
headers = {
"Authorization": f"Bearer {WHATSAPP_API_TOKEN}",
"Content-Type": "application/json"
}
payload = {
"messaging_product": "whatsapp",
"to": wa_id,
"type": "text",
"text": {"body": reply}
}
httpx.post(url, headers=headers, json=payload)
@app.post("/webhook") # incoming WhatsApp updates
async def receive_update(req: Request):
data = await req.json()
try:
change = data["entry"][0]["changes"][0]["value"]
contacts = change.get("contacts")
messages = change.get("messages")
if not contacts or not messages:
# Missing expected payload—ignore
print("Ignoring update, no contacts or messages field")
return JSONResponse({"status": "ignored", "reason": "no contacts/messages"})
wa_id = contacts[0]["wa_id"]
msg = messages[0]
except (KeyError, IndexError, TypeError) as e:
print("Malformed webhook payload:", e)
return JSONResponse({"status": "ignored", "reason": "malformed payload"})
# IMAGE: if user sent an image, convert to Base64 preview
if msg["type"] == "image":
media_id = msg["image"]["id"]
# 1) get media URL
media_resp = httpx.get(
f"https://graph.facebook.com/v23.0/{media_id}",
headers={"Authorization": f"Bearer {WHATSAPP_API_TOKEN}"}
)
media_resp.raise_for_status()
media_url = media_resp.json()["url"]
# 2) download bytes with auth
media_bytes = httpx.get(
media_url,
headers={"Authorization": f"Bearer {WHATSAPP_API_TOKEN}"}
).content
# 3) encode & preview
preview = base64.b64encode(media_bytes).decode("utf-8")[:10]
# send back preview
send_whatsapp_message(wa_id, f"Base64: {preview}")
return JSONResponse({"status": "image_preview_sent", "preview": preview})
# TEXT: simple echo
if msg["type"] == "text":
incoming = msg["text"]["body"]
reply = f"You said: {incoming}"
send_whatsapp_message(wa_id, reply)
return JSONResponse({"status": "replied"})
# OTHER: ignore
return JSONResponse({"status": "ignored", "type": msg["type"]})
# health-check endpoint for token
@app.post("/send")
async def check_token():
token = os.getenv("WHATSAPP_API_TOKEN", "")
if token:
return {"token_loaded": True}
raise HTTPException(status_code=400, detail="Token not found")
@app.get("/ping") # health check
def ping():
return {"pong": True} |