Update main.py

main.py

@@ -1,320 +1,406 @@
-import time
-import json
+from fastapi import FastAPI, HTTPException, Query
+from typing import List, Optional
+from pydantic import BaseModel
+from time import sleep
+from curl_cffi.requests import Session
+from urllib.parse import urlencode, unquote, urlparse, parse_qs
 import base64
-from typing import List, Optional, Dict, Any
-from urllib.parse import urlencode, urlparse, parse_qs
-
-import uvicorn
-from fastapi import FastAPI, HTTPException, Query, Request, Response
-from pydantic import BaseModel, Field
-from starlette.middleware.base import BaseHTTPMiddleware, RequestResponseEndpoint
-from curl_cffi.requests import AsyncSession
+from typing import Dict, Any
+from concurrent.futures import ThreadPoolExecutor
+from webscout.litagent import LitAgent
 from bs4 import BeautifulSoup
+import json
 
+app = FastAPI(
+    title="Bing Search API",
+    description="A FastAPI wrapper for the BingSearch library with advanced features.",
+    version="1.0.0",
+)
 
-# --- Pydantic Models for API Responses ---
-
-class SearchResultMetadata(BaseModel):
-    """
-    Defines the structure for rich metadata associated with a search result,
-    such as sitelinks and the display URL.
-    """
-    sitelinks: Optional[List[Dict[str, str]]] = Field(
-        None, description="A list of sitelinks (title and URL) found under the main result."
-    )
-    displayed_url: Optional[str] = Field(
-        None, description="The user-friendly display URL or breadcrumb shown on the search page."
-    )
+# --- BingSearch Library Code ---
+# The provided BingSearch code is integrated here directly.
 
 class BingSearchResult(BaseModel):
-    """Represents a single text search result from Bing."""
-    url: str = Field(..., description="The direct URL of the search result.")
-    title: str = Field(..., description="The title of the search result.")
-    description: str = Field(..., description="A brief description or snippet of the search result.")
-    metadata: SearchResultMetadata = Field(
-        default_factory=SearchResultMetadata, description="Additional rich metadata scraped for the result."
-    )
+    url: str
+    title: str
+    description: str
+    metadata: Dict[str, Any] = {}
 
 class BingImageResult(BaseModel):
-    """Represents a single image search result from Bing."""
-    title: str = Field(..., description="The title or description of the image.")
-    image_url: str = Field(..., description="The direct URL to the full-size image.")
-    thumbnail_url: str = Field(..., description="The URL to the thumbnail of the image.")
-    page_url: str = Field(..., description="The URL of the page where the image was found.")
-    source: str = Field(..., description="The source or domain of the image.")
+    title: str
+    image: str
+    thumbnail: str
+    url: str
+    source: str
 
 class BingNewsResult(BaseModel):
-    """Represents a single news article search result from Bing."""
-    title: str = Field(..., description="The headline of the news article.")
-    url: str = Field(..., description="The URL to the full news article.")
-    description: str = Field(..., description="A snippet from the news article.")
-    source: str = Field("", description="The publisher or source of the news article.")
-
-
-# --- Custom Middleware for Response Headers ---
-
-class CustomHeaderMiddleware(BaseHTTPMiddleware):
-    """
-    This middleware adds custom headers to every API response, including
-    the processing time and a 'Powered-By' header.
-    """
-    async def dispatch(self, request: Request, call_next: RequestResponseEndpoint) -> Response:
-        start_time = time.time()
-        response = await call_next(request)
-        process_time = time.time() - start_time
-        response.headers["X-Response-Time"] = f"{process_time:.4f}s"
-        response.headers["X-Powered-By"] = "NiansuhAI"
-        return response
-
-
-# --- Bing Search Service ---
+    title: str
+    url: str
+    description: str
+    source: str = ""
 
 class BingSearch:
-    """
-    An asynchronous service class for scraping search results from Bing.
-    It handles text, image, news, and suggestion searches using curl_cffi
-    for efficient, non-blocking HTTP requests.
-    """
+    """Bing search implementation with configurable parameters and advanced features."""
+    _executor: ThreadPoolExecutor = ThreadPoolExecutor()
 
     def __init__(
         self,
-        timeout: int = 15,
+        timeout: int = 10,
         proxies: Optional[Dict[str, str]] = None,
+        verify: bool = True,
         lang: str = "en-US",
+        sleep_interval: float = 0.0,
         impersonate: str = "chrome110"
     ):
         self.timeout = timeout
