onboarding integration

components/gateways/headlines_to_wa.py

@@ -4,7 +4,7 @@ import logging
 import os
 import re
 import time
-from typing import Dict, Optional
+from typing import Dict, List, Optional
 
 import redis
 import requests
@@ -233,8 +233,8 @@ def _post_to_gupshup(payload: Dict[str, object], action: str) -> Dict[str, objec
 # -----------------------------
 # Business logic
 # -----------------------------
-def fetch_cached_headlines() -> str:
-    """Fetch and format the daily headlines from Redis into a WhatsApp-friendly message."""
+def fetch_cached_headlines(selected_topics: Optional[List[str]] = None) -> str:
+    """Fetch and format the daily headlines into a WhatsApp-friendly message."""
     try:
         raw = redis_client.get("daily_news_feed_cache")
         if not raw:
@@ -252,7 +252,27 @@ def fetch_cached_headlines() -> str:
     message_parts.append(f"Nuse Daily - {formatted_date}")
     message_parts.append("")
 
-    for topic_key, display_info in TOPIC_DISPLAY_MAP.items():
+    normalized_topics: List[str] = []
+    if selected_topics:
+        seen = set()
+        for topic in selected_topics:
+            slug = (topic or "").strip().lower()
+            if slug in TOPIC_DISPLAY_MAP and slug not in seen:
+                normalized_topics.append(slug)
+                seen.add(slug)
+
+    topics_to_render = normalized_topics or list(TOPIC_DISPLAY_MAP.keys())
+
+    if selected_topics and not normalized_topics:
+        message_parts.append(
+            "(I couldn’t match those preferences yet, so here’s the full digest.)"
+        )
+        message_parts.append("")
+
+    for topic_key in topics_to_render:
+        display_info = TOPIC_DISPLAY_MAP.get(topic_key)
+        if not display_info:
+            continue
         stories_for_topic = data.get(topic_key)
         if not stories_for_topic:
             continue
@@ -285,28 +305,12 @@ def send_to_whatsapp(message_text: str, destination_number: str) -> Dict[str, ob
     return _post_to_gupshup(payload, "standard text WhatsApp message")
 
 
-def send_seen_receipt(
-    destination_number: str,
-    message_id: Optional[str],
-    provider: str = "gupshup",
-) -> Dict[str, object]:
-    """
-    Mark an inbound message as read.
-
-    provider:
-      - "meta_cloud": message originated via Meta WhatsApp Cloud (your Cloud webhook & phone number)
-      - "gupshup": message originated via Gupshup (default). In this case we SKIP the Cloud call.
-
-    Returns structured result with 'status' in {"success","failed","skipped"}.
-    """
+def send_seen_receipt(destination_number: str, message_id: Optional[str]) -> Dict[str, object]:
+    """Mark an inbound message as read via the WhatsApp Cloud API."""
     if not message_id:
         logging.debug("Skipping seen receipt for %s due to missing message id", destination_number)
         return {"status": "skipped", "reason": "missing_message_id"}
 
-    if (provider or "").lower() != "meta_cloud":
-        logging.debug("Skipping seen: provider is %s (not meta_cloud)", provider)
-        return {"status": "skipped", "reason": "non_meta_origin"}
-
     logging.info(
         "Seen receipt intent (Meta Cloud): msg_id=%s | phone_id=%s | token_prefix=%s",
         message_id,
@@ -329,6 +333,7 @@ def send_seen_receipt(
         "messaging_product": "whatsapp",
         "status": "read",
         "message_id": message_id,
+        "typing_indicator": {"type": "text"},
     }
 
     try:

components/handlers/wa/__init__.py

@@ -7,6 +7,11 @@ from .shortcuts import (
     handle_small_talk,
     handle_unsubscribe,
 )
+from .onboarding import (
+    initiate_onboarding,
+    handle_onboarding_preferences_text,
+    handle_onboarding_preferences_flow_submission,
+)
 from .chat import handle_chat_question
 from .research import handle_general_research
 
@@ -20,4 +25,7 @@ __all__ = [
     "handle_small_talk",
     "handle_chat_question",
     "handle_general_research",
+    "initiate_onboarding",
+    "handle_onboarding_preferences_text",
+    "handle_onboarding_preferences_flow_submission",
 ]

components/handlers/wa/onboarding.py

@@ -0,0 +1,258 @@
+import logging
+import os
+import re
+from typing import List, Optional, Dict, Any
+
+import requests
+from fastapi.responses import JSONResponse
+
+from components.gateways.headlines_to_wa import TOPIC_DISPLAY_MAP, fetch_cached_headlines
+from storage.users import set_onboarding_stage, set_preferences
+from .common import safe_send
+
+logger = logging.getLogger(__name__)
+
+# ----------------------------------
+# Configuration for WhatsApp Flows
+# ----------------------------------
+META_GRAPH_VERSION = os.environ.get("META_GRAPH_VERSION", "v23.0").strip()
+META_PHONE_NUMBER_ID = (os.environ.get("WHATSAPP_CLOUD_PHONE_NUMBER_ID") or os.environ.get("META_PHONE_NUMBER_ID") or "").strip()
+META_ACCESS_TOKEN = (os.environ.get("WHATSAPP_CLOUD_TOKEN") or os.environ.get("META_ACCESS_TOKEN") or "").strip()
+TOPICS_FLOW_ID = (os.environ.get("TOPICS_FLOW_ID") or "").strip()
+TOPICS_FLOW_MODE = (os.environ.get("TOPICS_FLOW_MODE") or "").strip()  # e.g., "draft" while testing
+
+# ----------------------------------
+# Topics config
+# ----------------------------------
+_TOPICS_ORDER = ["india", "world", "tech", "finance", "sports"]
+_TOPIC_SYNONYMS = {
+    "india": ["india"],
+    "world": ["world", "global"],
+    "tech": ["tech", "technology"],
+    "finance": [
+        "finance",
+        "business",
+        "markets",
+        "business and markets",
+        "business markets",
+        "business & markets",
+        "business&markets",
+    ],
+    "sports": ["sports", "sport"],
+}
+
+# ----------------------------------
+# Copy
+# ----------------------------------
+_WELCOME_MESSAGE_GENERIC = "Hi there! I’m Nuse — your personal guide to the news you care about."
+_PREFERENCES_PROMPT = (
+    "To get you started, pick the topics you’d like to follow:\n"
+    "• India\n• World\n• Technology\n• Business & Markets\n• Sports\n\n"
+    "Reply with one or more (comma or space separated)."
+)
+_INVALID_TOPICS_MESSAGE = (
+    "I didn’t recognise those topics. Please choose from India, World, Technology, Business & Markets, Sports."
+)
+
+
+# ----------------------------------
+# Name retrieval (best-effort)
+# ----------------------------------
+def _get_display_name(wa_id: str) -> Optional[str]:
+    """
+    Try to fetch a friendly display name for the user.
+    We attempt common helper functions if your storage layer exposes them.
+    Return None if unavailable.
+    """
+    try:
+        # Prefer a dedicated accessor if your storage layer provides it
+        from storage.users import get_profile  # type: ignore
+        profile = get_profile(wa_id) or {}
+        name = (profile.get("name") or profile.get("first_name") or profile.get("display_name") or "").strip()
+        return name or None
+    except Exception:
+        pass
+
+    try:
+        # Some projects expose a direct helper
+        from storage.users import get_name  # type: ignore
+        name = (get_name(wa_id) or "").strip()
+        return name or None
+    except Exception:
+        pass
+
+    # Nothing found
+    return None
+
+
+# ----------------------------------
+# WhatsApp Flow sending (Checkbox Group)
+# ----------------------------------
+def _flows_available() -> bool:
+    return bool(META_PHONE_NUMBER_ID and META_ACCESS_TOKEN and TOPICS_FLOW_ID)
+
+
+def _send_topics_flow(wa_id: str) -> Dict[str, Any]:
+    """
+    Send a WhatsApp Flow (interactive: flow) that contains a Checkbox Group
+    for multi-select topics. Returns a dict with status: success/failed.
+    """
+    if not _flows_available():
+        return {"status": "failed", "reason": "flows_unavailable"}
+
+    url = f"https://graph.facebook.com/{META_GRAPH_VERSION}/{META_PHONE_NUMBER_ID}/messages"
+    headers = {
+        "Authorization": f"Bearer {META_ACCESS_TOKEN}",
+        "Content-Type": "application/json",
+    }
+
+    # NOTE:
+    # - The Flow you create in WhatsApp Manager should have a Checkbox Group with options
+    #   whose IDs match our topic slugs (india, world, tech, finance, sports).
+    # - The "parameters" below assume a basic Flow. You may need to align keys (flow_message_version, etc.)
+    #   with your Flow definition and any data_exchange/init endpoints you configure.
+    interactive = {
+        "type": "flow",
+        "header": {"type": "text", "text": "Pick your news topics"},
+        "body": {"text": "Choose one or more topics and submit."},
+        "action": {
+            "name": "flow",
+            "parameters": {
+                "flow_id": TOPICS_FLOW_ID,
+                # In testing, many teams use "draft" mode; remove/omit for production
+                **({"mode": TOPICS_FLOW_MODE} if TOPICS_FLOW_MODE else {}),
+                # If you use data_exchange/INIT, you can add tokens or context here as needed.
+                "flow_message_version": "3"
+            },
+        },
+    }
+
+    payload = {
+        "messaging_product": "whatsapp",
+        "recipient_type": "individual",
+        "to": wa_id,
+        "type": "interactive",
+        "interactive": interactive,
+    }
+
+    try:
+        resp = requests.post(url, json=payload, headers=headers, timeout=10)
+        if resp.status_code >= 400:
+            logger.error("Flow send failed: %s | %s", resp.status_code, resp.text)
+            return {"status": "failed", "code": resp.status_code, "error": resp.text}
+        return {"status": "success", "details": resp.json()}
+    except requests.RequestException as e:
+        logger.error("Flow send error: %s", e)
+        return {"status": "failed", "error": str(e)}
+
+
+# ----------------------------------
+# Free-text parsing fallback
+# ----------------------------------
+def _normalise_preferences(message_text: str) -> List[str]:
+    cleaned = (message_text or "").lower().replace("&", " and ")
+    cleaned = re.sub(r"[^a-z\s,]", " ", cleaned)
+    cleaned = re.sub(r"\s+", " ", cleaned).strip()
+    if not cleaned:
+        return []
+
+    if " all " in f" {cleaned} ":
+        return list(_TOPICS_ORDER)
+
+    padded = f" {cleaned} "
+    selections: List[str] = []
+    for slug in _TOPICS_ORDER:
+        if slug in selections:
+            continue
+        for term in _TOPIC_SYNONYMS.get(slug, []):
+            term_norm = re.sub(r"\s+", " ", term.replace("&", " and ").lower()).strip()
+            if not term_norm:
+                continue
+            if f" {term_norm} " in padded:
+                selections.append(slug)
+                break
+    return selections
+
+
+# ----------------------------------
+# Public API
+# ----------------------------------
+def initiate_onboarding(wa_id: str) -> JSONResponse:
+    # Personalized welcome
+    name = _get_display_name(wa_id)
+    if name:
+        welcome = f"Hi {name}! I’m Nuse — your personal guide to the news you care about."
+    else:
+        welcome = _WELCOME_MESSAGE_GENERIC
+
+    safe_send(welcome, wa_id)
+
+    # Try sending the Flow (multi-select). If unavailable, fall back to text prompt.
+    flow_result = _send_topics_flow(wa_id)
+    if flow_result.get("status") == "success":
+        set_onboarding_stage(wa_id, "awaiting_preferences_flow")
+        return JSONResponse(status_code=200, content={"status": "onboarding_started", "mode": "flow"})
+    else:
+        logger.info("Flows unavailable or send failed (%s); falling back to text prompt.", flow_result.get("reason") or flow_result.get("error"))
+        safe_send(_PREFERENCES_PROMPT, wa_id)
+        set_onboarding_stage(wa_id, "awaiting_preferences_text")
+        return JSONResponse(status_code=200, content={"status": "onboarding_started", "mode": "text"})
+
+
+def handle_onboarding_preferences_text(wa_id: str, message_text: str) -> JSONResponse:
+    """
+    Fallback path: user replied with free text listing topics.
+    """
+    set_onboarding_stage(wa_id, "awaiting_preferences_text")
+    topics = _normalise_preferences(message_text)
+    if not topics:
+        safe_send(_INVALID_TOPICS_MESSAGE, wa_id)
+        safe_send(_PREFERENCES_PROMPT, wa_id)
+        return JSONResponse(status_code=200, content={"status": "awaiting_preferences_text"})
+
+    set_preferences(wa_id, topics=topics)
+    set_onboarding_stage(wa_id, "onboarded")
+
+    display_labels = [TOPIC_DISPLAY_MAP[slug]["name"] for slug in _TOPICS_ORDER if slug in topics]
+    confirmation = "Great! I’ll keep you posted on " + ", ".join(display_labels) + "."
+    safe_send(confirmation, wa_id)
+
+    # Your fetch_cached_headlines function in your codebase seems to support filtering
+    digest = fetch_cached_headlines(selected_topics=topics)
+    safe_send(digest, wa_id)
+
+    return JSONResponse(status_code=200, content={"status": "preferences_captured", "topics": topics})
+
+
+def handle_onboarding_preferences_flow_submission(wa_id: str, selected_topic_ids: List[str]) -> JSONResponse:
+    """
+    Call this from your webhook handler when you receive the Flow submission payload.
+    `selected_topic_ids` should contain the IDs of the checked boxes in your Flow.
+    Make sure those IDs match our topic slugs: india, world, tech, finance, sports.
+    """
+    # Normalize & validate against the allow-list
+    topics = [t for t in _TOPICS_ORDER if t in set(selected_topic_ids)]
+    if not topics:
+        # If user submitted nothing, nudge with the flow again
+        safe_send("Looks like you didn’t select any topics. Please choose at least one.", wa_id)
+        res = _send_topics_flow(wa_id)
+        if res.get("status") != "success":
+            # Fall back to text prompt if flow fails here
+            safe_send(_PREFERENCES_PROMPT, wa_id)
+            set_onboarding_stage(wa_id, "awaiting_preferences_text")
+        else:
+            set_onboarding_stage(wa_id, "awaiting_preferences_flow")
+        return JSONResponse(status_code=200, content={"status": "awaiting_preferences_flow"})
+
+    # Persist and confirm
+    set_preferences(wa_id, topics=topics)
+    set_onboarding_stage(wa_id, "onboarded")
+
+    display_labels = [TOPIC_DISPLAY_MAP[slug]["name"] for slug in _TOPICS_ORDER if slug in topics]
+    confirmation = "Great! I’ll keep you posted on " + ", ".join(display_labels) + "."
+    safe_send(confirmation, wa_id)
+
+    digest = fetch_cached_headlines(selected_topics=topics)
+    safe_send(digest, wa_id)
+
+    return JSONResponse(status_code=200, content={"status": "preferences_captured", "topics": topics})

components/handlers/wa/shortcuts.py

@@ -2,6 +2,7 @@ import logging
 from fastapi.responses import JSONResponse
 
 from components.gateways.headlines_to_wa import fetch_cached_headlines
+from storage.users import get_user
 
 from .common import safe_send
 
@@ -10,7 +11,11 @@ logger = logging.getLogger(__name__)
 
 
 def handle_headlines(from_number: str) -> JSONResponse:
-    full_message_text = fetch_cached_headlines()
+    user = get_user(from_number) or {}
+    prefs = user.get("preferences", {}) if isinstance(user, dict) else {}
+    preferred_topics = prefs.get("topics") if isinstance(prefs, dict) else None
+
+    full_message_text = fetch_cached_headlines(selected_topics=preferred_topics)
 
     if full_message_text.startswith("❌") or full_message_text.startswith("⚠️"):
         logger.error("Failed to fetch digest for %s: %s", from_number, full_message_text)
@@ -38,8 +43,8 @@ def handle_headlines(from_number: str) -> JSONResponse:
 def handle_preferences(from_number: str) -> JSONResponse:
     msg = (
         "Let’s tune your feed. Reply with topics you like:\n"
-        "• world • india • finance • sports • entertainment\n\n"
-        "You can send multiple, e.g.: india, finance"
+        "• India • World • Technology • Business & Markets • Sports\n\n"
+        "You can send multiple, e.g.: India, Technology"
     )
     safe_send(msg, to=from_number)
     return JSONResponse(status_code=200, content={"status": "success", "message": "Preferences prompt sent"})

routes/api/whatsapp_webhook.py

@@ -26,6 +26,9 @@ from components.handlers.wa import (
     handle_small_talk,
     handle_chat_question,
     handle_general_research,
+    initiate_onboarding,
+    handle_onboarding_preferences_text,
+    handle_onboarding_preferences_flow_submission,
 )
 
 try:  # Optional memory integration
@@ -93,6 +96,7 @@ def _extract_from_number_and_text(payload: dict) -> Tuple[Optional[str], Optiona
     Extracts (from_number, message_text) from a WhatsApp webhook payload.
     Returns (None, None) if not a user message (e.g., statuses, billing-event).
     """
+    last_sender: Optional[str] = None
     try:
         entries = payload.get("entry", [])
         for entry in entries:
@@ -113,6 +117,8 @@ def _extract_from_number_and_text(payload: dict) -> Tuple[Optional[str], Optiona
                 # Only look at the first message in the list
                 msg = messages_list[0]
                 from_number = msg.get("from")
+                if from_number:
+                    last_sender = from_number
                 mtype = msg.get("type")
 
                 # 1) Plain text
@@ -141,12 +147,70 @@ def _extract_from_number_and_text(payload: dict) -> Tuple[Optional[str], Optiona
                             return from_number, intent
 
         # No usable user message found
-        return None, None
+        return last_sender, None
 
     except Exception as e:
         # Always return a tuple on error
         logging.exception(f"_extract_from_number_and_text error: {e}")
-        return None, None
+        return last_sender, None
+
+
+def _extract_flow_topic_ids(payload: dict) -> Optional[list]:
+    try:
+        entries = payload.get("entry", [])
+        for entry in entries:
+            changes = entry.get("changes", [])
+            for change in changes:
+                if change.get("field") != "messages":
+                    continue
+                value = change.get("value", {}) or {}
+                messages_list = value.get("messages", []) or []
+                if not messages_list:
+                    continue
+                msg = messages_list[0]
+                if msg.get("type") != "interactive":
+                    continue
+                interactive = msg.get("interactive", {}) or {}
+                if interactive.get("type") != "flow":
+                    continue
+
+                flow_payload = (
+                    interactive.get("flow")
+                    or interactive.get("flow_response")
+                    or interactive.get("flow_reply")
+                    or {}
+                )
+
+                selected: list = []
+
+                def _collect(node):
+                    if isinstance(node, dict):
+                        node_type = str(node.get("type", "")).lower()
+                        if node_type in {"checkbox", "checkboxes", "checkbox_group"}:
+                            for option in node.get("options", []) or []:
+                                if not isinstance(option, dict):
+                                    continue
+                                if option.get("selected") or option.get("checked"):
+                                    val = (
+                                        option.get("id")
+                                        or option.get("value")
+                                        or option.get("title")
+                                    )
+                                    if val:
+                                        selected.append(str(val).strip().lower())
+                        for value in node.values():
+                            _collect(value)
+                    elif isinstance(node, list):
+                        for item in node:
+                            _collect(item)
+
+                _collect(flow_payload)
+
+                if selected:
+                    return selected
+    except Exception as exc:  # pragma: no cover - defensive path
+        logging.debug("_extract_flow_topic_ids error: %s", exc)
+    return None
 
 
 def _extract_message_timestamp(payload: dict) -> Optional[float]:
@@ -206,9 +270,15 @@ async def whatsapp_webhook_receiver(request: Request):
                 logging.info(f"Duplicate inbound message {msg_id} ignored.")
                 return JSONResponse(status_code=200, content={"status": "ignored", "message": "Duplicate message"})
 
-        # Extract sender + message
+        # Extract sender + message/flow payload
         from_number, message_text = _extract_from_number_and_text(incoming_message)
-        if not from_number or not message_text:
+        flow_topic_ids = _extract_flow_topic_ids(incoming_message) or []
+
+        if not from_number:
+            logging.info("Ignoring webhook without sender id.")
+            return JSONResponse(status_code=200, content={"status": "ignored", "message": "No sender"})
+
+        if not message_text and not flow_topic_ids:
             logging.info("Ignoring non-message webhook or missing sender/text.")
             return JSONResponse(status_code=200, content={"status": "ignored", "message": "No user message"})
 
@@ -223,10 +293,39 @@ async def whatsapp_webhook_receiver(request: Request):
         message_ts = _extract_message_timestamp(incoming_message)
 
         # Upsert user (profile + last_seen + last_msg_id)
+        user: Optional[dict] = None
+        user_created = False
         try:
-            ensure_user_from_webhook(incoming_message, from_number, msg_id)
+            user, user_created = ensure_user_from_webhook(incoming_message, from_number, msg_id)
         except Exception as e:
             logging.warning(f"ensure_user_from_webhook failed (continuing): {e}")
+            user = None
+            user_created = False
+
+        onboarding_stage = (user or {}).get("onboarding_stage") if isinstance(user, dict) else None
+
+        if user_created:
+            if typing_started:
+                send_typing_indicator(from_number, status="paused")
+                typing_started = False
+            return initiate_onboarding(from_number)
+
+        if onboarding_stage == "awaiting_preferences_flow":
+            if flow_topic_ids:
+                if typing_started:
+                    send_typing_indicator(from_number, status="paused")
+                    typing_started = False
+                return handle_onboarding_preferences_flow_submission(from_number, flow_topic_ids)
+            if typing_started:
+                send_typing_indicator(from_number, status="paused")
+                typing_started = False
+            return handle_onboarding_preferences_text(from_number, message_text)
+
+        if onboarding_stage == "awaiting_preferences_text":
+            if typing_started:
+                send_typing_indicator(from_number, status="paused")
+                typing_started = False
+            return handle_onboarding_preferences_text(from_number, message_text)
 
         if _store_chat_turn:
             try:

storage/users.py

@@ -3,7 +3,7 @@ import os
 import time
 import json
 import logging
-from typing import Any, Dict, List, Optional
+from typing import Any, Dict, List, Optional, Tuple
 
 import redis
 
@@ -129,6 +129,21 @@ def set_preferences(wa_id: str, topics: Optional[List[str]] = None,
         logging.exception(f"set_preferences error for {wa_id}: {e}")
         return {"wa_id": wa_id, "error": str(e)}
 
+
+def set_onboarding_stage(wa_id: str, stage: str) -> None:
+    try:
+        redis_client.hset(_user_key(wa_id), mapping={"onboarding_stage": stage})
+    except Exception as e:
+        logging.exception(f"set_onboarding_stage error for {wa_id}: {e}")
+
+
+def get_onboarding_stage(wa_id: str) -> Optional[str]:
+    try:
+        return redis_client.hget(_user_key(wa_id), "onboarding_stage")
+    except Exception as e:
+        logging.exception(f"get_onboarding_stage error for {wa_id}: {e}")
+        return None
+
 # ------------- Webhook helpers -------------
 
 def _extract_contact_name(payload: Dict[str, Any]) -> Optional[str]:
@@ -153,12 +168,19 @@ def _extract_contact_name(payload: Dict[str, Any]) -> Optional[str]:
     return None
 
 
-def ensure_user_from_webhook(payload: Dict[str, Any], from_number: str, message_id: Optional[str]) -> Dict[str, Any]:
+def ensure_user_from_webhook(
+    payload: Dict[str, Any],
+    from_number: str,
+    message_id: Optional[str],
+) -> Tuple[Dict[str, Any], bool]:
     """
     Ensure the user exists/updated using webhook payload signals.
     - from_number: WhatsApp 'from' (wa_id / E.164)
     - message_id: wamid... (for last message stamping)
     """
+    existing = get_user(from_number)
+    created = existing is None
+
     display_name = _extract_contact_name(payload)
     user = upsert_user(
         wa_id=from_number,
@@ -170,4 +192,6 @@ def ensure_user_from_webhook(payload: Dict[str, Any], from_number: str, message_
     set_last_seen(from_number)
     if message_id:
         record_inbound_message(from_number, message_id)
-    return user
+    if created:
+        user = get_user(from_number) or user
+    return user, created