Update app.py

app.py

@@ -1,77 +1,249 @@
-import gradio as gr
+from typing import Dict, List, Optional
+import aiohttp
+import asyncio
+from bs4 import BeautifulSoup
 from transformers import pipeline
+import pandas as pd
+from datetime import datetime
+import json
+import sqlite3
+import re
 import urllib.parse
 
-def extract_keywords(text: str, nlp_pipeline) -> list:
-    """Extract relevant keywords from the input text"""
-    prompt = f"Extract 3-4 most relevant gift-related keywords from: {text}\nKeywords:"
-    response = nlp_pipeline(prompt, max_new_tokens=30, num_return_sequences=1)
-    keywords = response[0]['generated_text'].split('Keywords:')[-1].strip()
-    return [k.strip() for k in keywords.split(',') if k.strip()]
-
-def generate_search_urls(keywords: list) -> dict:
-    """Generate search URLs for various e-commerce platforms"""
-    # Encode queries properly for URLs
-    query = urllib.parse.quote(" ".join(keywords))
-
-    return {
-        "Amazon India": f'<a href="https://www.amazon.in/s?k={query}" target="_blank">Amazon</a>',
-        "Flipkart": f'<a href="https://www.flipkart.com/search?q={query}" target="_blank">Flipkart</a>',
-        "IGP Gifts": f'<a href="https://www.igp.com/search?q={query}" target="_blank">IGP</a>',
-        "IndiaMart": f'<a href="https://www.indiamart.com/find?q={query}" target="_blank">IndiaMart</a>',
-    }
-
-def recommend_gifts(text: str):
-    """Main function to generate gift recommendations"""
-    if not text:
-        return "⚠️ Please provide a description."
-
-    try:
-        # Load GPT-2 as a text-generation model
-        nlp = pipeline(
-            "text-generation",
-            model="gpt2",
-            device_map="auto"
+class ProductKnowledgeBase:
+    def __init__(self, db_path: str = "product_knowledge.db"):
+        self.db_path = db_path
+        self.setup_database()
+        
+    def setup_database(self):
+        """Initialize the SQLite database with required tables"""
+        conn = sqlite3.connect(self.db_path)
+        cursor = conn.cursor()
+        
+        # Create products table
+        cursor.execute("""
+        CREATE TABLE IF NOT EXISTS products (
+            id INTEGER PRIMARY KEY,
+            name TEXT NOT NULL,
+            category TEXT NOT NULL,
+            subcategory TEXT,
+            features TEXT,
+            target_audience TEXT,
+            price_range TEXT,
+            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
         )
+        """)
+        
+        # Create price history table
+        cursor.execute("""
+        CREATE TABLE IF NOT EXISTS price_history (
+            id INTEGER PRIMARY KEY,
+            product_id INTEGER,
+            platform TEXT NOT NULL,
+            price REAL NOT NULL,
+            timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+            FOREIGN KEY (product_id) REFERENCES products (id)
+        )
+        """)
+        
+        # Create recommendations table for feedback
+        cursor.execute("""
+        CREATE TABLE IF NOT EXISTS recommendations (
+            id INTEGER PRIMARY KEY,
+            user_input TEXT NOT NULL,
+            product_id INTEGER,
+            success_rating INTEGER,
+            timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+            FOREIGN KEY (product_id) REFERENCES products (id)
+        )
+        """)
+        
+        conn.commit()
+        conn.close()
 
-        # Extract relevant keywords
-        keywords = extract_keywords(text, nlp)
-
-        # Generate search URLs
-        search_links = generate_search_urls(keywords)
+class PriceFetcher:
+    def __init__(self):
+        self.headers = {
+            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
+        }
+    
+    async def fetch_price(self, url: str) -> Optional[float]:
+        """Fetch price from a given URL"""
+        try:
+            async with aiohttp.ClientSession() as session:
+                async with session.get(url, headers=self.headers) as response:
+                    if response.status == 200:
+                        html = await response.text()
+                        return self._extract_price(html)
+            return None
+        except Exception as e:
+            print(f"Error fetching price: {str(e)}")
+            return None
+    
+    def _extract_price(self, html: str) -> Optional[float]:
+        """Extract price from HTML content"""
+        soup = BeautifulSoup(html, 'html.parser')
+        # Add platform-specific price extraction logic here
+        return None
 
-        # Format the output as clickable links
-        formatted_output = f"""
-        <h3>🔍 Predicted Interests: {", ".join(keywords)}</h3>
-        <h3>🛒 Gift Suggestions:</h3>
-        <ul>
-        <li>{search_links["Amazon India"]}</li>
-        <li>{search_links["Flipkart"]}</li>
-        <li>{search_links["IGP Gifts"]}</li>
-        <li>{search_links["IndiaMart"]}</li>
-        </ul>
+class RecommendationEngine:
+    def __init__(self, knowledge_base: ProductKnowledgeBase, price_fetcher: PriceFetcher):
+        self.kb = knowledge_base
+        self.price_fetcher = price_fetcher
+        self.nlp = pipeline("text-generation", model="gpt2", device_map="auto")
+    
+    def analyze_user_input(self, text: str) -> Dict:
+        """Analyze user input for context and requirements"""
+        # Extract age if mentioned
+        age_match = re.search(r'age\s+(?:is\s+)?(\d+)', text.lower())
+        age = age_match.group(1) if age_match else None
+        
+        # Extract budget if mentioned
+        budget_match = re.search(r'(?:budget|cost|price)\s*(?:is|:)?\s*(?:rs|₹)?\s*(\d+)', text.lower())
+        budget = budget_match.group(1) if budget_match else None
+        
+        # Generate categories and context
+        prompt = f"Extract gift categories and context from: {text}\nFormat: category1, category2 | context"
+        response = self.nlp(prompt, max_new_tokens=50)[0]['generated_text']
+        
+        categories, context = response.split('|')
+        categories = [c.strip() for c in categories.split(',')]
+        
+        return {
+            "age": age,
+            "budget": budget,
+            "categories": categories,
+            "context": context.strip()
+        }
+    
+    def find_matching_products(self, analysis: Dict) -> List[Dict]:
+        """Find products matching the analysis"""
+        conn = sqlite3.connect(self.kb.db_path)
+        cursor = conn.cursor()
+        
+        query = """
+        SELECT p.*, GROUP_CONCAT(DISTINCT ph.price) as prices
+        FROM products p
+        LEFT JOIN price_history ph ON p.id = ph.product_id
+        WHERE p.category IN ({})
+        GROUP BY p.id
+        """.format(','.join('?' * len(analysis['categories'])))
+        
+        cursor.execute(query, analysis['categories'])
+        products = cursor.fetchall()
+        
+        conn.close()
+        
+        return [self._format_product(p, analysis) for p in products]
+    
+    def _format_product(self, product_data: tuple, analysis: Dict) -> Dict:
+        """Format product data with explanation"""
+        return {
+            "name": product_data[1],
+            "category": product_data[2],
+            "features": json.loads(product_data[4]),
+            "why_recommended": self._generate_explanation(product_data, analysis),
+            "price_info": self._process_price_info(product_data[-1]),
+            "target_audience": json.loads(product_data[5])
+        }
+    
+    def _generate_explanation(self, product_data: tuple, analysis: Dict) -> str:
+        """Generate personalized explanation for recommendation"""
+        prompt = f"""
+        Product: {product_data[1]}
+        Category: {product_data[2]}
+        User Context: {analysis['context']}
+        
+        Generate a brief explanation why this product is recommended:
         """
-        return formatted_output
+        
+        response = self.nlp(prompt, max_new_tokens=100)[0]['generated_text']
+        return response.split('Generate a brief explanation why this product is recommended:')[-1].strip()
+    
+    def _process_price_info(self, prices: str) -> Dict:
+        """Process and format price information"""
+        if not prices:
+            return {"min": None, "max": None, "average": None}
+            
+        price_list = [float(p) for p in prices.split(',')]
+        return {
+            "min": min(price_list),
+            "max": max(price_list),
+            "average": sum(price_list) / len(price_list)
+        }
 
-    except Exception as e:
-        return f"❌ Error: {str(e)}"
+class GiftRecommenderAPI:
+    def __init__(self):
+        self.kb = ProductKnowledgeBase()
+        self.price_fetcher = PriceFetcher()
+        self.engine = RecommendationEngine(self.kb, self.price_fetcher)
+    
+    async def get_recommendations(self, text: str) -> Dict:
+        """Main method to get gift recommendations"""
+        try:
+            # Analyze user input
+            analysis = self.engine.analyze_user_input(text)
+            
+            # Find matching products
+            recommendations = self.engine.find_matching_products(analysis)
+            
+            # Fetch current prices
+            for rec in recommendations:
+                current_prices = await self._fetch_current_prices(rec['name'])
+                rec['current_prices'] = current_prices
+            
+            return {
+                "analysis": analysis,
+                "recommendations": recommendations
+            }
+            
+        except Exception as e:
+            return {"error": f"An error occurred: {str(e)}"}
+    
+    async def _fetch_current_prices(self, product_name: str) -> Dict:
+        """Fetch current prices from various platforms"""
+        encoded_name = urllib.parse.quote(product_name)
+        urls = {
+            "amazon": f"https://www.amazon.in/s?k={encoded_name}",
+            "flipkart": f"https://www.flipkart.com/search?q={encoded_name}",
+            "igp": f"https://www.igp.com/search?q={encoded_name}"
+        }
+        
+        prices = {}
+        for platform, url in urls.items():
+            price = await self.price_fetcher.fetch_price(url)
+            if price:
+                prices[platform] = price
+        
+        return prices
+
+# Create Gradio interface
+import gradio as gr
 
-# Create Gradio interface with HTML output
-demo = gr.Interface(
-    fn=recommend_gifts,
-    inputs=gr.Textbox(
-        lines=3,
-        placeholder="Describe who you're buying a gift for (age, interests, occasion, etc.)"
-    ),
-    outputs=gr.HTML(),  # Change output type to HTML
-    title="🎁 Smart Gift Recommender",
-    description="Get personalized gift suggestions with direct shopping links!",
-    examples=[
-        ["a small kid of age 3 want him to have something like a toy that teaches alphabets"],
-        ["age is 25 and he loves puzzle and online FPS games"],
-        ["Looking for a gift for my mom who enjoys gardening and cooking"]
-    ]
-)
+def create_gradio_interface():
+    recommender = GiftRecommenderAPI()
+    
+    def recommend(text: str) -> Dict:
+        return asyncio.run(recommender.get_recommendations(text))
+    
+    demo = gr.Interface(
+        fn=recommend,
+        inputs=gr.Textbox(
+            lines=3,
+            placeholder="Describe who you're buying a gift for (age, interests, occasion, etc.)"
+        ),
+        outputs=gr.JSON(),
+        title="🎁 Smart Gift Recommender",
+        description="Get personalized gift suggestions with real-time prices and explanations!",
+        examples=[
+            ["need a fifa latest game of EA"],
+            ["a small kid of age 3 want him to have something like toy that teaches alphabets"],
+            ["Looking for a gift for my mom who enjoys gardening and cooking"]
+        ]
+    )
+    
+    return demo
 
 if __name__ == "__main__":
-    demo.launch()
+    demo = create_gradio_interface()
+    demo.launch()