-        self.proxies = proxies or {}
+        self.proxies = proxies if proxies else {}
+        self.verify = verify
         self.lang = lang
+        self.sleep_interval = sleep_interval
         self._base_url = "https://www.bing.com"
-        self.session = AsyncSession(
+        self.session = Session(
             proxies=self.proxies,
+            verify=self.verify,
             timeout=self.timeout,
             impersonate=impersonate
         )
-        # Use a realistic User-Agent to mimic a real browser
+        # 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/125.0.0.0 Safari/537.36",
-            "Accept-Language": "en-US,en;q=0.9",
+            "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"
         })
 
-    async def _fetch_html(self, url: str) -> str:
-        """Asynchronously fetches HTML content from a given URL."""
-        try:
-            resp = await self.session.get(url)
-            resp.raise_for_status()
-            return resp.text
-        except Exception as e:
-            # Raise an HTTPException that FastAPI can handle gracefully
-            raise HTTPException(status_code=502, detail=f"Failed to fetch Bing content: {e}")
-
-    def _parse_url(self, url: Optional[str]) -> str:
-        """Decodes Bing's redirect URLs to find the actual destination URL."""
-        if not url:
-            return ""
+    # FIX: Updated selectors to be more robust against Bing UI changes.
+    def _selectors(self, element):
+        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
+        }
+        return selectors[element]
+
+    def _first_page(self, query):
+        url = f'{self._base_url}/search?q={query}&search=&form=QBLH'
+        return {'url': url, 'data': None}
+
+    def _next_page(self, soup):
+        selector = self._selectors('next')
+        next_page_tag = soup.select_one(selector)
+        url = None
+        if next_page_tag and next_page_tag.get('href'):
+            url = self._base_url + next_page_tag['href']
+        return {'url': url, 'data': None}
+
+    def _get_url(self, tag):
+        url = tag.get('href', '')
+        resp = url
         try:
             parsed_url = urlparse(url)
             query_params = parse_qs(parsed_url.query)
             if "u" in query_params:
-                encoded_url = query_params["u"][0].split("&")[0]
-                # Decode the Base64-encoded URL
-                decoded_bytes = base64.urlsafe_b64decode(encoded_url + '===')
-                return decoded_bytes.decode('utf-8', errors='ignore')
-        except (KeyError, IndexError, Exception):
-            # Fallback to the original URL if parsing fails
-            return url
-        return url
+                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')
+        except Exception as e:
+            print(f"Error decoding Base64 string: {e}")
+        return resp
 
-    async def text(
-        self, keywords: str, region: Optional[str], max_results: int
+    # FIX: The entire text parsing logic is updated to handle modern Bing HTML structure.
+    def text(
+        self,
+        keywords: str,
+        region: str = None,
+        safesearch: str = "moderate",
+        max_results: int = 10,
+        unique: bool = True
     ) -> List[BingSearchResult]:
-        """Performs a text search and scrapes the results page."""
         if not keywords:
-            raise ValueError("Search keywords cannot be empty.")
-
-        params = {"q": keywords, "form": "QBLH"}
-        url = f'{self._base_url}/search?{urlencode(params)}'
-        if region:
-            url += f"&setmkt={region}"
+            raise ValueError("Search keywords cannot be empty")
 
-        html = await self._fetch_html(url)
-        soup = BeautifulSoup(html, "html.parser")
         fetched_results = []
+        fetched_links = set()
 
-        for result in soup.select('li.b_algo'):
+        def fetch_page(url):
+            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)}")
+
+        current_url = self._first_page(keywords)['url']
+        
+        while current_url and len(fetched_results) < max_results:
+            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
+                
+                link_tag = title_tag.find('a')
+                if not link_tag or not link_tag.has_attr('href'):
+                    continue
+                
+                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'
+                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)
+                
+                # 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)
+
+                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)
+                    
+                    if len(fetched_results) >= max_results:
+                        break
+            
             if len(fetched_results) >= max_results:
                 break
 
-            title_tag = result.find('h2')
-            link_tag = title_tag.find('a') if title_tag else None
-            if not link_tag or not link_tag.has_attr('href'):
-                continue
-
-            url_val = self._parse_url(link_tag.get('href'))
-            title = link_tag.get_text(strip=True)
-            description = result.find('p').get_text(strip=True) if result.find('p') else ""
+            # Find the next page URL
+            next_page_info = self._next_page(soup)
+            current_url = next_page_info['url']
+            if current_url:
+                sleep(self.sleep_interval)
 
