""" Enhanced ConversationManager — the central brain of ACA. Improvements: - Strict no-external-platform policy enforced at every stage - Comparison mode: generates side-by-side product comparisons - Better image search flow with richer attribute extraction - Collective learning integration - Smarter product selection and deduplication - Clear "not found on ACA" messaging instead of redirecting elsewhere - Enhanced product relevance filtering """ import logging from urllib.parse import urlparse from typing import Dict, List, Optional from django.conf import settings as dj_settings from chat.models import Conversation, Message from cart.models import Order as CartOrder from gateway.services import ProductAggregator from stores.models import StoreRegistry from .llm_client import LLMClient, UNAVAILABLE_MESSAGE from .intent_layerer import enrich_query, build_prompt_context from .attribute_weighter import rank_products from .feedback_learner import get_user_context_summary from .query_builder import QueryBuilder logger = logging.getLogger(__name__) CREATOR_RESPONSE = ( "ACA was created by Mirjy Technologies Ltd. " "Learn more at https://mirjy.com or contact support@mirjy.com." ) ACA_INFO_RESPONSE = ( "ACA means AI Commerce Aggregator. " "You can visit ACA at https://aca.mirjy.com." "You can WhatsApp ACA at +233597224964." ) SYSTEM_PROMPT = """\ You are ACA, a friendly AI shopping assistant. You help users find products from partner stores that are registered on the ACA platform. PERSONALITY: Warm, casual, and conversational. Talk like a helpful friend. Keep things short and natural. No corporate speak. FORMATTING RULES (STRICT — NEVER VIOLATE): 1. NEVER use emojis. Not a single one. Ever. 2. NEVER use markdown. No **bold**, no *italics*, no ## headings. 3. NEVER use bullet points or dashes to make lists. Write in natural sentences and short paragraphs instead. 4. Use plain text only. Line breaks between paragraphs are fine. CRITICAL BEHAVIOR — NO EXCEPTIONS: - You ONLY recommend products that exist in the search data given to you. - NEVER suggest external platforms: Google, Jumia, Jiji, Amazon, eBay, Alibaba, AliExpress, Konga, Tonaton, Facebook Marketplace, Instagram, TikTok Shop, or ANY service outside ACA. This is absolutely forbidden. - If the user asks "where else can I find it?" — you say you can only help them search within ACA's partner stores. - If no products are found on ACA, say clearly: "I couldn't find that on ACA right now." Suggest trying different keywords, a broader category, or a different budget range. NEVER say "try Google" or "check Amazon" or anything of that sort. - NEVER invent products, prices, or descriptions. Everything you say about products must come from the data you receive. - NEVER change or round prices. CRITICAL SEARCH BEHAVIOR: - The search has ALREADY happened before you respond. - NEVER say "let me search", "give me a moment", "searching now", "hold on", or anything implying you are about to search. You either have results or you don't. - Present results directly. WHEN CHATTING CASUALLY: Respond warmly and naturally. Have a real conversation. You can mention that you help with shopping on ACA, but do not force it. Don't entertain non-shopping questions for too long, keep it short and sweet. CREATOR / COMPANY QUESTIONS: If the user asks who created you, who built this, who owns this, your company, developer, maker, founder, or anything similar, respond with: "ACA was created by Mirjy Technologies Ltd. Learn more at https://mirjy.com or contact support@mirjy.com." WHEN THE USER WANTS TO SHOP: Ask one or two follow-up questions at a time to narrow things down. Keep it conversational. Ask about category, budget, size, color, brand, or purpose. Never assume gender from product type — ask if it matters and is not stated. WHEN SHOWING PRODUCTS: Be concise. Mention what stands out about each option, the price, and the store name. When products are available, present 2-5 options — include variety when possible. WHEN NOTHING IS FOUND: Be straight about it. Do NOT suggest external platforms. Say something like: "Nothing matched that on ACA right now. Want to try with a different color, a wider budget, or a different category? I can search again." COMPARISON MODE: When the user asks to compare products, the comparison text is provided to you. Present it naturally, then ask if they want to narrow down further. """ INTENT_PROMPT = """\ Classify the user's latest message into one of these intents: - "shopping" — they mention a product, want to buy something, ask about prices, brands, categories, sizes, comparisons, or anything purchase-related. - "chat" — casual conversation, greetings, jokes, questions about you, small talk, feelings, opinions, or anything NOT related to shopping. - "unclear" — could go either way. Consider the full conversation context. If someone was shopping earlier and says "what about a blue one?" that is still "shopping". Respond with ONLY one word: shopping, chat, or unclear. """ ALTERNATIVE_INTENT_PROMPT = """\ Decide if the user's latest message indicates they want DIFFERENT product options than what was just shown. Return "yes" if they reject/dislike shown items or ask for alternatives. Return "no" if they ask details about shown items or show interest. Respond with ONLY one word: yes or no. """ class ConversationManager: """Stateless orchestrator — all state lives in the DB.""" def __init__(self): self.llm = LLMClient() self.query_builder = QueryBuilder(self.llm) self.aggregator = ProductAggregator() def handle_message( self, conversation: Conversation, user_text: str, forced_query: Optional[dict] = None, ) -> dict: # Save user message Message.objects.create( conversation=conversation, role="user", content=user_text, ) # ── Order status path ────────────────────────────────────── order, used_number = self._find_order_by_text(user_text) if used_number: if order: order_card = self._build_order_card_payload(order, used_number) Message.objects.create( conversation=conversation, role="assistant", content="", product_data=None, metadata={ "intent": "chat", "source": "order_status", "order_number": used_number, "order_card": order_card, }, ) return { "content": "", "products": None, "metadata": { "intent": "chat", "source": "order_status", "order_number": used_number, "order_card": order_card, }, } else: assistant_text = ( f"I couldn't find an order with the number you provided. " "Please check the format: ACA-1234-5678-YY and try again." ) Message.objects.create( conversation=conversation, role="assistant", content=assistant_text, product_data=None, metadata={"intent": "chat", "source": "order_status_not_found"}, ) return {"content": assistant_text, "products": None, "metadata": {"intent": "chat"}} if self._is_order_status_request(user_text): assistant_text = ( "Sure, I can check your order. " "Please share your order number in this format: ACA-1234-5678-YY." ) Message.objects.create( conversation=conversation, role="assistant", content=assistant_text, product_data=None, metadata={"intent": "chat", "source": "order_status_prompt"}, ) return {"content": assistant_text, "products": None, "metadata": {"intent": "chat"}} # ── Creator question ─────────────────────────────────────── if self._is_creator_question(user_text): assistant_text = CREATOR_RESPONSE Message.objects.create( conversation=conversation, role="assistant", content=assistant_text, product_data=None, metadata={"intent": "chat", "source": "creator_rule"}, ) return {"content": assistant_text, "products": None, "metadata": {"intent": "chat"}} # ── ACA brand/website question ───────────────────────────── if self._is_aca_info_question(user_text): assistant_text = ACA_INFO_RESPONSE Message.objects.create( conversation=conversation, role="assistant", content=assistant_text, product_data=None, metadata={"intent": "chat", "source": "aca_info_rule"}, ) return {"content": assistant_text, "products": None, "metadata": {"intent": "chat"}} # ── Store link request ───────────────────────────────────── requested_store = self._extract_referenced_store(user_text) if requested_store and self._is_store_link_request(user_text): domain_url = self._store_domain_url(requested_store.base_api_url) assistant_text = f"{requested_store.name} website is {domain_url}" Message.objects.create( conversation=conversation, role="assistant", content=assistant_text, product_data=None, metadata={"intent": "chat", "source": "store_link_rule"}, ) return {"content": assistant_text, "products": None, "metadata": {"intent": "chat"}} history = self._build_history(conversation) # ── Social / small-talk guard ────────────────────────────── if self._is_social_request(user_text): if not self.llm.available: assistant_text = "Hi! Happy to chat. How are you doing today?" else: llm_messages = [{"role": "system", "content": SYSTEM_PROMPT}] llm_messages.extend(history) llm_messages.append({ "role": "system", "content": ( "This is social conversation. Respond warmly and naturally. " "Do NOT pivot to shopping. Do NOT mention external platforms." ), }) assistant_text = self.llm.chat(llm_messages) if assistant_text == UNAVAILABLE_MESSAGE: assistant_text = ( "Hi! Happy to chat. How are you doing today?" ) Message.objects.create( conversation=conversation, role="assistant", content=assistant_text, product_data=None, metadata={"intent": "chat", "source": "social_chat"}, ) return {"content": assistant_text, "products": None, "metadata": {"intent": "chat"}} # ── Detect intent ────────────────────────────────────────── intent = "shopping" if forced_query else self._detect_intent(history) # ── Comparison mode detection ────────────────────────────── is_comparison_request = False if intent in ("shopping", "unclear"): is_comparison_request = self.llm.classify_comparison_intent(history, user_text) products = None selected_products = None product_context = "" comparison_text = "" forced_catalog_request = bool(requested_store and self._is_store_catalog_request(user_text)) query_params = {} if intent in ("shopping", "unclear"): base_slots = dict(forced_query or self.query_builder.build(history)) enriched_slots = enrich_query(base_slots, user_text or "") query_params = dict(enriched_slots) if requested_store: query_params["store_slug"] = requested_store.slug if forced_catalog_request: query_params = {"store_slug": requested_store.slug} has_enough_info = bool(query_params) if has_enough_info: search_query = { k: v for k, v in query_params.items() if k not in ("image_search", "image_match_mode", "image_b64") and not str(k).startswith("_") } products = self.aggregator.search(search_query) # ── Image search fallback chain (broaden query if empty) ── if query_params.get("image_search"): if not products: products = self._image_search_with_fallbacks( search_query, query_params ) # ── Filter out previously shown products if user wants alternatives ── if products and self._wants_other_options(history, user_text): seen_keys = self._collect_previously_suggested_keys(conversation) products = [ p for p in products if self._product_key(p) not in seen_keys ] # ── LLM relevance filter — skip for photo search (visual pass handles it) ── if products and not query_params.get("image_search"): products = self._filter_relevant_products(products, query_params) # ── Attribute-based re-ranking with personal + collective signals ── if products: user_ctx = get_user_context_summary( user=getattr(conversation, "user", None), session_key=getattr(conversation, "session_key", "") or "", ) products = rank_products( products, enriched_slots, user_ctx, raw_text=user_text or "" ) # ── Visual comparison vs user's image (web + WhatsApp) ───────────── if ( products and query_params.get("image_search") and query_params.get("image_b64") ): products, _mq, _notice = self.llm.filter_products_by_reference_image( query_params["image_b64"], products, max_items=3, pool_limit=12, ) query_params["_image_match_quality"] = _mq query_params["_image_user_notice"] = _notice _max_cards = 3 if query_params.get("image_search") else 6 if products: # For comparison: pick top candidates if is_comparison_request: compare_candidates = self._select_for_comparison( products, max_items=_max_cards ) comparison_text = self.llm.generate_comparison( compare_candidates, user_query=user_text ) selected_products = compare_candidates else: selected_products = self._select_product_options( products, max_items=_max_cards ) product_context = self._format_products_for_llm(selected_products) # ── Build LLM response ───────────────────────────────────── llm_messages = [{"role": "system", "content": SYSTEM_PROMPT}] llm_messages.extend(history) if comparison_text and selected_products: llm_messages.append({ "role": "system", "content": ( f"Comparison data for {len(selected_products)} products:\n\n" f"{comparison_text}\n\n" f"Present this comparison naturally in plain text. " f"Be concise and helpful. Ask if they want to narrow down further. " f"Do NOT mention any platforms outside ACA." ), }) elif product_context: intent_context_str = "" try: intent_context_str = build_prompt_context(query_params) except Exception: pass image_note = "" if query_params.get("image_search"): image_note = ( " These items were chosen after comparing YOUR uploaded image to each " "product photo. Only related product types are included. " "Do not mention products outside this list or unrelated categories." ) safe_query = { k: v for k, v in query_params.items() if not str(k).startswith("_") and k not in ("image_b64", "image_search", "image_match_mode") } llm_messages.append({ "role": "system", "content": ( f"Product search results for query {safe_query}:\n\n" f"{product_context}\n\n" f"Present these results NOW in ONE short line (8-15 words). " f"Do NOT describe products individually — the UI shows product cards. " f"Do NOT mention any platforms outside ACA. " f"Offer to compare or help decide only if the user asks." f"{image_note}" + (f"\nContext: {intent_context_str}" if intent_context_str else "") ), }) elif intent == "shopping" and not products: # Nothing found — clear messaging, no external redirects category = str(query_params.get("category", "")).strip() brand = str(query_params.get("brand", "")).strip() what = brand or category or "that" llm_messages.append({ "role": "system", "content": ( f"Search for '{what}' returned no results on ACA. " f"Tell the user clearly that nothing was found on ACA right now. " f"Suggest they try: a broader category, different color, wider budget, " f"or removing the brand filter. " f"NEVER suggest external platforms like Google, Jumia, Amazon, etc. " f"Keep it short and warm. One or two sentences max." ), }) # ── Generate response text ───────────────────────────────── if not self.llm.available: if selected_products: assistant_text = self._build_fallback_product_response( selected_products, query_params ) elif comparison_text: assistant_text = comparison_text[:500] else: assistant_text = self._build_fallback_chat_prompt( intent, query_params, user_text or "" ) else: assistant_text = self.llm.chat(llm_messages) if assistant_text == UNAVAILABLE_MESSAGE: if selected_products: assistant_text = self._build_fallback_product_response( selected_products, query_params ) else: assistant_text = self._build_fallback_chat_prompt( intent, query_params, user_text or "" ) img_notice = (query_params or {}).get("_image_user_notice", "").strip() if img_notice and selected_products and query_params.get("image_search"): assistant_text = f"{img_notice}\n\n{assistant_text}" # ── Save assistant message ───────────────────────────────── safe_meta_query = { k: v for k, v in query_params.items() if not str(k).startswith("_") and k not in ("image_b64",) } if query_params else {} meta = {"query": safe_meta_query, "intent": intent} if safe_meta_query else {"intent": intent} if is_comparison_request: meta["mode"] = "comparison" if comparison_text: meta["comparison_text"] = comparison_text if query_params.get("_image_match_quality"): meta["image_match_quality"] = query_params["_image_match_quality"] Message.objects.create( conversation=conversation, role="assistant", content=assistant_text, product_data=selected_products if selected_products else None, metadata=meta, ) return { "content": assistant_text, "products": selected_products if selected_products else None, "metadata": meta, } # ── Image search helpers ─────────────────────────────────────── def _image_search_with_fallbacks( self, search_query: dict, query_params: dict ) -> List[dict]: """ Progressive fallback chain for image-based searches. Tries increasingly broad queries until results are found. """ # Try 1: exact query minus brand broadened = self._broaden_image_query(search_query) if broadened: results = self.aggregator.search(broadened) if results: return results # Try 2: category-only guess from q tokens cat_guess = self._guess_category_from_text(search_query.get("q", "")) if cat_guess: results = self.aggregator.search({"category": cat_guess}) if results: return results # Try 3: extract key style words from q q_text = str(search_query.get("q", "")).strip() if q_text: style_words = self._extract_style_keywords(q_text) if style_words: results = self.aggregator.search({"q": style_words}) if results: return results return [] @staticmethod def _extract_style_keywords(text: str) -> str: """Extract meaningful style/aesthetic keywords from image query text.""" style_terms = [ "maxi", "mini", "midi", "crop", "oversized", "fitted", "slim", "floral", "striped", "plaid", "solid", "printed", "embroidered", "vintage", "retro", "modern", "classic", "streetwear", "casual", "formal", "elegant", "sporty", "bohemian", "minimal", "leather", "denim", "suede", "linen", "silk", "velvet", "knit", "lace", "sequin", "mesh", ] tokens = text.lower().split() found = [t for t in style_terms if any(t in tok for tok in tokens)] return " ".join(found[:4]) if found else "" # ── Comparison helpers ───────────────────────────────────────── @staticmethod def _select_for_comparison(products: List[dict], max_items: int = 4) -> List[dict]: """ Select the best candidates for comparison — prefer variety of brands/stores. """ if not products: return [] if len(products) <= max_items: return products selected = [] seen_brands = set() seen_stores = set() # First pass: one per brand for p in products: brand = (p.get("brand") or "").lower() store = (p.get("store_slug") or p.get("store_name") or "").lower() if brand and brand not in seen_brands: selected.append(p) seen_brands.add(brand) seen_stores.add(store) if len(selected) >= max_items: return selected # Second pass: fill by store diversity for p in products: if p not in selected: store = (p.get("store_slug") or p.get("store_name") or "").lower() if store not in seen_stores: selected.append(p) seen_stores.add(store) if len(selected) >= max_items: break # Final fill for p in products: if p not in selected: selected.append(p) if len(selected) >= max_items: break return selected # ── Product filtering ────────────────────────────────────────── def _filter_relevant_products( self, products: List[dict], query_params: dict ) -> List[dict]: """ Use the LLM to remove products that don't match the user's intent. Strictly limits results to ACA products only. """ if not products or not self.llm.available: return products category = str(query_params.get("category", "")).strip() q_text = str(query_params.get("q", "")).strip() search_intent = f"{category} {q_text}".strip() if not search_intent: return products lines = [] for i, p in enumerate(products, 1): name = p.get("name", "Unknown") cat = p.get("category", "") brand = p.get("brand", "") desc_snippet = (p.get("description") or "")[:80] parts = [name] if brand: parts.append(f"brand:{brand}") if cat: parts.append(f"cat:{cat}") if desc_snippet: parts.append(f"desc:{desc_snippet}") lines.append(f"{i}. {' | '.join(parts)}") product_list = "\n".join(lines) filter_messages = [ { "role": "system", "content": ( "You are a product relevance classifier for an online shopping assistant. " f'The user is looking for: "{search_intent}".\n\n' "Here are the search results:\n" f"{product_list}\n\n" "Using your knowledge of products and brands, return ONLY the numbers " "of products that are genuinely relevant to what the user wants.\n\n" "Rules:\n" "- Be strict: a necklace is NOT a watch.\n" "- A product from a known relevant brand IS relevant even if category is vague.\n" "- Return comma-separated numbers, e.g.: 1,3,5\n" "- If NONE are relevant, return: none" ), }, {"role": "user", "content": "Which products match?"}, ] try: result = self.llm.chat(filter_messages, temperature=0.0) result = result.strip().lower() if result == "none": return [] relevant_indices = set() for part in result.replace("\n", ",").split(","): part = part.strip().rstrip(".") if part.isdigit(): idx = int(part) - 1 if 0 <= idx < len(products): relevant_indices.add(idx) if relevant_indices: filtered = [products[i] for i in sorted(relevant_indices)] logger.info( "Relevance filter: %d/%d products kept for '%s'", len(filtered), len(products), search_intent, ) return filtered except Exception as exc: logger.warning("LLM relevance filter failed: %s", exc) return products @staticmethod def _select_product_options( products: List[dict], max_items: int = 6 ) -> List[dict]: """Pick a diverse set: prefer one per store, then fill by rank.""" if not products: return [] selected = [] seen_stores = set() for p in products: store_key = p.get("store_slug") or p.get("store_name") or "store" if store_key not in seen_stores: selected.append(p) seen_stores.add(store_key) if len(selected) >= max_items: return selected for p in products: if p not in selected: selected.append(p) if len(selected) >= max_items: break return selected def _wants_other_options( self, history: List[dict], user_text: str ) -> bool: """Detect when user wants alternatives.""" text = (user_text or "").lower().strip() if not text: return False # LLM semantic detection llm_signal = self._llm_wants_other_options(history) if llm_signal is not None: return llm_signal alt_signals = [ "other option", "other options", "more option", "more options", "another option", "another one", "something else", "else", "show me more", "show me other", "different one", "different options", "i don't like", "i dont like", "not this", "not these", "don't want this", "not interested", "next one", "any others", ] return any(sig in text for sig in alt_signals) def _llm_wants_other_options(self, history: List[dict]) -> Optional[bool]: """LLM-based classifier for alternative-seeking intent.""" if not self.llm.available: return None messages = [ {"role": "system", "content": ALTERNATIVE_INTENT_PROMPT}, *history[-8:], {"role": "user", "content": "Does the latest user message ask for different options?"}, ] try: result = self.llm.chat(messages, temperature=0.0).strip().lower() if result == "yes": return True if result == "no": return False if "yes" in result: return True if "no" in result: return False except Exception: return None return None # ── Intent detection ─────────────────────────────────────────── def _detect_intent(self, history: List[dict]) -> str: """Use the LLM to classify intent, with keyword fallback.""" if not self.llm.available: return self._naive_intent(history) last_user = "" for m in reversed(history): if m["role"] == "user": last_user = m["content"].lower().strip() break if not last_user: return "chat" messages = [ {"role": "system", "content": INTENT_PROMPT}, *history[-6:], {"role": "user", "content": "Classify the intent of the latest user message."}, ] try: result = self.llm.chat(messages, temperature=0.0).strip().lower() if result in ("shopping", "chat", "unclear"): return result if "shopping" in result: return "shopping" if "chat" in result: return "chat" return "unclear" except Exception: return self._naive_intent(history) @staticmethod def _naive_intent(history: List[dict]) -> str: """Keyword-based intent detection fallback.""" last_user = "" for m in reversed(history): if m.get("role") == "user": last_user = m["content"].lower().strip() break chat_signals = { "hi", "hello", "hey", "how are you", "what's up", "sup", "yo", "good morning", "good evening", "good night", "thanks", "thank you", "bye", "goodbye", "who are you", "what are you", "what can you do", "lol", "haha", "nice", "cool", "okay", "ok", "sure", "great", } for signal in chat_signals: if (last_user == signal or last_user.startswith(signal + " ") or last_user.startswith(signal + "!")): return "chat" shop_signals = [ "buy", "price", "cost", "cheap", "expensive", "budget", "looking for", "i need", "i want", "show me", "find me", "recommend", "suggest", "search", "shop", "order", "size", "color", "brand", "compare", "alternative", "shoes", "shirt", "dress", "laptop", "phone", "bag", "watch", "perfume", "jewelry", "bag", "jacket", "under $", "less than", "more than", ] for signal in shop_signals: if signal in last_user: return "shopping" return "unclear" # ── Order status helpers ─────────────────────────────────────── @staticmethod def _extract_order_candidate(text: str) -> Optional[str]: import re if not text: return None t = text.upper() m = re.search(r"ACA\D*([0-9]{4})\D*([0-9]{4})\D*([0-9]{2})", t) if m: return f"ACA-{m.group(1)}-{m.group(2)}-{m.group(3)}" m2 = re.search(r"ACA\D*([0-9]{10,})", t) if m2: digits = re.sub(r"\D", "", m2.group(1)) if len(digits) >= 10: d = digits[:10] return f"ACA-{d[0:4]}-{d[4:8]}-{d[8:10]}" return None @staticmethod def _build_order_card_payload(order: "CartOrder", used_number: Optional[str]) -> dict: from django.utils import timezone placed = timezone.localtime(order.created_at).strftime("%b %d, %Y %H:%M") card = { "order_number": used_number or order.order_number, "placed": placed, "total": float(order.total_amount), "vendors": [], } vorders = list(order.vendor_orders.select_related("vendor").all()) items = list(order.items.select_related("vendor").all()) for vo in vorders: vendor_name = vo.vendor.store_name if vo.vendor else "Vendor" phone = (vo.vendor.phone if vo.vendor else "") or "N/A" vendor_entry = { "name": vendor_name, "status": vo.get_status_display(), "amount": float(vo.total_amount), "phone": phone, "items": [], } for it in items: if it.vendor_id == vo.vendor_id: vendor_entry["items"].append({ "name": it.product_name, "qty": it.quantity, "currency": it.currency, "subtotal": float(it.subtotal), }) card["vendors"].append(vendor_entry) return card @staticmethod def _derive_overall_status(order: "CartOrder") -> str: statuses = list(order.vendor_orders.values_list("status", flat=True)) if not statuses: return order.get_status_display() unique = set(statuses) if len(unique) == 1: s = unique.pop() return dict(order.STATUS_CHOICES).get(s, s.title()) return "Mixed" @staticmethod def _normalize_order(num: str) -> str: import re t = (num or "").upper() return re.sub(r"[^A-Z0-9]", "", t) @staticmethod def _lev(a: str, b: str) -> int: if a == b: return 0 if len(a) < len(b): a, b = b, a prev = list(range(len(b) + 1)) for i, ca in enumerate(a, 1): cur = [i] for j, cb in enumerate(b, 1): cur.append(min(cur[j-1]+1, prev[j]+1, prev[j-1]+(0 if ca == cb else 1))) prev = cur return prev[-1] def _find_order_by_text( self, text: str ) -> tuple: cand = self._extract_order_candidate(text) if cand: order = CartOrder.objects.filter(order_number=cand).first() if order: return order, cand if cand: target_norm = self._normalize_order(cand) numbers = list( CartOrder.objects.order_by("-created_at") .values_list("order_number", flat=True)[:500] ) best = None best_d = 3 for n in numbers: nn = self._normalize_order(n) d = self._lev(target_norm, nn) if d < best_d: best = n best_d = d if d == 0: break if best and best_d <= 2: return CartOrder.objects.filter(order_number=best).first(), best return None, cand return None, None @staticmethod def _is_order_status_request(text: str) -> bool: t = (text or "").lower() signals = [ "order status", "check my order", "track my order", "track order", "where is my order", "order update", "order tracking", "check on my order", "status of my order", ] return any(sig in t for sig in signals) # ── Creator / store detection ────────────────────────────────── @staticmethod def _is_creator_question(user_text: str) -> bool: text = (user_text or "").lower().strip() if not text: return False strong_phrases = [ "who created you", "who made you", "who built you", "who owns you", "who is your owner", "who is behind aca", "who is behind this", "who developed you", "who is your developer", "what company made you", "which company made you", "who founded", "parent company", ] if any(p in text for p in strong_phrases): return True company_patterns = [ "about your company", "about the company", "your company name", "tell me about mirjy", "tell me about your company", "what is mirjy", "who is mirjy", ] if any(p in text for p in company_patterns): return True return False @staticmethod def _is_social_request(user_text: str) -> bool: t = (user_text or "").lower().strip() if not t: return False social_signals = [ "i want a friend", "i need a friend", "be my friend", "i'm bored", "im bored", "talk to me", "let's chat", "lets chat", "i feel lonely", "i'm lonely", "how are you", "how's your day", "hello", "hi", "hey", "what's up", "whats up", ] return any(sig in t for sig in social_signals) @staticmethod def _is_aca_info_question(user_text: str) -> bool: text = (user_text or "").lower().strip() if not text: return False asks_meaning = any( p in text for p in ( "what is aca", "what does aca mean", "full meaning of aca", "aca meaning", "meaning of aca", "aca stands for", ) ) asks_link = ( "aca" in text and any(k in text for k in ("link", "website", "site", "url", "web address", "domain")) ) asks_direct_domain = "aca.mirjy.com" in text return asks_meaning or asks_link or asks_direct_domain @staticmethod def _is_store_link_request(user_text: str) -> bool: text = (user_text or "").lower() if not text: return False keywords = ["domain", "link", "url", "website", "site", "web address"] ask_terms = ["what", "give", "send", "share", "show"] return any(k in text for k in keywords) and any(a in text for a in ask_terms) @staticmethod def _is_store_catalog_request(user_text: str) -> bool: text = (user_text or "").lower() if not text: return False catalog_terms = ["catalog", "catalogue", "products", "items", "inventory", "collection"] request_terms = ["show", "list", "see", "get", "what", "have", "from"] return any(t in text for t in catalog_terms) and any(r in text for r in request_terms) @staticmethod def _extract_referenced_store(user_text: str) -> Optional[StoreRegistry]: text = (user_text or "").lower().strip() if not text: return None stores = list(StoreRegistry.objects.filter(is_active=True)) if not stores: return None for store in stores: name = (store.name or "").lower().strip() slug = (store.slug or "").lower().strip() if name and name in text: return store if slug and slug in text: return store domain = ConversationManager._store_domain_url(store.base_api_url).lower() host = urlparse(domain).netloc if host and host in text: return store return None # ── Static helpers ───────────────────────────────────────────── @staticmethod def _broaden_image_query(q: dict) -> Optional[dict]: if not isinstance(q, dict): return None out = {k: v for k, v in q.items() if v not in (None, "", [], {})} out.pop("brand", None) out.pop("search_keywords", None) text = str(out.get("q", "")).lower() cat = str(out.get("category", "")).lower() if not cat: cat_guess = ConversationManager._guess_category_from_text(text) if cat_guess: out["category"] = cat_guess if out.get("category"): out["q"] = out["category"] elif text: for token in ( "watch", "watches", "shoe", "shoes", "bag", "bags", "perfume", "fragrance", "jewelry", "dress", "shirt", "jacket", "pants", "sneaker", ): if token in text: out["q"] = token break return out @staticmethod def _guess_category_from_text(text: str) -> Optional[str]: t = (text or "").lower() if any(k in t for k in ["watch", "wristwatch", "chronograph", "timepiece"]): return "watches" if any(k in t for k in ["shoe", "shoes", "sneaker", "trainer", "boot", "footwear", "heel", "sandal"]): return "shoes" if any(k in t for k in ["bag", "handbag", "tote", "backpack", "purse", "crossbody"]): return "bags" if any(k in t for k in ["perfume", "fragrance", "cologne", "scent"]): return "perfumes" if any(k in t for k in ["jewelry", "jewellery", "necklace", "bracelet", "ring", "earring"]): return "jewelry" if any(k in t for k in ["dress", "gown", "maxi"]): return "dresses" if any(k in t for k in ["shirt", "top", "blouse", "tee"]): return "shirts" if any(k in t for k in ["pants", "trouser", "jeans", "chino"]): return "pants" if any(k in t for k in ["jacket", "coat", "hoodie", "blazer"]): return "jackets" return None @staticmethod def _product_key(product: dict) -> str: store = product.get("store_slug") or product.get("store_name") or "store" ext = product.get("external_id") if ext: return f"{store}:{ext}" url = product.get("product_url") if url: return f"{store}:{url}" name = product.get("name", "") price = product.get("price", "") return f"{store}:{name}:{price}" def _collect_previously_suggested_keys(self, conversation: Conversation) -> set: keys = set() past_msgs = conversation.messages.filter(role="assistant").exclude( product_data__isnull=True ) for msg in past_msgs: pdata = msg.product_data or [] if isinstance(pdata, list): for p in pdata: if isinstance(p, dict): keys.add(self._product_key(p)) return keys def _build_history(self, conversation: Conversation) -> List[dict]: """ Sliding window of the *most recent* user/assistant messages only. Long-lived WhatsApp threads (session_key wa:…) use tighter limits so input tokens stay bounded across days of chatting. Note: older code used the oldest N messages by mistake; we now take the latest N in chronological order. """ sk = (conversation.session_key or "") wa_prefix = getattr(dj_settings, "WHATSAPP_SESSION_PREFIX", "wa:") if sk.startswith(wa_prefix): max_msgs = int(getattr(dj_settings, "WHATSAPP_CHAT_HISTORY_MAX_MESSAGES", 18)) max_msg_chars = int(getattr(dj_settings, "WHATSAPP_CHAT_HISTORY_MAX_MSG_CHARS", 1800)) max_total = int(getattr(dj_settings, "WHATSAPP_CHAT_HISTORY_MAX_TOTAL_CHARS", 28000)) else: max_msgs = int(getattr(dj_settings, "CHAT_HISTORY_MAX_MESSAGES", 30)) max_msg_chars = int(getattr(dj_settings, "CHAT_HISTORY_MAX_MSG_CHARS", 4000)) max_total = int(getattr(dj_settings, "CHAT_HISTORY_MAX_TOTAL_CHARS", 45000)) qs = ( conversation.messages.filter(role__in=("user", "assistant")) .order_by("-created_at")[:max_msgs] ) msgs = list(reversed(qs)) out: List[dict] = [] for m in msgs: c = (m.content or "") if len(c) > max_msg_chars: c = c[: max_msg_chars - 24] + "\n…[message truncated]" out.append({"role": m.role, "content": c}) while out and sum(len(x["content"]) for x in out) > max_total: out.pop(0) return out @staticmethod def _build_fallback_product_response(products: List[dict], query: dict) -> str: count = len(products) search_terms = [] if query.get("category"): search_terms.append(str(query["category"]).strip()) if query.get("q"): search_terms.append(str(query["q"]).strip()) desc = " ".join([t for t in search_terms if t]) or "your search" return f"Found {count} option{'s' if count != 1 else ''} for {desc}. Click a card to view details." @staticmethod def _build_fallback_chat_prompt(intent: str, query: dict, user_text: str) -> str: t = (user_text or "").strip() if intent == "chat": if t: return "Got it. Tell me more — what's on your mind?" return "Hi! Happy to chat. How can I help today?" hints = [] if query.get("category"): hints.append(str(query.get("category"))) if query.get("brand"): hints.append(str(query.get("brand"))) hint = " ".join([h for h in hints if h.strip()]) or "that" return f"Okay, looking for {hint}. Any budget or color in mind?" @staticmethod def _format_products_for_llm(products: List[dict]) -> str: lines = [] for i, p in enumerate(products, 1): lines.append( f"{i}. {p.get('name', 'N/A')} — " f"{p.get('currency', '$')}{p.get('price', '?')} " f"at {p.get('store_name', 'Unknown Store')} " f"(Brand: {p.get('brand', 'N/A')}, " f"In stock: {p.get('in_stock', '?')}, " f"Category: {p.get('category', 'N/A')})" ) return "\n".join(lines) @staticmethod def _store_domain_url(base_api_url: str) -> str: parsed = urlparse((base_api_url or "").strip()) if not parsed.scheme or not parsed.netloc: return base_api_url return f"{parsed.scheme}://{parsed.netloc}"