Add Gemnini Flash

.env

@@ -0,0 +1,28 @@
+WHATSAPP_API_URL = "adsads"
+API_URL = "adsads"
+ACCESS_TOKEN = "EAAQNYsHYo2cBO30HZCtSkpRppS8ZA6Du2MV9PPFFepInIb2og5cjP1ZAZBJrJhHTlkmcrcN0BS6NZAqMltXhOZBzAyGdNaZCL6XtC6ZCdcug5JR5fZCdXVMzOBbuwPKPBwZABhFsWDIAORYaCI97ajvUxrZCSALNZAIH1E5hs2ZApETvWoVMiZCnoT5MqpyAZBoFFJ84K6vFgZDZD"
+
+ENV="development"
+usernameDb = 'hidh4125_admin'
+password = 'Alberto471'
+host = 'hidigi.asia'  
+port = '3306' 
+MySQLDb_name = 'hidh4125_speechRecognition'
+
+mongodb_uri = "mongodb+srv://dedyariansyah:Haleluyah30@test.c5ul8at.mongodb.net/?retryWrites=true&w=majority&appName=Test"
+
+DB_NAME='traveloka'
+mongodb_name = "llm_wa_hf"
+WHATSAPP_API_URL ='https://graph.facebook.com/v21.0/360447720489034/messages'
+SECRET_KEY = "cf6352b5ac05bfc2761e2973da54c4c1751ddcb765c0036197d9799f79257296"
+
+client = "http://localhost:5000"
+
+fastapiusername = "PPR"
+fastapipassword = "HiDigi"
+MEDIA_UPLOAD_URL = "http://localhost:5000/upload"
+
+GEMINI_API = "AIzaSyAJxmpiEGSDGAPHiJDbIuclq6EUaQCpPyg"
+OPENAI_API = "sk-proj-gYAJTf4yUP55p9KBrQNMT3BlbkFJTT8TNjrp2qaZbbnqszwM"
+ACCESS_TOKEN = "EAAQNYsHYo2cBO30HZCtSkpRppS8ZA6Du2MV9PPFFepInIb2og5cjP1ZAZBJrJhHTlkmcrcN0BS6NZAqMltXhOZBzAyGdNaZCL6XtC6ZCdcug5JR5fZCdXVMzOBbuwPKPBwZABhFsWDIAORYaCI97ajvUxrZCSALNZAIH1E5hs2ZApETvWoVMiZCnoT5MqpyAZBoFFJ84K6vFgZDZD"
+

app/core/config.py

@@ -0,0 +1,22 @@
+
+# app/core/config.py
+import os
+from pydantic_settings import BaseSettings
+# from dotenv import load_dotenv
+
+# load_dotenv()
+
+class Settings(BaseSettings):
+    MONGO_DETAILS: str = os.getenv("mongodb_uri")
+    MongoDB_NAME: str = "llm_wa_hf"
+    COLLECTION_NAMES: list = ["users", "transactions"]
+    ACCESS_TOKEN:str = os.getenv("ACCESS_TOKEN")
+    API_URL: str = os.getenv("API_URL")
+    MEDIA_UPLOAD_URL: str = os.getenv("MEDIA_UPLOAD_URL")
+    SECRET_KEY: str = os.getenv("SECRET_KEY")
+    ALGORITHM: str = "HS256"
+    ACCESS_TOKEN_EXPIRE_MINUTES: int = 100
+
+settings = Settings()
+
+print(f"SECRET_KEY: {settings.SECRET_KEY}")  # Add this line to verify the SECRET_KEY

app/crud/transaction.py

@@ -0,0 +1,24 @@
+from app.db.database import db
+from app.schemas.user import TransactionResponse
+from bson.objectid import ObjectId
+from datetime import datetime
+
+async def record_transaction(user_id: str, amount: float, description: str):
+    transaction = {
+        "user_id": ObjectId(user_id),
+        "amount": amount,
+        "description": description,
+        "timestamp": datetime.utcnow()
+    }
+    result = await db.transactions.insert_one(transaction)
+    transaction["_id"] = result.inserted_id
+    transaction["id"] = str(transaction.pop("_id"))
+    return transaction
+
+async def get_user_transactions(user_id: str):
+    cursor = db.transactions.find({"user_id": ObjectId(user_id)})
+    transactions = []
+    async for transaction in cursor:
+        transaction["id"] = str(transaction.pop("_id"))
+        transactions.append(transaction)
+    return transactions

app/crud/user.py

