"""
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(),
}