Coverage for web_search / providers / duckduckgo.py: 48%

1import time 

2from typing import List 

3from ddgs import DDGS 

4from qrclaw.web_search.types import WebSearchProvider, SearchResult, WebSearchResponse 

5from qrclaw.logger import get_logger 

6 

7logger = get_logger("qrclaw.web_search.providers.duckduckgo") 

8 

9class DuckDuckGoSearchProvider(WebSearchProvider): 

10 @property 

11 def id(self) -> str: 

12 return "duckduckgo" 

13 

14 @property 

15 def name(self) -> str: 

16 return "DuckDuckGo Search" 

17 

18 def is_available(self) -> bool: 

19 # 始终可用，不需要 Key 

20 return True 

21 

22 def search(self, query: str, max_results: int = 5, **kwargs) -> WebSearchResponse: 

23 start_time = time.time() 

24 logger.debug(f"DuckDuckGo searching: {query}") 

25 

26 results: List[SearchResult] = [] 

27 try: 

28 with DDGS() as ddgs: 

29 # region="wt-wt" 表示全球搜索，timelimit="y" 表示最近一年(可选) 

30 ddg_results = ddgs.text(query, max_results=max_results, region="wt-wt") 

31 

32 for r in ddg_results: 

33 results.append(SearchResult( 

34 title=r.get("title", ""), 

35 url=r.get("href", ""), 

36 snippet=r.get("body", "") 

37 )) 

38 except Exception as e: 

39 logger.error(f"DuckDuckGo search failed: {e}") 

40 raise RuntimeError(f"DuckDuckGo API error: {e}") 

41 

42 took_ms = (time.time() - start_time) * 1000 

43 return WebSearchResponse( 

44 query=query, 

45 provider=self.id, 

46 count=len(results), 

47 took_ms=took_ms, 

48 results=results 

49 )