Coverage for open_webui/retrieval/web/openserp.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_openserp( 

12 base_url: str, 

13 query: str, 

14 count: int, 

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

16) -> list[SearchResult]: 

17 """Query an OpenSERP instance and return normalised results. 

18 

19 OpenSERP aggregates results from 6 engines (google, bing, yandex, 

20 baidu, duckduckgo, ecosia) at once via ``/mega/search``. 

21 

22 No API key is required -- only a reachable OpenSERP base URL. 

23 """ 

24 url = f'{base_url.rstrip("/")}/mega/search' 

25 params = {'text': query, 'limit': count} 

26 

27 log.debug('searching OpenSERP at %s', url) 

28 

29 session = await get_session() 

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

31 response.raise_for_status() 

32 payload = await response.json() 

33 

34 results = payload.get('results', []) 

35 if filter_list: 

36 results = get_filtered_results(results, filter_list) 

37 

38 return [ 

39 SearchResult( 

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

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

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

43 ) 

44 for item in results[:count] 

45 ]