Coverage for open_webui/retrieval/web/microsoft_web_iq.py: 31%

1from __future__ import annotations 

2 

3import logging 

4from urllib.parse import urlparse 

5 

6import requests 

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

8from open_webui.utils.headers import include_user_info_headers 

9 

10log = logging.getLogger(__name__) 

11 

12 

13def search_microsoft_web_iq( 

14 api_base_url: str, 

15 api_key: str, 

16 query: str, 

17 count: int, 

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

19 language: str = 'en', 

20 user=None, 

21) -> list[SearchResult]: 

22 try: 

23 api_base_url = api_base_url.rstrip('/') 

24 headers = { 

25 'host': urlparse(api_base_url).netloc or 'api.microsoft.ai', 

26 'x-apikey': api_key, 

27 'content-type': 'application/json', 

28 } 

29 if user is not None: 

30 headers = include_user_info_headers(headers, user) 

31 

32 response = requests.post( 

33 f'{api_base_url}/search/web', 

34 json={ 

35 'query': query, 

36 'maxResults': count, 

37 'language': language, 

38 'contentFormat': 'passage', 

39 }, 

40 headers=headers, 

41 ) 

42 response.raise_for_status() 

43 

44 results = response.json().get('webResults', []) 

45 if filter_list: 

46 results = get_filtered_results(results, filter_list) 

47 

48 return [ 

49 SearchResult( 

50 link=result['url'], 

51 title=result.get('title'), 

52 snippet=result.get('content'), 

53 ) 

54 for result in results 

55 ] 

56 except Exception as e: 

57 log.error(f'Error searching with Microsoft Web IQ API: {e}') 

58 return []