Update app.py

app.py

@@ -1,6 +1,3 @@
-# --------------------
-# IMPORTS & SETUP
-# --------------------
 import re
 import os
 import time
@@ -11,23 +8,44 @@ from datetime import datetime, timedelta
 from typing import Dict, List, Optional
 from fastapi import FastAPI, Request, HTTPException, BackgroundTasks, UploadFile, File, Form
 from fastapi.responses import JSONResponse, StreamingResponse, RedirectResponse
-from sqlalchemy import select, update
+from fastapi.encoders import jsonable_encoder
+from sqlalchemy import select, update, func
 from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
 from sqlalchemy.orm import sessionmaker, declarative_base
 from sqlalchemy import Column, Integer, String, DateTime, Text, Float, Boolean
+import logging
+import hmac
+import hashlib
+import json
+
+# Configure logging
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger("delivery_service")
 
 # --------------------
-# DATABASE MODELS
+# DATABASE SETUP
 # --------------------
+DATABASE_URL = os.getenv("DATABASE_URL", "sqlite+aiosqlite:///./test.db")
+engine = create_async_engine(DATABASE_URL, echo=True)
+async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
+
 Base = declarative_base()
 
+# --------------------
+# DATABASE MODELS
+# --------------------
 class UserProfile(Base):
     __tablename__ = "user_profiles"
     id = Column(Integer, primary_key=True, index=True)
     user_id = Column(String(36), unique=True, index=True)
-    # ... [keep previous fields] ...
+    phone_number = Column(String(20), nullable=True)
+    name = Column(String(100), nullable=True)
+    email = Column(String(100), nullable=True)
+    preferences = Column(Text, default="{}")
+    last_interaction = Column(DateTime, default=datetime.utcnow)
     ux_mode = Column(String(20), default="default")
     accessibility_prefs = Column(Text, default="{}")
+    current_order_status = Column(String(50), default="None")
 
 class UXPreferences(Base):
     __tablename__ = "ux_preferences"
@@ -36,6 +54,38 @@ class UXPreferences(Base):
     color_scheme = Column(String(20), default="light")
     input_preference = Column(String(20), default="buttons")
 
+class ChatHistory(Base):
+    __tablename__ = "chat_history"
+    id = Column(Integer, primary_key=True, index=True)
+    user_id = Column(String(36), index=True)
+    message = Column(Text)
+    timestamp = Column(DateTime, default=datetime.utcnow)
+
+class Order(Base):
+    __tablename__ = "orders"
+    order_id = Column(String(36), primary_key=True, index=True)
+    user_id = Column(String(36), index=True)
+    status = Column(String(20), default="pending")
+    payment_reference = Column(String(100), nullable=True)
+    last_location = Column(Text, nullable=True)  # Stored as a JSON string
+
+class OrderTracking(Base):
+    __tablename__ = "order_tracking"
+    id = Column(Integer, primary_key=True, index=True)
+    order_id = Column(String(36), index=True)
+    status = Column(String(50))
+    message = Column(Text)
+    timestamp = Column(DateTime, default=datetime.utcnow)
+    location = Column(Text, nullable=True)  # Stored as a JSON string
+
+# --------------------
+# INITIALIZATION
+# --------------------
+async def init_db():
+    async with engine.begin() as conn:
+        await conn.run_sync(Base.metadata.create_all)
+    logger.info("Database tables created.")
+
 # --------------------
 # STATE MANAGEMENT
 # --------------------
@@ -56,10 +106,13 @@ class ConversationState:
             "role": role,
             "message": message
         })
-        # Keep only last 5 messages
+        # Keep only the last 5 messages
         if len(self.context) > 5:
             self.context = self.context[-5:]
 
+# Global state dictionary
+user_state: Dict[str, ConversationState] = {}
+
 # --------------------
 # CORE UTILITIES
 # --------------------
@@ -97,7 +150,7 @@ async def get_or_create_user_profile(user_id: str, phone: str = None) -> UserPro
 class DeliveryUXManager:
     @staticmethod
     async def generate_response_template(user_id: str) -> Dict:
