| import os |
| import requests |
| from typing import List, Dict, Optional |
|
|
|
|
| class ShopSmartAgent: |
| def __init__(self): |
| self.serpapi_key = os.getenv("SERPAPI_API_KEY") |
|
|
| def search_products(self, query: str, max_price: Optional[float] = None) -> List[Dict]: |
| fallback_products = [ |
| { |
| "name": "Samsung Galaxy S23", |
| "price": 699, |
| "rating": 4.5, |
| "reviews_count": 2847, |
| "image_url": "", |
| "url": "https://www.samsung.com/us/smartphones/galaxy-s23/", |
| "seller": "Samsung", |
| "seller_rating": 4.8 |
| }, |
| { |
| "name": "iPhone 15", |
| "price": 799, |
| "rating": 4.6, |
| "reviews_count": 3254, |
| "image_url": "", |
| "url": "https://www.apple.com/iphone-15/", |
| "seller": "Apple Store", |
| "seller_rating": 4.9 |
| }, |
| { |
| "name": "Google Pixel 8", |
| "price": 599, |
| "rating": 4.3, |
| "reviews_count": 1892, |
| "image_url": "", |
| "url": "https://store.google.com/", |
| "seller": "Google Store", |
| "seller_rating": 4.7 |
| } |
| ] |
|
|
| if not self.serpapi_key: |
| print("SERPAPI_API_KEY not set. Using fallback products.") |
| return [p for p in fallback_products if max_price is None or p["price"] <= max_price] |
|
|
| params = { |
| "engine": "google_shopping", |
| "q": query, |
| "api_key": self.serpapi_key, |
| "gl": "us", |
| "hl": "en", |
| "num": 20, |
| "no_cache": "true", |
| } |
|
|
| try: |
| response = requests.get("https://serpapi.com/search", params=params, timeout=20) |
| response.raise_for_status() |
| data = response.json() |
|
|
| shopping_results = data.get("shopping_results", []) |
| normalized_products = [] |
|
|
| for item in shopping_results: |
| price = item.get("extracted_price") |
| if price is None: |
| raw_price = item.get("price", "") |
| price = self._extract_price_number(raw_price) |
|
|
| if price is None: |
| continue |
|
|
| if max_price is not None and price > max_price: |
| continue |
|
|
| normalized_products.append({ |
| "name": item.get("title", "Unknown Product"), |
| "price": price, |
| "rating": float(item.get("rating", 0) or 0), |
| "reviews_count": int(item.get("reviews", 0) or 0), |
| "image_url": item.get("thumbnail", ""), |
| "url": item.get("product_link") or item.get("link") or "#", |
| "seller": item.get("source", "Unknown Seller"), |
| "seller_rating": 4.5 |
| }) |
|
|
| normalized_products = self._dedupe_products(normalized_products) |
|
|
| if not normalized_products: |
| print("No SerpApi products found. Using fallback products.") |
| return [p for p in fallback_products if max_price is None or p["price"] <= max_price] |
|
|
| return normalized_products |
|
|
| except Exception as e: |
| print(f"SerpApi search failed: {e}. Using fallback products.") |
| return [p for p in fallback_products if max_price is None or p["price"] <= max_price] |
|
|
| def _extract_price_number(self, raw_price: str): |
| if not raw_price: |
| return None |
| cleaned = raw_price.replace("$", "").replace(",", "").strip() |
| parts = cleaned.split() |
| try: |
| return float(parts[0]) |
| except Exception: |
| return None |
|
|
| def _dedupe_products(self, products: List[Dict]) -> List[Dict]: |
| seen = set() |
| deduped = [] |
|
|
| for product in products: |
| name = str(product.get("name", "")).strip().lower() |
| seller = str(product.get("seller", "")).strip().lower() |
| key = (name, seller) |
|
|
| if key not in seen: |
| seen.add(key) |
| deduped.append(product) |
|
|
| return deduped |
|
|
| def analyze_reviews(self, product_name: str) -> Dict: |
| name = product_name.lower() |
|
|
| if any(k in name for k in ["iphone", "galaxy", "pixel", "oneplus"]): |
| return { |
| "sentiment": "positive", |
| "pros": ["Strong performance", "Good camera quality", "Reliable everyday use"], |
| "cons": ["Can be expensive", "Battery life varies by model"], |
| "summary": "Generally strong smartphone pick with solid mainstream appeal." |
| } |
|
|
| if any(k in name for k in ["headphone", "earbud", "airpods", "bose", "sony"]): |
| return { |
| "sentiment": "positive", |
| "pros": ["Strong audio quality", "Comfortable design", "Useful everyday features"], |
| "cons": ["Price may be high", "Battery life varies"], |
| "summary": "Well-reviewed audio option with strong consumer appeal." |
| } |
|
|
| if any(k in name for k in ["mask", "serum", "cleanser", "tonic", "moisturizer"]): |
| return { |
| "sentiment": "mixed", |
| "pros": ["Popular product type", "Affordable options available", "Easy to compare"], |
| "cons": ["Results vary by skin type", "Some products may be overhyped"], |
| "summary": "Promising beauty option, but personal fit matters more than ratings alone." |
| } |
|
|
| return { |
| "sentiment": "mixed", |
| "pros": ["Popular option", "Reasonable value", "Accessible price range"], |
| "cons": ["Not perfect for every user", "Feature tradeoffs may apply"], |
| "summary": "Solid option overall with a few tradeoffs depending on user needs." |
| } |
|
|
| def assess_risk(self, product: Dict) -> Dict: |
| risk_score = 0 |
| risk_factors = [] |
|
|
| if product["price"] < 50: |
| risk_score += 2 |
| risk_factors.append("Unusually low price") |
|
|
| if product["rating"] and product["rating"] < 3.5: |
| risk_score += 3 |
| risk_factors.append("Low customer rating") |
|
|
| if product["reviews_count"] < 50: |
| risk_score += 2 |
| risk_factors.append("Limited reviews available") |
|
|
| if product.get("seller_rating", 5) < 4.0: |
| risk_score += 2 |
| risk_factors.append("Low seller rating") |
|
|
| risk_level = "Low" if risk_score <= 2 else "Medium" if risk_score <= 5 else "High" |
|
|
| return { |
| "level": risk_level, |
| "score": risk_score, |
| "factors": risk_factors |
| } |
|
|
| def rank_products(self, products: List[Dict], user_query: str) -> List[Dict]: |
| if not products: |
| return [] |
|
|
| query_lower = user_query.lower() |
|
|
| for product in products: |
| product["risk"] = self.assess_risk(product) |
|
|
| review_weight = max(product["reviews_count"], 1) |
| rating_weight = max(product["rating"], 0.1) |
| price_weight = max(product["price"], 1) |
|
|
| base_score = (rating_weight * review_weight) / price_weight |
|
|
| name_lower = product["name"].lower() |
| keyword_bonus = 0 |
| for token in query_lower.split(): |
| if token in name_lower: |
| keyword_bonus += 20 |
|
|
| risk_penalty = product["risk"]["score"] * 10 |
| product["value_score"] = base_score + keyword_bonus - risk_penalty |
|
|
| ranked = sorted(products, key=lambda x: (x["value_score"], x["rating"]), reverse=True) |
|
|
| if ranked: |
| ranked[0]["is_best"] = True |
| for product in ranked[1:]: |
| product["is_best"] = False |
|
|
| return ranked |
|
|
| def generate_recommendation(self, product: Dict, is_best: bool = False) -> str: |
| if is_best: |
| return "Best overall value" |
|
|
| if product["rating"] >= 4.5: |
| return "Premium option with strong reviews" |
| elif product["price"] < 600: |
| return "Budget-friendly choice with good value" |
| else: |
| return "Solid mid-range option" |
|
|
| def process_query(self, query: str, max_price: float = None) -> Dict: |
| products = self.search_products(query, max_price) |
| ranked_products = self.rank_products(products, query) |
|
|
| for product in ranked_products[:3]: |
| product["review_analysis"] = self.analyze_reviews(product["name"]) |
| product["recommendation"] = self.generate_recommendation( |
| product, |
| product.get("is_best", False) |
| ) |
|
|
| return { |
| "query": query, |
| "total_found": len(ranked_products), |
| "products": ranked_products |
| } |