@@ -0,0 +1,30 @@
+from app.db.database import db
+from app.schemas.user import UserCreate
+from bson.objectid import ObjectId
+
+async def create_user(user: UserCreate):
+    result = await db.users.insert_one(user.dict())
+    return str(result.inserted_id)
+
+async def get_user_by_id(user_id: str):
+    user = await db.users.find_one({"_id": ObjectId(user_id)})
+    if user:
+        user["id"] = str(user.pop("_id"))
+    return user
+
+async def update_user_credits(user_id: str, amount: float):
+    user = await db.users.find_one({"_id": ObjectId(user_id)})
+    if not user:
+        return None
+
+    new_remaining_credits = user["remaining_credits"] + amount
+    payment_status = new_remaining_credits > 0
+
+    await db.users.update_one(
+        {"_id": ObjectId(user_id)},
+        {"$set": {"remaining_credits": new_remaining_credits, "payment_status": payment_status}}
+    )
+
+    user["remaining_credits"] = new_remaining_credits
+    user["payment_status"] = payment_status
+    return user

app/db/database.py

@@ -0,0 +1,46 @@
+# app/core/database.py
+from motor.motor_asyncio import AsyncIOMotorClient
+from app.core.config import settings
+from pymongo import IndexModel, ASCENDING
+import os
+from dotenv import load_dotenv
+
+load_dotenv()
+
+
+client = AsyncIOMotorClient(settings.MONGO_DETAILS)
+db = client[settings.MongoDB_NAME]
+
+def get_database(db_name:str):
+    return client[db_name]
+
+async def create_collection(db_name:str, collection_name:str):
+    database = get_database(db_name)
+    existing_collections = await database.list_collection_names()
+    if collection_name not in existing_collections:
+        await database.create_collection(collection_name)
+    else:
+        print(f"Collection '{collection_name}' already exists in database '{db_name}'")
+
+
+async def list_collection_names(db_name: str):
+    database = get_database(db_name)
+    collection_names = await database.list_collection_names()
+    return collection_names
+
+async def init_db():
+    print(f"Initializing database '{settings.MongoDB_NAME}'")
+    for collection_name in settings.COLLECTION_NAMES:
+        await create_collection(settings.MongoDB_NAME, collection_name)
+    
+    collections = await list_collection_names(settings.MongoDB_NAME)
+
+    await create_indexes()
+    print(f"Collections in '{settings.MongoDB_NAME}': {collections}")
+ 
+# Indexes for users and transactions
+async def create_indexes():
+    user_index = IndexModel([("email", ASCENDING)], unique=True)
+    transaction_index = IndexModel([("user_id", ASCENDING), ("timestamp", ASCENDING)])
+    await db.users.create_indexes([user_index])
+    await db.transactions.create_indexes([transaction_index])

app/endpoints/v1/users.py

@@ -0,0 +1,48 @@
+from fastapi import APIRouter, HTTPException, Depends
+from app.schemas.user import UserCreate, UserResponse, TransactionResponse
+from app.crud.user import create_user, get_user_by_id, update_user_credits
+from app.crud.transaction import record_transaction, get_user_transactions
+
+router = APIRouter()
+
+@router.post("/", response_model=UserResponse)
+async def create_new_user(user: UserCreate):
+    user_id = await create_user(user)
+    user_response = await get_user_by_id(user_id)
+    return user_response
+
+@router.get("/{user_id}", response_model=UserResponse)
+async def get_user(user_id: str):
+    user = await get_user_by_id(user_id)
+    if not user:
+        raise HTTPException(status_code=404, detail="User not found")
+    return user
+
+@router.post("/{user_id}/top-up", response_model=TransactionResponse)
+async def top_up_credits(user_id: str, amount: float):
+    if amount <= 0:
+        raise HTTPException(status_code=400, detail="Amount must be greater than 0")
+
+    transaction = await record_transaction(user_id, amount, "Top-up credits")
+    user = await update_user_credits(user_id, amount)
+    if not user:
+        raise HTTPException(status_code=404, detail="User not found")
+    return transaction
+
+@router.post("/{user_id}/deduct", response_model=TransactionResponse)
+async def deduct_credits(user_id: str, amount: float):
+    if amount <= 0:
+        raise HTTPException(status_code=400, detail="Amount must be greater than 0")
+
+    transaction = await record_transaction(user_id, -amount, "Deduct credits")
+    user = await update_user_credits(user_id, -amount)
+    if not user:
+        raise HTTPException(status_code=404, detail="User not found")
+    return transaction
+
+@router.get("/{user_id}/transactions", response_model=list[TransactionResponse])
+async def list_transactions(user_id: str):
+    transactions = await get_user_transactions(user_id)
+    if not transactions:
+        raise HTTPException(status_code=404, detail="No transactions found")
+    return transactions

app/main.py

@@ -7,121 +7,166 @@ from app.services.message import generate_reply, send_reply, process_message_wit
 import logging
 from datetime import datetime
 import time
+from contextlib import asynccontextmanager
+from app.db.database import create_indexes, init_db
+from app.endpoints.v1 import users
+from app.services.webhook_handler import webhook, verify_webhook
 
 # Configure logging
 logging.basicConfig(
     level=logging.INFO,
     format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
 )
+
 logger = logging.getLogger(__name__)
 # Initialize FastAPI app
