Update main.py

main.py

@@ -1,84 +1,104 @@
+from typing import List, Optional
+
 import httpx
-import json
+from bs4 import BeautifulSoup
 from fastapi import FastAPI, HTTPException, Query
-from pydantic import BaseModel, Field
-from typing import Optional, List, Dict, Any
 
-# Initialize the FastAPI app
+UPSTREAM_URL = "https://lite.duckduckgo.com/lite/"
+DEFAULT_KL = "wt-wt"
+
 app = FastAPI(
-    title="FastAPI DuckDuckGo Instant Answer API Proxy",
-    description="A proxy for the DuckDuckGo Instant Answer API. This API provides direct answers and summaries, but not a list of general web search results.",
-    version="1.1.0",
+    title="DuckDuckGo Lite Web Search Proxy",
+    description=(
+        "A tiny FastAPI wrapper that proxies search queries to DuckDuckGo Lite and returns"
+        " structured JSON results."
+    ),
+    version="1.0.0",
 )
 
-# Define Pydantic models for a cleaner response structure
-class SearchResult(BaseModel):
-    heading: Optional[str] = Field(None, description="The title of the answer.")
-    answer: Optional[str] = Field(None, description="A direct answer to the query.")
-    abstract: Optional[str] = Field(None, description="A summary or abstract of the topic.")
-    abstract_url: Optional[str] = Field(None, description="The URL for the abstract source.")
-    image_url: Optional[str] = Field(None, description="A relevant image URL.")
-    related_topics: List[Dict[str, Any]] = Field([], description="A list of related topics.")
-    raw_response: Dict[str, Any] = Field({}, description="The full, raw JSON response from the API.")
-
-# Define the base URL for the DuckDuckGo Instant Answer API
-DUCKDUCKGO_API_URL = "https://api.duckduckgo.com/"
-
-@app.get("/search", 
-         tags=["Search"],
-         response_model=SearchResult,
-         summary="Get an Instant Answer from DuckDuckGo")
-async def search_duckduckgo(
-    q: str = Query(..., description="The search query. Try 'what is the capital of France' or 'Google' to see results."),
+
+def _extract_results(html: str) -> List[dict]:
+    """Parse DuckDuckGo Lite HTML into a list of search results."""
+    soup = BeautifulSoup(html, "html.parser")
+    results: List[dict] = []
+
+    for table in soup.select("table.result"):
+        link_tag = table.select_one("td.result-link a")
+        if not link_tag:
+            continue
+
+        title = link_tag.get_text(strip=True)
+        url = link_tag.get("href")
+        if not url:
+            continue
+
+        snippet_tag = table.select_one("td.result-snippet")
+        snippet = snippet_tag.get_text(" ", strip=True) if snippet_tag else None
+
+        results.append({
+            "title": title,
+            "url": url,
+            "snippet": snippet,
+        })
+
+    if not results:
+        # As a fallback, try to find plain links if the expected structure changes.
+        for link_tag in soup.select("a.result-link"):
+            title = link_tag.get_text(strip=True)
+            url = link_tag.get("href")
+            if not title or not url:
+                continue
+            sibling = link_tag.find_parent().find_next_sibling() if link_tag.find_parent() else None
+            snippet = sibling.get_text(" ", strip=True) if sibling else None
+            results.append({
+                "title": title,
+                "url": url,
+                "snippet": snippet,
+            })
+
+    return results
+
+
+@app.post("/lite/")
+async def search_duckduckgo_lite(
+    q: str = Query(..., description="keywords for query", min_length=1),
+    s: Optional[int] = Query(None, description="can be `0`"),
+    o: Optional[str] = Query(None, description="can be `json`"),
+    api: Optional[str] = Query(None, description="can be `d.js`"),
+    kl: Optional[str] = Query(None, description="market/locale code"),
+    bing_market: Optional[str] = Query(None, description="market/locale code"),
 ):
-    """
-    Performs a search using the DuckDuckGo Instant Answer API.
-    
-    This endpoint returns structured information for queries that have a direct answer. 
-    It will return an empty response for general web search queries that do not trigger an Instant Answer.
-    """
-    params = {
-        "q": q,
-        "format": "json",
-        "no_html": 1,
+    params = {"q": q}
+    if s is not None:
+        params["s"] = str(s)
+    if o:
+        params["o"] = o
+    if api:
+        params["api"] = api
+    params["kl"] = kl or DEFAULT_KL
+    if bing_market:
+        params["bing_market"] = bing_market
+
+    headers = {
+        "User-Agent": (
+            "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
+            "(KHTML, like Gecko) Chrome/116.0.0.0 Safari/537.36"
+        )
     }
 
-    async with httpx.AsyncClient() as client:
-        try:
-            response = await client.get(DUCKDUCKGO_API_URL, params=params)
+    try:
+        async with httpx.AsyncClient(timeout=httpx.Timeout(10.0)) as client:
+            response = await client.get(UPSTREAM_URL, params=params, headers=headers)
             response.raise_for_status()
+    except httpx.HTTPError as exc:
+        raise HTTPException(status_code=502, detail="Upstream search failed") from exc
+
+    results = _extract_results(response.text)
 
-            if not response.text:
-                return SearchResult(raw_response={"message": "No results found."})
-
-            data = response.json()
-            
-            # Check if a meaningful answer is present.
-            # The 'Type' field is 'A' for Article, 'D' for Disambiguation, 'C' for category. Empty for no result.
-            if data.get("Type") or data.get("AbstractText"):
-                return SearchResult(
-                    heading=data.get("Heading"),
-                    answer=data.get("Answer"),
-                    abstract=data.get("AbstractText"),
-                    abstract_url=data.get("AbstractURL"),
-                    image_url=f'https://duckduckgo.com{data.get("Image")}' if data.get("Image") else None,
-                    related_topics=data.get("RelatedTopics", []),
-                    raw_response=data
-                )
-            else:
-                # Return an empty but valid SearchResult if no instant answer was found
-                return SearchResult(
-                    raw_response=data,
-                    related_topics=[{"message": "No direct Instant Answer found for this query."}]
-                )
-
-        except json.JSONDecodeError:
-            raise HTTPException(status_code=500, detail="Failed to decode JSON from DuckDuckGo API.")
-        except httpx.HTTPStatusError as e:
-            raise HTTPException(status_code=e.response.status_code, detail=f"Error from DuckDuckGo API: {e.response.text}")
-        except httpx.RequestError as e:
-            raise HTTPException(status_code=500, detail=f"Failed to connect to DuckDuckGo API: {str(e)}")
-
-@app.get("/", tags=["Root"])
-async def read_root():
-    return {"message": "Welcome to the DuckDuckGo Instant Answer API proxy!"}
+    return {
+        "query": q,
+        "kl": params.get("kl"),
+        "bing_market": params.get("bing_market"),
+        "count": len(results),
+        "results": results,
+    }