import requests import streamlit as st import torch import clip import os import sys from PIL import Image import math import statistics from datetime import datetime import pandas as pd from inference import inference def resource_path(relative_path): try: base_path = sys._MEIPASS except Exception: base_path = os.path.abspath(".") return os.path.join(base_path, relative_path) st.session_state.device = "cuda" if torch.cuda.is_available() else "cpu" def get_clip_model(device): CACHE_DIR = "/tmp/clip_cache" os.makedirs(CACHE_DIR, exist_ok=True) model, preprocess = clip.load("ViT-B/32", device=device, download_root=CACHE_DIR, jit=False) model.to(device).eval() return model, preprocess model, preprocess = get_clip_model(st.session_state.device) product_labels = [ "laptop", "headphones", "smartphone", "tablet", "wireless mouse", "gaming keyboard", "refrigerator", "microwave", "television", "air conditioner", "washing machine", "vacuum cleaner", "running shoes", "leather shoes", "formal shirt", "hoodie", "t-shirt", "jeans", "jacket", "sneakers", "wristwatch", "smartwatch", "sunglasses", "handbag", "backpack", "wallet", "duffel bag", "blender", "water bottle", "camera", "tripod", "drone", "dslr camera", "hair dryer", "makeup kit", "perfume", "book", "notebook", "pen set", "electric kettle", "rice cooker", "pressure cooker", "fan", "heater", "toaster", "gaming console", "joystick", "earbuds", "power bank", "router", "monitor", "projector" ] st.set_page_config(page_title="Smart Deal Hunter", layout="wide") st.title("🛒 Smart Deal Hunter") st.markdown("Compare product prices across platforms and get ML-backed 'Buy or Wait' decisions.") image_file = st.file_uploader("📷 Upload product image (optional)", type=["jpg", "jpeg", "png"]) query = None if image_file: image = preprocess(Image.open(image_file)).unsqueeze(0).to(st.session_state.device) text = clip.tokenize(product_labels).to(st.session_state.device) with torch.no_grad(): image_features = model.encode_image(image) text_features = model.encode_text(text) image_features /= image_features.norm(dim=-1, keepdim=True) text_features /= text_features.norm(dim=-1, keepdim=True) similarity = (100.0 * image_features @ text_features.T).squeeze(0) top_prob, top_idx = similarity.topk(1) predicted_label = product_labels[top_idx.item()] st.success(f"✅ Predicted product: **{predicted_label}**") query = predicted_label else: query = st.text_input("🔎 Enter product name manually", "") # with open("model/.auth", 'r') as f: # API_KEY = f.read().strip() API_KEY = os.getenv("auth") def search_google_shopping(query): params = { "engine": "google_shopping", "q": query, "location": "India", "hl": "en", "gl": "in", "api_key": API_KEY } response = requests.get("https://serpapi.com/search", params=params) if response.status_code != 200: st.error(f"❌ Request failed: {response.status_code}") return [] return response.json().get("shopping_results", []) def get_smart_score(item): try: price = item.get("extracted_price", 1) rating = float(item.get("rating", 0)) reviews = float(item.get("reviews", 1)) if price == 0 or rating == 0 or reviews == 0: return 0 score = round((rating * math.log(reviews + 1)) / price, 5) scaled = 1 + 4 * (1 - math.exp(-score / 0.1)) return round(max(1, min(5, scaled)), 2) except Exception: raise ValueError("Invalid data for score calculation") def remove_outlier_prices(items): prices = [item["extracted_price"] for item in items if item.get("extracted_price") is not None] if len(prices) < 3: return items avg = statistics.mean(prices) std_dev = statistics.stdev(prices) lower = avg - std_dev upper = avg + std_dev return [item for item in items if lower <= item.get("extracted_price", 0) <= upper] def get_card_color(score, rating): try: rating = float(rating) except Exception: rating = 0 if score > 2.5 or rating > 3.8: return "#eaffea" elif score < 0.5 or rating < 2.5: return "#ffeaea" return "#ffffff" def show_single_card(item, index=None, highlight=False, decision=None): score = get_smart_score(item) title = item.get("title", "N/A") price = item.get("extracted_price", "--") rating = item.get("rating", "--") reviews = item.get("reviews", "--") delivery = item.get("delivery", "--") platform = item.get("source", "N/A") image = item.get("thumbnail", "") link = item.get("link") or item.get("product_link") or item.get("serpapi_product_link") bg_color = get_card_color(score, rating) badge = "" if decision is not None: badge = "🟢 BUY" if decision == 1 else "🟡 WAIT" st.markdown(f"""
{index or ''}. {title[:60]}{'...' if len(title) > 60 else ''}

