Spaces:
Sleeping
Sleeping
| import os | |
| import httpx | |
| from dotenv import load_dotenv | |
| load_dotenv() | |
| class WebClient: | |
| """ | |
| Communicates with the Google Custom Search API. | |
| """ | |
| def __init__(self) -> None: | |
| self.search_endpoint = "https://www.googleapis.com/customsearch/v1" | |
| async def search(self, query: str, max_results: int = 5, region: str = "us"): | |
| """ | |
| Sends the query to Google Custom Search and returns search results. | |
| """ | |
| max_results_value = self._sanitize_max_results(max_results) | |
| api_key = os.getenv("GOOGLE_SEARCH_API_KEY") | |
| cx_id = os.getenv("GOOGLE_SEARCH_CX_ID") | |
| if not api_key or not cx_id: | |
| raise RuntimeError("Google Custom Search credentials not configured.") | |
| params = { | |
| "key": api_key, | |
| "cx": cx_id, | |
| "q": query, | |
| "num": max_results_value, | |
| "gl": self._sanitize_region(region), | |
| } | |
| try: | |
| async with httpx.AsyncClient(timeout=10) as client: | |
| response = await client.get(self.search_endpoint, params=params) | |
| response.raise_for_status() | |
| except Exception as exc: | |
| raise RuntimeError(f"Google Custom Search request failed: {exc}") from exc | |
| data = response.json() | |
| items = data.get("items", []) | |
| return [ | |
| { | |
| "title": item.get("title"), | |
| "link": item.get("link"), | |
| "snippet": item.get("snippet"), | |
| } | |
| for item in items | |
| ] | |
| def _sanitize_max_results(value: int) -> int: | |
| try: | |
| return max(1, min(int(value), 10)) | |
| except (TypeError, ValueError): | |
| raise RuntimeError("max_results must be an integer between 1 and 10.") | |
| def _sanitize_region(region: str) -> str: | |
| region_value = (region or "us").lower().split("-", 1)[0] | |
| if len(region_value) != 2: | |
| return "us" | |
| return region_value | |