-            sitelinks = [
-                {"title": a.get_text(strip=True), "url": self._parse_url(a.get('href'))}
-                for a in result.select('ul.b_vlist li a')
-            ]
-            displayed_url = result.cite.get_text(strip=True) if result.cite else None
-            metadata = SearchResultMetadata(sitelinks=sitelinks or None, displayed_url=displayed_url)
+        return fetched_results[:max_results]
 
-            if url_val and title:
-                fetched_results.append(
-                    BingSearchResult(url=url_val, title=title, description=description, metadata=metadata)
-                )
-        return fetched_results
 
-    async def suggestions(self, query: str, region: Optional[str]) -> List[str]:
-        """Fetches auto-complete suggestions for a given query."""
+    def suggestions(self, query: str, region: str = None) -> List[str]:
         if not query:
-            raise ValueError("Search query cannot be empty.")
-        params = {"query": query, "mkt": region or "en-US"}
+            raise ValueError("Search query cannot be empty")
+        params = {
+            "query": query,
+            "mkt": region if region else "en-US"
+        }
         url = f"https://api.bing.com/osjson.aspx?{urlencode(params)}"
         try:
-            resp = await self.session.get(url)
+            resp = self.session.get(url)
             resp.raise_for_status()
             data = resp.json()
-            return data[1] if isinstance(data, list) and len(data) > 1 else []
+            if isinstance(data, list) and len(data) > 1 and isinstance(data[1], list):
+                return data[1]
+            return []
         except Exception as e:
-            raise HTTPException(status_code=502, detail=f"Failed to fetch suggestions: {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)}")
 
-    async def images(self, keywords: str, max_results: int) -> List[BingImageResult]:
-        """Performs an image search and scrapes the results."""
+    def images(
+        self,
+        keywords: str,
+        region: str = None,
+        safesearch: str = "moderate",
+        max_results: int = 10
+    ) -> List[BingImageResult]:
         if not keywords:
-            raise ValueError("Search keywords cannot be empty.")
-        
-        params = {"q": keywords, "count": max_results}
+            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,
+        }
+        if region:
+            params["mkt"] = region
         url = f"{self._base_url}/images/search?{urlencode(params)}"
-        html = await self._fetch_html(url)
+        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)}")
         soup = BeautifulSoup(html, "html.parser")
         results = []
-
         for item in soup.select("a.iusc"):
-            if len(results) >= max_results:
-                break
             try:
-                meta = json.loads(item["m"])
-                if "murl" in meta:
-                    results.append(
-                        BingImageResult(
-                            title=meta.get("t", ""),
-                            image_url=meta["murl"],
-                            thumbnail_url=meta.get("turl", ""),
-                            page_url=meta.get("purl", ""),
-                            source=urlparse(meta.get("purl", "")).netloc
-                        )
-                    )
-            except (json.JSONDecodeError, KeyError):
+                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
+            except Exception:
                 continue
-        return results
+        return results[:max_results]
 
-    async def news(self, keywords: str, region: Optional[str], max_results: int) -> List[BingNewsResult]:
-        """Performs a news search and scrapes the results."""
+    def news(
+        self,
+        keywords: str,
+        region: str = None,
+        safesearch: str = "moderate",
+        max_results: int = 10,
+    ) -> List['BingNewsResult']:
         if not keywords:
-            raise ValueError("Search keywords cannot be empty.")
-            
-        params = {"q": keywords, "form": "QBNH"}
+            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,
+        }
         if region:
             params["mkt"] = region
-        
         url = f"{self._base_url}/news/search?{urlencode(params)}"
-        html = await self._fetch_html(url)
-        soup = BeautifulSoup(html, "html.parser")
+        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)}")
+        soup = BeautifulSoup(resp.text, "html.parser")
         results = []
