Coverage for open_webui/retrieval/web/duckduckgo.py: 28%

1from __future__ import annotations 

2 

3import logging 

4import urllib.request 

5 

6from open_webui.env import USE_SLIM 

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

8 

9log = logging.getLogger(__name__) 

10 

11 

12def search_duckduckgo( 

13 query: str, 

14 count: int, 

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

16 concurrent_requests: int | None = None, 

17 backend: str | None = 'auto', 

18) -> list[SearchResult]: 

19 """ 

20 Search using DuckDuckGo's Search API and return the results as a list of SearchResult objects. 

21 Args: 

22 query (str): The query to search for 

23 count (int): The number of results to return 

24 backend (str): The search backend to use (auto, duckduckgo, google, brave, etc.) 

25 

26 Returns: 

27 list[SearchResult]: A list of search results 

28 """ 

29 if USE_SLIM: 

30 raise ValueError( 

31 'DDGS is unavailable in slim. Configure another web search provider in Admin Settings > Web Search.' 

32 ) 

33 

34 from ddgs import DDGS 

35 

36 # The ddgs library (primp-based) does not auto-detect proxy env vars. 

37 # Resolve via stdlib getproxies() — same pattern as the other loaders. 

38 env_proxies = urllib.request.getproxies() 

39 proxy = env_proxies.get('https') or env_proxies.get('http') 

40 with DDGS(proxy=proxy) as ddgs: 

41 if concurrent_requests: 

42 ddgs.threads = concurrent_requests 

43 

44 search_results = ddgs.text(query, safesearch='moderate', max_results=count, backend=backend or 'auto') 

45 if filter_list: 

46 search_results = get_filtered_results(search_results, filter_list) 

47 

48 # Return the list of search results 

49 return [ 

50 SearchResult( 

51 link=result['href'], 

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

53 snippet=result.get('body'), 

54 ) 

55 for result in search_results 

56 ]