Before google search

app/services/message.py

@@ -13,6 +13,7 @@ from typing import List, Dict, Any, Optional
 from app.utils.load_env import ACCESS_TOKEN, WHATSAPP_API_URL, GEMNI_API, OPENAI_API
 from app.utils.system_prompt import system_prompt
 
+
 # Load environment variables
 load_dotenv()
 

app/utils/load_env.py

@@ -16,6 +16,8 @@ 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")
+CX_CODE = os.getenv("CX_CODE")
+CUSTOM_SEARCH_API_KEY = os.getenv("CUSTOM_SEARCH_API_KEY")
 
 # Debugging: Print the retrieved ACCESS_TOKEN (for development only)
 if ENV == "development":

tools/search_engine.py

@@ -0,0 +1,73 @@
+import requests
+
+import logging
+import httpx
+from dotenv import load_dotenv
+from typing import List, Dict, Optional
+
+from app.utils.load_env import CX_CODE, CUSTOM_SEARCH_API_KEY
+# Configure logging
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+# Load environment variables
+load_dotenv()
+
+async def google_search(query: str, num_results: str = "3") -> Optional[List[Dict[str, str]]]:
+    """
+    Perform a Google search using the Custom Search JSON API.
+
+    Args:
+        query: The search query string.
+        num_results: The number of results to fetch as a string. Defaults to "10".
+
+    Returns:
+        Optional: A list of dictionaries containing the title and link of each search result.
+        Returns None if an error occurs or no results are found.
+    """
+    if not query:
+        logger.error("Search query is empty.")
+        return None
+
+    try:
+        # Convert num_results to an integer
+        num_results_int = int(num_results)
+        if num_results_int > 10:
+            logger.warning("Google Custom Search API supports a maximum of 10 results per request.")
+            num_results_int = 10
+        elif num_results_int < 1:
+            logger.warning("num_results must be at least 1. Setting it to 1.")
+            num_results_int = 1
+    except ValueError:
+        logger.error("Invalid value for num_results. Must be a valid integer string.")
+        return None
+
+    # Define the URL and parameters
+    url = "https://www.googleapis.com/customsearch/v1"
+    params = {
+        "key": CUSTOM_SEARCH_API_KEY,
+        "cx": CX_CODE,
+        "q": query,
+        "num": num_results_int,
+    }
+    print(CUSTOM_SEARCH_API_KEY)
+    print(CX_CODE)
+    try:
+        # Make the HTTP request
+        async with httpx.AsyncClient() as client:
+            response = await client.get(url, params=params)  # Properly encode query parameters
+
+        # Process the response
+        if response.status_code == 200:
+            results = response.json()
+            items = results.get("items", [])
+            return [{"title": item["title"], "link": item["link"]} for item in items]
+
+        else:
+            logger.error(f"Google Search API error: {response.status_code} - {response.text}")
+            return None
+
+    except httpx.RequestError as e:
+        logger.error("A network error occurred while performing the Google search.")
+        logger.error(f"Error details: {e}")
+        return None