""" PriceOye AI — Conversation Manager ==================================== Handles multi-turn conversation with persistent session memory. User preferences update dynamically — no restart needed. Language: Professional Roman Urdu throughout. """ import re from dataclasses import dataclass, field from typing import Optional from src.scoring_engine import UserPreferences, recommend, find_closest_alternative, get_priority_reason, get_category_scores, get_sub_scores from src.phone_database import build_database PHONE_DB = build_database() IPHONE_MIN_PRICE = 280000 # ───────────────────────────────────────────── # SESSION MEMORY # ───────────────────────────────────────────── @dataclass class SessionMemory: """ Cross-turn memory — survives budget/priority changes. Allows: "always prefer Samsung", "avoid Huawei", etc. """ brand_preference: Optional[str] = None brand_avoid: list[str] = field(default_factory=list) previous_recommendations: list[str] = field(default_factory=list) # phone IDs user_name: Optional[str] = None notes: list[str] = field(default_factory=list) # ───────────────────────────────────────────── # CONVERSATION STATE # ───────────────────────────────────────────── @dataclass class ConversationState: prefs: UserPreferences = field(default_factory=UserPreferences) memory: SessionMemory = field(default_factory=SessionMemory) step: str = "greet" # greet | collecting | result awaiting_new_budget: bool = False last_result_count: int = 1 # ───────────────────────────────────────────── # NLP EXTRACTION # ───────────────────────────────────────────── NON_PHONE_PATTERNS = [ r'\bweather\b|\bcricket\b|\bnews\b|\bpolitics\b|\brecipe\b|\bfootball\b|\bfilm\b|\bsong\b|\bmusic\b', r'\bwhat is\b|\bwho is\b|\bhow to\b|\bexplain\b|\btell me about\b|\bhistory of\b', r'\blaptop\b|\bcomputer\b|\btablet\b|\bipad\b|\bsmartwatch\b|\bheadphone\b|\bspeaker\b', r'\bjoke\b|\bfunny\b|\bstory\b|\bpoem\b|\bessay\b|\btranslate\b', ] PHONE_SIGNAL = r'phone|mobile|camera|battery|gaming|budget|samsung|iphone|android|recommend|suggest|buy|khareedna|lena|chahiye' BRANDS = ["Samsung", "Apple", "OnePlus", "Xiaomi", "Realme", "Huawei", "ASUS", "ROG", "Nothing", "Tecno", "Infinix", "Oppo", "Vivo", "Nokia", "Redmi"] def is_non_phone_query(text: str) -> bool: if re.search(PHONE_SIGNAL, text, re.IGNORECASE): return False return any(re.search(p, text, re.IGNORECASE) for p in NON_PHONE_PATTERNS) def extract_preferences(text: str, current: UserPreferences) -> UserPreferences: """ Parse user text and return UPDATED preferences. Only overwrites fields explicitly mentioned — preserves others. """ t = text.lower() prefs = UserPreferences( budget=current.budget, os_preference=current.os_preference, priority=current.priority, brand_preference=current.brand_preference, brand_avoid=current.brand_avoid.copy(), session_memory=current.session_memory.copy(), ) # ── BUDGET ── lakh_m = re.search(r'([\d.]+)\s*lakh', t) k_m = re.search(r'(\d+(?:\.\d+)?)\s*k\b', t) num_m = re.search(r'\b(\d[\d,]{3,})\b', t) if lakh_m: prefs.budget = int(float(lakh_m.group(1)) * 100000) elif k_m: prefs.budget = int(float(k_m.group(1)) * 1000) elif num_m: n = int(num_m.group(1).replace(',', '')) if 10000 <= n <= 1000000: prefs.budget = n # Quick-reply budget labels qr_budgets = { '30k se kam': 31999, 'under 30k': 31999, '50k se kam': 49999, 'under 50k': 49999, '50k': 99999, '1 lakh': 99999, '1 – 2 lakh': 199999,'2 lakh': 199999, '2 – 3 lakh': 299999,'3 lakh': 299999, '3 lakh se zyada': 500000, } for label, budget in qr_budgets.items(): if label in t: prefs.budget = budget break # ── OS ── if re.search(r'\biphone\b|\bios\b|\bapple\b', t): prefs.os_preference = 'ios' elif re.search(r'\bandroid\b|samsung|xiaomi|oneplus|realme|nothing|tecno|huawei|rog|asus|infinix|redmi|oppo|vivo', t): prefs.os_preference = 'android' elif re.search(r'koi bhi|chalega|any|either|no preference', t): prefs.os_preference = 'any' # ── PRIORITY ── if re.search(r'gaming|game|pubg|cod|bgmi|fps|esport', t): prefs.priority = 'gaming' elif re.search(r'\bcamera\b|photo|selfie|vlog|photography|pic', t): prefs.priority = 'photography' elif re.search(r'battery|long.?last|endurance', t): prefs.priority = 'battery' elif re.search(r'\bvalue\b|sasta|cheap|afford|economical|money|paisa', t): prefs.priority = 'value' elif re.search(r'business|work|professional|office|email', t): prefs.priority = 'business' elif re.search(r'balanced|everyday|general|normal|all.?round|overall', t): prefs.priority = 'balanced' elif re.search(r'fast charg|charg.*fast', t): prefs.priority = 'battery' # ── BRAND PREFERENCE ── for brand in BRANDS: if re.search(rf'always.*{brand.lower()}|prefer.*{brand.lower()}|{brand.lower()}.*prefer', t): prefs.brand_preference = brand if re.search(rf'avoid.*{brand.lower()}|not.*{brand.lower()}|{brand.lower()}.*nahi', t): if brand not in prefs.brand_avoid: prefs.brand_avoid.append(brand) return prefs def extract_count(text: str) -> Optional[int]: t = text.lower() m = re.search(r'(\d+)\s*(phone|option|recomm)', t) if m: n = int(m.group(1)) if 2 <= n <= 5: return n if re.search(r'\bmore\b|another|compar|vs\b|versus|alternative|dikhayein', t): return 3 return None # ───────────────────────────────────────────── # RESPONSE FORMATTERS # ───────────────────────────────────────────── def format_budget(n: int) -> str: if n >= 100000: return f"{n/100000:.1f} lakh" return f"{n//1000}k" def format_phone_summary(phone, final_score: float, priority: str) -> dict: """Returns structured data for frontend rendering.""" cats = get_category_scores(phone) priority_cat_map = { 'photography': 'camera', 'gaming': 'performance', 'battery': 'battery', 'value': 'camera', 'business': 'software', 'balanced': 'camera', 'ios': 'camera', } deep_cat = priority_cat_map.get(priority, 'camera') return { "name": phone.name, "brand": phone.brand, "price_label": phone.price_label, "emoji": phone.emoji, "ai_score": round(final_score, 1), "priceoye_url": phone.priceoye_url, "whatmobile_url": phone.whatmobile_url, "highlights": phone.highlights, "category_scores": {k: round(v, 1) for k, v in cats.items()}, "deep_dive_category": deep_cat, "deep_dive_scores": get_sub_scores(phone, deep_cat), "priority_reason": get_priority_reason(phone, priority), } # ───────────────────────────────────────────── # MAIN CONVERSATION HANDLER # ───────────────────────────────────────────── def handle_message(user_text: str, state: ConversationState) -> dict: """ Process one user message. Returns a response dict: { "text": str, # bot message text "quick_replies": list[str], # button labels "phones": list[dict] | None, # phone card data if recommendation "state": ConversationState, # updated state } """ t = user_text.lower().strip() # ── RESET ── if re.search(r'\bnew search\b|\brestart\b|\bnaya search\b|\breset\b', t): new_state = ConversationState(memory=state.memory) # preserve session memory return _respond( "Theek hai, naya search shuru karte hain.
" "Budget, OS preference aur use case batayein.", ["Camera phone 1 lakh mein", "Gaming phone 2 lakh mein", "iPhone 3 lakh mein", "Budget phone 30k mein"], state=new_state ) # ── NON-PHONE GUARD ── if is_non_phone_query(user_text): return _respond( "Main sirf mobile phone recommendations ke liye design kiya gaya hoon.

