""" WhatsApp Bot - Powered by Hugging Face 💬 A FastAPI webhook handler for WhatsApp Cloud API with AI-powered responses. """ import os import json import asyncio import logging from datetime import datetime, timezone from collections import deque import httpx from contextlib import asynccontextmanager from fastapi import FastAPI, Request, Response, HTTPException from fastapi.responses import PlainTextResponse, HTMLResponse # ─── Logging ──────────────────────────────────────────────────────────────── logging.basicConfig(level=logging.INFO) logger = logging.getLogger("whatsapp-bot") # ─── App ──────────────────────────────────────────────────────────────────── # ─── Lifespan (startup) ──────────────────────────────────────────────────── @asynccontextmanager async def lifespan(app: FastAPI): logger.info("🚀 WhatsApp Bot starting up...") logger.info(f" Verify Token: {'✅ Set' if VERIFY_TOKEN else '❌ Not set'}") logger.info(f" WhatsApp Token: {'✅ Set' if WHATSAPP_TOKEN else '❌ Not set'}") logger.info(f" Phone Number ID: {'✅ Set' if PHONE_NUMBER_ID else '❌ Not set'}") logger.info(f" HF Token: {'✅ Set' if HF_TOKEN else '❌ Not set'}") logger.info(f" Webhook URL: https://miftahulkhairim-whatsapp-bot.hf.space/webhook") logger.info("✅ Bot is ready to receive messages!") yield app = FastAPI(title="WhatsApp Bot", version="1.0.0", lifespan=lifespan) # ─── Configuration (loaded from HF Space Secrets) ────────────────────────── VERIFY_TOKEN = os.environ.get("WHATSAPP_VERIFY_TOKEN", "") WHATSAPP_TOKEN = os.environ.get("WHATSAPP_TOKEN", "") PHONE_NUMBER_ID = os.environ.get("PHONE_NUMBER_ID", "") HF_TOKEN = os.environ.get("HF_TOKEN", "") WHATSAPP_API_URL = f"https://graph.facebook.com/v21.0/{PHONE_NUMBER_ID}/messages" # ─── In-memory message log (last 100 messages) ───────────────────────────── message_log = deque(maxlen=100) bot_stats = {"messages_received": 0, "messages_sent": 0, "errors": 0, "started_at": datetime.now(timezone.utc).isoformat()} # ═══════════════════════════════════════════════════════════════════════════ # WEBHOOK ENDPOINTS # ═══════════════════════════════════════════════════════════════════════════ @app.get("/webhook") async def verify_webhook(request: Request): """ Meta webhook verification endpoint. When you set the webhook URL in Meta Developer Portal, it sends a GET request with hub.mode, hub.verify_token, and hub.challenge to verify ownership. """ params = dict(request.query_params) mode = params.get("hub.mode") token = params.get("hub.verify_token") challenge = params.get("hub.challenge") logger.info(f"Webhook verification: mode={mode}, token={'***' if token else 'none'}") if mode == "subscribe" and token == VERIFY_TOKEN: logger.info("✅ Webhook verified successfully!") return PlainTextResponse(content=challenge, status_code=200) logger.warning("❌ Webhook verification failed") raise HTTPException(status_code=403, detail="Verification failed") @app.post("/webhook") async def receive_message(request: Request): """ Receive incoming WhatsApp messages via webhook. CRITICAL: Always return 200 quickly - WhatsApp retries on timeout. Heavy processing (LLM calls) runs in background. """ try: body = await request.json() except Exception: return Response(status_code=200) # Navigate the WhatsApp Cloud API payload structure try: entry = body["entry"][0] changes = entry["changes"][0] value = changes["value"] # Check if this is a message event (not a status update) if "messages" not in value: return Response(status_code=200) message = value["messages"][0] msg_type = message["type"] from_number = message["from"] # Get contact name if available contact_name = "Unknown" if "contacts" in value and value["contacts"]: contact_name = value["contacts"][0].get("profile", {}).get("name", "Unknown") if msg_type == "text": text_body = message["text"]["body"] else: text_body = f"[{msg_type} message - not supported yet]" logger.info(f"📩 Message from {contact_name} ({from_number}): {text_body[:100]}") bot_stats["messages_received"] += 1 message_log.append({ "timestamp": datetime.now(timezone.utc).isoformat(), "from": from_number, "name": contact_name, "message": text_body, "direction": "incoming" }) # Process text messages - run LLM in background to respond quickly if msg_type == "text": asyncio.create_task(process_and_reply(from_number, contact_name, text_body)) except (KeyError, IndexError) as e: logger.debug(f"Non-message event received: {e}") return Response(status_code=200) # ═══════════════════════════════════════════════════════════════════════════ # MESSAGE PROCESSING # ═══════════════════════════════════════════════════════════════════════════ async def process_and_reply(to_number: str, contact_name: str, user_message: str): """Generate AI reply and send it back via WhatsApp.""" try: reply = await generate_reply(user_message, contact_name) await send_whatsapp_message(to_number, reply) bot_stats["messages_sent"] += 1 message_log.append({ "timestamp": datetime.now(timezone.utc).isoformat(), "from": "bot", "name": "Bot", "message": reply[:200], "direction": "outgoing" }) logger.info(f"📤 Reply sent to {to_number}: {reply[:100]}") except Exception as e: bot_stats["errors"] += 1 logger.error(f"❌ Error processing message: {e}") # Send a fallback message try: await send_whatsapp_message( to_number, "Sorry, I'm having trouble processing your message right now. Please try again later! 🙏" ) except Exception: logger.error("❌ Failed to send fallback message") async def generate_reply(user_message: str, contact_name: str = "User") -> str: """ Generate an AI response using Hugging Face Inference API. Uses Meta Llama model for high-quality conversational responses. """ if not HF_TOKEN: return "⚠️ Bot is not fully configured yet. Please set up the HF_TOKEN in Space secrets." try: from huggingface_hub import InferenceClient client = InferenceClient( provider="hf-inference", api_key=HF_TOKEN, ) system_prompt = ( "You are a helpful, friendly WhatsApp assistant. " "Keep responses concise and conversational (under 300 words). " "Use emojis occasionally to be friendly. " "If asked about your capabilities, mention you're an AI assistant " "powered by Hugging Face." ) messages = [ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_message} ] response = client.chat_completion( model="meta-llama/Llama-3.1-8B-Instruct", messages=messages, max_tokens=512, temperature=0.7, ) return response.choices[0].message.content except Exception as e: logger.error(f"LLM error: {e}") return f"I received your message but had trouble generating a response. Error: {str(e)[:100]}" async def send_whatsapp_message(to: str, text: str): """Send a text message via WhatsApp Cloud API.""" if not WHATSAPP_TOKEN or not PHONE_NUMBER_ID: logger.warning("WhatsApp credentials not configured - skipping send") return headers = { "Authorization": f"Bearer {WHATSAPP_TOKEN}", "Content-Type": "application/json", } # WhatsApp has a ~4096 char limit per message; split if needed chunks = [text[i:i+4000] for i in range(0, len(text), 4000)] async with httpx.AsyncClient(timeout=30.0) as client: for chunk in chunks: payload = { "messaging_product": "whatsapp", "recipient_type": "individual", "to": to, "type": "text", "text": {"preview_url": False, "body": chunk}, } resp = await client.post(WHATSAPP_API_URL, json=payload, headers=headers) if resp.status_code != 200: logger.error(f"WhatsApp API error: {resp.status_code} - {resp.text}") resp.raise_for_status() # ═══════════════════════════════════════════════════════════════════════════ # DASHBOARD UI # ═══════════════════════════════════════════════════════════════════════════ DASHBOARD_HTML = """ WhatsApp Bot Dashboard

