Coverage for open_webui/retrieval/web/serphouse.py: 27%

1from __future__ import annotations 

2 

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

4from open_webui.utils.session_pool import get_session 

5 

6 

7async def search_serphouse( 

8 api_key: str, 

9 domain: str, 

10 query: str, 

11 count: int, 

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

13) -> list[SearchResult]: 

14 """Query SERPHouse and return normalised organic results.""" 

15 session = await get_session() 

16 async with session.get( 

17 'https://api.serphouse.com/serp/live', 

18 params={ 

19 'q': query, 

20 'domain': (domain or 'google.com').strip() or 'google.com', 

21 'device': 'desktop', 

22 'serp_type': 'web', 

23 'page': 1, 

24 'num_result': count, 

25 }, 

26 headers={'Authorization': f'Bearer {api_key}', 'Accept': 'application/json'}, 

27 ) as response: 

28 response.raise_for_status() 

29 payload = await response.json() 

30 

31 organic = payload.get('results', {}).get('results', {}).get('organic', []) 

32 organic = sorted(organic, key=lambda item: item.get('position', 0)) 

33 if filter_list: 

34 organic = get_filtered_results(organic, filter_list) 

35 

36 return [ 

37 SearchResult( 

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

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

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

41 ) 

42 for item in organic[:count] 

43 if item.get('link') 

44 ]