-
-        for item in soup.select("div.news-card"):
-            if len(results) >= max_results:
-                break
-            a_tag = item.find("a", class_="title")
-            snippet_tag = item.find("div", class_="snippet")
+        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")
-
-            if a_tag and a_tag.has_attr('href'):
-                results.append(
-                    BingNewsResult(
-                        title=a_tag.get_text(strip=True),
-                        url=a_tag['href'],
-                        description=snippet_tag.get_text(strip=True) if snippet_tag else "",
-                        source=source_tag.get_text(strip=True).split('·')[0].strip() if source_tag else "",
-                    )
-                )
-        return results
-
-
-# --- FastAPI Application Setup ---
-
-app = FastAPI(
-    title="Bing Search API",
-    description="An advanced, asynchronous FastAPI wrapper to scrape Bing search results, powered by NiansuhAI.",
-    version="3.1.0",
-)
-
-app.add_middleware(CustomHeaderMiddleware)
-bing_search_service = BingSearch()
-
-
-# --- API Endpoints ---
-
-@app.get("/search", response_model=List[BingSearchResult], summary="Perform a text search with rich metadata")
+            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
+        return results[:max_results]
+
+
+bing = BingSearch()
+
+@app.get("/search", response_model=List[BingSearchResult])
 async def text_search(
     query: str = Query(..., description="The search keywords."),
-    region: Optional[str] = Query(None, description="The market/region for the search (e.g., 'en-US')."),
-    max_results: int = Query(10, ge=1, le=30, description="Maximum number of results to return."),
+    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."),
 ):
     """
-    Performs a text search on Bing and returns a list of results,
-    each enriched with metadata like sitelinks.
+    Perform a text search on Bing.
     """
     try:
-        return await bing_search_service.text(keywords=query, region=region, max_results=max_results)
-    except ValueError as e:
-        raise HTTPException(status_code=400, detail=str(e))
+        results = bing.text(
+            keywords=query,
+            region=region,
+            safesearch=safesearch,
+            max_results=max_results,
+        )
+        return results
     except Exception as e:
-        # Catch-all for any other unexpected errors
-        raise HTTPException(status_code=500, detail=f"An unexpected internal error occurred: {e}")
+        raise HTTPException(status_code=500, detail=str(e))
 
-@app.get("/suggestions", response_model=List[str], summary="Get real-time search suggestions")
+@app.get("/suggestions", response_model=List[str])
 async def get_suggestions(
-    query: str = Query(..., description="The partial search query for which to fetch 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')."),
 ):
-    """Fetches real-time search suggestions from Bing's autocomplete service."""
+    """
+    Fetches search suggestions for a given query.
+    """
     try:
-        return await bing_search_service.suggestions(query=query, region=region)
-    except ValueError as e:
-        raise HTTPException(status_code=400, detail=str(e))
+        suggestions = bing.suggestions(query=query, region=region)
+        return suggestions
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
 
-@app.get("/images", response_model=List[BingImageResult], summary="Perform an image search")
+@app.get("/images", response_model=List[BingImageResult])
 async def image_search(
     query: str = Query(..., description="The search keywords for images."),
-    max_results: int = Query(20, ge=1, le=100, description="Maximum number of image results to return."),
+    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."),
 ):
-    """Performs an image search on Bing and returns a list of image results."""
+    """
+    Perform an image search on Bing.
+    """
     try:
-        return await bing_search_service.images(keywords=query, max_results=max_results)
-    except ValueError as e:
-        raise HTTPException(status_code=400, detail=str(e))
+        results = bing.images(
+            keywords=query,
+            region=region,
+            safesearch=safesearch,
+            max_results=max_results,
+        )
+        return results
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
 
-@app.get("/news", response_model=List[BingNewsResult], summary="Perform a news search")
+@app.get("/news", response_model=List[BingNewsResult])
 async def news_search(
-    query: str = Query(..., description="The search keywords for news articles."),
-    region: Optional[str] = Query(None, description="The region for the news search (e.g., 'en-US')."),
-    max_results: int = Query(15, ge=1, le=50, description="Maximum number of news results to return."),
+    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."),
 ):
-    """Performs a news search on Bing and returns a list of recent articles."""
+    """
+    Perform a news search on Bing.
+    """
     try:
-        return await bing_search_service.news(keywords=query, region=region, max_results=max_results)
-    except ValueError as e:
-        raise HTTPException(status_code=400, detail=str(e))
-
-@app.get("/", include_in_schema=False)
-async def root():
-    return {"message": "Bing Search API is running. Visit /docs for documentation."}
-
+        results = bing.news(
+            keywords=query,
+            region=region,
+            safesearch=safesearch,
+            max_results=max_results,
+        )
+        return results
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
 
 if __name__ == "__main__":
-    # Standard entry point to run the FastAPI application using Uvicorn
+    import uvicorn
     uvicorn.run(app, host="0.0.0.0", port=8000)