searchapi.py 1.3 KB

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