Update main.py

main.py

@@ -1,32 +1,36 @@
 from fastapi import FastAPI, HTTPException, Query
-from typing import List, Optional
+from typing import List, Optional, Dict, Any
 from pydantic import BaseModel
 from time import sleep
 from curl_cffi.requests import Session
-from urllib.parse import urlencode, unquote, urlparse, parse_qs
+from urllib.parse import urlencode, urlparse, parse_qs
 import base64
-from typing import Dict, Any
-from concurrent.futures import ThreadPoolExecutor
-from webscout.litagent import LitAgent
-from bs4 import BeautifulSoup
 import json
+import logging
 
+# --- Setup Logging ---
+logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
+logger = logging.getLogger(__name__)
+
+
+# --- FastAPI App Initialization ---
 app = FastAPI(
     title="Snapzion Search API",
-    description="A FastAPI wrapper for the Search library with advanced features.",
-    version="1.0.0",
+    description="An improved FastAPI wrapper for a robust Bing Search library.",
+    version="1.1.0",
 )
 
-# --- BingSearch Library Code ---
-# The provided BingSearch code is integrated here directly.
+# --- Pydantic Models for API Responses ---
 
 class BingSearchResult(BaseModel):
+    """Represents a single text search result."""
     url: str
     title: str
     description: str
     metadata: Dict[str, Any] = {}
 
 class BingImageResult(BaseModel):
+    """Represents a single image search result."""
     title: str
     image: str
     thumbnail: str
@@ -34,15 +38,19 @@ class BingImageResult(BaseModel):
     source: str
 
 class BingNewsResult(BaseModel):
+    """Represents a single news search result."""
     title: str
     url: str
     description: str
     source: str = ""
 
-class BingSearch:
-    """Bing search implementation with configurable parameters and advanced features."""
-    _executor: ThreadPoolExecutor = ThreadPoolExecutor()
 
+# --- Improved BingSearch Library Code ---
+
+class BingSearch:
+    """
+    Bing search implementation with more robust selectors, better error handling, and advanced features.
+    """
     def __init__(
         self,
         timeout: int = 10,
@@ -64,24 +72,31 @@ class BingSearch:
             timeout=self.timeout,
             impersonate=impersonate
         )
-        # It's good practice to set a realistic User-Agent
         self.session.headers.update({
-            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/117.0.0.0 Safari/537.36"
+            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/117.0.0.0 Safari/537.36",
+            "Accept-Language": "en-US,en;q=0.9",
         })
 
-    # FIX: Updated selectors to be more robust against Bing UI changes.
-    def _selectors(self, element):
+    def _selectors(self, element: str) -> str:
+        """
+        Returns the CSS selector for a given element, updated for modern Bing layout.
+        """
         selectors = {
-            'links': 'ol#b_results > li',  # More generic selector for any list item in results
-            'next': 'a.sb_pagN' # Selector for the "Next" page button
+            # IMPROVEMENT: Use 'li.b_algo' to specifically target organic search results
+            # and ignore ads, "People also ask", etc.
+            'links': 'li.b_algo',
+            'next': 'a.sb_pagN'
         }
         return selectors[element]
 
-    def _first_page(self, query):
-        url = f'{self._base_url}/search?q={query}&search=&form=QBLH'
+    def _first_page(self, query: str) -> Dict[str, Any]:
+        """Constructs the URL for the first search page."""
+        params = {'q': query, 'form': 'QBLH', 'setlang': self.lang}
+        url = f'{self._base_url}/search?{urlencode(params)}'
         return {'url': url, 'data': None}
 
-    def _next_page(self, soup):
+    def _next_page(self, soup: 'BeautifulSoup') -> Dict[str, Any]:
+        """Finds the URL for the next page of results."""
         selector = self._selectors('next')
         next_page_tag = soup.select_one(selector)
         url = None
@@ -89,46 +104,47 @@ class BingSearch:
             url = self._base_url + next_page_tag['href']
         return {'url': url, 'data': None}
 
-    def _get_url(self, tag):
+    def _get_url(self, tag: 'Tag') -> str:
+        """Decodes the tracked URL from Bing's search results if possible."""
         url = tag.get('href', '')
