|
|
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) |