" "Apna budget, OS preference aur main use case batayein — " "main aap ke liye Pakistan mein best available phone select karunga.", ["50k se kam mein phone dikhayein", "Best camera phone 1 lakh mein", "Gaming phone 2 lakh mein"], state=state ) # ── AWAITING NEW BUDGET ── if state.awaiting_new_budget: state.prefs = extract_preferences(user_text, state.prefs) state.awaiting_new_budget = False if not state.prefs.budget: return _respond( "Meherbani karke valid budget enter karein. " "Jaise: 1.5 lakh, 80k, ya 200000.", state=state ) state.step = 'collecting' return _advance(state) # ── UPDATE PREFS FROM TEXT ── state.prefs = extract_preferences(user_text, state.prefs) # ── RESULT FOLLOWUP ── if state.step == 'result': return _handle_result_followup(t, state) # ── ADVANCE ── return _advance(state) def _advance(state: ConversationState) -> dict: """Check what's missing and ask, or show result.""" p = state.prefs # iOS + budget too low → guide user if p.os_preference == 'ios' and p.budget and p.budget < IPHONE_MIN_PRICE: bl = format_budget(p.budget) state.prefs.os_preference = None return _respond( f"Pakistan mein sabse sasti iPhone abhi PKR 2,80,000 (iPhone 15) se shuru hoti hai.

