""" aca/whatsapp/views.py — CONVERSATIONAL, GUIDED, FRIENDLY Flow ----- Menu is shown as an interactive list (tap to choose — no "reply with a number"). After selection, the user is guided clearly through each path. AI is called for: text search (path 1), image search (path 3), support (path 5). Cart, orders, checkout are handled with pure logic. UX principles: - Friendly, local Ghanaian tone — not robotic - Always guide the user on what to do next via interactive buttons/lists - No "reply 0 to go back" — use a persistent footer button or quick reply - Image search uses broader matching against vendor products """ import json import logging import re import time from typing import Optional from django.conf import settings from django.core.cache import cache from django.http import HttpResponse from django.shortcuts import redirect from django.views.decorators.csrf import csrf_exempt from django.views.decorators.http import require_http_methods from chat.models import Conversation from ai_engine.services import ConversationManager from gateway.services import ProductAggregator from .session import ( set_last_products, get_last_products, get_product_by_number, add_to_cart, remove_from_cart, get_cart, clear_cart, cart_total, get_checkout_state, set_checkout_state, normalize_phone, ) from .media import ( download_whatsapp_image, send_whatsapp_image, send_whatsapp_text, send_whatsapp_list, ) from .checkout import format_cart_summary, start_checkout, process_checkout_step logger = logging.getLogger(__name__) _manager = ConversationManager() _aggregator = ProductAggregator() # ── Rate limiting ───────────────────────────────────────────────────────── _RATE_WINDOW = 2 _DEDUP_WINDOW = 30 _MAX_BURST = 5 # ── Session mode keys ───────────────────────────────────────────────────── _MODE_SEARCH = "search" _MODE_IMAGE = "image" _MODE_SUPPORT = "support" _MODE_ORDER = "order" _MODE_MENU = "menu" _MODE_CART = "cart" _MODE_STORE = "store" _SEARCH_STOP_WORDS = { "i", "me", "my", "we", "us", "you", "your", "want", "need", "find", "search", "show", "get", "for", "a", "an", "the", "to", "of", "with", "please", "item", "items", "product", "products", "cart", "add", "that", "this", "it", "on", "in", "from", "at", "and", "or", } _CATEGORY_KEYWORDS = { "watches": ("watch", "watches", "timepiece", "wristwatch", "chronograph"), "shirts": ("shirt", "shirts", "tee", "t-shirt", "polo", "top", "blouse"), "dresses": ("dress", "dresses", "gown", "maxi", "mini"), "shoes": ("shoe", "shoes", "sneaker", "sneakers", "boot", "boots", "sandals"), "bags": ("bag", "bags", "handbag", "purse", "tote", "backpack"), "perfumes": ("perfume", "perfumes", "fragrance", "cologne", "scent"), "jewelry": ("jewelry", "jewellery", "ring", "necklace", "bracelet", "earring"), "phones": ("phone", "phones", "iphone", "android", "smartphone"), "laptops": ("laptop", "laptops", "notebook", "macbook"), } # ───────────────────────────────────────────────────────────────────────── # Helpers # ───────────────────────────────────────────────────────────────────────── def whatsapp_chat_entry(request): """ Public entrypoint for shoppers while web chat is disabled. Redirects to configured click-to-chat URL when available. """ chat_url = (getattr(settings, "WHATSAPP_CLICK_TO_CHAT_URL", "") or "").strip() if chat_url: return redirect(chat_url) return HttpResponse( "WhatsApp chat is currently the only supported channel. " "Please ask support to configure WHATSAPP_CLICK_TO_CHAT_URL.", status=503, ) def _rate_check(phone: str, text: str) -> bool: now = time.time() last_ts_key = f"wa:ts:{phone}" last_ts = cache.get(last_ts_key, 0) if now - last_ts < _RATE_WINDOW: return False cache.set(last_ts_key, now, 60) import hashlib msg_hash = hashlib.md5(text.encode()).hexdigest()[:12] dedup_key = f"wa:dedup:{phone}:{msg_hash}" if cache.get(dedup_key): return False cache.set(dedup_key, 1, _DEDUP_WINDOW) burst_key = f"wa:burst:{phone}" count = cache.get(burst_key, 0) if count >= _MAX_BURST: return False cache.set(burst_key, count + 1, 30) return True def _get_mode(phone: str) -> str: return cache.get(f"wa:mode:{phone}", _MODE_MENU) def _set_mode(phone: str, mode: str): cache.set(f"wa:mode:{phone}", mode, 60 * 60 * 6) def _clear_mode(phone: str): cache.delete(f"wa:mode:{phone}") def _set_await_choice(phone: str, choice_type: str): cache.set(f"wa:await:{phone}", choice_type, 60 * 20) def _get_await_choice(phone: str) -> str: return cache.get(f"wa:await:{phone}", "") def _clear_await_choice(phone: str): cache.delete(f"wa:await:{phone}") def _set_store_filter(phone: str, vendor_id: str, store_name: str): cache.set( f"wa:store:{phone}", { "scope": "specific", "vendor_id": str(vendor_id), "store_name": str(store_name or "").strip(), }, 60 * 60 * 6, ) def _get_store_filter(phone: str) -> dict: return cache.get(f"wa:store:{phone}", {}) or {} def _clear_store_filter(phone: str): cache.delete(f"wa:store:{phone}") def _set_all_store_scope(phone: str): cache.set( f"wa:store:{phone}", {"scope": "all", "vendor_id": "", "store_name": "All stores"}, 60 * 60 * 6, ) def _has_store_scope(phone: str) -> bool: scope = _get_store_filter(phone) if not scope: return False if scope.get("scope") in ("all", "specific"): return True # Backward compatibility with older cached shape. return bool(str(scope.get("vendor_id", "")).strip()) def _get_or_create_conversation(phone: str) -> Conversation: key = f"wa:{phone}" conv = Conversation.objects.filter(session_key=key).order_by("-updated_at").first() if not conv: conv = Conversation.objects.create(session_key=key, title=f"WA {phone}") return conv def _abs_url(url: str) -> str: if not url: return "" if url.startswith(("http://", "https://")): return url base = getattr(settings, "SITE_URL", "https://aca.mirjy.com").rstrip("/") return f"{base}{url}" def _is_menu_trigger(t: str) -> bool: """Detect when user wants to go back to the main menu.""" triggers = ( "menu", "main menu", "home", "back", "start over", "restart", "start", "hi", "hello", "hey", "hola", "good morning", "good evening", "good afternoon", "howdy", "yo", "what can you do", "help", ) return t.strip().lower() in triggers def _is_greeting_only(text: str) -> bool: """True when the message is mostly a greeting/small talk opener.""" t = re.sub(r"[^\w\s]", " ", text.lower()).strip() if not t: return False tokens = [w for w in t.split() if w] if len(tokens) > 5: return False greetings = { "hi", "hello", "hey", "yo", "hola", "morning", "afternoon", "evening", "good", "sup", "howdy", } # Require all tokens to look like greeting words so # "hello i need a phone" does not get treated as greeting-only. return all(tok in greetings for tok in tokens) def _classify_text_intent(text: str) -> str: """ Lightweight intent classifier for menu-mode routing. Returns one of: greeting, search, order, support, image, cart, checkout, unknown. """ t = text.lower().strip() if not t: return "unknown" if _is_greeting_only(t): return "greeting" if re.search(r"aca[-\s]?\d{4}", t, re.IGNORECASE): return "order" if any(k in t for k in ("cart", "basket")): return "cart" if any(k in t for k in ("checkout", "check out", "pay now", "place order", "buy now")): return "checkout" if any(k in t for k in ("track order", "where is my order", "order status", "delivery status")): return "order" if any(k in t for k in ("photo", "picture", "image", "snap")): return "image" if any(k in t for k in ("help", "issue", "problem", "complaint", "refund", "support", "agent")): return "support" search_markers = ( "looking for", "find", "search", "need", "show me", "i want", "do you have", "buy", "price of", "cheap", "best", "where can i get", ) if any(m in t for m in search_markers): return "search" if len(t.split()) >= 2: # Natural free-text with 2+ words is usually a product query in this flow. return "search" return "unknown" def _send_back_to_menu(phone: str, message: str = ""): """Send a message with a tap-to-go-back option.""" rows = [ {"id": "go_menu", "title": "Back to main menu", "description": "See all options"}, {"id": "menu_1", "title": "Search for something else", "description": "Find another product"}, ] cart = get_cart(phone) if cart: rows.insert(0, { "id": "view_cart", "title": f"View cart ({len(cart)} items)", "description": f"GHS {cart_total(phone):.2f} total", }) body = message or "What would you like to do next?" send_whatsapp_list(phone, body, "Choose", rows[:10]) _set_await_choice(phone, "next_action") # ───────────────────────────────────────────────────────────────────────── # Main menu — friendly, local, no "reply with a number" # ───────────────────────────────────────────────────────────────────────── FIRST_VISIT_TEXT = ( "Hey! Welcome to ACA — your fastest way to find anything here in Ghana. " "I search across multiple stores at once and show you the best options. " "What would you like to do?" ) RETURNING_TEXT = ( "Good to have you back! What are we doing today?" ) CART_HINT_TEXT = ( "You have {count} item{s} in your cart worth GHS {total:.2f}. " "What would you like to do?" ) def _send_main_menu(phone: str, returning: bool = False, show_cart_hint: bool = False): if not _has_store_scope(phone): _start_store_scope_mode(phone) return _set_mode(phone, _MODE_MENU) cart = get_cart(phone) store_filter = _get_store_filter(phone) if show_cart_hint and cart: count = len(cart) body = CART_HINT_TEXT.format( count=count, s="s" if count != 1 else "", total=cart_total(phone) ) elif returning: body = RETURNING_TEXT else: body = FIRST_VISIT_TEXT if store_filter.get("scope") == "specific" and store_filter.get("store_name"): body = f"{body}\n\nShopping scope: *{store_filter['store_name']}* only." else: body = f"{body}\n\nShopping scope: *All stores*." rows = [ {"id": "menu_1", "title": "Find a product", "description": "Search by name, brand or description"}, {"id": "menu_6", "title": "Change shopping scope", "description": "All stores or one specific store"}, {"id": "menu_2", "title": "My cart", "description": f"{len(cart)} item(s) — GHS {cart_total(phone):.2f}" if cart else "Empty"}, {"id": "menu_3", "title": "Search by photo", "description": "Send a picture and I'll find it"}, {"id": "menu_4", "title": "Track my order", "description": "Check where your order is"}, {"id": "menu_5", "title": "Talk to us", "description": "Questions, issues or feedback"}, ] send_whatsapp_list(phone, body, "Choose option", rows) _set_await_choice(phone, "main_menu") # ───────────────────────────────────────────────────────────────────────── # Webhook entry points # ───────────────────────────────────────────────────────────────────────── @csrf_exempt @require_http_methods(["GET", "POST"]) def whatsapp_webhook(request): if request.method == "GET": return _verify(request) return _incoming(request) def _verify(request): mode = request.GET.get("hub.mode") token = request.GET.get("hub.verify_token") challenge = request.GET.get("hub.challenge") if mode == "subscribe" and token == getattr(settings, "WHATSAPP_VERIFY_TOKEN", ""): return HttpResponse(challenge, content_type="text/plain") return HttpResponse("Forbidden", status=403) def _incoming(request): try: data = json.loads(request.body) except json.JSONDecodeError: return HttpResponse("Bad Request", status=400) try: _process(data) except Exception as exc: logger.error("WhatsApp webhook error: %s", exc, exc_info=True) return HttpResponse("OK", status=200) def _process(data: dict): for entry in data.get("entry", []): for change in entry.get("changes", []): value = change.get("value", {}) for msg in value.get("messages", []): phone = normalize_phone(msg.get("from", "")) msg_type = msg.get("type") if not phone: continue if msg_type == "text": text = msg["text"]["body"].strip() if not _rate_check(phone, text): continue _handle_text(phone, text) elif msg_type == "image": _handle_image(phone, msg["image"]) elif msg_type == "interactive": _handle_interactive(phone, msg.get("interactive", {})) # ───────────────────────────────────────────────────────────────────────── # Interactive reply handler # ───────────────────────────────────────────────────────────────────────── def _handle_interactive(phone: str, interactive: dict): i_type = (interactive.get("type") or "").strip() payload = interactive.get(i_type, {}) if i_type else {} reply_id = str(payload.get("id", "")).strip() _clear_await_choice(phone) # Main menu if reply_id == "menu_1": _start_search_mode(phone) return if reply_id == "menu_2" or reply_id == "view_cart": _show_cart(phone) return if reply_id == "menu_6": _start_store_scope_mode(phone) return if reply_id == "menu_3": _start_image_mode(phone) return if reply_id == "menu_4": _start_order_mode(phone) return if reply_id == "menu_5": _start_support_mode(phone) return if reply_id in ("go_menu", "main_menu"): _send_main_menu(phone, returning=True) return if reply_id == "scope_all": _set_all_store_scope(phone) send_whatsapp_text(phone, "Great, you'll shop across all stores.") _send_main_menu(phone, returning=True) return if reply_id == "scope_specific": _start_store_mode(phone) return if reply_id == "store_all": _set_all_store_scope(phone) send_whatsapp_text(phone, "Great, you'll shop across all stores.") _send_main_menu(phone, returning=True) return if reply_id.startswith("store_"): vendor_id = reply_id.split("store_", 1)[-1] _select_store(phone, vendor_id) return # Checkout if reply_id == "checkout_now": msg = start_checkout(phone) send_whatsapp_text(phone, msg) return # Cart add/remove if reply_id.startswith("addpid_"): pid = reply_id.split("addpid_", 1)[-1] if pid: _add_by_product_id(phone, pid) return if reply_id.startswith("remove_"): pid = reply_id.split("remove_", 1)[-1] if pid: _remove_interactive(phone, pid) return if reply_id.startswith("add_"): suffix = reply_id.split("add_", 1)[-1] if suffix.isdigit(): _handle_add_number(phone, int(suffix)) return # Order actions if reply_id == "dispute_order": _handle_dispute(phone) return if reply_id.startswith("track_"): order_num = reply_id.split("track_", 1)[-1] _show_order_detail(phone, order_num) return # Keep searching if reply_id == "keep_searching": _set_mode(phone, _MODE_SEARCH) send_whatsapp_text(phone, "Sure, what else are you looking for?") return # Fall back: treat title as text title = str(payload.get("title", "")).strip() if title: _handle_text(phone, title) # ───────────────────────────────────────────────────────────────────────── # Text dispatcher # ───────────────────────────────────────────────────────────────────────── def _handle_text(phone: str, text: str): t = text.lower().strip() # Menu trigger if _is_menu_trigger(t): _send_main_menu(phone, returning=True) return # Active checkout flow takes priority checkout_state = get_checkout_state(phone) if checkout_state: reply = process_checkout_step(phone, text) if reply: send_whatsapp_text(phone, reply) return # Convenience keywords if t in ("cart", "my cart", "view cart", "show cart", "basket"): _show_cart(phone) return if t in ("checkout", "check out", "place order", "pay now", "i want to pay", "buy now"): msg = start_checkout(phone) send_whatsapp_text(phone, msg) return if any(p in t for p in ("clear cart", "empty cart", "remove all")): clear_cart(phone) send_whatsapp_text(phone, "Done, your cart is empty now.") _send_main_menu(phone, returning=True) return # Route by current mode mode = _get_mode(phone) await_choice = _get_await_choice(phone) # Strict guided mode: while a list/menu is waiting for a tap, avoid free-form routing. if await_choice: if mode == _MODE_SEARCH and await_choice in ("search_next", "next_action", "refine"): if _looks_like_search_query(text): send_whatsapp_text(phone, "Tap *Search for something else* first, then type your new search.") _send_search_followup(phone, get_last_products(phone)) return if _is_add_intent(phone, text): _do_add_to_cart(phone, text) return send_whatsapp_text(phone, "Please tap one of the options so I know what to do next.") if await_choice == "refine": _send_refine_options(phone) else: _send_search_followup(phone, get_last_products(phone)) return if mode == _MODE_MENU and await_choice == "main_menu": if _is_greeting_only(text): send_whatsapp_text(phone, "Hi! Please choose one option from the menu to continue.") else: send_whatsapp_text(phone, "Please choose an option from the menu to continue.") _send_main_menu(phone, returning=True) return if mode == _MODE_SEARCH: _run_text_search(phone, text) return if mode == _MODE_SUPPORT: _handle_support_query(phone, text) return if mode == _MODE_ORDER: _resolve_order(phone, text) return if mode == _MODE_STORE: _handle_store_mode_text(phone, text) return if mode == _MODE_IMAGE: # They typed instead of sending a photo send_whatsapp_text( phone, "Just send me a photo and I'll get right on it. " "Or if you'd rather type what you're looking for, tap the menu button to switch." ) _send_back_to_menu(phone) return # Detect add-to-cart intent from number if _is_add_intent(phone, text): _do_add_to_cart(phone, text) return # Order number pattern if re.search(r'ACA[-\s]?\d{4}', text, re.IGNORECASE): _show_order_detail(phone, text) return # Default — show menu send_whatsapp_text(phone, "Please choose an option from the menu to continue.") _send_main_menu(phone, returning=True) # ───────────────────────────────────────────────────────────────────────── # Path 1 — Text search (uses AI) # ───────────────────────────────────────────────────────────────────────── def _start_search_mode(phone: str): _set_mode(phone, _MODE_SEARCH) _clear_await_choice(phone) store_filter = _get_store_filter(phone) if store_filter.get("vendor_id"): send_whatsapp_text( phone, f"Okay, you're shopping in *{store_filter.get('store_name', 'this store')}* only.\n" "What are you looking for? You can be specific — brand, color, price range, anything." ) else: send_whatsapp_text( phone, "Okay, what are you looking for? " "You can be as specific as you want — brand, color, price range, anything." ) def _run_text_search(phone: str, text: str): _clear_await_choice(phone) send_whatsapp_text(phone, "Searching...") query = _build_catalog_query_from_text(text) products = _apply_store_filter(phone, _aggregator.search(query)) products = _filter_catalog_results(products, text, query) products = _products_with_images(products) store_filter = _get_store_filter(phone) if products: set_last_products(phone, products) send_whatsapp_text(phone, _build_search_intro(products, store_filter)) _send_product_cards(phone, products) # Guide the user on next steps — no "type 0" _send_search_followup(phone, products) else: if store_filter.get("vendor_id"): send_whatsapp_text( phone, f"I couldn't find that in *{store_filter.get('store_name', 'this store')}* right now." ) send_whatsapp_text( phone, "Hmm, I couldn't find anything for that. " "Want to try different keywords? Maybe a broader term or different brand?" ) _send_refine_options(phone) # Stay in search mode so follow-up messages keep searching _set_mode(phone, _MODE_SEARCH) def _send_search_followup(phone: str, products: list): """After showing search results, guide user on what to do.""" vendor_count = sum(1 for p in products if p.get("source") == "aca_vendor" and p.get("in_stock", True)) rows = [] if vendor_count: rows.append({ "id": "keep_searching", "title": "Search for something else", "description": "Find another product", }) rows.append({ "id": "view_cart", "title": "View my cart", "description": f"GHS {cart_total(phone):.2f} total", }) else: rows.append({ "id": "keep_searching", "title": "Try a different search", "description": "Different keywords or brand", }) rows.append({"id": "go_menu", "title": "Back to main menu", "description": "See all options"}) send_whatsapp_list( phone, "Use the Add to cart list above, or choose what to do next:", "What's next?", rows[:10], ) _set_await_choice(phone, "search_next") def _send_refine_options(phone: str): """When no results — offer to refine or go back.""" rows = [ {"id": "keep_searching", "title": "Try again", "description": "Different keywords"}, {"id": "menu_3", "title": "Search by photo", "description": "Send a picture instead"}, {"id": "go_menu", "title": "Back to main menu", "description": "See all options"}, ] send_whatsapp_list(phone, "What would you like to do?", "Options", rows) _set_await_choice(phone, "refine") # ───────────────────────────────────────────────────────────────────────── # Path 3 — Image search (uses AI vision + broader vendor matching) # ───────────────────────────────────────────────────────────────────────── def _start_image_mode(phone: str): _set_mode(phone, _MODE_IMAGE) _clear_await_choice(phone) send_whatsapp_text( phone, "Nice! Just send me a clear photo of what you're looking for " "and I'll find the closest matches from our stores." ) def _handle_image(phone: str, image_msg: dict): """ Download image, extract product attributes, then do a broad search against vendor products matching any of the extracted fields. """ mode = _get_mode(phone) media_id = image_msg.get("id", "") caption = (image_msg.get("caption") or "").strip() if mode != _MODE_IMAGE: _set_mode(phone, _MODE_IMAGE) send_whatsapp_text(phone, "Let me take a look at that...") image_bytes, mime_type = download_whatsapp_image(media_id) if not image_bytes: send_whatsapp_text( phone, "I had trouble loading that image. " "Could you send it again, or just describe what you're looking for?" ) _send_back_to_menu(phone) return # Strict visual flow: compare uploaded image against catalog product photos first. from gateway.services import ProductAggregator import base64 send_whatsapp_text(phone, "Comparing your photo with product images...") candidates = _apply_store_filter(phone, ProductAggregator().search({})) if not candidates: send_whatsapp_text( phone, "I could not load product images to compare right now. Please try again shortly." ) _send_back_to_menu(phone) return matched, quality, notice = _manager.llm.filter_products_by_reference_image( base64.b64encode(image_bytes).decode("utf-8"), candidates, max_items=4, pool_limit=40, ) matched = _products_with_images(matched) if notice: send_whatsapp_text(phone, notice) if not matched or quality == "none": send_whatsapp_text( phone, "I could not find a close visual match in our catalog from this image." ) _send_refine_options(phone) _set_mode(phone, _MODE_IMAGE) return set_last_products(phone, matched) _send_product_cards(phone, matched) _send_search_followup(phone, matched) _set_mode(phone, _MODE_IMAGE) def _enrich_image_query(query: dict) -> dict: """ Make the image query broader so it matches vendor products better. Adds search_keywords from category hints and broadens the q field. """ from ai_engine.services.query_builder import QueryBuilder category = (query.get("category") or "").lower().strip() color = (query.get("color") or "").strip() brand = (query.get("brand") or "").strip() style = (query.get("style") or "").strip() # Build a rich q string combining all useful attributes q_parts = [p for p in [color, category, style, brand] if p] if q_parts: query["q"] = " ".join(q_parts[:5]) # Add keyword hints from category cat_key = category.rstrip("s") for hint_key, hints in QueryBuilder.BRAND_HINTS.items(): if hint_key in cat_key or cat_key in hint_key: query["search_keywords"] = hints[:8] break # If no search_keywords yet, build some from the category word if not query.get("search_keywords") and category: # Generic expansion: plural/singular + common synonyms words = [category, category.rstrip("s"), category + "s"] if color: words.append(f"{color} {category}") query["search_keywords"] = list(set(words))[:6] return query def _image_fallback_search(query: dict) -> list: """ Broader text-only fallback when image visual matching returns nothing. Strips image fields and tries progressively looser queries. """ from gateway.services import ProductAggregator agg = ProductAggregator() # Remove image-specific keys safe_query = {k: v for k, v in query.items() if k not in ("image_search", "image_b64", "image_match_mode")} # Try 1: full query results = agg.search(safe_query) if results: return results # Try 2: category + color only light = {} if query.get("category"): light["category"] = query["category"] if query.get("color"): light["color"] = query["color"] if query.get("q"): light["q"] = query["q"] if light: results = agg.search(light) if results: return results # Try 3: category alone if query.get("category"): results = agg.search({"category": query["category"]}) if results: return results return [] # ───────────────────────────────────────────────────────────────────────── # Path 4 — Orders (no AI) # ───────────────────────────────────────────────────────────────────────── def _start_order_mode(phone: str): _set_mode(phone, _MODE_ORDER) send_whatsapp_text( phone, "Sure! Send me your order number and I'll pull it up. " "It looks something like ACA-1234-5678-26." ) def _resolve_order(phone: str, text: str): conv = _get_or_create_conversation(phone) result = _manager.handle_message(conv, text) meta = result.get("metadata", {}) if meta.get("source") == "order_status" and meta.get("order_card"): send_whatsapp_text(phone, _format_order_card(meta["order_card"])) _send_order_actions(phone, meta["order_card"]) elif meta.get("source") == "order_status_not_found": send_whatsapp_text( phone, "I couldn't find that order number. " "Double-check the format — it should be ACA-XXXX-XXXX-YY. " "Send it again whenever you're ready." ) # Stay in order mode so next message is tried again _set_mode(phone, _MODE_ORDER) else: send_whatsapp_text( phone, "Send me your order number and I'll look it up straight away. " "Format: ACA-1234-5678-26" ) _set_mode(phone, _MODE_ORDER) def _show_order_detail(phone: str, text: str): _set_mode(phone, _MODE_ORDER) _resolve_order(phone, text) def _send_order_actions(phone: str, order_card: dict): order_num = order_card.get("order_number", "") rows = [ { "id": f"track_{order_num}", "title": "Check delivery status", "description": "See vendor details and progress", }, { "id": "dispute_order", "title": "Report an issue", "description": "Something wrong with this order?", }, { "id": "go_menu", "title": "Back to main menu", "description": "See all options", }, ] send_whatsapp_list(phone, "What would you like to do with this order?", "Choose", rows) def _handle_dispute(phone: str): send_whatsapp_text( phone, "I'm sorry to hear there's an issue. Here's how to get it sorted:\n\n" "If you already confirmed delivery by tapping the SMS link, " "the payment has been released and would need to be handled directly with the seller.\n\n" "If you haven't confirmed yet and there's a problem, " "please reach us directly and we'll step in:\n\n" "WhatsApp: +233597224964\n" "Email: support@mirjy.com\n\n" "Include your order number and we'll get back to you quickly." ) _send_back_to_menu(phone, "Anything else I can help you with?") def _format_order_card(card: dict) -> str: lines = [ f"Here's your order:\n", f"Order number: {card.get('order_number')}", f"Placed: {card.get('placed')}", f"Total: GHS {card.get('total', 0):.2f}", ] for v in card.get("vendors", []): lines.append(f"\n{v.get('name')}") lines.append(f"Status: {v.get('status')}") lines.append(f"Amount: GHS {v.get('amount', 0):.2f}") if v.get("phone") and v["phone"] != "N/A": lines.append(f"Seller phone: {v.get('phone')}") for it in v.get("items", []): lines.append( f" - {it.get('name')} x{it.get('qty')} " f"({it.get('currency')} {it.get('subtotal')})" ) return "\n".join(lines) # ───────────────────────────────────────────────────────────────────────── # Path 5 — Support (uses AI + escalation) # ───────────────────────────────────────────────────────────────────────── _SUPPORT_KEYWORDS_ESCALATE = ( "refund", "stolen", "fraud", "not delivered", "wrong item", "speak to human", "speak to someone", "real person", "agent", "urgent", "complaint", "legal", ) def _start_support_mode(phone: str): _set_mode(phone, _MODE_SUPPORT) send_whatsapp_text( phone, "Of course! What's on your mind? " "Ask me anything about your orders, products, or how ACA works." ) def _handle_support_query(phone: str, text: str): t = text.lower() if any(kw in t for kw in _SUPPORT_KEYWORDS_ESCALATE): send_whatsapp_text( phone, "This sounds like something our team should handle directly. " "I've noted your message and a support agent will reach out to you shortly.\n\n" "You can also contact us right now:\n" "WhatsApp: +233597224964\n" "Email: support@mirjy.com" ) _notify_admin_support(phone, text) _send_back_to_menu(phone) return conv = _get_or_create_conversation(phone) result = _manager.handle_message(conv, text) content = (result.get("content") or "").strip() if content: send_whatsapp_text(phone, content) else: send_whatsapp_text( phone, "I'm not sure about that one. " "Our support team can help — reach us at support@mirjy.com " "or WhatsApp +233597224964." ) _set_mode(phone, _MODE_SUPPORT) _send_back_to_menu(phone, "Anything else I can help you with?") def _notify_admin_support(phone: str, message: str): try: from escrow.services.sms import _send admin_phone = getattr(settings, "ESCROW_ADMIN_PHONE", "") if admin_phone: _send(admin_phone, f"ACA Support escalation from {phone}: {message[:160]}") except Exception as exc: logger.warning("Admin support notification failed: %s", exc) # ───────────────────────────────────────────────────────────────────────── # Cart helpers # ───────────────────────────────────────────────────────────────────────── def _show_cart(phone: str): _set_mode(phone, _MODE_CART) cart = get_cart(phone) if cart: summary = format_cart_summary(phone) send_whatsapp_text(phone, summary) _send_cart_actions(phone) else: rows = [ {"id": "menu_1", "title": "Find something to buy", "description": "Search our stores"}, {"id": "menu_6", "title": "Shop by store", "description": "Choose a specific store"}, {"id": "go_menu", "title": "Main menu", "description": "See all options"}, ] send_whatsapp_list(phone, "Your cart is empty — let's fix that!", "Options", rows) def _send_cart_actions(phone: str): cart = get_cart(phone) if not cart: return rows = [ {"id": "checkout_now", "title": "Checkout", "description": f"Pay GHS {cart_total(phone):.2f}"}, {"id": "menu_1", "title": "Keep shopping", "description": "Find more items"}, {"id": "menu_6", "title": "Shop by store", "description": "Choose a specific store"}, {"id": "go_menu", "title": "Main menu", "description": "See all options"}, ] for item in cart[:5]: pid = str(item.get("product_id", "")).strip() if not pid: continue name = str(item.get("name", ""))[:20] rows.append({ "id": f"remove_{pid}", "title": f"Remove: {name}", "description": f"x{item.get('quantity', 1)} — GHS {item.get('price', 0)}", }) send_whatsapp_list( phone, f"Your cart has {len(cart)} item(s) — GHS {cart_total(phone):.2f} total.", "Cart options", rows[:10], ) def _is_add_intent(phone: str, text: str) -> bool: t = text.lower().strip() if any(s in t for s in ("add to cart", "add it", "add this", "add number", "add item", "i want this", "buy this")): return True if re.search(r"\badd\s*#?\s*\d+\b", t): return True if re.search(r"\b(?:item|number|option)\s*#?\s*\d+\b", t): return True if re.fullmatch(r"#?\d+", t) and get_last_products(phone): return True return False def _do_add_to_cart(phone: str, text: str): products = get_last_products(phone) if not products: send_whatsapp_text( phone, "Let's find something first! What are you looking for?" ) _start_search_mode(phone) return num = _extract_num(text) if num: _handle_add_number(phone, num) return vendor_prods = [p for p in products if p.get("source") == "aca_vendor" and p.get("in_stock", True)] if len(vendor_prods) == 1: _add_product_to_cart(phone, vendor_prods[0]) elif vendor_prods: _send_add_list(phone, products) else: send_whatsapp_text( phone, "These items are from partner stores — use the buy link to purchase directly from them." ) def _handle_add_number(phone: str, num: int): product = get_product_by_number(phone, num) if not product: send_whatsapp_text( phone, f"I don't see item number {num} in the results. " "Check the number and try again." ) return if product.get("source") != "aca_vendor": url = product.get("product_url", "") name = product.get("name", "This item") send_whatsapp_text( phone, f"{name} is sold by a partner store. " f"Buy it directly here: {url}" ) return if not product.get("in_stock", True): send_whatsapp_text( phone, f"Sorry, {product.get('name', 'that item')} is currently out of stock." ) return _add_product_to_cart(phone, product) def _add_product_to_cart(phone: str, product: dict): name = product.get("name", "this item") cart = add_to_cart(phone, product, 1) total_qty = sum(i["quantity"] for i in cart) send_whatsapp_text( phone, f"Done! *{name}* is in your cart.\n" f"You have {total_qty} item(s) — GHS {cart_total(phone):.2f} total." ) _send_cart_actions(phone) def _add_by_product_id(phone: str, product_id: str): for p in get_last_products(phone): pid = str(p.get("product_id") or p.get("id") or "").strip() if pid == product_id: if p.get("source") != "aca_vendor": send_whatsapp_text( phone, f"Buy {p.get('name')} directly here: {p.get('product_url', '')}" ) return if not p.get("in_stock", True): send_whatsapp_text(phone, f"Sorry, {p.get('name')} is out of stock right now.") return _add_product_to_cart(phone, p) return send_whatsapp_text( phone, "That item is no longer in the current results. " "Do a new search to find it again." ) def _remove_interactive(phone: str, product_id: str): before = get_cart(phone) removed_name = next( (i.get("name", "") for i in before if str(i.get("product_id")) == product_id), "" ) after = remove_from_cart(phone, product_id) if len(after) == len(before): send_whatsapp_text(phone, "Couldn't find that item in your cart.") else: if after: send_whatsapp_text( phone, f"Removed {removed_name or 'that item'}. " f"Cart: {len(after)} item(s) — GHS {cart_total(phone):.2f}" ) _send_cart_actions(phone) else: send_whatsapp_text(phone, f"Removed {removed_name or 'that item'}. Your cart is now empty.") _send_main_menu(phone, returning=True) def _send_add_list(phone: str, products: list): rows = [] for i, p in enumerate(products[:6], 1): if p.get("source") != "aca_vendor" or not p.get("in_stock", True): continue pid = str(p.get("product_id") or p.get("id") or "").strip() if not pid: continue rows.append({ "id": f"addpid_{pid}", "title": f"{i}. {str(p.get('name', ''))[:22]}", "description": f"{p.get('currency', 'GHS')} {p.get('price', '')}", }) if rows: send_whatsapp_list(phone, "Which one would you like to add?", "Add to cart", rows) # ───────────────────────────────────────────────────────────────────────── # Product card renderer # ───────────────────────────────────────────────────────────────────────── def _send_product_cards(phone: str, products: list): shown = _products_with_images(products, limit=5) if not shown: send_whatsapp_text( phone, "I couldn't find products with images to show right now. Please try another search." ) return for i, p in enumerate(shown, 1): is_vendor = p.get("source") == "aca_vendor" name = p.get("name", "Product") price = p.get("price", "") currency = p.get("currency", "GHS") store = p.get("store_name", "") in_stock = p.get("in_stock", True) image_url = _abs_url(_pick_image_url(p)) parts = [f"{i}. *{name}*", f"{currency} {price} | {store}"] if not in_stock: parts.append("(out of stock)") elif is_vendor: parts.append("Tap below to add to cart") elif p.get("product_url"): parts.append(f"Buy: {p['product_url']}") caption = "\n".join(parts) if image_url: if not send_whatsapp_image(phone, image_url, caption[:1024]): send_whatsapp_text(phone, caption) else: send_whatsapp_text(phone, caption) # Tap-to-add list for ACA vendor products vendor_rows = [] for i, p in enumerate(shown, 1): if p.get("source") != "aca_vendor" or not p.get("in_stock", True): continue pid = str(p.get("product_id") or p.get("id") or "").strip() if not pid: continue vendor_rows.append({ "id": f"addpid_{pid}", "title": f"{i}. {str(p.get('name', ''))[:22]}", "description": f"{p.get('currency', 'GHS')} {p.get('price', '')}", }) if vendor_rows: send_whatsapp_list( phone, "Tap to add to your cart:", "Add to cart", vendor_rows, ) # ───────────────────────────────────────────────────────────────────────── # Utilities # ───────────────────────────────────────────────────────────────────────── def _extract_num(text: str) -> Optional[int]: m = re.search(r'\b([1-9])\b', text) return int(m.group(1)) if m else None def _looks_like_search_query(text: str) -> bool: t = text.lower().strip() if not t: return False if _is_menu_trigger(t): return False if t in ("cart", "my cart", "view cart", "checkout", "check out"): return False return len(t.split()) >= 2 def _is_image_request(text: str) -> bool: t = text.lower().strip() return any(k in t for k in ("show image", "show me image", "show me an image", "image", "photo", "picture")) def _pick_image_url(product: dict) -> str: url = str(product.get("image_url") or "").strip() if url: return url image_urls = product.get("image_urls") if isinstance(image_urls, list): for candidate in image_urls: c = str(candidate or "").strip() if c: return c images = product.get("images") if isinstance(images, list): for candidate in images: c = str(candidate or "").strip() if c: return c return "" def _products_with_images(products: list, limit: Optional[int] = None) -> list: with_images = [p for p in (products or []) if _pick_image_url(p)] return with_images[:limit] if limit else with_images def _send_best_image_from_results(phone: str) -> bool: products = get_last_products(phone) or [] for p in products: image_url = _abs_url(_pick_image_url(p)) if not image_url: continue name = p.get("name", "Product") price = p.get("price", "") currency = p.get("currency", "GHS") caption = f"{name}\n{currency} {price}" if send_whatsapp_image(phone, image_url, caption[:1024]): return True return False def _start_store_mode(phone: str): _set_mode(phone, _MODE_STORE) _clear_await_choice(phone) _send_store_list(phone) def _start_store_scope_mode(phone: str): _set_mode(phone, _MODE_STORE) _clear_await_choice(phone) rows = [ {"id": "scope_all", "title": "Shop all stores", "description": "Search products from every store"}, {"id": "scope_specific", "title": "Specific store", "description": "Pick one store first"}, ] send_whatsapp_list( phone, "Before we continue, choose how you want to shop:", "Shopping scope", rows, ) _set_await_choice(phone, "store_scope") def _send_store_list(phone: str): from vendors.models import Vendor rows = [ {"id": "store_all", "title": "All stores", "description": "Search across every store"}, ] vendors = ( Vendor.objects.filter(is_approved=True, products__in_stock=True) .distinct() .order_by("store_name")[:9] ) for v in vendors: rows.append({ "id": f"store_{v.id}", "title": str(v.store_name)[:24], "description": "Shop only this store", }) send_whatsapp_list( phone, "Choose a specific store. Your main menu actions (search, image search, etc.) will use this store.", "Pick store", rows[:10], ) _set_await_choice(phone, "store_select") def _select_store(phone: str, vendor_id: str): from vendors.models import Vendor try: vendor = Vendor.objects.filter(is_approved=True).get(id=int(vendor_id)) except (ValueError, Vendor.DoesNotExist): send_whatsapp_text(phone, "I couldn't find that store. Please choose from the list.") _send_store_list(phone) return _set_store_filter(phone, str(vendor.id), vendor.store_name) send_whatsapp_text( phone, f"Perfect. I'll use *{vendor.store_name}* for your shopping actions." ) _send_main_menu(phone, returning=True) def _handle_store_mode_text(phone: str, text: str): t = text.lower().strip() await_choice = _get_await_choice(phone) if await_choice == "store_scope": if any(k in t for k in ("all store", "all stores", "all", "every store")): _set_all_store_scope(phone) send_whatsapp_text(phone, "Great, you'll shop across all stores.") _send_main_menu(phone, returning=True) return if any(k in t for k in ("specific", "one store", "single store", "store")): _send_store_list(phone) return send_whatsapp_text(phone, "Please choose either *Shop all stores* or *Specific store*.") _start_store_scope_mode(phone) return if any(k in t for k in ("all store", "all stores", "every store", "all")): _set_all_store_scope(phone) send_whatsapp_text(phone, "Great, you'll shop across all stores.") _send_main_menu(phone, returning=True) return from vendors.models import Vendor vendor = Vendor.objects.filter( is_approved=True, store_name__icontains=text.strip(), ).order_by("store_name").first() if vendor: _set_store_filter(phone, str(vendor.id), vendor.store_name) send_whatsapp_text(phone, f"I found *{vendor.store_name}*. I'll search only this store now.") _start_search_mode(phone) return send_whatsapp_text(phone, "Please tap a store from the list so I can lock search to it.") _send_store_list(phone) def _apply_store_filter(phone: str, products: list) -> list: store_filter = _get_store_filter(phone) vendor_id = str(store_filter.get("vendor_id", "")).strip() if not vendor_id: return products filtered = [] for p in products: p_vendor_id = str(p.get("vendor_id", "")).strip() if p_vendor_id and p_vendor_id == vendor_id: filtered.append(p) return filtered def _build_catalog_query_from_text(text: str) -> dict: t = (text or "").lower() words = [w for w in re.findall(r"[a-z0-9]+", t) if w and w not in _SEARCH_STOP_WORDS] query = {"q": " ".join(words[:6]) or text.strip()} for category, keys in _CATEGORY_KEYWORDS.items(): if any(k in t for k in keys): query["category"] = category break nums = re.findall(r"\b\d+(?:\.\d+)?\b", t) if nums: try: query["max_price"] = float(nums[0]) except (TypeError, ValueError): pass return query def _filter_catalog_results(products: list, _raw_text: str, query: dict) -> list: if not products: return [] # Prevent cross-category leakage (e.g. watch search showing shirts). category = str(query.get("category", "")).strip().lower() if category: aliases = set(_CATEGORY_KEYWORDS.get(category, (category,))) filtered = [] for p in products: hay = " ".join([ str(p.get("name", "")), str(p.get("description", "")), str(p.get("category", "")), ]).lower() if any(a in hay for a in aliases): filtered.append(p) if filtered: products = filtered return products[:6] def _build_search_intro(products: list, store_filter: dict) -> str: names = [str(p.get("name", "")).strip() for p in products[:3] if str(p.get("name", "")).strip()] preview = ", ".join(names) if names else "matching products" if store_filter.get("vendor_id"): return ( f"Here are matches from *{store_filter.get('store_name', 'your selected store')}*:\n" f"{preview}." ) return f"Here are matches on ACA right now:\n{preview}."