Update app.py

app.py

@@ -1,6 +1,7 @@
 import os
 import re
 import time
+import random
 import logging
 import io
 import json
@@ -103,82 +104,98 @@ def detect_tools(message):
     return tools
 
 # --- Internet Search Feature ---
-def web_search(query: str, max_results: int = 5) -> list:
-    """Search the web using DuckDuckGo for free. No API key required."""
-    try:
-        # Added timeout to DDGS to prevent long hangs
-        with DDGS(timeout=10) as ddgs:
-            results = list(ddgs.text(query, max_results=max_results))
-            return results
-    except Exception as e:
-        logger.error(f"Search failed: {str(e)}")
-        return [{"error": f"Search failed: {str(e)}"}]
-
-def search_news(query: str, max_results: int = 5) -> list:
-    """Search news using DuckDuckGo for free."""
-    try:
-        # Added timeout to DDGS to prevent long hangs
-        with DDGS(timeout=10) as ddgs:
-            results = list(ddgs.news(query, max_results=max_results))
-            return results
-    except Exception as e:
-        logger.error(f"News search failed: {str(e)}")
-        return [{"error": f"News search failed: {str(e)}"}]
+def ddg_search(query, max_results=5): 
+    try: 
+        from duckduckgo_search import DDGS 
+        time.sleep(random.uniform(1, 3)) 
+        with DDGS(headers={"User-Agent": "Mozilla/5.0 (X11; Linux x86_64)"}) as ddgs: 
+            return list(ddgs.text(query, max_results=max_results))
+    except: 
+        return [] 
+
+def bing_search(query, max_results=5): 
+    headers = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.0"} 
+    try: 
+        url = f"https://www.bing.com/search?q={requests.utils.quote(query)}" 
+        r = requests.get(url, headers=headers, timeout=10) 
+        links = re.findall(r'<a[^>]*href="(https?://[^"]+)"[^>]*h="[^"]*"', r.text) 
+        titles = re.findall(r'<a[^>]*href="https?://[^"]+"[^>]*h="[^"]*"[^>]*>(.*?)</a>', r.text) 
+        results = [] 
+        for i in range(min(max_results, len(links))): 
+            results.append({ 
+                "title": re.sub(r'<[^>]+>', '', titles[i]) if i < len(titles) else "No title", 
+                "href": links[i], 
+                "body": "" 
+            }) 
+        return results 
+    except: 
+        return [] 
+
+def wiki_search(query, max_results=3): 
+    try: 
+        url = "https://en.wikipedia.org/w/api.php" 
+        params = {"action": "query", "list": "search", "srsearch": query, "format": "json", "srlimit": max_results} 
+        r = requests.get(url, params=params, timeout=10) 
+        data = r.json() 
+        return [{"title": i["title"], "href": f"https://en.wikipedia.org/wiki/{i['title'].replace(' ', '_')}", 
+                 "body": i["snippet"].replace('<span class="searchmatch">', '').replace('</span>', '')} 
+                for i in data["query"]["search"]] 
+    except: 
+        return [] 
+
+def web_search(query, max_results=5): 
+    # Try DDG 
+    logger.info(f"🔍 Searching DDG: {query}")
+    ddg = ddg_search(query, max_results) 
+    if ddg: 
+        return {"engine": "DuckDuckGo", "results": ddg} 
+    
+    logger.warning("⚠️ DDG failed, trying Bing in 2s...")
+    time.sleep(2) 
+    # Try Bing 
+    bing = bing_search(query, max_results) 
+    if bing: 
+        return {"engine": "Bing", "results": bing} 
+    
+    logger.warning("⚠️ Bing failed, trying Wikipedia in 2s...")
+    time.sleep(2) 
+    # Try Wikipedia 
+    wiki = wiki_search(query, max_results) 
+    if wiki: 
+        return {"engine": "Wikipedia", "results": wiki} 
+    
+    return {"engine": "Failed", "results": [{"error": "All engines failed"}]} 
 
-def format_results(results: list) -> str:
-    """Format search results with citations."""
-    try:
-        if not results:
-            return "No results found."
-        if isinstance(results, list) and len(results) > 0 and isinstance(results[0], dict) and "error" in results[0]:
-            return results[0]["error"]
-        
-        formatted = []
-        # Ensure results is a list of dicts
-        if not isinstance(results, list):
-            return "Unexpected search result format."
-
-        for i, r in enumerate(results, 1):
-            if not isinstance(r, dict): continue
-            title = r.get("title", "No title")
-            url = r.get("href", r.get("url", "#"))
-            body = r.get("body", r.get("snippet", "No description"))[:300]
-            formatted.append(f"{i}. [{title}]({url})\n   {body}")
-        
-        return "\n\n".join(formatted) if formatted else "No results found."
-    except Exception as e:
-        logger.error(f"Formatting failed: {e}")
-        return f"Error formatting results: {str(e)}"
+def format_results(data): 
+    if not data or not data.get("results") or (isinstance(data["results"][0], dict) and "error" in data["results"][0]): 
+        return "No results found." 
+    
+    out = [f"Source: {data['engine']}\n"] 
+    for i, r in enumerate(data["results"], 1): 
+        title = r.get("title", "No title") 
+        url = r.get("href", r.get("url", "#")) 
+        body = r.get("body", r.get("snippet", ""))[:200] 
+        out.append(f"{i}. [{title}]({url})\n   {body}") 
+    return "\n\n".join(out)
 
 def perform_search(query):
     try:
         # Clean query for better search results
         search_query = query.lower().replace('search for', '').replace('find out about', '').replace('/search', '').strip()
-        logger.info(f"🔍 Searching for: {search_query}")
-        
-        # Try text search first
-        results = web_search(search_query)
         
-        # If no results or error, try news search as fallback
-        if not results or (isinstance(results[0], dict) and "error" in results[0]):
-            logger.info("Web search empty or failed, trying News search...")
-            news_results = search_news(search_query)
-            # If news search returned actual results, use them
-            if news_results and not (isinstance(news_results[0], dict) and "error" in news_results[0]):
-                results = news_results
+        search_data = web_search(search_query)
         
-        # Check if we have valid results now
-        if results and isinstance(results[0], dict) and "error" not in results[0]:
-            logger.info(f"✅ Found {len(results)} results")
-            return results, None
+        if search_data["engine"] != "Failed":
+            logger.info(f"✅ Found {len(search_data['results'])} results via {search_data['engine']}")
+            return search_data, None
         else:
-            error_msg = results[0]["error"] if results and isinstance(results[0], dict) and "error" in results[0] else "No results found."
+            error_msg = search_data["results"][0].get("error", "All search engines failed.")
             logger.warning(f"❌ Search failed: {error_msg}")
-            return [], error_msg
+            return None, error_msg
             
     except Exception as e:
         logger.error(f"⚠️ Global Search Error: {str(e)}")
-        return [], str(e)
+        return None, str(e)
 
 # --- Image Generation Feature ---
 @app.route('/api/image', methods=['POST'])
@@ -412,15 +429,16 @@ def search_api():
         if not query:
             return jsonify({"error": "No query provided"}), 400
         
-        results, error = perform_search(query)
-        if error and not results:
+        search_data, error = perform_search(query)
+        if error and not search_data:
             return jsonify({"response": error, "results": []})
-        if not results:
+        if not search_data:
             return jsonify({"response": "No results found.", "results": []})
             
         return jsonify({
-            "response": format_results(results),
-            "results": results,
+            "response": format_results(search_data),
+            "results": search_data["results"],
+            "engine": search_data["engine"],
             "status": "success"
         })
     except Exception as e: