Coverage for open_webui/retrieval/web/serpstack.py: 32%

1from __future__ import annotations 

2 

3import logging 

4 

5from open_webui.retrieval.web.main import SearchResult, get_filtered_results 

6from open_webui.utils.session_pool import get_session 

7 

8log = logging.getLogger(__name__) 

9 

10 

11async def search_serpstack( 

12 api_key: str, 

13 query: str, 

14 count: int, 

15 filter_list: list[str] | None = None, 

16 https_enabled: bool = True, 

17) -> list[SearchResult]: 

18 """Query the serpstack.com API and return normalised results. 

19 

20 Uses HTTPS by default; set ``https_enabled=False`` for free-tier HTTP access. 

21 """ 

22 scheme = 'https' if https_enabled else 'http' 

23 url = f'{scheme}://api.serpstack.com/search' 

24 params = {'access_key': api_key, 'query': query} 

25 

26 session = await get_session() 

27 async with session.get(url, params=params) as response: 

28 response.raise_for_status() 

29 payload = await response.json() 

30 

31 organic = sorted(payload.get('organic_results', []), key=lambda x: x.get('position', 0)) 

32 if filter_list: 

33 organic = get_filtered_results(organic, filter_list) 

34 

35 return [ 

36 SearchResult( 

37 link=item.get('url', ''), 

38 title=item.get('title'), 

39 snippet=item.get('snippet'), 

40 ) 

41 for item in organic[:count] 

42 ]