Price: ₹{price}
Rating: {rating} ({reviews} reviews)
Score: {score}
Store: {platform}
Delivery: {delivery}
{badge}

""", unsafe_allow_html=True) def show_cards_in_grid(items, predictions=None, cols=3): rows = (len(items) + cols - 1) // cols for r in range(rows): columns = st.columns(cols) for i in range(cols): idx = r * cols + i if idx < len(items): item = items[idx] pred = predictions[idx] if predictions is not None and idx < len(predictions) else None with columns[i]: show_single_card(item, index=idx + 1, decision=pred) results = [] if query: with st.spinner("🔍 Searching Google Shopping..."): results = search_google_shopping(query) if query and results: filtered = remove_outlier_prices(results) filtered = [item for item in filtered if item.get("extracted_price")] now = datetime.now().strftime("%Y-%m-%d %H:%M:%S") used_items = [] model_input = [] for item in filtered: try: entry = { "name": item.get("title", "N/A")[:30], "price": item.get("extracted_price", 0), "rating": float(item.get("rating", 0)), "smart_score": get_smart_score(item), "review_count": int(float(item.get("reviews", 0))), "date": now } model_input.append(entry) used_items.append(item) except Exception: st.warning(f"⚠️ Skipping item due to missing data: {item.get('title', 'N/A')}") predictions = inference(pd.DataFrame(model_input)) if model_input else [] predicted_pairs = predictions st.markdown(f"
Last updated: {now}
", unsafe_allow_html=True) tabs = st.tabs(["🔍 Filtered Deals", "💸 Cheapest", "🔝 Highest Rated", "🧠 Smart Score Rank", "🎯 By Platform"]) with tabs[0]: st.success(f"✅ Showing {len(used_items)} filtered products (no price anomalies)") show_cards_in_grid(used_items, predicted_pairs) with tabs[1]: cheapest = min(filtered, key=lambda x: x.get("extracted_price", float('inf'))) idx = used_items.index(cheapest) if cheapest in used_items else None st.markdown("### 💸 Cheapest Option") show_single_card(cheapest, highlight=True, decision=predicted_pairs[idx] if idx is not None else None) with tabs[2]: rated = [x for x in filtered if x.get("rating") and x.get("reviews")] if rated: top_rated = max(rated, key=lambda x: float(x.get("rating", 0))) idx = used_items.index(top_rated) if top_rated in used_items else None st.markdown("### 🔝 Highest Rated Option") show_single_card(top_rated, highlight=True, decision=predicted_pairs[idx] if idx is not None else None) with tabs[3]: scored = sorted(filtered, key=lambda x: get_smart_score(x), reverse=True) top_scored = scored[:9] top_preds = [predicted_pairs[used_items.index(x)] if x in used_items else None for x in top_scored] st.markdown("### 🧠 Top Smart Scores") show_cards_in_grid(top_scored, top_preds) with tabs[4]: platforms = {} for item in filtered: platform = item.get("source", "Other") platforms.setdefault(platform, []).append(item) for platform, items in platforms.items(): st.markdown(f"### 🛍️ {platform}") items_to_show = items[:6] preds_to_show = [predicted_pairs[used_items.index(x)] if x in used_items else None for x in items_to_show] show_cards_in_grid(items_to_show, preds_to_show) elif query: st.warning("⚠️ No valid results found.") st.markdown("---") st.markdown("WARNING: This Deployment has limited API requests per month and is for showcasing only") st.markdown("
🚀 Developed by Tech Titans at AndinoHack2025
", unsafe_allow_html=True)