-app = FastAPI()
-
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+   
+    
+    try:
+        await init_db()
+        logger.info("Connected to the MongoDB database!")
+        # collections = app.database.list_collection_names()
+        # print(f"Collections in {db_name}: {collections}")
+        yield
+    except Exception as e:
+        logger.error(e)
+        
+app = FastAPI(lifespan=lifespan)
+
+
+app.include_router(users.router, prefix="/users", tags=["Users"])
+
+# Register webhook routes
+app.post("/webhook")(webhook)
+app.get("/webhook")(verify_webhook)
 # In-memory cache with timestamp cleanup
-class MessageCache:
-    def __init__(self, max_age_hours: int = 24):
-        self.messages: Dict[str, float] = {}
-        self.max_age_seconds = max_age_hours * 3600
-
-    def add(self, message_id: str) -> None:
-        self.cleanup()
-        self.messages[message_id] = time.time()
-
-    def exists(self, message_id: str) -> bool:
-        self.cleanup()
-        return message_id in self.messages
-
-    def cleanup(self) -> None:
-        current_time = time.time()
-        self.messages = {
-            msg_id: timestamp 
-            for msg_id, timestamp in self.messages.items()
-            if current_time - timestamp < self.max_age_seconds
-        }
-
-message_cache = MessageCache()
-
-@app.post("/webhook")
-async def webhook(request: Request):
-    request_id = f"req_{int(time.time()*1000)}"
-    logger.info(f"Processing webhook request {request_id}")
-    payload = await request.json()
-
-    processed_count = 0
-    error_count = 0
-    results = []
-
-    entries = payload.get("entry", [])
-
-    for entry in entries:
-        entry_id = entry.get("id")
-        logger.info(f"Processing entry_id: {entry_id}")
-
-        changes = entry.get("changes", [])
-        for change in changes:
-            messages = change.get("value", {}).get("messages", [])
-            for message in messages:
-                message_id = message.get("id")
-                timestamp = message.get("timestamp")
-                content = message.get("text", {}).get("body")
-                sender_id = message.get("from")
-
-                # Deduplicate messages based on message_id
-                if message_cache.exists(message_id):
-                    logger.info(f"Duplicate message detected and skipped: {message_id}")
-                    continue
-
-                try:
-                    # Process message with retry logic
-                    result = await process_message_with_retry(
-                        sender_id,
-                        content,
-                        timestamp
-                    )
-                    # Add the message ID to the cache
-                    message_cache.add(message_id)
-                    processed_count += 1
-                    results.append(result)
-                except Exception as e:
-                    error_count += 1
-                    logger.error(
-                        f"Failed to process message {message_id}: {str(e)}",
-                        exc_info=True
-                    )
-                    results.append({
-                        "status": "error",
-                        "message_id": message_id,
-                        "error": str(e)
-                    })
-
-    response_data = {
-        "request_id": request_id,
-        "processed": processed_count,
-        "errors": error_count,
-        "results": results
-    }
-
-    logger.info(
-        f"Webhook processing completed - "
-        f"Processed: {processed_count}, Errors: {error_count}"
-    )
-
-    return JSONResponse(
-        content=response_data,
-        status_code=status.HTTP_200_OK
-    )
-
-
-
-@app.get("/webhook")
-async def verify_webhook(request: Request):
-    mode = request.query_params.get('hub.mode')
-    token = request.query_params.get('hub.verify_token')
-    challenge = request.query_params.get('hub.challenge')
-
-    # Replace 'your_verification_token' with the token you set in Facebook Business Manager
-    if mode == 'subscribe' and token == 'test':
-           # Return the challenge as plain text
-        return Response(content=challenge, media_type="text/plain")
-    else:
-        raise HTTPException(status_code=403, detail="Verification failed")
+# class MessageCache:
+#     def __init__(self, max_age_hours: int = 24):
+#         self.messages: Dict[str, float] = {}
+#         self.max_age_seconds = max_age_hours * 3600
+
+#     def add(self, message_id: str) -> None:
+#         self.cleanup()
+#         self.messages[message_id] = time.time()
+
+#     def exists(self, message_id: str) -> bool:
+#         self.cleanup()
+#         return message_id in self.messages
+
+#     def cleanup(self) -> None:
+#         current_time = time.time()
+#         self.messages = {
+#             msg_id: timestamp 
+#             for msg_id, timestamp in self.messages.items()
+#             if current_time - timestamp < self.max_age_seconds
+#         }
+
+# message_cache = MessageCache()
+# user_chats = {}
+
+# @app.post("/webhook")
+# async def webhook(request: Request):
+#     request_id = f"req_{int(time.time()*1000)}"
+#     logger.info(f"Processing webhook request {request_id}")
+#     payload = await request.json()
+
+#     print("Webhook received:", payload)
+
+#     processed_count = 0
+#     error_count = 0
+#     results = []
+
+#     entries = payload.get("entry", [])
+
+#     for entry in entries:
+#         entry_id = entry.get("id")
+#         logger.info(f"Processing entry_id: {entry_id}")
+
+#         changes = entry.get("changes", [])
+#         for change in changes:
+#             messages = change.get("value", {}).get("messages", [])
+#             for message in messages:
+#                 message_id = message.get("id")
+#                 timestamp = message.get("timestamp")
+#                 content = message.get("text", {}).get("body")
+#                 sender_id = message.get("from")
+#                 msg_type = message.get('type')
+
+#                 # Deduplicate messages based on message_id
+#                 if message_cache.exists(message_id):
+#                     logger.info(f"Duplicate message detected and skipped: {message_id}")
+#                     continue
+                
+#                 if sender_id not in user_chats:
+#                     user_chats[sender_id] = []
+#                 user_chats[sender_id].append({
+                   
+#                     "role": "user",
+#                     "content": content
+#                 })
+
+#                 history = "".join([f"{item['role']}: {item['content']}\n" for item in user_chats[sender_id]])
+#                 print(f"history: {history}")
+
+#                 try:
+#                     # Process message with retry logic
+#                     result = await process_message_with_retry(
+#                         sender_id,content,
+#                         history,
+#                         timestamp, 
+#                     )
+#                     user_chats[sender_id].append({
+              
+#                     "role": "assistant",
+#                     "content": result
+#                     })
+
+#                     # Add the message ID to the cache
+#                     message_cache.add(message_id)
+#                     processed_count += 1
+#                     results.append(result)
+#                 except Exception as e:
+#                     error_count += 1
+#                     logger.error(
+#                         f"Failed to process message {message_id}: {str(e)}",
+#                         exc_info=True
+#                     )
+#                     results.append({
+#                         "status": "error",
+#                         "message_id": message_id,
+#                         "error": str(e)
+#                     })
+
+#     response_data = {
+#         "request_id": request_id,
+#         "processed": processed_count,
+#         "errors": error_count,
+#         "results": results
+#     }
+
+#     logger.info(
+#         f"Webhook processing completed - "
+#         f"Processed: {processed_count}, Errors: {error_count}"
+#     )
+
+#     return JSONResponse(
+#         content=response_data,
+#         status_code=status.HTTP_200_OK
+#     )
+
+
+
+# @app.get("/webhook")
+# async def verify_webhook(request: Request):
+#     mode = request.query_params.get('hub.mode')
+#     token = request.query_params.get('hub.verify_token')
+#     challenge = request.query_params.get('hub.challenge')
+
+#     # Replace 'your_verification_token' with the token you set in Facebook Business Manager
+#     if mode == 'subscribe' and token == 'test':
+#            # Return the challenge as plain text
+#         return Response(content=challenge, media_type="text/plain")
+#     else:
+#         raise HTTPException(status_code=403, detail="Verification failed")

