diff --git a/axon/services/sources/serper.py b/axon/services/sources/serper.py new file mode 100644 index 0000000..980a08e --- /dev/null +++ b/axon/services/sources/serper.py @@ -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 []