Spaces:
Running
Running
| from typing import Dict, List | |
| import requests | |
| from bs4 import BeautifulSoup | |
| import aiohttp | |
| import asyncio | |
| import json | |
| from sentence_transformers import SentenceTransformer | |
| import numpy as np | |
| import re | |
| class DynamicRecommender: | |
| def __init__(self): | |
| self.headers = { | |
| 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36' | |
| } | |
| self.model = SentenceTransformer('all-mpnet-base-v2') | |
| async def search_amazon(self, query: str) -> List[Dict]: | |
| """Search Amazon for products""" | |
| 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 [] | |
| async def search_flipkart(self, query: str) -> List[Dict]: | |
| """Search Flipkart for products""" | |
| 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_amazon_results(self, html: str) -> List[Dict]: | |
| soup = BeautifulSoup(html, 'html.parser') | |
| products = [] | |
| for item in soup.select('.s-result-item'): | |
| try: | |
| name = item.select_one('.a-text-normal') | |
| price = item.select_one('.a-price-whole') | |
| if name and price: | |
| products.append({ | |
| 'name': name.text.strip(), | |
| 'price': price.text.strip(), | |
| 'source': 'Amazon', | |
| 'url': 'https://amazon.in' + item.select_one('a')['href'] | |
| }) | |
| except Exception: | |
| continue | |
| return products[:5] | |
| def _parse_flipkart_results(self, html: str) -> List[Dict]: | |
| soup = BeautifulSoup(html, 'html.parser') | |
| products = [] | |
| for item in soup.select('._1AtVbE'): | |
| try: | |
| name = item.select_one('._4rR01T') | |
| price = item.select_one('._30jeq3') | |
| if name and price: | |
| products.append({ | |
| 'name': name.text.strip(), | |
| 'price': price.text.strip(), | |
| 'source': 'Flipkart', | |
| 'url': 'https://flipkart.com' + item.select_one('a')['href'] | |
| }) | |
| except Exception: | |
| continue | |
| return products[:5] | |
| def _extract_keywords(self, text: str) -> List[str]: | |
| """Extract relevant search keywords from input""" | |
| age_match = re.search(r'age\s+(\d+)', text.lower()) | |
| age = age_match.group(1) if age_match else None | |
| interests = [] | |
| if 'software' in text.lower() or 'engineer' in text.lower(): | |
| interests.extend(['programming books', 'tech gadgets']) | |
| if 'books' in text.lower(): | |
| interests.append('books') | |
| if 'successful' in text.lower(): | |
| interests.extend(['self help books', 'business books']) | |
| return [f"{interest} for {age} year old" if age else interest for interest in interests] | |
| async def get_recommendations(self, text: str) -> Dict: | |
| """Get personalized recommendations""" | |
| try: | |
| keywords = self._extract_keywords(text) | |
| all_products = [] | |
| for keyword in keywords: | |
| amazon_products = await self.search_amazon(keyword) | |
| flipkart_products = await self.search_flipkart(keyword) | |
| all_products.extend(amazon_products + flipkart_products) | |
| # Remove duplicates and sort by relevance | |
| seen = set() | |
| unique_products = [] | |
| for product in all_products: | |
| if product['name'] not in seen: | |
| seen.add(product['name']) | |
| unique_products.append(product) | |
| return unique_products[:5] | |
| except Exception as e: | |
| print(f"Error in recommendations: {str(e)}") | |
| return [] |