feat(sprint-16): implement Coverage Intelligence Dashboard and WhatsApp expiry alerts

whatsapp/expiry_alerts.py

@@ -0,0 +1,104 @@
+import os
+import sys
+import httpx
+from datetime import date, timedelta
+from typing import List, Dict, Any, cast
+from supabase import create_client, Client
+
+sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+from config import settings
+
+SUPABASE_URL = settings.SUPABASE_URL
+SUPABASE_KEY = settings.SUPABASE_KEY
+# META_API_TOKEN = os.environ.get("META_API_TOKEN") # In a real scenario, this is the WhatsApp API token
+
+def generate_alert_payload(scheme: Dict[str, Any], phone_number: str) -> Dict[str, Any]:
+    """
+    Format the WhatsApp API payload for the expiry alert.
+    """
+    title = scheme.get("title", "Government Scheme")
+    deadline = scheme.get("deadline_date", "Unknown Date")
+    url = scheme.get("source_url", "https://myscheme.gov.in")
+    
+    return {
+        "messaging_product": "whatsapp",
+        "to": phone_number,
+        "type": "template",
+        "template": {
+            "name": "scheme_expiry_alert",
+            "language": {
+                "code": "en_US"
+            },
+            "components": [
+                {
+                    "type": "body",
+                    "parameters": [
+                        {
+                            "type": "text",
+                            "text": title
+                        },
+                        {
+                            "type": "text",
+                            "text": deadline
+                        }
+                    ]
+                },
+                {
+                    "type": "button",
+                    "sub_type": "url",
+                    "index": "0",
+                    "parameters": [
+                        {
+                            "type": "text",
+                            "text": url
+                        }
+                    ]
+                }
+            ]
+        }
+    }
+
+def main() -> None:
+    if not SUPABASE_URL or not SUPABASE_KEY:
+        print("❌ Missing SUPABASE_URL or SUPABASE_KEY")
+        return
+
+    supabase: Client = create_client(SUPABASE_URL, SUPABASE_KEY)
+    today = date.today()
+    alert_date = (today + timedelta(days=7)).isoformat()
+    today_str = today.isoformat()
+
+    print(f"📡 Scanning for schemes expiring between {today_str} and {alert_date}...")
+
+    try:
+        response = supabase.table("schemes") \
+            .select("id, title, deadline_date, source_url") \
+            .lte("deadline_date", alert_date) \
+            .gte("deadline_date", today_str) \
+            .eq("is_active", True) \
+            .execute()
+            
+        schemes = cast(List[Dict[str, Any]], response.data or [])
+
+        if not schemes:
+            print("✅ No schemes expiring within the next 7 days.")
+            return
+
+        print(f"⚠️ Found {len(schemes)} schemes approaching their deadline.")
+        
+        # Simulate fetching subscribed users (in a full setup, this comes from whatsapp_subscriptions table)
+        mock_subscribers = ["+919876543210"] 
+
+        for scheme in schemes:
+            print(f"   Generating alerts for: {scheme.get('title')} (Deadline: {scheme.get('deadline_date')})")
+            
+            for user in mock_subscribers:
+                payload = generate_alert_payload(scheme, user)
+                # In production, this would be an httpx.post to graph.facebook.com/v17.0/.../messages
+                print(f"      [DRY RUN] Sent WhatsApp payload to {user} for scheme ID: {scheme.get('id')}")
+
+    except Exception as e:
+        print(f"❌ Alert generation failed: {e}")
+
+if __name__ == "__main__":
+    main()