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 }