File size: 2,832 Bytes
b269676 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 |
import httpx
from bs4 import BeautifulSoup
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import uvicorn
app = FastAPI()
class SearchQuery(BaseModel):
query: str
class BraveSearch:
def __init__(self, query):
self.query = query
self.url = f"https://search.brave.com/search?q={query.replace(' ', '+')}&source=web"
async def fetch_results(self):
try:
async with httpx.AsyncClient() as client:
response = await client.get(self.url)
response.raise_for_status()
soup = BeautifulSoup(response.text, "html.parser")
results = {
"searchResults": [],
"additionalData": [],
"faq": []
}
for el in soup.select(".snippet"):
title = el.select_one(".title")
description = el.select_one(".snippet-description")
link = el.select_one("a")
results["searchResults"].append({
"title": title.text.strip() if title else "No Title",
"description": description.text.strip() if description else "No Description",
"link": link["href"] if link else "#"
})
for el in soup.select(".t-tertiary.svelte-1yt5tdo"):
attribution = el.select_one(".attribution")
citation_link = el.select_one("cite a")
results["additionalData"].append({
"attribution": attribution.text.strip() if attribution else "No Attribution",
"citationLink": citation_link["href"] if citation_link else "#"
})
for el in soup.select(".fq-item"):
question = el.select_one(".faq-q")
answer = el.select_one(".faq-a")
faq_link = el.select_one("a")
results["faq"].append({
"question": question.text.strip() if question else "No Question",
"answer": answer.text.strip() if answer else "No Answer",
"faqLink": faq_link["href"] if faq_link else "#"
})
return results
except httpx.HTTPStatusError as e:
raise HTTPException(status_code=e.response.status_code, detail=f"HTTP error occurred: {e}")
except Exception as e:
raise HTTPException(status_code=500, detail=f"An error occurred: {e}")
@app.post("/search")
async def search(search_query: SearchQuery):
if not search_query.query:
raise HTTPException(status_code=400, detail="Query is required")
search = BraveSearch(search_query.query)
data = await search.fetch_results()
return data
if __name__ == "__main__":
uvicorn.run(app, host="0.0.0.0", port=8000) |