Update app.py

app.py

@@ -1,26 +1,30 @@
 # chatbot_api.py
+
 import os
 import time
 import requests
 import base64
-from datetime import datetime
+from datetime import datetime, timedelta
 from bs4 import BeautifulSoup
 
-from fastapi import FastAPI, Request, HTTPException, BackgroundTasks, UploadFile, File
+from fastapi import FastAPI, Request, HTTPException, BackgroundTasks, UploadFile, File, Form
 from fastapi.responses import JSONResponse, StreamingResponse
 
 import openai
 
+# For sentiment analysis using TextBlob
+from textblob import TextBlob
+
+# SQLAlchemy Imports (Async)
 from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
 from sqlalchemy.orm import sessionmaker, declarative_base
-from sqlalchemy import Column, Integer, String, DateTime, Text
+from sqlalchemy import Column, Integer, String, DateTime, Text, Float
 
 # --- Configuration & Environment Variables ---
 SPOONACULAR_API_KEY = os.getenv("SPOONACULAR_API_KEY", "815bf76e0764456293f0e96e080e8f60")
 PAYSTACK_SECRET_KEY = os.getenv("PAYSTACK_SECRET_KEY", "pk_test_3222fb257041f1f2fd5ef33eafd19e1db4bdb634")
-DATABASE_URL = os.getenv("DATABASE_URL", "postgresql+asyncpg://postgres.lgbnxplydqdymepehirg:Lovyelias5584.@aws-0-eu-central-1.pooler.supabase.com:5432/postgres")
-NVIDIA_API_KEY = os.getenv("NVIDIA_API_KEY", "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")  # For NVIDIA LLM endpoints
-
+DATABASE_URL = os.getenv("DATABASE_URL", "postgresql+asyncpg://postgres.lgbnxplydqdymepehirg:YourPassword@aws-0-eu-central-1.pooler.supabase.com:5432/postgres")
+NVIDIA_API_KEY = os.getenv("NVIDIA_API_KEY", "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")
 openai.api_key = os.getenv("OPENAI_API_KEY", "your_openai_api_key")
 
 # --- Database Setup ---
@@ -41,12 +45,29 @@ class Order(Base):
     user_id = Column(String, index=True)
     dish = Column(String)
     quantity = Column(String)
-    price = Column(String, default="0")  # Price as string (or use a numeric type)
+    price = Column(String, default="0")  # Price as string (or numeric type)
     status = Column(String, default="Pending Payment")  # e.g., Pending Payment, Paid, Completed
     payment_reference = Column(String, nullable=True)
     timestamp = Column(DateTime, default=datetime.utcnow)
 
-# Create the asynchronous engine. Make sure DATABASE_URL is configured correctly.
+class UserProfile(Base):
+    __tablename__ = "user_profiles"
+    id = Column(Integer, primary_key=True, index=True)
+    user_id = Column(String, unique=True, index=True)
+    name = Column(String, default="Valued Customer")
+    email = Column(String, default="unknown@example.com")
+    preferences = Column(Text, default="")  # e.g., favorite dishes, dietary restrictions
+    last_interaction = Column(DateTime, default=datetime.utcnow)
+
+class SentimentLog(Base):
+    __tablename__ = "sentiment_logs"
+    id = Column(Integer, primary_key=True, index=True)
+    user_id = Column(String, index=True)
+    timestamp = Column(DateTime, default=datetime.utcnow)
+    sentiment_score = Column(Float)
+    message = Column(Text)
+
+# Create the asynchronous engine.
 engine = create_async_engine(DATABASE_URL, echo=True)
 async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
 
@@ -54,8 +75,10 @@ async def init_db():
     async with engine.begin() as conn:
         await conn.run_sync(Base.metadata.create_all)
 
