Update app.py

app.py

@@ -11,6 +11,7 @@ import random
 import textract
 import boto3
 from botocore.exceptions import NoCredentialsError
+from duckduckgo_search import DDGS
 
 app = FastAPI()
 
@@ -181,4 +182,82 @@ async def extract_text(file: UploadFile, api_key: str = Depends(verify_api_key))
     if not result['success']:
         raise HTTPException(status_code=500, detail=result['error'])
     return {"extracted_text": result['extracted_text']}
-    
+
+def web_search(query: str, num_results: int = 5) -> dict:
+    """
+    Perform a web search using DuckDuckGo and return the results.
+    Args:
+    query (str): The search query.
+    num_results (int, optional): The number of results to return. Defaults to 5.
+    Returns:
+    dict: A dictionary containing the success status and either the search results or an error message.
+    """
+    try:
+        with DDGS() as ddgs:
+            results = list(ddgs.text(query, max_results=num_results))
+            formatted_results = [
+                {
+                    'title': result['title'],
+                    'body': result['body'],
+                    'href': result['href']
+                }
+                for result in results
+            ]
+        return {
+            'success': True,
+            'results': formatted_results
+        }
+    except Exception as e:
+        logger.error(f"Error performing web search: {e}")
+        return {
+            'success': False,
+            'error': f"Error performing web search: {str(e)}"
+        }
+
+def image_search(query: str, num_results: int = 5) -> dict:
+    """
+    Perform an image search using DuckDuckGo and return the results.
+    Args:
+    query (str): The search query.
+    num_results (int, optional): The number of results to return. Defaults to 5.
+    Returns:
+    dict: A dictionary containing the success status and either the image search results or an error message.
+    """
+    try:
+        with DDGS() as ddgs:
+            results = list(ddgs.images(query, max_results=num_results))
+            formatted_results = [
+                {
+                    'title': result['title'],
+                    'image_url': result['image'],
+                    'thumbnail_url': result['thumbnail'],
+                    'source_url': result['url'],
+                    'width': result['width'],
+                    'height': result['height']
+                }
+                for result in results
+            ]
+        return {
+            'success': True,
+            'results': formatted_results
+        }
+    except Exception as e:
+        logger.error(f"Error performing image search: {e}")
+        return {
+            'success': False,
+            'error': f"Error performing image search: {str(e)}"
+        }
+
+@app.get("/web-search/")
+async def web_search_endpoint(query: str, num_results: int = 5, api_key: str = Depends(verify_api_key)):
+    result = web_search(query, num_results)
+    if not result['success']:
+        raise HTTPException(status_code=500, detail=result['error'])
+    return result
+
+@app.get("/image-search/")
+async def image_search_endpoint(query: str, num_results: int = 5, api_key: str = Depends(verify_api_key)):
+    result = image_search(query, num_results)
+    if not result['success']:
+        raise HTTPException(status_code=500, detail=result['error'])
+    return result