Coverage for open_webui/retrieval/web/yacy.py: 25%

1import logging 

2from typing import Optional 

3 

4import requests 

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

6from requests.auth import HTTPDigestAuth 

7 

8log = logging.getLogger(__name__) 

9 

10 

11def search_yacy( 

12 query_url: str, 

13 username: Optional[str], 

14 password: Optional[str], 

15 query: str, 

16 count: int, 

17 filter_list: Optional[list[str]] = None, 

18) -> list[SearchResult]: 

19 """ 

20 Search a Yacy instance for a given query and return the results as a list of SearchResult objects. 

21 

22 The function accepts username and password for authenticating to Yacy. 

23 

24 Args: 

25 query_url (str): The base URL of the Yacy server. 

26 username (str): Optional YaCy username. 

27 password (str): Optional YaCy password. 

28 query (str): The search term or question to find in the Yacy database. 

29 count (int): The maximum number of results to retrieve from the search. 

30 

31 Returns: 

32 list[SearchResult]: A list of SearchResults sorted by relevance score in descending order. 

33 

34 Raise: 

35 requests.exceptions.RequestException: If a request error occurs during the search process. 

36 """ 

37 

38 # Use authentication if either username or password is set 

39 yacy_auth = None 

40 if username or password: 

41 yacy_auth = HTTPDigestAuth(username, password) 

42 

43 params = { 

44 'query': query, 

45 'contentdom': 'text', 

46 'resource': 'global', 

47 'maximumRecords': count, 

48 'nav': 'none', 

49 } 

50 

51 # Check if provided a json API URL 

52 if not query_url.endswith('yacysearch.json'): 

53 # Strip all query parameters from the URL 

54 query_url = query_url.rstrip('/') + '/yacysearch.json' 

55 

56 log.debug('searching %s', query_url) 

57 

58 response = requests.get( 

59 query_url, 

60 auth=yacy_auth, 

61 headers={ 

62 # LICENSE covers this Open WebUI user-agent identifier. 

63 # Do not alter, remove, obscure, or replace it except as LICENSE permits: 

64 # https://docs.openwebui.com/license. 

65 'User-Agent': 'Open WebUI (https://github.com/open-webui/open-webui) RAG Bot', 

66 'Accept': 'text/html', 

67 'Accept-Encoding': 'gzip, deflate', 

68 'Accept-Language': 'en-US,en;q=0.5', 

69 'Connection': 'keep-alive', 

70 }, 

71 params=params, 

72 ) 

73 

74 response.raise_for_status() # Raise an exception for HTTP errors. 

75 

76 json_response = response.json() 

77 results = json_response.get('channels', [{}])[0].get('items', []) 

78 sorted_results = sorted(results, key=lambda x: x.get('ranking', 0), reverse=True) 

79 if filter_list: 

80 sorted_results = get_filtered_results(sorted_results, filter_list) 

81 return [ 

82 SearchResult( 

83 link=result['link'], 

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

85 snippet=result.get('description'), 

86 ) 

87 for result in sorted_results[:count] 

88 ]