-# --- Global In-Memory Stores for Ephemeral Data ---
-user_state = {}  # Example: { user_id: { "flow": "order", "step": int, "data": dict } }
+# --- Global In-Memory Stores ---
+user_state = {}       # For active conversation flows: { user_id: { "flow": str, "step": int, "data": dict, "last_active": datetime } }
+conversation_context = {}  # Optionally store extended conversation history (in-memory for demo)
+proactive_timer = {}   # Track last interaction times to send proactive greetings
 
 # Local menu with nutritional details
 menu_items = [
@@ -74,14 +97,21 @@ async def log_chat_to_db(user_id: str, direction: str, message: str):
         session.add(entry)
         await session.commit()
 
+async def log_sentiment(user_id: str, message: str, score: float):
+    """Store sentiment analysis results in the database."""
+    async with async_session() as session:
+        entry = SentimentLog(user_id=user_id, sentiment_score=score, message=message)
+        session.add(entry)
+        await session.commit()
+
+def analyze_sentiment(text: str) -> float:
+    """Analyze text sentiment using TextBlob. Returns polarity between -1 and 1."""
+    blob = TextBlob(text)
+    return blob.sentiment.polarity
+
 def google_image_scrape(query: str) -> str:
-    """
-    Scrape Google Images using BeautifulSoup to get an image URL for the query.
-    Note: This basic scraper may break if Google changes its markup.
-    """
-    headers = {
-        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)"
-    }
+    """Scrape Google Images using BeautifulSoup to get an image URL."""
+    headers = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)"}
     search_url = f"https://www.google.com/search?tbm=isch&q={query}"
     try:
         response = requests.get(search_url, headers=headers, timeout=5)
@@ -97,13 +127,7 @@ def google_image_scrape(query: str) -> str:
     return ""
 
 def create_paystack_payment_link(email: str, amount: int, reference: str) -> dict:
-    """
-    Call Paystack to initialize a transaction.
-    - email: customer's email
-    - amount: in kobo (multiply NGN amount by 100)
-    - reference: unique order reference
-    Returns a dict with the payment link and status.
-    """
+    """Initialize a transaction via Paystack."""
     url = "https://api.paystack.co/transaction/initialize"
     headers = {
         "Authorization": f"Bearer {PAYSTACK_SECRET_KEY}",
@@ -125,7 +149,6 @@ def create_paystack_payment_link(email: str, amount: int, reference: str) -> dic
         return {"status": False, "message": str(e)}
 
 # --- NVIDIA LLM Streaming Functions ---
-
 def stream_text_completion(prompt: str):
     """
     Stream text completion using NVIDIA's text-only model.
@@ -176,13 +199,13 @@ def stream_image_completion(image_b64: str):
         if line:
             yield line.decode("utf-8") + "\n"
 
-# --- Internal Flow: Order Processing & Payment Integration ---
-def process_internal_flow(user_id: str, message: str) -> str:
+# --- Advanced Internal Flow: Order Processing & Payment Integration ---
+def process_order_flow(user_id: str, message: str) -> str:
     """
-    A simple two-step order flow:
+    A multi-step order process:
       - Step 1: Ask for dish.
       - Step 2: Ask for quantity.
-    After collecting these details, the order is saved and a payment link is generated.
+    After details are collected, save order and generate payment link.
     """
     if user_id in user_state:
         state = user_state[user_id]
@@ -198,11 +221,10 @@ def process_internal_flow(user_id: str, message: str) -> str:
                 data["quantity"] = message
                 order_id = f"ORD-{int(time.time())}"
                 data["order_id"] = order_id
-                # Price calculation example (₦1500 per serving)
-                price_per_serving = 1500
+                price_per_serving = 1500  # ₦1500 per serving
                 total_price = int(data["quantity"]) * price_per_serving
                 data["price"] = str(total_price)
-                # Save order details to the DB asynchronously.
+                # Save order asynchronously
                 import asyncio
                 async def save_order():
                     async with async_session() as session:
@@ -217,10 +239,10 @@ def process_internal_flow(user_id: str, message: str) -> str:
                         session.add(order)
                         await session.commit()
                 asyncio.create_task(save_order())
-                # Clear the in-memory state.
+                # Clear conversation state for order flow.
                 del user_state[user_id]
-                # Assume we have the customer's email; using a placeholder.
-                email = "customer@example.com"
+                # For personalization, try retrieving the user's email.
+                email = "customer@example.com"  # In a real system, use UserProfile info.
                 payment_data = create_paystack_payment_link(email, total_price * 100, order_id)
                 if payment_data.get("status"):
                     payment_link = payment_data["data"]["authorization_url"]
@@ -230,10 +252,44 @@ def process_internal_flow(user_id: str, message: str) -> str:
                     return f"Your order has been placed with Order ID {order_id}, but we could not initialize payment. Please try again later."
     else:
         if "order" in message.lower():
-            user_state[user_id] = {"flow": "order", "step": 1, "data": {}}
+            user_state[user_id] = {"flow": "order", "step": 1, "data": {}, "last_active": datetime.utcnow()}
             return "Sure! What dish would you like to order?"
     return ""
 
+# --- User Profile Functions ---
+async def get_or_create_user_profile(user_id: str) -> UserProfile:
+    """Retrieve an existing profile or create a new one."""
+    async with async_session() as session:
+        result = await session.execute(
+            UserProfile.__table__.select().where(UserProfile.user_id == user_id)
+        )
+        profile = result.scalar_one_or_none()
+        if profile is None:
+            profile = UserProfile(user_id=user_id, last_interaction=datetime.utcnow())
+            session.add(profile)
+            await session.commit()
+        return profile
+
+async def update_user_last_interaction(user_id: str):
+    """Update the user's last interaction timestamp."""
+    async with async_session() as session:
+        result = await session.execute(
+            UserProfile.__table__.select().where(UserProfile.user_id == user_id)
+        )
+        profile = result.scalar_one_or_none()
+        if profile:
+            profile.last_interaction = datetime.utcnow()
+            await session.commit()
+
+# --- Proactive Engagement: Warm Greetings ---
+async def send_proactive_greeting(user_id: str):
+    """Simulate sending a proactive greeting if the user has been inactive."""
+    # In a real system, you might schedule this using a job scheduler.
+    # Here we just simulate a warm greeting message.
+    greeting = "Hi again! We miss you. Would you like to see our new menu items or get personalized recommendations?"
+    await log_chat_to_db(user_id, "outbound", greeting)
+    return greeting
+
 # --- FastAPI Setup & Endpoints ---
 app = FastAPI()
 
@@ -241,15 +297,16 @@ app = FastAPI()
 async def on_startup():
     await init_db()
 
+# --- Chatbot Endpoint ---
 @app.post("/chatbot")
 async def chatbot_response(request: Request, background_tasks: BackgroundTasks):
     """
     Main chatbot endpoint.
-    Expects a JSON payload with:
+    Supports text queries, image queries, and advanced logic.
+    Expects JSON payload with:
       - 'user_id'
-      - 'message' (text query)
-      - Optionally, 'is_image': true and 'image_base64': <base64 string> for image queries.
-    Streaming responses will be returned.
+      - 'message'
+      - Optionally, 'is_image': true and 'image_base64'
     """
     data = await request.json()
     user_id = data.get("user_id")
@@ -260,35 +317,50 @@ async def chatbot_response(request: Request, background_tasks: BackgroundTasks):
     if not user_id:
         raise HTTPException(status_code=400, detail="Missing user_id in payload.")
 
-    # Log inbound message if it's a text query (for image queries, you might log separately).
-    if user_message:
-        background_tasks.add_task(log_chat_to_db, user_id, "inbound", user_message)
-    if is_image and image_b64 is None:
-        raise HTTPException(status_code=400, detail="is_image is true but no image_base64 provided.")
+    # Log inbound message
+    background_tasks.add_task(log_chat_to_db, user_id, "inbound", user_message)
+    # Update user last interaction and profile
+    await update_user_last_interaction(user_id)
+    await get_or_create_user_profile(user_id)
 
-    # If an image is provided, use the image model.
+    # Handle voice/image queries if applicable
     if is_image and image_b64:
-        # Verify the image is small enough.
         if len(image_b64) >= 180_000:
-            raise HTTPException(status_code=400, detail="Image too large. Use a smaller image or the assets API.")
-        # Return a streaming response from the image-based LLM.
+            raise HTTPException(status_code=400, detail="Image too large.")
         return StreamingResponse(stream_image_completion(image_b64), media_type="text/plain")
-    
-    # --- Process textual queries (menu, nutritional facts, internal flows) ---
+
+    # --- Advanced Textual Processing ---
+    # Analyze sentiment and log it
+    sentiment_score = analyze_sentiment(user_message)
+    background_tasks.add_task(log_sentiment, user_id, user_message, sentiment_score)
+
+    # Adjust response tone based on sentiment (for demonstration)
+    sentiment_modifier = ""
+    if sentiment_score < -0.3:
+        sentiment_modifier = "I'm sorry if you're having a tough time. "
+    elif sentiment_score > 0.3:
+        sentiment_modifier = "Great to hear from you! "
+
+    # Check for specialized commands:
     if "menu" in user_message.lower():
+        # Return menu with images
         menu_with_images = []
         for item in menu_items:
             image_url = google_image_scrape(item["name"])
-            menu_with_images.append({"name": item["name"], "description": item["description"], "price": item["price"], "image_url": image_url})
+            menu_with_images.append({
+                "name": item["name"],
+                "description": item["description"],
+                "price": item["price"],
+                "image_url": image_url
+            })
         response_payload = {
-            "response": "Here’s our delicious menu:",
+            "response": sentiment_modifier + "Here’s our delicious menu:",
             "menu": menu_with_images,
-            "follow_up": ("Would you like to see nutritional facts for any dish? "
-                          "Just type, for example, 'Nutritional facts for Jollof Rice'.")
+            "follow_up": "Would you like nutritional facts for any dish? Just type, for example, 'Nutritional facts for Jollof Rice'."
         }
         background_tasks.add_task(log_chat_to_db, user_id, "outbound", str(response_payload))
         return JSONResponse(content=response_payload)
-    
+
     if "nutritional facts for" in user_message.lower():
         dish_name = user_message.lower().replace("nutritional facts for", "").strip().title()
         dish = next((item for item in menu_items if item["name"].lower() == dish_name.lower()), None)
@@ -297,28 +369,30 @@ async def chatbot_response(request: Request, background_tasks: BackgroundTasks):
         else:
             response_text = f"Sorry, I couldn't find nutritional facts for {dish_name}."
         background_tasks.add_task(log_chat_to_db, user_id, "outbound", response_text)
-        return JSONResponse(content={"response": response_text})
-    
-    internal_response = process_internal_flow(user_id, user_message)
-    if internal_response:
-        background_tasks.add_task(log_chat_to_db, user_id, "outbound", internal_response)
-        return JSONResponse(content={"response": internal_response})
-    
-    # --- Fallback: Use NVIDIA text LLM with streaming ---
-    prompt = f"User query: {user_message}\nGenerate a helpful response for a restaurant chatbot."
-    
-    # Create a streaming response using the NVIDIA text model.
+        return JSONResponse(content={"response": sentiment_modifier + response_text})
+
+    # Check if this is an order flow request
+    order_response = process_order_flow(user_id, user_message)
+    if order_response:
+        background_tasks.add_task(log_chat_to_db, user_id, "outbound", order_response)
+        return JSONResponse(content={"response": sentiment_modifier + order_response})
+
+    # For context-aware conversation: store conversation context
+    conversation_context.setdefault(user_id, []).append({"timestamp": datetime.utcnow().isoformat(), "message": user_message})
+
+    # Fallback: use NVIDIA text LLM streaming for a response
+    prompt = f"User query: {user_message}\nGenerate a helpful, personalized response for a restaurant chatbot."
     def stream_response():
         for chunk in stream_text_completion(prompt):
             yield chunk
-
     background_tasks.add_task(log_chat_to_db, user_id, "outbound", f"LLM fallback response for prompt: {prompt}")
     return StreamingResponse(stream_response(), media_type="text/plain")
 
+# --- Chat History Endpoint ---
 @app.get("/chat_history/{user_id}")
 async def get_chat_history(user_id: str):
     """
-    Retrieve the chat history for a given user from the database.
+    Retrieve chat history for a user.
     """
     async with async_session() as session:
         result = await session.execute(
@@ -327,10 +401,11 @@ async def get_chat_history(user_id: str):
         history = result.fetchall()
         return [dict(row) for row in history]
 
+# --- Order Details Endpoint ---
 @app.get("/order/{order_id}")
 async def get_order(order_id: str):
     """
-    Retrieve details for a specific order from the database.
+    Retrieve details for a specific order.
     """
     async with async_session() as session:
         result = await session.execute(
@@ -342,6 +417,84 @@ async def get_order(order_id: str):
         else:
             raise HTTPException(status_code=404, detail="Order not found.")
 
+# --- User Profile Endpoint ---
+@app.get("/user_profile/{user_id}")
+async def get_user_profile(user_id: str):
+    """
+    Retrieve the user profile.
+    """
+    profile = await get_or_create_user_profile(user_id)
+    return {
+        "user_id": profile.user_id,
+        "name": profile.name,
+        "email": profile.email,
+        "preferences": profile.preferences,
+        "last_interaction": profile.last_interaction.isoformat()
+    }
+
+# --- Analytics Endpoint ---
+@app.get("/analytics")
+async def get_analytics():
+    """
+    Simple analytics dashboard endpoint.
+    Returns counts of messages, orders, and average sentiment.
+    """
+    async with async_session() as session:
+        # Total messages count
+        msg_result = await session.execute(ChatHistory.__table__.count())
+        total_messages = msg_result.scalar() or 0
+
+        # Total orders count
+        order_result = await session.execute(Order.__table__.count())
+        total_orders = order_result.scalar() or 0
+
+        # Average sentiment score
+        sentiment_result = await session.execute("SELECT AVG(sentiment_score) FROM sentiment_logs")
+        avg_sentiment = sentiment_result.scalar() or 0
+
+    return {
+        "total_messages": total_messages,
+        "total_orders": total_orders,
+        "average_sentiment": avg_sentiment
+    }
+
+# --- Voice Integration Endpoint ---
+@app.post("/voice")
+async def process_voice(file: UploadFile = File(...)):
+    """
+    Accept a voice file upload, perform speech-to-text (simulated), and process the resulting text.
+    In production, integrate with a real STT service.
+    """
+    # Simulated Speech-to-Text: read file bytes and decode (for demo, just return a fixed string)
+    contents = await file.read()
+    # In real implementation, send `contents` to an STT service.
+    simulated_text = "Simulated speech-to-text conversion result."
+    return {"transcription": simulated_text}
+
+# --- Payment Callback Endpoint (Stub) ---
+@app.post("/payment_callback")
+async def payment_callback(request: Request):
+    """
+    Endpoint to handle payment callbacks from Paystack.
+    Update order status based on callback data.
+    """
+    data = await request.json()
+    # Extract order reference and update order status accordingly.
+    # This is a stub – in production, verify callback signature, extract data, and update DB.
+    order_id = data.get("reference")
+    new_status = data.get("status", "Paid")
+    async with async_session() as session:
+        result = await session.execute(
+            Order.__table__.select().where(Order.order_id == order_id)
+        )
+        order = result.scalar_one_or_none()
+        if order:
+            order.status = new_status
+            await session.commit()
+            return JSONResponse(content={"message": "Order updated successfully."})
+        else:
+            raise HTTPException(status_code=404, detail="Order not found.")
+
 if __name__ == "__main__":
     import uvicorn
-    uvicorn.run(app, host="0.0.0.0", port=8000)
+    uvicorn.run(app, host="0.0.0.0", port=8000)