app/schemas/user.py

@@ -0,0 +1,33 @@
+from pydantic import BaseModel, EmailStr
+from datetime import datetime
+from bson import ObjectId
+from typing import Optional, List
+
+class UserBase(BaseModel):
+    username: str
+    phone_number: str
+    email: EmailStr
+
+class UserCreate(UserBase):
+    credits: float = 0.0
+    remaining_credits: float = 0.0
+    payment_status: bool = False
+
+class UserResponse(UserBase):
+    id: str
+    credits: float
+    remaining_credits: float
+    payment_status: bool
+
+    class Config:
+        orm_mode = True
+
+class TransactionResponse(BaseModel):
+    id: str
+    user_id: str
+    amount: float
+    description: str
+    timestamp: datetime
+
+    class Config:
+        orm_mode = True

app/services/cache.py

@@ -0,0 +1,23 @@
+import time
+from typing import Dict
+
+class MessageCache:
+    def __init__(self, max_age_hours: int = 24):
+        self.messages: Dict[str, float] = {}
+        self.max_age_seconds = max_age_hours * 3600
+
+    def add(self, message_id: str) -> None:
+        self.cleanup()
+        self.messages[message_id] = time.time()
+
+    def exists(self, message_id: str) -> bool:
+        self.cleanup()
+        return message_id in self.messages
+
+    def cleanup(self) -> None:
+        current_time = time.time()
+        self.messages = {
+            msg_id: timestamp 
+            for msg_id, timestamp in self.messages.items()
+            if current_time - timestamp < self.max_age_seconds
+        }

app/services/download_media.py

