import hmac import hashlib import os import httpx import datetime import asyncio from typing import Dict, List, Optional, Any, Union, cast from fastapi import APIRouter, Request, HTTPException, BackgroundTasks from fastapi.responses import PlainTextResponse from supabase import create_client router = APIRouter() # ── Dynamic Secret Loader ──────────────────────────────────────────────────── def get_secret(key: str) -> str: """Dynamically fetches secrets to prevent Hugging Face soft-reboot caching.""" return os.environ.get(key, "").strip() # ── Webhook Verification (GET) ──────────────────────────────────────────────── @router.get("/webhook/whatsapp") async def verify_webhook(request: Request): params = request.query_params if params.get("hub.mode") == "subscribe" and params.get("hub.verify_token") == get_secret("WHATSAPP_VERIFY_TOKEN"): print("✅ WhatsApp webhook verified") return PlainTextResponse(content=params.get("hub.challenge")) raise HTTPException(status_code=403, detail="Verification failed") # ── Incoming Message Handler (POST) ────────────────────────────────────────── @router.post("/webhook/whatsapp") async def receive_message(request: Request, background_tasks: BackgroundTasks): signature = request.headers.get("X-Hub-Signature-256", "") body = await request.body() app_secret = get_secret("WHATSAPP_APP_SECRET") expected = "sha256=" + hmac.new(app_secret.encode(), body, hashlib.sha256).hexdigest() if not hmac.compare_digest(signature, expected): raise HTTPException(status_code=403, detail="Invalid signature") data = await request.json() try: entry = data["entry"][0]["changes"][0]["value"] if "messages" not in entry: return {"status": "ok"} message = entry["messages"][0] from_number = message["from"] msg_type = message["type"] if msg_type == "text": query_text = message["text"]["body"] background_tasks.add_task(process_and_reply, from_number, query_text, "english") elif msg_type == "audio": audio_id = message["audio"]["id"] background_tasks.add_task(process_voice_and_reply, from_number, audio_id) except (KeyError, IndexError) as e: print(f"Webhook parsing error: {e}") return {"status": "ok"} # ── Outgoing Message Pipeline ──────────────────────────────────────────────── async def send_whatsapp_message(to: str, text: str): if len(text) > 4096: text = text[:4000] + "\n\n_[Reply truncated. Visit govbridge.in for full answer]_" clean_phone_id = get_secret("WHATSAPP_PHONE_NUMBER_ID") clean_token = get_secret("WHATSAPP_TOKEN") if not clean_phone_id or not clean_token: print("🚨 FATAL ERROR: API Keys are completely empty. Hugging Face failed to load secrets!") return api_url = f"https://graph.facebook.com/v25.0/{clean_phone_id}/messages" payload = { "messaging_product": "whatsapp", "to": to, "type": "text", "text": {"body": text} } headers = { "Authorization": f"Bearer {clean_token}", "Content-Type": "application/json" } try: # Primary Attempt: Fast Asynchronous Request async with httpx.AsyncClient(timeout=15.0) as client: resp = await client.post(api_url, json=payload, headers=headers) print(f"📤 Meta Send Response (ASYNC): {resp.status_code} - {resp.text}") except httpx.ConnectTimeout: print("⚠️ ConnectTimeout triggered! Falling back to synchronous network request...") # Backup Attempt: Bypasses async IPv6 network glitches try: resp = await asyncio.to_thread( lambda: httpx.Client().post(api_url, json=payload, headers=headers, timeout=15.0) ) print(f"📤 Meta Send Response (SYNC FALLBACK): {resp.status_code} - {resp.text}") except Exception as fallback_e: print(f"❌ Fallback completely failed: {fallback_e}") except Exception as e: print(f"❌ Unknown send error: {e}") def check_and_increment_quota() -> bool: from config import settings current_month = datetime.datetime.utcnow().strftime("%Y-%m") supabase_url = settings.SUPABASE_URL supabase_key = settings.SUPABASE_KEY if not supabase_url or not supabase_key: print("⚠️ Supabase keys missing, skipping quota check.") return True supabase = create_client(supabase_url, supabase_key) res = supabase.table("whatsapp_quota").select("conversation_count").eq("month", current_month).execute() if res.data and len(res.data) > 0: row = cast(Dict[str, Any], res.data[0]) count = int(row.get("conversation_count") or 0) if count >= 950: return False supabase.table("whatsapp_quota").update({"conversation_count": count + 1}).eq("month", current_month).execute() else: supabase.table("whatsapp_quota").insert({"month": current_month, "conversation_count": 1}).execute() return True # ── AI Brain Connection ────────────────────────────────────────────────────── async def process_and_reply(from_number: str, query: str, language: str): if not check_and_increment_quota(): await send_whatsapp_message(from_number, "⚠️ Our WhatsApp service has reached its monthly limit. Please visit govbridge.in for full access.") return # --- UPGRADED SMART GREETING INTERCEPTOR --- clean_query = query.strip().lower() # A list of common conversational triggers small_talk = [ "hi", "hello", "hey", "namaste", "pranam", "help", "who are you", "who are you?", "what are you", "what are you?" ] if len(clean_query) < 3 or clean_query in small_talk: greeting_msg = "🏛️ *GovBridge India*\n\nNamaste! 🙏 I am GovBridge, your AI assistant for Indian government schemes.\n\nAsk me a question like:\n_\"What are the eligibility criteria for PM Awas Yojana?\"_" await send_whatsapp_message(from_number, greeting_msg) return # ------------------------------------------- print(f"🧠 Querying GovBridge AI: '{query}'") try: # Internal API call to your existing RAG pipeline async with httpx.AsyncClient(timeout=60.0) as client: api_url = "http://127.0.0.1:7860/api/rag/query" payload = {"question": query, "language": language} resp = await client.post(api_url, json=payload) if resp.status_code == 200: answer = resp.text # Extract sources from the headers you built in api.py sources_raw = resp.headers.get("X-Sources", "") sources = [s for s in sources_raw.split("|") if s.strip()] else: answer = "My AI brain is temporarily offline for upgrades. Please try again soon!" sources = [] except Exception as e: print(f"❌ Internal AI Connection Error: {e}") answer = "I am having a little trouble thinking right now. Please try again in a minute!" sources = [] sources_text = "" if sources: unique_sources = list(dict.fromkeys(sources)) sources_text = "\n\n📄 *Sources:*\n" + "\n".join([f"• {s}" for s in unique_sources[:3]]) reply = f"🏛️ *GovBridge India*\n\n{answer}{sources_text}" await send_whatsapp_message(from_number, reply) async def process_voice_and_reply(from_number: str, audio_id: str): clean_token = get_secret("WHATSAPP_TOKEN") groq_key = get_secret("GROQ_API_KEY") async with httpx.AsyncClient() as client: meta_resp = await client.get( f"https://graph.facebook.com/v25.0/{audio_id}", headers={"Authorization": f"Bearer {clean_token}"} ) if meta_resp.status_code != 200: print(f"❌ Failed to get audio metadata: {meta_resp.text}") return audio_url = meta_resp.json()["url"] audio_resp = await client.get(audio_url, headers={"Authorization": f"Bearer {clean_token}"}) audio_bytes = audio_resp.content transcription_resp = await client.post( "https://api.groq.com/openai/v1/audio/transcriptions", headers={"Authorization": f"Bearer {groq_key}"}, files={"file": ("audio.ogg", audio_bytes, "audio/ogg")}, data={"model": "whisper-large-v3-turbo", "language": "hi"} ) transcript = transcription_resp.json().get("text", "") if transcript: await process_and_reply(from_number, transcript, "hindi") else: await send_whatsapp_message(from_number, "🙏 Sorry, I couldn't understand the voice message. Please type your question.")