-        resp = url
+        # Bing often uses a tracking redirect; try to extract the real URL.
         try:
             parsed_url = urlparse(url)
             query_params = parse_qs(parsed_url.query)
             if "u" in query_params:
-                encoded_url = query_params["u"][0][2:]
-                try:
-                    decoded_bytes = base64.urlsafe_b64decode(encoded_url + '===')
-                except base64.binascii.Error as e:
-                    print(f"Error decoding Base64 string: {e}")
-                    return url
-                resp = decoded_bytes.decode('utf-8')
+                # The real URL is often Base64 encoded in the 'u' parameter.
+                encoded_url = query_params["u"][0][2:]  # Strip leading 'a1'
+                # Add padding if necessary for correct decoding.
+                decoded_bytes = base64.urlsafe_b64decode(encoded_url + '===')
+                return decoded_bytes.decode('utf-8', errors='ignore')
         except Exception as e:
-            print(f"Error decoding Base64 string: {e}")
-        return resp
+            # IMPROVEMENT: Use logging instead of print for library code.
+            logger.warning(f"Could not decode Bing tracking URL '{url}': {e}")
+        return url
 
-    # FIX: The entire text parsing logic is updated to handle modern Bing HTML structure.
     def text(
         self,
         keywords: str,
-        region: str = None,
+        region: Optional[str] = None,
         safesearch: str = "moderate",
         max_results: int = 10,
         unique: bool = True
     ) -> List[BingSearchResult]:
+        """Performs a standard web search."""
         if not keywords:
             raise ValueError("Search keywords cannot be empty")
 
         fetched_results = []
-        fetched_links = set()
+        fetched_links = set() if unique else None
 
-        def fetch_page(url):
+        def fetch_page(url: str):
             try:
                 resp = self.session.get(url)
                 resp.raise_for_status()
                 return resp.text
             except Exception as e:
-                raise Exception(f"Bing search failed: {str(e)}")
+                logger.error(f"Failed to fetch Bing search page '{url}': {e}")
+                raise Exception(f"Bing search page request failed.")
 
         current_url = self._first_page(keywords)['url']
         
@@ -136,11 +152,9 @@ class BingSearch:
             html = fetch_page(current_url)
             soup = BeautifulSoup(html, "html.parser")
             
-            # Use the more generic selector for result blocks
             result_blocks = soup.select(self._selectors('links'))
             
             for result in result_blocks:
-                # Find the title and link, which are usually in an <h2> tag
                 title_tag = result.find('h2')
                 if not title_tag:
                     continue
@@ -152,29 +166,30 @@ class BingSearch:
                 url_val = self._get_url(link_tag)
                 title = title_tag.get_text(strip=True)
 
-                # Find the description, often in a div with class 'b_caption'
+                # Find the description in its container
                 desc_container = result.find('div', class_='b_caption')
-                description = ''
-                if desc_container:
-                    # Find the paragraph within the caption, or use the whole caption text
-                    desc_p = desc_container.find('p')
-                    if desc_p:
-                        description = desc_p.get_text(strip=True)
-                    else:
-                        description = desc_container.get_text(strip=True)
+                description = desc_container.get_text(strip=True) if desc_container else ''
                 
-                # Fallback if no 'b_caption' is found
-                if not description:
-                    p_tag = result.find('p')
-                    if p_tag:
-                        description = p_tag.get_text(strip=True)
+                # IMPROVEMENT: Extract metadata like the display URL.
+                metadata = {}
+                cite_tag = result.find('cite')
+                if cite_tag:
+                    metadata['display_url'] = cite_tag.get_text(strip=True)
 
                 if url_val and title:
                     if unique and url_val in fetched_links:
                         continue
                     
-                    fetched_results.append(BingSearchResult(url=url_val, title=title, description=description))
-                    fetched_links.add(url_val)
+                    fetched_results.append(
+                        BingSearchResult(
+                            url=url_val, 
+                            title=title, 
+                            description=description, 
+                            metadata=metadata
+                        )
+                    )
+                    if unique:
+                        fetched_links.add(url_val)
                     
                     if len(fetched_results) >= max_results:
                         break
@@ -182,7 +197,6 @@ class BingSearch:
             if len(fetched_results) >= max_results:
                 break
 
-            # Find the next page URL
             next_page_info = self._next_page(soup)
             current_url = next_page_info['url']
             if current_url:
