import gradio as gr
import requests
from bs4 import BeautifulSoup
import re
import os
import urllib.parse
from huggingface_hub import InferenceClient
MODEL_ID = "Qwen/Qwen2.5-7B-Instruct"
PRICE_RE = re.compile(r"(?:₹|Rs\.?|INR)\s*([\d,]+(?:\.\d+)?)")
ASIN_RE = re.compile(r"/(?:dp|gp/product)/([A-Z0-9]{10})")
DDG_HEADERS = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36",
"Accept-Language": "en-US,en;q=0.9",
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
}
def get_client():
token = os.environ.get("HF_TOKEN", "")
return InferenceClient(token=token) if token else InferenceClient()
def clean_price(text: str):
if not text:
return None
m = PRICE_RE.search(str(text))
if m:
raw = m.group(1).replace(",", "")
try:
val = int(float(raw))
if 100 < val < 10_000_000:
return f"₹{val:,}"
except ValueError:
pass
return None
def clean_amazon_link(raw_link: str) -> str:
"""Extract ASIN and return a clean, working Amazon.in product URL."""
if not raw_link:
return None
m = ASIN_RE.search(raw_link)
if m:
return f"https://www.amazon.in/dp/{m.group(1)}"
# If no ASIN, keep only the base path (strip all query params/tracking)
try:
parsed = urllib.parse.urlparse(raw_link)
if "amazon" in parsed.netloc:
clean = parsed._replace(query="", fragment="").geturl()
return clean
except Exception:
pass
return raw_link
def clean_flipkart_link(raw_link: str) -> str:
"""Keep only essential Flipkart URL params, strip tracking."""
if not raw_link:
return None
try:
parsed = urllib.parse.urlparse(raw_link)
qs = urllib.parse.parse_qs(parsed.query)
kept = {}
for k in ("pid", "lid", "marketplace"):
if k in qs:
kept[k] = qs[k]
new_q = urllib.parse.urlencode(kept, doseq=True)
return parsed._replace(query=new_q, fragment="").geturl()
except Exception:
return raw_link
def ddg_search(query: str, num: int = 12):
try:
resp = requests.post(
"https://html.duckduckgo.com/html/",
data={"q": query, "b": "", "kl": "in-en"},
headers=DDG_HEADERS,
timeout=15,
)
soup = BeautifulSoup(resp.text, "lxml")
results = []
for item in soup.select(".result")[:num]:
title_el = item.select_one(".result__title")
snippet_el = item.select_one(".result__snippet")
url_el = item.select_one(".result__url")
link_el = item.select_one(".result__title a")
title = title_el.get_text(" ", strip=True) if title_el else ""
snippet = snippet_el.get_text(" ", strip=True) if snippet_el else ""
url_txt = url_el.get_text(strip=True) if url_el else ""
link = link_el.get("href", "") if link_el else ""
if link and "duckduckgo.com" in link:
try:
qs = urllib.parse.urlparse(link).query
params = urllib.parse.parse_qs(qs)
link = urllib.parse.unquote(params.get("uddg", [link])[0])
except Exception:
pass
results.append({"title": title, "snippet": snippet, "url": url_txt, "link": link})
return results
except Exception:
return []
def normalize_query(raw: str) -> str:
try:
client = get_client()
resp = client.chat_completion(
messages=[
{
"role": "system",
"content": (
"You are a product search query cleaner. "
"Output ONLY a short, clean product name (max 8 words) suitable for searching on "
"Amazon India, Flipkart, and Myntra. No explanation, no punctuation at the end."
),
},
{"role": "user", "content": f"Clean this product name: {raw}"},
],
model=MODEL_ID,
max_tokens=25,
temperature=0.05,
)
cleaned = resp.choices[0].message.content.strip().strip('"').strip("'")
if cleaned and 3 < len(cleaned) < 100:
return cleaned
except Exception as e:
print(f"[normalize_query] {e}")
return raw.strip()
def hf_get_prices(query: str) -> dict:
try:
client = get_client()
resp = client.chat_completion(
messages=[
{
"role": "system",
"content": (
"You are a real-time Indian e-commerce price assistant. "
"You know current approximate prices on Amazon India, Flipkart, and Myntra. "
"Reply with ONLY three lines in this exact format:\n"
"Amazon: ₹PRICE\n"
"Flipkart: ₹PRICE\n"
"Myntra: ₹PRICE\n"
"If a product is not sold on a platform, write N/A. "
"No extra text. No explanations."
),
},
{
"role": "user",
"content": f"Current price of '{query}' on Amazon India, Flipkart, Myntra?",
},
],
model=MODEL_ID,
max_tokens=80,
temperature=0.05,
)
text = resp.choices[0].message.content.strip()
result = {}
for line in text.splitlines():
price = clean_price(line)
if not price:
continue
ll = line.lower()
if "amazon" in ll:
result["amazon"] = price
elif "flipkart" in ll:
result["flipkart"] = price
elif "myntra" in ll:
result["myntra"] = price
return result
except Exception as e:
print(f"[hf_get_prices] {e}")
return {}
def hf_ai_analysis(query: str, amazon: dict, flipkart: dict, myntra: dict) -> str:
lines = []
for r in [amazon, flipkart, myntra]:
p = r.get("price") or "N/A"
lines.append(f"- {r['platform']}: {p}")
scraped_str = "\n".join(lines)
try:
client = get_client()
resp = client.chat_completion(
messages=[
{
"role": "system",
"content": "You are a smart Indian price comparison assistant called 'Come & Compare'.",
},
{
"role": "user",
"content": (
f"Product: {query}\n\nPrices:\n{scraped_str}\n\n"
"Reply in this exact format:\n"
"🏆 BEST DEAL: [platform] at [price]\n\n"
"📊 PRICE RANKING:\n1. [platform] — [price]\n2. ...\n\n"
"💡 BUYING ADVICE:\n[2-3 line recommendation]\n\n"
"⚠️ NOTES:\n[any warnings about unavailable prices]"
),
},
],
model=MODEL_ID,
max_tokens=350,
temperature=0.3,
)
return resp.choices[0].message.content.strip()
except Exception as e:
return f"⚠️ AI analysis unavailable: {str(e)}"
def get_platform_link(results, domain: str, platform: str):
"""Return a clean, working link for the platform."""
for r in results:
url = r.get("url", "")
link = r.get("link", "")
if domain in url or domain in link:
raw = link if link.startswith("http") else ("https://" + url if url else None)
if not raw:
continue
if platform == "Amazon.in":
cleaned = clean_amazon_link(raw)
if cleaned:
return cleaned
elif platform == "Flipkart":
cleaned = clean_flipkart_link(raw)
if cleaned:
return cleaned
else:
return raw
return None
def get_platform_title(results, domain: str):
for r in results:
if domain in r.get("url", "") or domain in r.get("link", ""):
return r.get("title", "")
return ""
def get_product_image(query: str, ddg_results: list):
import json
for r in ddg_results:
link = r.get("link", "")
if "amazon.in" in link or "amazon.com" in link:
try:
resp = requests.get(link, headers=DDG_HEADERS, timeout=8)
soup = BeautifulSoup(resp.text, "lxml")
for sel in ["#landingImage", "#imgBlkFront", ".a-dynamic-image"]:
img = soup.select_one(sel)
if img:
src = img.get("src", "")
if src and src.startswith("http"):
return src
data = img.get("data-a-dynamic-image", "")
if data:
try:
d = json.loads(data)
return max(d.keys(), key=lambda u: d[u][0] * d[u][1])
except Exception:
pass
except Exception:
pass
for r in ddg_results[:5]:
link = r.get("link", "")
if not link or "duckduckgo" in link:
continue
try:
resp = requests.get(link, headers=DDG_HEADERS, timeout=6)
soup = BeautifulSoup(resp.text, "lxml")
og = soup.select_one("meta[property='og:image']")
if og and og.get("content", "").startswith("http"):
return og["content"]
except Exception:
pass
return None
def compare_prices(product_name, product_details, selected_platforms, progress=gr.Progress()):
if not product_name or not product_name.strip():
return (
"⚠️ Please enter a product name.
",
"❌ No product entered.",
"",
)
query = product_name.strip()
if product_details and product_details.strip():
query = f"{query} {product_details.strip()}"
progress(0.05, desc="🤖 Normalizing query with Qwen 7B...")
normalized = normalize_query(query)
progress(0.2, desc="🔍 Searching DuckDuckGo for product links...")
ddg_results = ddg_search(f"{normalized} buy online india amazon flipkart myntra price", num=12)
progress(0.5, desc="💰 Fetching prices via Qwen 7B...")
hf_prices = hf_get_prices(normalized)
progress(0.7, desc="🖼️ Finding product image...")
image_url = get_product_image(normalized, ddg_results)
progress(0.85, desc="🤖 Running AI analysis...")
enc = urllib.parse.quote_plus(normalized)
PLATFORMS = [
{"platform": "Amazon.in", "domain": "amazon.in", "color": "#FF9900", "bg": "#FFF8EE",
"search": f"https://www.amazon.in/s?k={enc}", "price_key": "amazon"},
{"platform": "Flipkart", "domain": "flipkart.com", "color": "#2874F0", "bg": "#EEF4FF",
"search": f"https://www.flipkart.com/search?q={enc}", "price_key": "flipkart"},
{"platform": "Myntra", "domain": "myntra.com", "color": "#FF3F6C", "bg": "#FFF0F4",
"search": f"https://www.myntra.com/{enc}", "price_key": "myntra"},
]
active_keys = {p.lower(): p for p in (selected_platforms or [])}
results = []
for p in PLATFORMS:
if active_keys and not any(k in p["platform"].lower() for k in active_keys):
continue
link = get_platform_link(ddg_results, p["domain"], p["platform"]) or p["search"]
title = get_platform_title(ddg_results, p["domain"])
price = hf_prices.get(p["price_key"])
results.append({**p, "price": price, "title": title, "link": link})
ai_out = hf_ai_analysis(normalized, *results[:3]) if len(results) >= 3 else "Need all 3 platforms for AI analysis."
progress(1.0, desc="✅ Done!")
table_html = _build_cards(results, image_url, normalized)
links_html = _build_links(normalized, results)
return table_html, ai_out, links_html
def _find_best(results):
found = [r for r in results if r.get("price")]
if not found:
return ""
def val(r):
return int(r["price"].replace("₹","").replace(",","").strip())
try:
return min(found, key=val)["platform"]
except Exception:
return ""
def _build_cards(results, image_url, query):
best = _find_best(results)
img_html = ""
if image_url:
img_html = (
f''
f'
🏆 BEST DEAL
' if is_best else ""
price_html = (
f'{price}
'
if price else
'Not Available
'
)
title_html = f'{title}
' if title else ''
btn_html = (
f'View on {r["platform"]} →'
) if price else ""
cards += f'''
{trophy}
{"🛒" if "Amazon" in r["platform"] else "🛍️" if "Flipkart" in r["platform"] else "👗"}
{r["platform"]}
{price_html}
{title_html}
{btn_html}
{cards}
'
has_price = any(r.get("price") for r in results)
no_token_warn = "" if has_price else (
''
'⚠️ No prices found — make sure HF_TOKEN is set in Space Secrets '
'(Settings → Variables and secrets)
'
)
heading = (
f''
f'📦 Results for: {query}
'
)
return f"{no_token_warn}{heading}{img_html}{cards_row}"
def _build_links(query, results):
q = urllib.parse.quote_plus(query)
chips = "".join(
f'{r["platform"]}'
for r in results
)
chips += (
f'🌐 Google Shopping'
)
return f'🔗 Search directly on each platform:
{chips}