Import source: axon/services/sources/serper

Import the existing local project snapshot. This commit organizes existing files; it does not represent newly developed functionality.
2026-09-18 23:06:38 +03:30
parent e2fbe90654
commit fa54cc0b7f
1 changed files with 55 additions and 0 deletions
@@ -0,0 +1,55 @@
+import logging
+
+import httpx
+
+from axon.config import settings
+from axon.models import SearchResult
+
+logger = logging.getLogger(__name__)
+
+_TBS = {"day": "qdr:d", "week": "qdr:w", "month": "qdr:m", "year": "qdr:y"}
+
+
+async def search_serper(
+    query: str,
+    count: int = 10,
+    topic: str = "general",
+    time_range: str | None = None,
+) -> list[SearchResult]:
+    if not settings.serper_api_key:
+        return []
+
+    endpoint = (
+        "https://google.serper.dev/news"
+        if topic == "news"
+        else "https://google.serper.dev/search"
+    )
+    payload: dict = {"q": query, "num": min(count, 10)}
+    if time_range and time_range in _TBS:
+        payload["tbs"] = _TBS[time_range]
+
+    headers = {"X-API-KEY": settings.serper_api_key, "Content-Type": "application/json"}
+
+    try:
+        async with httpx.AsyncClient() as client:
+            response = await client.post(endpoint, json=payload, headers=headers, timeout=10.0)
+            response.raise_for_status()
+            data = response.json()
+
+        raw = data.get("organic") or data.get("news") or []
+        results = []
+        for item in raw:
+            results.append(
+                SearchResult(
+                    title=item.get("title", ""),
+                    url=item.get("link", ""),
+                    content=item.get("snippet", ""),
+                    score=0.5,
+                    published_date=item.get("date"),
+                    favicon=None,
+                )
+            )
+        return results
+    except Exception as exc:
+        logger.error("Serper search error: %s", exc)
+        return []