💬 WhatsApp Bot

AI-Powered Assistant on Hugging Face

🟢 Bot Status

{messages_received}
Messages Received
{messages_sent}
Replies Sent
{errors}
Errors

⚙️ Configuration

Webhook URL https://miftahulkhairim-whatsapp-bot.hf.space/webhook
WhatsApp Verify Token {verify_label}
WhatsApp Access Token {wa_token_label}
Phone Number ID {phone_label}
HF Token (AI Inference) {hf_token_label}
AI Model meta-llama/Llama-3.1-8B-Instruct
Running Since {started_at}

📨 Recent Messages

{messages_html}

📖 Quick Setup Guide

1
Go to Meta Developer Portal → Create App → Add WhatsApp product
2
In WhatsApp → Configuration → Set webhook URL to: https://miftahulkhairim-whatsapp-bot.hf.space/webhook
3
Add Space Secrets in Settings: WHATSAPP_VERIFY_TOKEN, WHATSAPP_TOKEN, PHONE_NUMBER_ID, HF_TOKEN
4
Subscribe to messages webhook field in Meta Developer Portal
5
Send a WhatsApp message to your test number — the bot will reply with AI! 🎉
""" @app.get("/", response_class=HTMLResponse) async def dashboard(): """Render the bot dashboard with live stats.""" def secret_status(val, name): if val: return "status-ok", f"✅ Configured" return "status-warn", f"⚠️ Not set — add {name} in Space Secrets" verify_status, verify_label = secret_status(VERIFY_TOKEN, "WHATSAPP_VERIFY_TOKEN") wa_token_status, wa_token_label = secret_status(WHATSAPP_TOKEN, "WHATSAPP_TOKEN") phone_status, phone_label = secret_status(PHONE_NUMBER_ID, "PHONE_NUMBER_ID") hf_token_status, hf_token_label = secret_status(HF_TOKEN, "HF_TOKEN") # Build messages HTML if message_log: msgs_html = "" for msg in reversed(message_log): direction = "msg-incoming" if msg["direction"] == "incoming" else "msg-outgoing" icon = "📩" if msg["direction"] == "incoming" else "🤖" name = msg.get("name", "Unknown") text = msg["message"][:300] time_str = msg["timestamp"][:19].replace("T", " ") msgs_html += f'''
{icon}
{text}
{name} • {time_str} UTC
''' else: msgs_html = '
No messages yet. Send a WhatsApp message to get started! 📱
' html = DASHBOARD_HTML.format( messages_received=bot_stats["messages_received"], messages_sent=bot_stats["messages_sent"], errors=bot_stats["errors"], verify_status=verify_status, verify_label=verify_label, wa_token_status=wa_token_status, wa_token_label=wa_token_label, phone_status=phone_status, phone_label=phone_label, hf_token_status=hf_token_status, hf_token_label=hf_token_label, started_at=bot_stats["started_at"][:19].replace("T", " ") + " UTC", messages_html=msgs_html, ) return HTMLResponse(content=html) # ─── Health check API ────────────────────────────────────────────────────── @app.get("/health") async def health(): """Health check endpoint.""" return { "status": "running", "bot": "WhatsApp Bot", "stats": bot_stats, "config": { "verify_token_set": bool(VERIFY_TOKEN), "whatsapp_token_set": bool(WHATSAPP_TOKEN), "phone_number_id_set": bool(PHONE_NUMBER_ID), "hf_token_set": bool(HF_TOKEN), } }