@@ -190,217 +204,185 @@ class BingSearch:
 
         return fetched_results[:max_results]
 
-
-    def suggestions(self, query: str, region: str = None) -> List[str]:
+    def suggestions(self, query: str, region: Optional[str] = None) -> List[str]:
+        """Fetches auto-complete suggestions for a query."""
         if not query:
             raise ValueError("Search query cannot be empty")
-        params = {
-            "query": query,
-            "mkt": region if region else "en-US"
-        }
+        # This uses a stable API endpoint, so it's less likely to break.
+        params = {"query": query, "mkt": region if region else "en-US"}
         url = f"https://api.bing.com/osjson.aspx?{urlencode(params)}"
         try:
             resp = self.session.get(url)
             resp.raise_for_status()
             data = resp.json()
-            if isinstance(data, list) and len(data) > 1 and isinstance(data[1], list):
-                return data[1]
-            return []
+            return data[1] if isinstance(data, list) and len(data) > 1 else []
         except Exception as e:
-            if hasattr(e, 'response') and e.response is not None:
-                raise Exception(f"Bing suggestions failed with status {e.response.status_code}: {str(e)}")
-            else:
-                raise Exception(f"Bing suggestions failed: {str(e)}")
+            logger.error(f"Bing suggestions failed for query '{query}': {e}")
+            raise Exception("Bing suggestions failed.")
 
     def images(
-        self,
-        keywords: str,
-        region: str = None,
-        safesearch: str = "moderate",
-        max_results: int = 10
+        self, keywords: str, region: Optional[str] = None, safesearch: str = "moderate", max_results: int = 10
     ) -> List[BingImageResult]:
+        """Performs an image search."""
         if not keywords:
             raise ValueError("Search keywords cannot be empty")
-        safe_map = {
-            "on": "Strict",
-            "moderate": "Moderate",
-            "off": "Off"
-        }
-        safe = safe_map.get(safesearch.lower(), "Moderate")
+        
         params = {
-            "q": keywords,
-            "count": max_results,
-            "setlang": self.lang,
-            "safeSearch": safe,
+            "q": keywords, "count": max_results, "setlang": self.lang,
+            "safeSearch": safesearch.capitalize(), "form": "IRFLTR"
         }
         if region:
             params["mkt"] = region
+
         url = f"{self._base_url}/images/search?{urlencode(params)}"
         try:
             resp = self.session.get(url)
             resp.raise_for_status()
             html = resp.text
         except Exception as e:
-            if hasattr(e, 'response') and e.response is not None:
-                raise Exception(f"Bing image search failed with status {e.response.status_code}: {str(e)}")
-            else:
-                raise Exception(f"Bing image search failed: {str(e)}")
+            logger.error(f"Bing image search failed for '{keywords}': {e}")
+            raise Exception("Bing image search failed.")
+            
         soup = BeautifulSoup(html, "html.parser")
         results = []
+        # This method of parsing JSON from an attribute is quite stable.
         for item in soup.select("a.iusc"):
+            if len(results) >= max_results:
+                break
             try:
-                m = item.get("m")
-                meta = json.loads(m) if m else {}
-                image_url = meta.get("murl", "")
-                thumb_url = meta.get("turl", "")
-                title = meta.get("t", "")
-                page_url = meta.get("purl", "")
-                source = meta.get("surl", "")
-                if image_url:
-                    results.append(BingImageResult(title=title, image=image_url, thumbnail=thumb_url, url=page_url, source=source))
-                    if len(results) >= max_results:
-                        break
+                m_data = json.loads(item.get("m", "{}"))
+                if m_data.get("murl"):
+                    results.append(
+                        BingImageResult(
+                            title=m_data.get("t", ""), image=m_data.get("murl", ""),
+                            thumbnail=m_data.get("turl", ""), url=m_data.get("purl", ""),
+                            source=m_data.get("surl", "")
+                        )
+                    )
             except Exception:
                 continue
-        return results[:max_results]
+        return results
 
     def news(
-        self,
-        keywords: str,
-        region: str = None,
-        safesearch: str = "moderate",
-        max_results: int = 10,
-    ) -> List['BingNewsResult']:
+        self, keywords: str, region: Optional[str] = None, safesearch: str = "moderate", max_results: int = 10,
+    ) -> List[BingNewsResult]:
+        """Performs a news search."""
         if not keywords:
             raise ValueError("Search keywords cannot be empty")
