product_recommender.py

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 SentenceTransformer for embedding-based recommendations
        self.model = SentenceTransformer('all-mpnet-base-v2')

        # Pre‐define broad candidate categories. Adjust to your needs.
        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",  # covers user "art" interest
            "home decor",
            "pet supplies",
            "novels",
            "gaming consoles",
            "smartphones",
            "camera gear",
            "toys",
            "gift hamper"
        ]
        # Pre‐encode category texts
        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 = []
        
        # (Might need to tweak 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"From Amazon: {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 = []

        # (Might need to tweak 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"From Flipkart: {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 tweak if IGP changes 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"From IGP: {product_name}"
                    })
            except Exception:
                continue
        
        return products[:5]

    # ------------------------------------------------------------------
    # Embedding-based category extraction
    # ------------------------------------------------------------------
    def _extract_categories(self, text: str) -> List[str]:
        # 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. Encode user text
        user_emb = self.model.encode(text, convert_to_tensor=True)

        # 3. Cosine similarity with candidate categories
        sims = util.cos_sim(user_emb, self.category_embeddings)[0]
        top_k = min(3, len(self.candidate_categories))  # pick top 3
        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("Top categories chosen via embeddings:", best_categories)
        return best_categories

    # ------------------------------------------------------------------
    # Main recommendations
    # ------------------------------------------------------------------
    async def get_recommendations(self, text: str) -> List[Dict]:
        """
        Search across Amazon, Flipkart, IGP based on top embedding matches,
        then deduplicate, then return final list.
        """
        try:
            # 1) Get top matching categories from user text
            categories = self._extract_categories(text)

            # 2) For each category, search across sites
            all_products = []
            for c in categories:
                amazon_products = await self.search_amazon(c)
                flipkart_products = await self.search_flipkart(c)
                igp_products = await self.search_igp(c)

                all_products.extend(amazon_products + flipkart_products + igp_products)

            # 3) Deduplicate
            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 get_recommendations: {str(e)}")
            return []