Import source: axon/services/sources/brave

Import the existing local project snapshot. This commit organizes existing files; it does not represent newly developed functionality.

axon/services/sources/brave.py

@@ -0,0 +1,65 @@
+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 []