-        safe_map = {
-            "on": "Strict",
-            "moderate": "Moderate",
-            "off": "Off"
-        }
-        safe = safe_map.get(safesearch.lower(), "Moderate")
-        params = {
-            "q": keywords,
-            "form": "QBNH",
-            "safeSearch": safe,
-        }
+
+        params = {"q": keywords, "form": "QBNH", "safeSearch": safesearch.capitalize()}
         if region:
             params["mkt"] = region
+            
         url = f"{self._base_url}/news/search?{urlencode(params)}"
         try:
             resp = self.session.get(url)
             resp.raise_for_status()
         except Exception as e:
-            if hasattr(e, 'response') and e.response is not None:
-                raise Exception(f"Bing news search failed with status {e.response.status_code}: {str(e)}")
-            else:
-                raise Exception(f"Bing news search failed: {str(e)}")
+            logger.error(f"Bing news search failed for '{keywords}': {e}")
+            raise Exception("Bing news search failed.")
+        
         soup = BeautifulSoup(resp.text, "html.parser")
         results = []
-        for item in soup.select("div.news-card, div.card, div.newsitem, div.card-content, div.t_s_main"):
-            a_tag = item.find("a")
-            title = a_tag.get_text(strip=True) if a_tag else ''
-            url_val = a_tag['href'] if a_tag and a_tag.has_attr('href') else ''
-            desc_tag = item.find("div", class_="snippet") or item.find("div", class_="news-card-snippet") or item.find("div", class_="snippetText")
-            description = desc_tag.get_text(strip=True) if desc_tag else ''
-            source_tag = item.find("div", class_="source")
-            source = source_tag.get_text(strip=True) if source_tag else ''
-            if url_val and title:
-                results.append(BingNewsResult(title=title, url=url_val, description=description, source=source))
-                if len(results) >= max_results:
-                    break
-        if not results:
-            for item in soup.select("a.title"):
-                title = item.get_text(strip=True)
-                url_val = item['href'] if item.has_attr('href') else ''
-                description = ''
-                source = ''
-                if url_val and title:
-                    results.append(BingNewsResult(title=title, url=url_val, description=description, source=source))
-                    if len(results) >= max_results:
-                        break
+        # Use a more resilient selector for news cards
+        for item in soup.select("div.news-card"):
+            if len(results) >= max_results:
+                break
+            
+            title_tag = item.find("a", class_="title")
+            desc_tag = item.find("div", class_="snippet")
+            source_tag = item.find(class_="source")
+
+            if title_tag and title_tag.has_attr('href'):
+                results.append(
+                    BingNewsResult(
+                        title=title_tag.get_text(strip=True),
+                        url=title_tag['href'],
+                        description=desc_tag.get_text(strip=True) if desc_tag else "",
+                        source=source_tag.get_text(strip=True) if source_tag else ""
+                    )
+                )
         return results[:max_results]
 
 
+# --- API Endpoints ---
+
 bing = BingSearch()
 
-@app.get("/search", response_model=List[BingSearchResult])
+@app.get("/search", response_model=List[BingSearchResult], summary="Perform a web search")
 async def text_search(
-    query: str = Query(..., description="The search keywords."),
-    region: Optional[str] = Query(None, description="The region for the search (e.g., 'us-US')."),
-    safesearch: str = Query("moderate", description="Safe search level ('on', 'moderate', 'off')."),
-    max_results: int = Query(10, description="Maximum number of results to return."),
+    query: str = Query(..., description="The search keywords.", min_length=1),
+    region: Optional[str] = Query(None, description="Region for the search (e.g., 'en-US')."),
+    safesearch: str = Query("moderate", description="Safe search: 'on', 'moderate', 'off'."),
+    max_results: int = Query(10, description="Maximum number of results.", gt=0, le=50),
 ):
     """
-    Perform a text search on Bing.
+    Perform a standard text search on Bing and get a list of results.
     """
     try:
-        results = bing.text(
-            keywords=query,
-            region=region,
-            safesearch=safesearch,
-            max_results=max_results,
-        )
+        results = bing.text(keywords=query, region=region, safesearch=safesearch, max_results=max_results)
         return results