" f"Aap ka budget PKR {bl} kisi bhi iPhone ke liye kaafi nahi.

" "Budget barhayen ya Android try karein? Is budget mein kaafi acha Android cameras available hain.", ["Budget Barhana Hai", "Android Dikhayein", "Naya Search"], state=state ) # All 3 collected → recommend if p.budget and p.os_preference is not None and p.priority: state.step = 'result' return _show_result(state, count=1) # Ask what's missing if not p.budget: return _ask_budget(state) if p.os_preference is None: return _ask_os(state) if not p.priority: return _ask_priority(state) return _show_result(state, count=1) def _show_result(state: ConversationState, count: int = 1) -> dict: p = state.prefs results = recommend(p, PHONE_DB, count=count) if not results: # Diagnose if p.os_preference == 'ios' and p.budget and p.budget < IPHONE_MIN_PRICE: msg = ( f"PKR {format_budget(p.budget)} mein koi iPhone available nahi.
" "iPhone 15 ka minimum price PKR 2,84,999 hai.

" "Budget barhayen ya Android select karein." ) else: msg = ( f"PKR {format_budget(p.budget)} mein aap ki criteria ke mutabiq " "koi phone nahi mila.
Budget thoda barhayein ya preferences adjust karein." ) return _respond( msg, ["Budget Barhana Hai", "Preferences Change Karein", "Naya Search"], state=state ) bl = format_budget(p.budget) priority_label = (p.priority or 'balanced').title() if count == 1: phone, score = results[0] state.last_result_count = 1 state.memory.previous_recommendations.append(phone.id) phone_data = [format_phone_summary(phone, score, p.priority or 'balanced')] text = ( f"Aap ke liye main ne yeh best phone select kiya hai:
" f"Budget: PKR {bl}  ·  Priority: {priority_label} " f" ·  AI Score: {score:.1f}/10" ) followup = ( f"Yeh phone kyun best hai?
" f"• {get_priority_reason(phone, p.priority or 'balanced')}
" f"• Aap ke budget PKR {bl} ke andar hai
" f"• AI Score: {score:.1f}/10