@@ -0,0 +1,158 @@
+import aiohttp
+import asyncio
+from typing import Optional
+
+import aiohttp
+import logging
+from typing import Optional
+
+logger = logging.getLogger(__name__)
+
+import aiohttp
+import aiofiles
+import logging
+import os
+from typing import Optional
+
+logger = logging.getLogger(__name__)
+
+async def download_whatsapp_media(media_id: str, access_token: str, file_path: Optional[str] = None) -> Optional[str]:
+    """
+    Asynchronously download media from WhatsApp Cloud API using media ID and save it to a file.
+
+    Args:
+        media_id (str): The ID of the media to download.
+        access_token (str): Your WhatsApp access token.
+        file_path (Optional[str]): The path where the media file should be saved.
+
+    Returns:
+        str: The file path if the media is saved successfully.
+        None: If download fails.
+    """
+    timeout = aiohttp.ClientTimeout(total=30)  # Set an appropriate timeout
+
+    async with aiohttp.ClientSession(timeout=timeout) as session:
+        # Step 1: Get the media URL
+        try:
+            url = f"https://graph.facebook.com/v21.0/{media_id}"
+            headers = {
+                "Authorization": f"Bearer {access_token}"
+            }
+            async with session.get(url, headers=headers) as response:
+                data = await response.json()
+                if response.status != 200:
+                    logger.error(f"Failed to get media URL. Status: {response.status}, Response: {data}")
+                    return None
+                
+                media_url = data.get('url')
+                if not media_url:
+                    logger.error(f"Failed to get media URL from response: {data}")
+                    return None
+        except aiohttp.ClientError as e:
+            logger.error(f"Error getting media URL: {e}")
+            return None
+        except Exception as e:
+            logger.error(f"Unexpected error: {e}")
+            return None
+
+        # Step 2: Download the actual media
+        try:
+            headers = {
+                "Authorization": f"Bearer {access_token}"
+            }
+
+            async with session.get(media_url, headers=headers) as media_response:
+                if media_response.status != 200:
+                    media_data = await media_response.text()
+                    logger.error(f"Failed to download media. Status: {media_response.status}, Response: {media_data}")
+                    return None
+
+                if file_path:
+                    # Ensure the directory exists
+                    os.makedirs("user_media", exist_ok=True)
+                    file_path = os.path.join("user_media", file_path)
+
+                    # Write the media content to the file asynchronously
+                    async with aiofiles.open(file_path, 'wb') as f:
+                        async for chunk in media_response.content.iter_chunked(1024):
+                            await f.write(chunk)
+                    return file_path
+                else:
+                    logger.error("File path not provided for saving the media.")
+                    return None
+        except aiohttp.ClientError as e:
+            logger.error(f"Error downloading media: {e}")
+            return None
+        except Exception as e:
+            logger.error(f"Error saving media to file: {e}")
+            return None
+
+# async def download_whatsapp_media(media_id: str, access_token: str) -> Optional[bytes]:
+#     """
+#     Asynchronously download media from WhatsApp Cloud API using media ID
+    
+#     Args:
+#         media_id (str): The ID of the media to download
+#         access_token (str): Your WhatsApp access token
+    
+#     Returns:
+#         bytes: The media content if successful
+#         None: If download fails
+#     """
+#     headers = {
+#         "Authorization": f"Bearer {access_token}"
+#     }
+    
+#     async with aiohttp.ClientSession() as session:
+#         try:
+#             # Step 1: Get the media URL
+#             url = f"https://graph.facebook.com/v21.0/{media_id}"
+#             async with session.get(url, headers=headers) as response:
+#                 if response.status != 200:
+#                     print(f"Failed to get media URL. Status: {response.status}")
+#                     return None
+                
+#                 data = await response.json()
+#                 media_url = data.get('url')
+                
+#                 if not media_url:
+#                     print("Failed to get media URL from response")
+#                     return None
+            
+#             # Step 2: Download the actual media
+#             async with session.get(media_url, headers=headers) as media_response:
+#                 if media_response.status != 200:
+#                     print(f"Failed to download media. Status: {media_response.status}")
+#                     return None
+                
+#                 return await media_response.read()
+            
+#             # Step 3: Save the media to a file
+                
+#         except aiohttp.ClientError as e:
+#             print(f"Error downloading media: {e}")
+#             return None
+
+async def save_whatsapp_image(media_id: str, access_token: str, output_path: str) -> None:
+    """
+    Asynchronously download and save WhatsApp image to file
+    
+    Args:
+        media_id (str): The ID of the media to download
+        access_token (str): Your WhatsApp access token
+        output_path (str): Path where to save the image
+    """
+    media_content = await download_whatsapp_media(media_id, access_token)
+    
+    if media_content:
+        # File operations are blocking, so we use asyncio.to_thread for Python 3.9+
+        # For older Python versions, you might want to use loop.run_in_executor
+        await asyncio.to_thread(save_file, output_path, media_content)
+        print(f"Image saved to {output_path}")
+    else:
+        print("Failed to download image")
+
+def save_file(path: str, content: bytes) -> None:
+    """Helper function to save file content"""
+    with open(path, 'wb') as f:
+        f.write(content)

app/services/message.py

@@ -1,14 +1,25 @@
 import os
 import httpx
 from dotenv import load_dotenv
-from typing import Dict, Any
+from typing import Dict, Any, Optional, List
 from datetime import datetime
 import logging
 import asyncio