-        """Generate personalized response structure"""
+        """Generate a personalized response structure."""
         profile = await get_or_create_user_profile(user_id)
         return {
             "meta": {
@@ -115,14 +168,14 @@ class DeliveryUXManager:
 
     @staticmethod
     async def handle_ux_preferences(user_id: str, preference: str):
-        """Update UX preferences"""
+        """Update UX preferences."""
         async with async_session() as session:
             prefs = await session.get(UXPreferences, user_id)
             if not prefs:
                 prefs = UXPreferences(user_id=user_id)
                 session.add(prefs)
             
-            # Update specific preference
+            # Update specific preference based on input
             if preference.startswith("color_"):
                 prefs.color_scheme = preference.split("_")[1]
             elif preference.startswith("speed_"):
@@ -130,28 +183,72 @@ class DeliveryUXManager:
             
             await session.commit()
 
+# --------------------
+# HELPER FUNCTIONS FOR PAYMENT CALLBACK & TRACKING
+# --------------------
+async def log_order_tracking(order_id: str, status: str, message: str):
+    async with async_session() as session:
+        tracking_entry = OrderTracking(
+            order_id=order_id,
+            status=status,
+            message=message
+        )
+        session.add(tracking_entry)
+        await session.commit()
+
+async def send_whatsapp_message(number: str, message: str):
+    # Dummy implementation for sending WhatsApp messages
+    logger.info(f"Sending WhatsApp message to {number}: {message}")
+
+async def update_user_order_status(user_id: str, order_id: str, status: str):
+    async with async_session() as session:
+        await session.execute(
+            update(UserProfile)
+            .where(UserProfile.user_id == user_id)
+            .values(current_order_status=status)
+        )
+        await session.commit()
+
+async def get_order_details(order_id: str) -> Optional[Order]:
+    async with async_session() as session:
+        result = await session.execute(
+            select(Order).where(Order.order_id == order_id)
+        )
+        return result.scalar_one_or_none()
+
+def calculate_eta(last_location: dict) -> str:
+    # Implement actual ETA calculation logic here.
+    return "30 minutes"
+
 # --------------------
 # MAIN APPLICATION
 # --------------------
 app = FastAPI(title="Delivery Service Chatbot")
 
+@app.on_event("startup")
+async def on_startup():
+    await init_db()
+
 @app.post("/chatbot")
 async def enhanced_chatbot_handler(request: Request, bg: BackgroundTasks):
     data = await request.json()
     user_id = data["user_id"]
     message = data.get("message", "")
     
-    # Initialize state
+    # Initialize conversation state if it doesn't exist
     if user_id not in user_state:
         user_state[user_id] = ConversationState()
     
     state = user_state[user_id]
     state.update_last_active()
     
-    # Process message
+    # Dummy conversation handling logic
     response = await DeliveryUXManager.generate_response_template(user_id)
+    response["content"]["text"] = f"Received your message: {message}"
+    state.add_context("user", message)
     
-    # [Add conversation handling logic here]
+    # Background task to update user's last interaction in the DB
+    bg.add_task(update_user_last_interaction, user_id)
     
     return JSONResponse(response)
 
@@ -164,69 +261,34 @@ async def update_ux_preferences(request: Request):
     )
     return {"status": "preferences updated"}
 
-# --------------------
-# PROACTIVE FEATURES
-# --------------------
-async def send_proactive_update(user_id: str, update_type: str):
-    """Send unsolicited updates to users"""
-    state = user_state.get(user_id)
-    if not state:
-        return  # User not active
-    
-    response = await DeliveryUXManager.generate_response_template(user_id)
-    
-    if update_type == "delivery_eta":
-        response["content"]["text"] = "📦 Your package is arriving in 15 minutes!"
-        response["content"]["quick_replies"] = [
-            {"title": "Track Live", "action": "track_live"},
-            {"title": "Delay Delivery", "action": "delay_delivery"}
-        ]
-    
-    return response
-
-# --------------------
-# ERROR HANDLING
-# --------------------
-@app.exception_handler(HTTPException)
-async def ux_error_handler(request: Request, exc: HTTPException):
-    return JSONResponse({
-        "meta": {"error": True},
-        "content": {
-            "text": f"⚠️ Error: {exc.detail}",
-            "quick_replies": [{"title": "Main Menu", "action": "reset"}]
-        }
-    }, status_code=exc.status_code)
-
-if __name__ == "__main__":
-    import uvicorn
-    uvicorn.run(app, host="0.0.0.0", port=8000)
-# --- FastAPI Setup & Endpoints ---
-app = FastAPI()
-
-@app.on_event("startup")
-async def on_startup():
-    await init_db()
-
-
-# --- Other Endpoints (Chat History, Order Details, User Profile, Analytics, Voice, Payment Callback) ---
 @app.get("/chat_history/{user_id}")
 async def get_chat_history(user_id: str):
     async with async_session() as session:
         result = await session.execute(
-            ChatHistory.__table__.select().where(ChatHistory.user_id == user_id)
+            select(ChatHistory).where(ChatHistory.user_id == user_id)
         )
-        history = result.fetchall()
-        return [dict(row) for row in history]
+        history = result.scalars().all()
+        return [jsonable_encoder({
+            "user_id": entry.user_id,
+            "message": entry.message,
+            "timestamp": entry.timestamp.isoformat()
+        }) for entry in history]
 
 @app.get("/order/{order_id}")
 async def get_order(order_id: str):
     async with async_session() as session:
         result = await session.execute(
-            Order.__table__.select().where(Order.order_id == order_id)
+            select(Order).where(Order.order_id == order_id)
         )
-        order = result.fetchone()
+        order = result.scalar_one_or_none()
         if order:
-            return dict(order)
+            return jsonable_encoder({
+                "order_id": order.order_id,
+                "user_id": order.user_id,
+                "status": order.status,
+                "payment_reference": order.payment_reference,
+                "last_location": order.last_location
+            })
         else:
             raise HTTPException(status_code=404, detail="Order not found.")
 
@@ -239,16 +301,23 @@ async def get_user_profile(user_id: str):
         "name": profile.name,
         "email": profile.email,
         "preferences": profile.preferences,
-        "last_interaction": profile.last_interaction.isoformat()
+        "last_interaction": profile.last_interaction.isoformat(),
+        "ux_mode": profile.ux_mode,
+        "current_order_status": profile.current_order_status
     }
 
 @app.get("/analytics")
 async def get_analytics():
     async with async_session() as session:
-        msg_result = await session.execute(ChatHistory.__table__.count())
+        msg_result = await session.execute(
+            select(func.count()).select_from(ChatHistory)
+        )
         total_messages = msg_result.scalar() or 0
-        order_result = await session.execute(Order.__table__.count())
+        order_result = await session.execute(
+            select(func.count()).select_from(Order)
+        )
         total_orders = order_result.scalar() or 0
+        # This query assumes a sentiment_logs table exists with a sentiment_score column.
         sentiment_result = await session.execute("SELECT AVG(sentiment_score) FROM sentiment_logs")
         avg_sentiment = sentiment_result.scalar() or 0
     return {
@@ -263,10 +332,11 @@ async def process_voice(file: UploadFile = File(...)):
     simulated_text = "Simulated speech-to-text conversion result."
     return {"transcription": simulated_text}
 
-# Enhanced payment callback with security checks
 @app.api_route("/payment_callback", methods=["GET", "POST"])
 async def payment_callback(request: Request):
-    """Handle payment gateway callbacks with HMAC verification"""
+    PAYSTACK_SECRET_KEY = os.getenv("PAYSTACK_SECRET_KEY", "dummy_secret")
+    MANAGEMENT_WHATSAPP_NUMBER = os.getenv("MANAGEMENT_WHATSAPP_NUMBER", "+1234567890")
+    
     try:
         if request.method == "POST":
             # Verify Paystack signature
@@ -298,7 +368,6 @@ async def payment_callback(request: Request):
             raise HTTPException(status_code=400, detail="Missing order reference")
 
         async with async_session() as session:
-            # Get order with lock to prevent race conditions
             result = await session.execute(
                 select(Order)
                 .where(Order.order_id == order_id)
@@ -326,9 +395,9 @@ async def payment_callback(request: Request):
             
             await session.commit()
 
-            # Send notifications
+            # Send notifications concurrently
             await asyncio.gather(
-                log_order_tracking(order_id, "Payment Updated", status),
+                log_order_tracking(order_id, "Payment Updated", f"Payment status changed to {status}"),
                 send_whatsapp_message(
                     MANAGEMENT_WHATSAPP_NUMBER,
                     f"Payment Update: Order {order_id} - {status}"
@@ -337,38 +406,35 @@ async def payment_callback(request: Request):
             )
 
             if request.method == "GET":
+                redirect_url = os.getenv("PAYMENT_REDIRECT_URL", "https://default-redirect.com")
                 return RedirectResponse(
-                    url=os.getenv("PAYMENT_REDIRECT_URL", "https://default-redirect.com"),
+                    url=redirect_url,
                     status_code=303
                 )
             return JSONResponse(content={"status": "success", "order_id": order_id})
 
     except HTTPException as he:
-        raise
+        raise he
     except Exception as e:
         logger.error(f"Payment callback error: {str(e)}")
         raise HTTPException(status_code=500, detail="Internal server error")
 
-# Enhanced tracking endpoint with caching
 from fastapi_cache.decorator import cache
 
 @app.get("/track_order/{order_id}", response_model=List[dict])
 @cache(expire=60)  # Cache for 1 minute
-async def get_order_tracking(order_id: str, 
-                            page: int = 1, 
-                            limit: int = 10):
+async def get_order_tracking(order_id: str, page: int = 1, limit: int = 10):
     """
-    Get order tracking history with pagination
+    Get order tracking history with pagination.
     """
     try:
         async with async_session() as session:
-            # Get order first to verify existence
             order_result = await session.execute(
                 select(Order).where(Order.order_id == order_id)
+            )
             if not order_result.scalar_one_or_none():
                 raise HTTPException(status_code=404, detail="Order not found")
 
-            # Get tracking history
             tracking_result = await session.execute(
                 select(OrderTracking)
                 .where(OrderTracking.order_id == order_id)
@@ -387,40 +453,58 @@ async def get_order_tracking(order_id: str,
                     "status": update.status,
                     "message": update.message,
                     "timestamp": update.timestamp.isoformat(),
-                    "location": jsonable_encoder(update.location) if update.location else None
+                    "location": json.loads(update.location) if update.location else None
                 } 
                 for update in tracking_updates
             ]
 
-            # Add estimated delivery time
+            # Add estimated delivery time if order is shipped
             order = await get_order_details(order_id)
-            if order.status == "shipped":
-                response[0]["estimated_delivery"] = calculate_eta(order.last_location)
+            if order and order.status.lower() == "shipped":
+                loc = json.loads(order.last_location) if order.last_location else {}
+                response[0]["estimated_delivery"] = calculate_eta(loc)
 
             return JSONResponse(content=response)
 
     except HTTPException as he:
-        raise
+        raise he
     except Exception as e:
         logger.error(f"Tracking error: {str(e)}")
         raise HTTPException(status_code=500, detail="Internal server error")
 
-# Helper functions
-async def get_order_details(order_id: str):
-    async with async_session() as session:
-        result = await session.execute(
-            select(Order).where(Order.order_id == order_id)
-        return result.scalar_one_or_none()
+# --------------------
+# PROACTIVE FEATURES
+# --------------------
+async def send_proactive_update(user_id: str, update_type: str):
+    """Send unsolicited updates to users."""
+    state = user_state.get(user_id)
+    if not state:
+        return  # User not active
+    
+    response = await DeliveryUXManager.generate_response_template(user_id)
+    
+    if update_type == "delivery_eta":
+        response["content"]["text"] = "📦 Your package is arriving in 15 minutes!"
+        response["content"]["quick_replies"] = [
+            {"title": "Track Live", "action": "track_live"},
+            {"title": "Delay Delivery", "action": "delay_delivery"}
+        ]
+    
+    return response
 
-async def update_user_order_status(user_id: str, order_id: str, status: str):
-    async with async_session() as session:
-        await session.execute(
-            update(UserProfile)
-            .where(UserProfile.user_id == user_id)
-            .values(current_order_status=status)
-        )
-        await session.commit()
+# --------------------
+# ERROR HANDLING
+# --------------------
+@app.exception_handler(HTTPException)
+async def ux_error_handler(request: Request, exc: HTTPException):
+    return JSONResponse({
+        "meta": {"error": True},
+        "content": {
+            "text": f"⚠️ Error: {exc.detail}",
+            "quick_replies": [{"title": "Main Menu", "action": "reset"}]
+        }
+    }, status_code=exc.status_code)
 
-def calculate_eta(last_location: dict) -> str:
-    # Implement actual ETA calculation logic
-    return "30 minutes"
+if __name__ == "__main__":
+    import uvicorn
+    uvicorn.run(app, host="0.0.0.0", port=8000)