Update main.py

main.py

@@ -8,11 +8,14 @@ import google.generativeai as genai
 from heyoo import WhatsApp
 from dotenv import load_dotenv
 from flask import Flask, request, make_response
-from typing import Dict, List, Optional
+from typing import Dict, List, Union
+import json
 
 # Initialize Flask App
 app = Flask(__name__)
 
+
+
 # Configure logging
 logging.basicConfig(
     level=logging.INFO,
@@ -27,13 +30,6 @@ load_dotenv()
 nameserver1 = os.getenv('nameserver1', '8.8.8.8')
 nameserver2 = os.getenv('nameserver2', '8.8.4.4')
 
-# Configure Gemini AI
-genai.configure(api_key=os.getenv("google_api_key"))
-gemini = genai.GenerativeModel('gemini-2.0-flash-exp')
-
-products_df = pd.read_csv("supermarket_data1.csv")
-user_carts: Dict[str, List[dict]] = {}
-
 # Configure global DNS resolution
 def setup_dns():
     """Configure DNS resolution globally"""
@@ -59,74 +55,6 @@ def setup_dns():
 # Setup DNS resolution
 setup_dns()
 
-def parse_user_intent(text: str) -> dict:
-    """Use Gemini to parse user intent and context"""
-    try:
-        prompt = f""" You are an intelligent shopping assistant called Brave Shopper,
-        
-        Analyze this shopping query and return JSON with:
-        - intent: [product_search, cart_check, checkout, other]
-        - scenario: brief description
-        - budget: extracted number or null
-        - required_categories: list of relevant product categories
-        - constraints: list of important constraints
-        
-        Query: {text}"""
-        
-        response = gemini.generate_content(prompt)
-        return eval(response.text)
-    except Exception as e:
-        logger.error(f"Gemini parse error: {e}")
-        return {"intent": "other"}
-
-def search_products(query: dict, branch: str = "Harare") -> pd.DataFrame:
-    """Search products based on Gemini analysis"""
-    try:
-        filtered = products_df[
-            (products_df["Branch"] == branch) &
-            (products_df["Category"].isin(query.get("required_categories", []))) &
-            (products_df["Price (USD)"] <= (query.get("budget") or float('inf')))
-        ]
-        return filtered.sort_values("Price (USD)").head(5)
-    except Exception as e:
-        logger.error(f"Search error: {e}")
-        return pd.DataFrame()
-
-def create_cart_response(user_id: str, products: pd.DataFrame, total: float = 0):
-    """Create interactive cart response with buttons"""
-    products_text = "\n".join(
-        f"{row['Product']} - ${row['Price (USD)']:.2f}" 
-        for _, row in products.iterrows()
-    )
-    
-    buttons = []
-    if not products.empty:
-        for _, row in products.iterrows():
-            buttons.append({
-                "type": "reply",
-                "reply": {"id": f"add_{row.name}", "title": f"Add {row['Product']}"}
-            })
-        buttons.append({"type": "reply", "reply": {"id": "cancel", "title": "Cancel"}})
-    else:
-        return {"text": "No products found matching your criteria."}
-
-    cart = user_carts.get(user_id, [])
-    if cart:
-        cart_total = sum(item["Price (USD)"] for item in cart)
-        response_text = (
-            f"Suggested Products:\n{products_text}\n\n"
-            f"Current Cart Total: ${cart_total:.2f}\n"
-            "Select products to add:"
-        )
-        buttons.extend([
-            {"type": "reply", "reply": {"id": "checkout", "title": "Checkout"}},
-            {"type": "reply", "reply": {"id": "continue", "title": "Keep Shopping"}}
-        ])
-    else:
-        response_text = f"Suggested Products:\n{products_text}\n\nSelect products to add:"
-
-    return {"text": response_text, "buttons": buttons}
-
 # Initialize WhatsApp
 try:
     messenger = WhatsApp(
@@ -175,89 +103,185 @@ def verify_token():
         logger.error(f"Error in verify_token: {str(e)}")
         return "Internal server error", 500
 
+# Load .env file
+load_dotenv()
+
+# Initialize Gemini
+genai.configure(api_key=os.getenv("google_api_key"))
+model = genai.GenerativeModel('gemini-2.0-flash-exp')
+vision_model = genai.GenerativeModel('gemini-2.0-flash-exp')
+
+# Load product data
+try:
+    df = pd.read_csv("supermarket_data1")
+    product_data = df.to_dict('records')
+except Exception as e:
+    logger.error(f"Failed to load product data: {str(e)}")
+    raise
+
+# Cart management
+carts: Dict[str, List[Dict]] = {}
+
+class ShoppingAssistant:
+    def __init__(self):
+        self.product_data = product_data
+        
+    def analyze_content(self, content: Union[str, bytes], content_type: str, budget: float = None) -> dict:
+        try:
+            if content_type == "text":
+                response = model.generate_content(
+                    f"""As a retail shopping assistant, analyze this request and extract shopping context, 
+                    specific products needed, and any budget constraints. Request: {content}"""
+                )
+            else:  # For images, documents, etc.
+                response = vision_model.generate_content(
+                    [
+                        "Analyze this content and extract shopping-related information, products needed, and any specifications.",
+                        content
+                    ]
+                )
+            
+            analysis = response.text
+            
+            # Second pass to find specific products
+            products_prompt = f"""Based on this analysis: {analysis}
+            Find matching products from our inventory considering these details:
+            {json.dumps(self.product_data)}
+            Return only product names and prices that match the request."""
+            
+            products_response = model.generate_content(products_prompt)
+            return self.format_recommendations(products_response.text, budget)
+            
+        except Exception as e:
+            logger.error(f"Error in content analysis: {str(e)}")
+            return {"error": "Failed to analyze content"}
+
+    def format_recommendations(self, recommendations: str, budget: float = None) -> dict:
+        try:
+            # Parse recommendations and filter by budget if specified
+            suggested_products = []
+            total = 0
+            
+            for product in self.product_data:
+                if product['product'].lower() in recommendations.lower():
+                    if budget and (total + product['price']) > budget:
+                        continue
+                    suggested_products.append(product)
+                    total += product['price']
+            
+            return {
+                "products": suggested_products,
+                "total": total,
+                "message": "No products found" if not suggested_products else "Products found"
+            }
+        except Exception as e:
+            logger.error(f"Error formatting recommendations: {str(e)}")
+            return {"error": "Failed to format recommendations"}
+
+# Initialize WhatsApp and DNS setup (keeping your original configuration)
+
 @app.post("/")
 def hook():
     """Main webhook handler"""
     try:
         data = request.get_json()
+        logger.info("Received webhook data: %s", data)
+        
+        changed_field = messenger.changed_field(data)
+        if changed_field != "messages":
+            return "OK", 200
+            
+        if not messenger.is_message(data):
+            delivery = messenger.get_delivery(data)
+            if delivery:
+                logger.info(f"Message delivery status: {delivery}")
+            return "OK", 200
+
         mobile = messenger.get_mobile(data)
+        name = messenger.get_name(data)
+        message_type = messenger.get_message_type(data)
         
-        if messenger.is_message(data):
+        shopping_assistant = ShoppingAssistant()
+
+        if message_type == "text":
             message = messenger.get_message(data)
-            intent = parse_user_intent(message)
-            
-            if intent["intent"] == "checkout":
-                return handle_checkout(mobile)
-                
-            elif intent["intent"] == "cart_check":
-                return send_cart_summary(mobile)       
-            else:
-                products = search_products(intent)
-                response = create_cart_response(mobile, products)
-                messenger.send_reply_button(
-                    response["text"], 
-                    response["buttons"],
-                    mobile
-                )
-                
-        elif messenger.is_interactive(data):
-            response = messenger.get_interactive_response(data)
-            handle_button_click(mobile, response["id"])
+            result = shopping_assistant.analyze_content(message, "text")
             
+            if "error" in result:
+                messenger.send_message(f"Sorry, I couldn't process your request.", mobile)
+                return "OK", 200
+
+            if not result["products"]:
+                messenger.send_message("I couldn't find any matching products.", mobile)
+                return "OK", 200
+
+            # Format product message
+            products_text = "Here are the suggested products:\n\n"
+            for product in result["products"]:
+                products_text += f"- {product['product']}: ${product['price']}\n"
+            products_text += f"\nTotal: ${result['total']:.2f}"
+
+            # Send interactive message with buttons
+            buttons = [
+                {"id": "add_to_cart", "title": "Add to Cart"},
+                {"id": "cancel", "title": "Cancel"}
+            ]
+            messenger.send_reply_button(products_text, buttons, mobile)
+
+        elif message_type == "interactive":
+            handle_interactive_message(messenger, data, mobile, carts)
+
+        elif message_type in ["image", "document", "audio"]:
+            media_content = handle_media_message(messenger, data, message_type, mobile)
+            if media_content:
+                result = shopping_assistant.analyze_content(media_content, message_type)
+                # Handle result similar to text message
+
         return "OK", 200
 
     except Exception as e:
-        logger.error(f"Webhook error: {e}")
+        logger.error(f"Error in webhook handler: {str(e)}")
         return "Internal server error", 500
 
-def handle_button_click(user_id: str, button_id: str):
-    """Handle interactive button clicks"""
-    if button_id.startswith("add_"):
-        product_id = int(button_id.split("_")[1])
-        product = products_df.loc[product_id].to_dict()
-        user_carts.setdefault(user_id, []).append(product)
-        messenger.send_message(f"{product['Product']} added to cart!", user_id)
-        
-    elif button_id == "checkout":
-        messenger.send_interactive_button(
-            "Confirm checkout?",
-            [
-                {"type": "reply", "reply": {"id": "confirm", "title": "Confirm"}},
-                {"type": "reply", "reply": {"id": "cancel", "title": "Cancel"}}
-            ],
-            user_id
-        )
+def handle_interactive_message(messenger, data, mobile, carts):
+    """Handle interactive message responses"""
+    try:
+        response = messenger.get_interactive_response(data)
+        button_id = response.get("id")
         
-    elif button_id == "confirm":
-        cart = user_carts.pop(user_id, [])
-        total = sum(item["Price (USD)"] for item in cart)
-        messenger.send_message(
-            f"Order confirmed! Total: ${total:.2f}\n"
-            "Your items will be ready for pickup in 1 hour.",
-            user_id
-        )
-
-def send_cart_summary(user_id: str):
-    """Send current cart summary to user"""
-    cart = user_carts.get(user_id, [])
-    if not cart:
-        messenger.send_message("Your cart is empty!", user_id)
-        return
-    
-    cart_text = "\n".join(
-        f"{item['Product']} - ${item['Price (USD)']:.2f}" 
-        for item in cart
-    )
-    total = sum(item["Price (USD)"] for item in cart)
-    
-    messenger.send_interactive_button(
-        f"Current Cart:\n{cart_text}\nTotal: ${total:.2f}",
-        [
-            {"type": "reply", "reply": {"id": "checkout", "title": "Checkout"}},
-            {"type": "reply", "reply": {"id": "clear", "title": "Clear Cart"}}
-        ],
-        user_id
-    )
+        if button_id == "add_to_cart":
+            # Add products to cart
+            if mobile not in carts:
+                carts[mobile] = []
+            # Add current products to cart logic here
+            
+            buttons = [
+                {"id": "more_shopping", "title": "Do More Shopping"},
+                {"id": "checkout", "title": "Checkout"}
+            ]
+            messenger.send_reply_button("Products added to cart!", buttons, mobile)
+
+        elif button_id == "checkout":
+            if mobile not in carts or not carts[mobile]:
+                messenger.send_message("Your cart is empty!", mobile)
+                return
+
+            buttons = [
+                {"id": "confirm_checkout", "title": "Confirm"},
+                {"id": "cancel_checkout", "title": "Cancel"}
+            ]
+            messenger.send_reply_button("Ready to complete your purchase?", buttons, mobile)
+
+        elif button_id == "confirm_checkout":
+            # Process checkout
+            carts[mobile] = []  # Clear cart
+            messenger.send_message("Thank you for your purchase!", mobile)
+
+    except Exception as e:
+        logger.error(f"Error handling interactive message: {str(e)}")
+        messenger.send_message("Sorry, something went wrong.", mobile)
+
+# Keep your original media handling, DNS setup, and other utility functions
 
 def test_connection():
     """Test DNS resolution and connection to Graph API"""
@@ -274,8 +298,5 @@ def test_connection():
         logger.error(f"Connection test failed: {e}")
 
 if __name__ == "__main__":
-    # Test connection before starting
     test_connection()
-    
-    # Start Flask app
     app.run(debug=True, host="0.0.0.0", port=7860)