+from openai import AsyncOpenAI
+import google.generativeai as genai
+import PIL.Image
+from typing import List, Dict, Any, Optional
+
+from app.utils.load_env import ACCESS_TOKEN, WHATSAPP_API_URL, GEMNI_API, OPENAI_API
+
 
 # Load environment variables
 load_dotenv()
 
+genai.configure(api_key=GEMNI_API)
+client = AsyncOpenAI(api_key = OPENAI_API)
+# ACCESS_TOKEN = "EAAQNYsHYo2cBO30HZCtSkpRppS8ZA6Du2MV9PPFFepInIb2og5cjP1ZAZBJrJhHTlkmcrcN0BS6NZAqMltXhOZBzAyGdNaZCL6XtC6ZCdcug5JR5fZCdXVMzOBbuwPKPBwZABhFsWDIAORYaCI97ajvUxrZCSALNZAIH1E5hs2ZApETvWoVMiZCnoT5MqpyAZBoFFJ84K6vFgZDZD"
+
 # Configure logging
 logging.basicConfig(
     level=logging.INFO,
@@ -17,8 +28,9 @@ logging.basicConfig(
 logger = logging.getLogger(__name__)
 
 # Define the WhatsApp API URL and Access Token
-WHATSAPP_API_URL = os.environ.get("WHATSAPP_API_URL")
-ACCESS_TOKEN = os.environ.get("ACCESS_TOKEN")
+# WHATSAPP_API_URL = os.environ.get("WHATSAPP_API_URL")
+# WHATSAPP_API_URL = "https://graph.facebook.com/v21.0/360447720489034/messages"
+# ACCESS_TOKEN = os.environ.get("ACCESS_TOKEN")
 
 # Validate environment variables
 if not WHATSAPP_API_URL or not ACCESS_TOKEN:
@@ -69,24 +81,171 @@ async def generate_reply(sender: str, content: str, timestamp: int) -> str:
 async def process_message_with_retry(
     sender_id: str,
     content: str,
-    timestamp: int
+    history: List[str],
+    timestamp: Optional[int] = None,
+    media: Optional[Dict[str, Any]] = None,
+    image_file_path: Optional[str] = None,
+    doc_path: Optional[str] = None,
 ) -> Dict[str, Any]:
     """Process message with retry logic"""
-    retries = 3
-    delay = 2  # Initial delay in seconds
-
-    for attempt in range(retries):
-        try:
-            generated_reply = await generate_reply(sender_id, content, timestamp)
-            response = await send_reply(sender_id, generated_reply)
-            return {"status": "success", "reply": generated_reply, "response": response}
-        except Exception as e:
-            logger.error(f"Attempt {attempt + 1} failed: {str(e)}", exc_info=True)
-            if attempt < retries - 1:
-                await asyncio.sleep(delay)
-                delay *= 2  # Exponential backoff
-            else:
-                raise Exception(f"All {retries} attempts failed.") from e
+    retries = 1
+    delay = 0.1  # Initial delay in seconds
+
+    # for attempt in range(retries):
+    try:
+        logger.info(f"Sending message to the Gemini model...")
+        generated_reply = await generate_response_from_gemini(sender = sender_id, content=content, history = history, timestamp = timestamp, image_file_path = image_file_path, media=media, doc_path = doc_path)   
+        logger.info(f"Reply generated: {generated_reply}")
+        response = await send_reply(sender_id, generated_reply)
+        return generated_reply
+        return {"status": "success", "reply": generated_reply, "response": response}
+    except Exception as e:
+        logger.error(f"Error generating reply: {str(e)}", exc_info=True)
+        return {"status": "error", "reply": "Sorry, I couldn't generate a response at this time."}
+        # logger.error(f"Attempt {attempt + 1} failed: {str(e)}", exc_info=True)
+        # if attempt < retries - 1:
+        #     await asyncio.sleep(delay)
+        #     delay *= 2  # Exponential backoff
+        # else:
+        #     raise Exception(f"All {retries} attempts failed.") from e
 
 # Example usage
 # asyncio.run(process_message_with_retry("1234567890", "hello", 1700424056000))
+
+
+async def generate_response_from_gemini(sender: str, content: str, timestamp: str, history: List[Dict[str, str]], media: Optional[Dict[str, Any]] = None, image_file_path: Optional[str] = None, doc_path: Optional[str] = None) -> str:
+    try:
+        print(f"Sender: {sender}")
+        print(f"Content: {content}")
+        print(f"Timestamp: {timestamp}")
+        print(f"History: {history}")
+        print(f"Media: {media}")
+
+        # Initialize the model
+        model = genai.GenerativeModel("gemini-1.5-pro-002")
+
+        # Define the chat history
+        chat = model.start_chat(
+            history=history
+        )
+        logger.info(f"file_path: {image_file_path}")
+        if image_file_path: # Should be bytes or a file-like object
+    
+
+            prompt = "Describe the following image:"
+            image_data = PIL.Image.open(image_file_path)
+
+            print("Sending image to the Gemini model...")
+            response = await chat.send_message_async(image_data)
+            print(f"Model response: {response.text}")
+            return response.text
+        
+        if doc_path:
+            doc_data = genai.upload_file(doc_path)
+            print("Sending document to the Gemini model...")
+            response = await chat.send_message_async(doc_data)
+            print(f"Model response: {response.text}")
+            return response.text
+
+        # Send the user's message
+        print("Sending message to the Gemini model...")
+        response = await chat.send_message_async(content)
+        print(f"Model response: {response.text}")
+
+        return response.text
+
+    except Exception as e:
+        print("Error generating reply from Gemini:", e)
+        return "Sorry, I couldn't generate a response at this time."
+
+
+async def generate_response_from_chatgpt(sender: str, content: str, timestamp: str, history: str) -> str:
+    """
+    Generate a reply using OpenAI's ChatGPT API.
+    """
+    try:
+        # # Initialize chat history if not provided
+        # chat_history = chat_history or []
+
+        # # Append the current user message to the chat history
+        # chat_history.append({"role": "user", "content": f"From {sender} at {timestamp}: {content}"})
+          
+        messages = [
+            {"role": "system", "content": "You're an investor, a serial founder, and you've sold many startups. You understand nothing but business."},
+            {"role": "system", "content": f"Message History: {history}"},
+            {"role": "user", "content": f"From {sender} at {timestamp}: {content}"}
+        ]
+
+        print(f"Messages: {messages}")
+        
+        response = await client.chat.completions.create(
+            model="gpt-3.5-turbo",
+            messages=messages,
+            max_tokens=200,
+            temperature=0.5
+        )
+
+        chatgpt_response = response.choices[0].message.content.strip()
+        # Append the assistant's response to the chat history
+        # chat_history.append({"role": "assistant", "content": chatgpt_response})
+        return chatgpt_response
+
+    except Exception as e:
+        print("Error generating reply:", e)
+        return "Sorry, I couldn't generate a response at this time."
+
+# async def generate_response_from_chatgpt(
+#     sender: str,
+#     content: str,
+#     timestamp: str,
+#     image: Optional[bytes] = None,
+#     file: Optional[bytes] = None,
+#     file_name: Optional[str] = None,
+#     chat_history: Optional[List[Dict[str, str]]] = None,
+# ) -> Dict[str, Any]:
+#     """
+#     Generate a reply using OpenAI's GPT-4 API, including support for images, files, and maintaining chat history.
+#     """
+#     try:
+#         # Initialize chat history if not provided
+#         chat_history = chat_history or []
+
+#         # Append the current user message to the chat history
+#         chat_history.append({"role": "user", "content": f"From {sender} at {timestamp}: {content}"})
+
+#         # Prepare files for the request
+#         files = []
+#         if image:
+#             files.append({"name": "image.png", "type": "image/png", "content": image})
+#         if file:
+#             files.append({"name": file_name or "file.txt", "type": "application/octet-stream", "content": file})
+
+#         logger.debug(f"Chat History Before Response: {chat_history}")
+
+#         # Send the request to the GPT-4 API
+#         response = await client.chat.completions.create(
+#             model="gpt-4-vision",  # Ensure this is the correct model for multimodal support
+#             messages=chat_history,
+#             files=files if files else None,  # Include files if present
+#             max_tokens=200,
+#             temperature=0.5,
+#         )
+
+#         # Parse the assistant's response
+#         chatgpt_response = response.choices[0].message.content.strip()
+
+#         # Append the assistant's response to the chat history
+#         chat_history.append({"role": "assistant", "content": chatgpt_response})
+
+#         logger.debug(f"Chat History After Response: {chat_history}")
+
+#         # Return both the assistant's response and the updated chat history
+#         return {"response": chatgpt_response, "chat_history": chat_history}
+
+#     except Exception as e:
+#         logger.error("Error generating reply", exc_info=True)
+#         return {"response": "Sorry, I couldn't generate a response at this time.", "chat_history": chat_history}
+
+
+
+

app/services/webhook_handler.py

@@ -0,0 +1,58 @@
+import logging
+from fastapi import Request, status
+from fastapi.responses import JSONResponse, Response
+from fastapi.exceptions import HTTPException
+from app.services.cache import MessageCache
+from app.services.message import process_message_with_retry
+from app.services.download_media import download_whatsapp_media
+from app.utils.handle_message import handle_message
+from app.utils.load_env import ACCESS_TOKEN
+import time
+
+logger = logging.getLogger(__name__)
+message_cache = MessageCache()
+user_chats = {}
+
+async def webhook(request: Request):
+    request_id = f"req_{int(time.time()*1000)}"
+
+    payload = await request.json()
+    logger.info(f"Processing webhook request {payload}")
+    processed_count = 0
+    error_count = 0
+    results = []
+
+    entries = payload.get("entry", [])
+    for entry in entries:
+        entry_id = entry.get("id")
+        logger.info(f"Processing entry_id: {entry_id}")
+
+        changes = entry.get("changes", [])
+        for change in changes:
+            messages = change.get("value", {}).get("messages", [])
+            media = {}
+            for message in messages:
+                logger.info(f"Processing message: {message}")
+                response = await handle_message(message=message, user_chats = user_chats, message_cache = message_cache, access_token = ACCESS_TOKEN)
+
+                results.append(response)
+
+    response_data = {
+        "request_id": request_id,
+        # "processed": processed_count,
+        # "errors": error_count,
+        "results": results,
+    }
+
+    logger.info(f"Webhook processing completed - Processed: {processed_count}, Errors: {error_count}")
+    return JSONResponse(content=response_data, status_code=status.HTTP_200_OK)
+
+async def verify_webhook(request: Request):
+    mode = request.query_params.get('hub.mode')
+    token = request.query_params.get('hub.verify_token')
+    challenge = request.query_params.get('hub.challenge')
+
+    if mode == 'subscribe' and token == 'test':
+        return Response(content=challenge, media_type="text/plain")
+    else:
+        raise HTTPException(status_code=403, detail="Verification failed")

app/utils/handle_message.py

@@ -0,0 +1,75 @@
+from app.services.download_media import download_whatsapp_media
+from app.services.message import process_message_with_retry
+from app.services.cache import MessageCache
+import logging
+
+message_cache = MessageCache()
+logger = logging.getLogger(__name__)
+async def handle_message(message, user_chats, message_cache, access_token):
+    try:
+        # Process the message content
+        content = message.get("text", {}).get("body")
+        message_id = message.get("id")
+        image = message.get("image")
+        document = message.get("document")
+        video = message.get("video")
+        sender_id = message.get("from")
+
+        logger.info(f"Content: {content}")
+        logger.info(f"sender_id: {sender_id}")
+
+        # Download media if exists
+        if image:
+            file_path = f"{image.get('id')}.jpg"
+            logger.info(f"file_path: {file_path}")
+            image_file_path = await download_whatsapp_media(image.get("id"), access_token=access_token, file_path=file_path)
+            logger.info(f"image file_path: {image_file_path}")
+        else:
+            image_file_path = None
+
+        if document:
+            filename = message.get("document", {}).get("filename")
+            logger.info(f"file_path: {filename}")
+            document_file_path = await download_whatsapp_media(document.get("id"), access_token=access_token, file_path=filename)
+            logger.info(f"document file_path: {document_file_path}")
+        else:
+            document_file_path = None
+
+        if video:
+            video_type = message.get("video", {}).get("mime_type")
+            video_filename= f"{video.get('id')}.{video_type.split('/')[-1]}"
+            logger.info(f"video_type: {video_type}")
+            video_file_path = await download_whatsapp_media(video.get("id"), access_token=access_token, file_path=video_filename)
+        else:
+            video_file_path = None
+        # Check for duplicates
+        if message_cache.exists(message_id):
+            logger.info(f"Duplicate message detected and skipped: {message_id}")
+            return {"status": "duplicate", "message_id": message_id}
+
+        # Initialize chat history
+        if sender_id not in user_chats:
+            user_chats[sender_id] = []
+        
+
+        # Append content to user chat
+        if user_chats[sender_id] == []:
+            user_chats[sender_id].append({"role": "user", "parts": "This is the chat history so far"})
+
+        logger.info(f"user_chats: {user_chats}")
+        
+    
+    
+        # Process the message
+        result = await process_message_with_retry(sender_id, content, user_chats[sender_id], image_file_path = image_file_path, doc_path = document_file_path)
+        
+        if content != None:
+            user_chats[sender_id].append({"role": "user", "parts": content})
+
+        user_chats[sender_id].append({"role": "model", "parts": result})
+
+        message_cache.add(message_id)
+
+        return {"status": "success", "message_id": message_id, "result": result}
+    except Exception as e:
+        return {"status": "error", "message_id": message.get("id"), "error": str(e)}

app/utils/load_env.py

@@ -0,0 +1,25 @@
+from dotenv import load_dotenv
+import os
+
+
+# Load the .env file if the environment is development
+ENV = os.getenv("ENV", "production")  # Default to production if ENV is not set
+
+if ENV == "development":
+    print("Environment detected: Development. Loading .env file.")
+    load_dotenv()
+else:
+    print("Environment detected: Production. Using system environment variables.")
+
+# Access your environment variables
+ACCESS_TOKEN = os.getenv("ACCESS_TOKEN")
+WHATSAPP_API_URL = os.getenv("WHATSAPP_API_URL")
+OPENAI_API = os.getenv("OPENAI_API")
+GEMNI_API = os.getenv("GEMINI_API")
+
+# Debugging: Print the retrieved ACCESS_TOKEN (for development only)
+if ENV == "development":
+    print(f"ACCESS_TOKEN loaded: {ACCESS_TOKEN}")
+
+
+