import logging import httpx from axon.config import settings from axon.models import SearchResult logger = logging.getLogger(__name__) _FRESHNESS = {"day": "pd", "week": "pw", "month": "pm", "year": "py"} async def search_brave( query: str, count: int = 10, topic: str = "general", time_range: str | None = None, ) -> list[SearchResult]: if not settings.brave_api_key: return [] endpoint = ( "https://api.search.brave.com/res/v1/news/search" if topic == "news" else "https://api.search.brave.com/res/v1/web/search" ) params: dict = {"q": query, "count": min(count, 20)} if time_range and time_range in _FRESHNESS: params["freshness"] = _FRESHNESS[time_range] headers = { "Accept": "application/json", "Accept-Encoding": "gzip", "X-Subscription-Token": settings.brave_api_key, } try: async with httpx.AsyncClient() as client: response = await client.get( endpoint, params=params, headers=headers, timeout=10.0 ) response.raise_for_status() data = response.json() raw = ( data.get("results", []) # news endpoint if topic == "news" else data.get("web", {}).get("results", []) ) results = [] for item in raw: results.append( SearchResult( title=item.get("title", ""), url=item.get("url", ""), content=item.get("description") or (item.get("extra_snippets") or [""])[0], score=0.5, published_date=item.get("page_age") or item.get("age"), favicon=item.get("meta_url", {}).get("favicon"), ) ) return results except Exception as exc: logger.error("Brave search error: %s", exc) return []