+    except ValueError as e:
+        raise HTTPException(status_code=400, detail=str(e))
     except Exception as e:
+        logger.exception(f"Unhandled error in text_search for query: {query}")
         raise HTTPException(status_code=500, detail=str(e))
 
-@app.get("/suggestions", response_model=List[str])
+@app.get("/suggestions", response_model=List[str], summary="Get search suggestions")
 async def get_suggestions(
-    query: str = Query(..., description="The search query for which to fetch suggestions."),
-    region: Optional[str] = Query(None, description="The region for the suggestions (e.g., 'en-US')."),
+    query: str = Query(..., description="The query to get suggestions for.", min_length=1),
+    region: Optional[str] = Query(None, description="Region for suggestions (e.g., 'en-US')."),
 ):
     """
-    Fetches search suggestions for a given query.
+    Fetches auto-complete search suggestions for a given query.
     """
     try:
-        suggestions = bing.suggestions(query=query, region=region)
-        return suggestions
+        return bing.suggestions(query=query, region=region)
+    except ValueError as e:
+        raise HTTPException(status_code=400, detail=str(e))
     except Exception as e:
+        logger.exception(f"Unhandled error in get_suggestions for query: {query}")
         raise HTTPException(status_code=500, detail=str(e))
 
-@app.get("/images", response_model=List[BingImageResult])
+@app.get("/images", response_model=List[BingImageResult], summary="Perform an image search")
 async def image_search(
-    query: str = Query(..., description="The search keywords for images."),
-    region: Optional[str] = Query(None, description="The region for the image search (e.g., 'us-US')."),
-    safesearch: str = Query("moderate", description="Safe search level ('on', 'moderate', 'off')."),
-    max_results: int = Query(10, description="Maximum number of image results to return."),
+    query: str = Query(..., description="The search keywords for images.", min_length=1),
+    region: Optional[str] = Query(None, description="Region for the search (e.g., 'en-US')."),
+    safesearch: str = Query("moderate", description="Safe search: 'on', 'moderate', 'off'."),
+    max_results: int = Query(10, description="Maximum number of results.", gt=0, le=50),
 ):
     """
     Perform an image search on Bing.
     """
     try:
-        results = bing.images(
-            keywords=query,
-            region=region,
-            safesearch=safesearch,
-            max_results=max_results,
-        )
+        results = bing.images(keywords=query, region=region, safesearch=safesearch, max_results=max_results)
         return results
+    except ValueError as e:
+        raise HTTPException(status_code=400, detail=str(e))
     except Exception as e:
+        logger.exception(f"Unhandled error in image_search for query: {query}")
         raise HTTPException(status_code=500, detail=str(e))
 
-@app.get("/news", response_model=List[BingNewsResult])
+@app.get("/news", response_model=List[BingNewsResult], summary="Perform a news search")
 async def news_search(
-    query: str = Query(..., description="The search keywords for news."),
-    region: Optional[str] = Query(None, description="The region for the news search (e.g., 'us-US')."),
-    safesearch: str = Query("moderate", description="Safe search level ('on', 'moderate', 'off')."),
-    max_results: int = Query(10, description="Maximum number of news results to return."),
+    query: str = Query(..., description="The search keywords for news.", min_length=1),
+    region: Optional[str] = Query(None, description="Region for the search (e.g., 'en-US')."),
+    safesearch: str = Query("moderate", description="Safe search: 'on', 'moderate', 'off'."),
+    max_results: int = Query(10, description="Maximum number of results.", gt=0, le=50),
 ):
     """
     Perform a news search on Bing.
     """
     try:
-        results = bing.news(
-            keywords=query,
-            region=region,
-            safesearch=safesearch,
-            max_results=max_results,
-        )
+        results = bing.news(keywords=query, region=region, safesearch=safesearch, max_results=max_results)
         return results
+    except ValueError as e:
+        raise HTTPException(status_code=400, detail=str(e))
     except Exception as e:
+        logger.exception(f"Unhandled error in news_search for query: {query}")
         raise HTTPException(status_code=500, detail=str(e))
 
 if __name__ == "__main__":
     import uvicorn
+    # To run: uvicorn your_filename:app --reload
     uvicorn.run(app, host="0.0.0.0", port=8000)