from typing import Dict, List import aiohttp import asyncio import re import torch from sentence_transformers import SentenceTransformer, util from bs4 import BeautifulSoup class DynamicRecommender: def __init__(self): self.headers = { 'User-Agent': ( 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) ' 'AppleWebKit/537.36 (KHTML, like Gecko) ' 'Chrome/100.0.4896.75 Safari/537.36' ) } # Load your model self.model = SentenceTransformer('all-mpnet-base-v2') # Pre‐define some candidate categories you might want to search for. # Adjust these to suit your domain. The more you add, the more "general" # your coverage becomes. They can be as broad or as niche as you like. self.candidate_categories = [ "tech gadgets", "programming books", "self help books", "business books", "leadership novels", "fashion accessories", "beauty products", "board games", "music instruments", "cooking utensils", "cookbooks", "art and painting supplies", "home decor", "pet supplies", "novels", "gaming consoles", "smartphones", "camera gear", "toys", "gift hamper" ] # Pre‐encode those categories for faster scoring. self.category_embeddings = self.model.encode(self.candidate_categories, convert_to_tensor=True) # ------------------------------------------------------------------ # Amazon search # ------------------------------------------------------------------ async def search_amazon(self, query: str) -> List[Dict]: print(f"Searching Amazon for: {query}") search_url = f"https://www.amazon.in/s?k={query}" async with aiohttp.ClientSession() as session: async with session.get(search_url, headers=self.headers) as response: if response.status == 200: html = await response.text() return self._parse_amazon_results(html) return [] def _parse_amazon_results(self, html: str) -> List[Dict]: soup = BeautifulSoup(html, 'html.parser') products = [] # These selectors may need updating if Amazon changes HTML search_items = soup.select('.s-result-item') for item in search_items: try: name_elem = item.select_one('.a-text-normal') price_elem = item.select_one('.a-price-whole') link_elem = item.select_one('a.a-link-normal') if name_elem and price_elem and link_elem: product_name = name_elem.get_text(strip=True) product_price = price_elem.get_text(strip=True) product_url = link_elem.get('href') products.append({ 'name': product_name, 'price': product_price, 'source': 'Amazon', 'url': 'https://www.amazon.in' + product_url, 'description': f"This item is from Amazon related to '{product_name}'." }) except Exception: continue return products[:5] # ------------------------------------------------------------------ # Flipkart search # ------------------------------------------------------------------ async def search_flipkart(self, query: str) -> List[Dict]: print(f"Searching Flipkart for: {query}") search_url = f"https://www.flipkart.com/search?q={query}" async with aiohttp.ClientSession() as session: async with session.get(search_url, headers=self.headers) as response: if response.status == 200: html = await response.text() return self._parse_flipkart_results(html) return [] def _parse_flipkart_results(self, html: str) -> List[Dict]: soup = BeautifulSoup(html, 'html.parser') products = [] # These selectors may need updating if Flipkart changes HTML item_cards = soup.select('._1AtVbE') for item in item_cards: try: name_elem = item.select_one('._4rR01T') price_elem = item.select_one('._30jeq3') link_elem = item.select_one('a') if name_elem and price_elem and link_elem: product_name = name_elem.get_text(strip=True) product_price = price_elem.get_text(strip=True) product_url = link_elem.get('href') products.append({ 'name': product_name, 'price': product_price, 'source': 'Flipkart', 'url': 'https://www.flipkart.com' + product_url, 'description': f"This item is from Flipkart related to '{product_name}'." }) except Exception: continue return products[:5] # ------------------------------------------------------------------ # IGP search # ------------------------------------------------------------------ async def search_igp(self, query: str) -> List[Dict]: print(f"Searching IGP for: {query}") search_url = f"https://www.igp.com/search/{query}" async with aiohttp.ClientSession() as session: async with session.get(search_url, headers=self.headers) as response: if response.status == 200: html = await response.text() return self._parse_igp_results(html) return [] def _parse_igp_results(self, html: str) -> List[Dict]: soup = BeautifulSoup(html, 'html.parser') products = [] # Likely need to update based on actual IGP HTML item_cards = soup.select('.product-item') for item in item_cards: try: name_elem = item.select_one('.product-title') price_elem = item.select_one('.product-price') link_elem = item.select_one('a') if name_elem and price_elem and link_elem: product_name = name_elem.get_text(strip=True) product_price = price_elem.get_text(strip=True) product_url = link_elem.get('href') products.append({ 'name': product_name, 'price': product_price, 'source': 'IGP', 'url': 'https://www.igp.com' + product_url, 'description': f"This item is from IGP related to '{product_name}'." }) except Exception: continue return products[:5] # ------------------------------------------------------------------ # Extract categories from user text using embeddings # ------------------------------------------------------------------ def _extract_keywords(self, text: str) -> List[str]: """ 1. Parse out age if present 2. Use embeddings to find top 2-3 matching categories from self.candidate_categories. 3. Combine them with the age if found. """ # 1) Check for age with a regex age_match = re.search(r'age\s+(\d+)', text.lower()) age = age_match.group(1) if age_match else None # 2) Use the entire user text as an embedding user_emb = self.model.encode(text, convert_to_tensor=True) # Compute similarity with each candidate category sims = util.cos_sim(user_emb, self.category_embeddings)[0] # shape: [num_categories] # Grab top 3 indices top_k = min(3, len(self.candidate_categories)) top_results = torch.topk(sims, k=top_k) best_categories = [] for idx in top_results.indices: cat_text = self.candidate_categories[idx] if age: cat_text = f"{cat_text} for {age} year old" best_categories.append(cat_text) print("Embedding-based categories:", best_categories) return best_categories # ------------------------------------------------------------------ # Main recommendations # ------------------------------------------------------------------ async def get_recommendations(self, text: str) -> List[Dict]: """ Search across Amazon, Flipkart, and IGP based on the top category matches. """ try: # 1) Figure out best categories (queries) from user text queries = self._extract_keywords(text) # 2) Search each site for each query all_products = [] for query in queries: # For each query, hit Amazon, Flipkart, IGP amazon_products = await self.search_amazon(query) flipkart_products = await self.search_flipkart(query) igp_products = await self.search_igp(query) all_products.extend(amazon_products) all_products.extend(flipkart_products) all_products.extend(igp_products) # 3) De‐duplicate by product name seen = set() unique_products = [] for product in all_products: if product['name'] not in seen: seen.add(product['name']) unique_products.append(product) # 4) Optionally slice or sort further return unique_products[:5] except Exception as e: print(f"Error in recommendations: {str(e)}") return []