" "Aur options dekhna chahenge ya alternatives compare karna hai?" ) return { "text": text, "followup": followup, "quick_replies": ["3 Options Dikhayein", "Top 2 Compare Karein", "Naya Search"], "phones": phone_data, "state": state, } else: medals = ["🥇 Behtar in Sab Mein", "🥈 Doosra Best", "🥉 Aur Ek Option", "4️⃣ Yeh Bhi Dekh Lein"] phones_data = [] for i, (phone, score) in enumerate(results): d = format_phone_summary(phone, score, p.priority or 'balanced') d['medal'] = medals[i] if i < len(medals) else "" phones_data.append(d) state.memory.previous_recommendations.append(phone.id) state.last_result_count = len(results) return { "text": f"Aap ki requirements ke mutabiq AI score se ranked top {len(results)} phones yeh hain:", "followup": "Kisi bhi phone ke baare mein aur detail chahiye? Ya naya search karna hai?", "quick_replies": ["Naya Search"], "phones": phones_data, "state": state, } def _handle_result_followup(t: str, state: ConversationState) -> dict: if re.search(r'budget barhana|increase.*budget|barhao|naya budget', t): state.awaiting_new_budget = True state.prefs.budget = None return _respond( "Naya budget batayein. Jaise: 2 lakh, 150k, ya 300000.", state=state ) if re.search(r'android|dikhayein.*android|switch.*android', t): state.prefs.os_preference = 'android' state.step = 'result' return _show_result(state, count=1) if re.search(r'preferences.*change|change.*prefer|priority.*change', t): state.prefs.priority = None return _ask_priority(state) if re.search(r'3.*option|compare|top.*2|2.*option|vs\b|versus|alternative|dikhayein', t): count = 2 if re.search(r'2.*option|top.*2|compare.*2', t) else 3 return _show_result(state, count=count) if re.search(r'naya search|restart|reset', t): new_state = ConversationState(memory=state.memory) return _respond( "Theek hai, naya search shuru karte hain. Budget, OS aur use case batayein.", ["Camera phone 1 lakh mein", "Gaming phone 2 lakh mein", "iPhone 3 lakh mein", "Budget phone 30k mein"], state=new_state ) return _respond( "Main aap ki aur kya madad kar sakta hoon?", ["3 Options Dikhayein", "Budget Barhana Hai", "Naya Search"], state=state ) def _ask_budget(state: ConversationState) -> dict: return _respond( "Aap ka mobile ka budget kya hai?
" "Jaise likh sakte hain: 50k, 1.5 lakh, ya 200000", ["30k se Kam", "50k se Kam", "50k – 1 Lakh", "1 – 2 Lakh", "2 – 3 Lakh", "3 Lakh se Zyada"], state=state ) def _ask_os(state: ConversationState) -> dict: bl = format_budget(state.prefs.budget) return _respond( f"Budget confirm ho gaya: PKR {bl}

" "Aap iPhone (iOS) prefer karte hain ya Android?", ["iPhone (iOS)", "Android", "Koi Bhi Chalega"], state=state ) def _ask_priority(state: ConversationState) -> dict: p = state.prefs bl = format_budget(p.budget) os_label = "iPhone" if p.os_preference == 'ios' else "Android" if p.os_preference == 'android' else "Any OS" return _respond( f"Budget PKR {bl} ✅  ·  {os_label}

" "Is phone mein aap ko sabse zyada kya chahiye?", ["Camera aur Photography", "Gaming Performance", "Battery Life", "Value for Money", "Balanced All-Round", "Business aur Productivity"], state=state ) def _respond(text: str, quick_replies: list = None, phones=None, state: ConversationState = None) -> dict: return { "text": text, "quick_replies": quick_replies or [], "phones": phones, "state": state, } # ───────────────────────────────────────────── # GREETING # ───────────────────────────────────────────── def get_greeting() -> dict: return { "text": ( "PriceOye AI Phone Advisor mein khush aamdeed! 📱

" "Main aap ki madad karunga Pakistan mein aap ke budget aur requirements " "ke mutabiq best mobile phone dhundhne mein.

" "Data sources: WhatMobile.com.pk aur PriceOye.pk se verified.

" "Seedha bhi bata sakte hain, jaise:
" "\"2 lakh mein best camera Android chahiye\"

" "Ya main aap se step-by-step poochunga." ), "quick_replies": [ "Camera phone 1 lakh mein", "Gaming phone 2 lakh mein", "iPhone 3 lakh mein", "Budget phone 30k mein", ], "phones": None, "state": ConversationState(), }