searchapi.py 1.3 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849
  1. import logging
  2. from typing import Optional
  3. from urllib.parse import urlencode
  4. import requests
  5. from apps.rag.search.main import SearchResult, get_filtered_results
  6. from env import SRC_LOG_LEVELS
  7. log = logging.getLogger(__name__)
  8. log.setLevel(SRC_LOG_LEVELS["RAG"])
  9. def search_searchapi(
  10. api_key: str, engine: str, query: str, count: int, filter_list: Optional[list[str]] = None
  11. ) -> list[SearchResult]:
  12. """Search using searchapi.io's API and return the results as a list of SearchResult objects.
  13. Args:
  14. api_key (str): A searchapi.io API key
  15. query (str): The query to search for
  16. """
  17. url = "https://www.searchapi.io/api/v1/search"
  18. engine = engine or "google"
  19. payload = {
  20. "engine": engine,
  21. "q": query,
  22. "api_key": api_key
  23. }
  24. url = f"{url}?{urlencode(payload)}"
  25. response = requests.request("GET", url)
  26. json_response = response.json()
  27. log.info(f"results from searchapi search: {json_response}")
  28. results = sorted(
  29. json_response.get("organic_results", []), key=lambda x: x.get("position", 0)
  30. )
  31. if filter_list:
  32. results = get_filtered_results(results, filter_list)
  33. return [
  34. SearchResult(
  35. link=result["link"],
  36. title=result["title"],
  37. snippet=result["snippet"]
  38. )
  39. for result in results[:count]
  40. ]