Spaces:
Running
Running
| from __future__ import annotations | |
| import asyncio | |
| import logging | |
| from typing import Any, Dict, List, Optional | |
| import aiohttp | |
| from app.config import get_settings | |
| _logger = logging.getLogger(__name__) | |
| _settings = get_settings() | |
| def safe_text(text: Optional[str], maxlen: int = 0) -> str: | |
| if not text: | |
| return "" | |
| text = text.encode("utf-8", errors="replace").decode("utf-8", errors="replace") | |
| text = text.replace("\ufffd", "?").replace("\uFFFD", "?") | |
| text = "".join(c if ord(c) < 0x10000 else "?" for c in text) | |
| if maxlen and len(text) > maxlen: | |
| text = text[:maxlen] | |
| return text | |
| class WebSearchService: | |
| def __init__(self) -> None: | |
| self._base_url: str = _settings.searxng_base_url.rstrip("/") | |
| self._timeout: int = _settings.searxng_timeout | |
| self._max_retries: int = _settings.searxng_max_retries | |
| self._max_results: int = _settings.searxng_max_results | |
| async def search( | |
| self, | |
| query: str, | |
| categories: str = "general", | |
| language: str = "en", | |
| pageno: int = 1, | |
| time_range: Optional[str] = None, | |
| safesearch: int = 0, | |
| engines: Optional[str] = None, | |
| max_results: int = 10, | |
| ) -> Dict[str, Any]: | |
| params: Dict[str, Any] = { | |
| "q": query, | |
| "format": "json", | |
| "categories": categories, | |
| "language": language, | |
| "pageno": pageno, | |
| "safesearch": safesearch, | |
| } | |
| if time_range: | |
| params["time_range"] = time_range | |
| if engines: | |
| params["engines"] = engines | |
| url = f"{self._base_url}/search" | |
| last_error: Optional[str] = None | |
| for attempt in range(1 + self._max_retries): | |
| try: | |
| timeout = aiohttp.ClientTimeout(total=self._timeout) | |
| async with aiohttp.ClientSession(timeout=timeout) as session: | |
| async with session.get(url, params=params) as resp: | |
| if resp.status == 403: | |
| last_error = ( | |
| "SearXNG instance returned 403 Forbidden. " | |
| "JSON format must be enabled in settings.yml " | |
| "(search.formats: [html, json])." | |
| ) | |
| continue | |
| if resp.status != 200: | |
| text = await resp.text() | |
| last_error = f"SearXNG HTTP {resp.status}: {text[:200]}" | |
| continue | |
| data = await resp.json() | |
| return self._normalize_response(query, categories, data, max_results) | |
| except asyncio.TimeoutError: | |
| last_error = f"SearXNG request timed out after {self._timeout}s" | |
| except aiohttp.ClientConnectorError: | |
| last_error = ( | |
| f"Cannot connect to SearXNG at {self._base_url}. " | |
| "Ensure the SearXNG instance is running." | |
| ) | |
| except Exception as exc: | |
| last_error = f"Search request failed: {exc}" | |
| if attempt < self._max_retries: | |
| await asyncio.sleep(1.0 * (attempt + 1)) | |
| return { | |
| "success": False, | |
| "query": query, | |
| "number_of_results": 0, | |
| "results": [], | |
| "suggestions": [], | |
| "infoboxes": [], | |
| "error": last_error or "Unknown error", | |
| } | |
| def _normalize_response( | |
| self, | |
| query: str, | |
| categories: str, | |
| data: Dict[str, Any], | |
| max_results: int, | |
| ) -> Dict[str, Any]: | |
| raw_results = data.get("results", []) | |
| results: List[Dict[str, Any]] = [] | |
| for r in raw_results[:max_results]: | |
| results.append({ | |
| "title": safe_text(r.get("title", ""), 500), | |
| "url": safe_text(r.get("url", "")), | |
| "content": safe_text(r.get("content", ""), 1000), | |
| "engine": safe_text(r.get("engine", "")), | |
| "category": safe_text(r.get("category", "")), | |
| "published_date": r.get("publishedDate"), | |
| }) | |
| suggestions = [safe_text(s, 200) for s in data.get("suggestions", [])] | |
| infoboxes_raw = data.get("infoboxes", []) | |
| infoboxes: List[Dict[str, Any]] = [] | |
| for ib in infoboxes_raw: | |
| infoboxes.append({ | |
| "title": safe_text(ib.get("title", "")), | |
| "content": safe_text(ib.get("content", "")), | |
| "infobox": safe_text(ib.get("infobox", "")), | |
| "engine": safe_text(ib.get("engine", "")), | |
| "urls": ib.get("urls", []), | |
| }) | |
| return { | |
| "success": True, | |
| "query": query, | |
| "number_of_results": len(results), | |
| "results": results, | |
| "suggestions": suggestions, | |
| "infoboxes": infoboxes, | |
| "error": None, | |
| } | |
| async def get_config(self) -> Dict[str, Any]: | |
| url = f"{self._base_url}/config" | |
| try: | |
| timeout = aiohttp.ClientTimeout(total=self._timeout) | |
| async with aiohttp.ClientSession(timeout=timeout) as session: | |
| async with session.get(url) as resp: | |
| if resp.status != 200: | |
| return { | |
| "success": False, | |
| "error": f"HTTP {resp.status} from SearXNG config", | |
| } | |
| data = await resp.json() | |
| engines = data.get("engines", []) | |
| categories = list(data.get("categories_as_tabs", {}).keys()) | |
| plugins = data.get("plugins", []) | |
| return { | |
| "success": True, | |
| "instance_name": data.get("instance_name"), | |
| "version": data.get("version"), | |
| "engines": [ | |
| {"name": e.get("name"), "engine": e.get("engine"), | |
| "categories": e.get("categories", []), | |
| "shortcut": e.get("shortcut")} | |
| for e in engines | |
| ], | |
| "categories": categories, | |
| "plugins": plugins, | |
| "error": None, | |
| } | |
| except asyncio.TimeoutError: | |
| return {"success": False, "error": f"Config request timed out after {self._timeout}s"} | |
| except aiohttp.ClientConnectorError: | |
| return {"success": False, "error": f"Cannot connect to SearXNG at {self._base_url}"} | |
| except Exception as exc: | |
| return {"success": False, "error": f"Config request failed: {exc}"} | |
| async def autocomplete(self, query: str) -> Dict[str, Any]: | |
| url = f"{self._base_url}/autocompleter" | |
| params = {"q": query} | |
| try: | |
| timeout = aiohttp.ClientTimeout(total=10) | |
| async with aiohttp.ClientSession(timeout=timeout) as session: | |
| async with session.get(url, params=params) as resp: | |
| if resp.status != 200: | |
| return {"success": False, "error": f"HTTP {resp.status} from SearXNG autocompleter"} | |
| data = await resp.json() | |
| if isinstance(data, list) and len(data) >= 2: | |
| suggestions = [safe_text(s, 200) for s in data[1]] | |
| else: | |
| suggestions = [] | |
| return {"success": True, "query": query, "suggestions": suggestions, "error": None} | |
| except asyncio.TimeoutError: | |
| return {"success": False, "error": "Autocomplete request timed out"} | |
| except aiohttp.ClientConnectorError: | |
| return {"success": False, "error": f"Cannot connect to SearXNG at {self._base_url}"} | |
| except Exception as exc: | |
| return {"success": False, "error": f"Autocomplete request failed: {exc}"} | |
| async def get_engine_descriptions(self) -> Dict[str, Any]: | |
| url = f"{self._base_url}/engine_descriptions.json" | |
| try: | |
| timeout = aiohttp.ClientTimeout(total=self._timeout) | |
| async with aiohttp.ClientSession(timeout=timeout) as session: | |
| async with session.get(url) as resp: | |
| if resp.status != 200: | |
| return {"success": False, "error": f"HTTP {resp.status}"} | |
| data = await resp.json() | |
| return {"success": True, "engines": data, "error": None} | |
| except Exception as exc: | |
| return {"success": False, "error": f"Engine descriptions request failed: {exc}"} | |
| async def get_stats(self) -> Dict[str, Any]: | |
| url = f"{self._base_url}/stats" | |
| try: | |
| timeout = aiohttp.ClientTimeout(total=self._timeout) | |
| async with aiohttp.ClientSession(timeout=timeout) as session: | |
| async with session.get(url) as resp: | |
| if resp.status != 200: | |
| return {"success": False, "error": f"HTTP {resp.status}"} | |
| data = await resp.json() | |
| return {"success": True, "stats": data, "error": None} | |
| except Exception as exc: | |
| return {"success": False, "error": f"Stats request failed: {exc}"} | |
| async def health(self) -> Dict[str, Any]: | |
| url = f"{self._base_url}/healthz" | |
| try: | |
| timeout = aiohttp.ClientTimeout(total=10) | |
| async with aiohttp.ClientSession(timeout=timeout) as session: | |
| async with session.get(url) as resp: | |
| return { | |
| "success": resp.status == 200, | |
| "status": "healthy" if resp.status == 200 else f"unhealthy (HTTP {resp.status})", | |
| "error": None, | |
| } | |
| except asyncio.TimeoutError: | |
| return {"success": False, "status": "unhealthy", "error": "Connection timed out"} | |
| except aiohttp.ClientConnectorError: | |
| return {"success": False, "status": "unhealthy", "error": f"Cannot connect to SearXNG at {self._base_url}"} | |
| except Exception as exc: | |
| return {"success": False, "status": "unhealthy", "error": str(exc)} | |