Coverage for src/local_deep_research/web_search_engines/engines/search_engine_ddg.py: 100%

1from typing import Any, Dict, List, Optional 

2 

3from langchain_community.utilities import DuckDuckGoSearchAPIWrapper 

4from langchain_core.language_models import BaseLLM 

5 

6from ...security.secure_logging import logger 

7from ..rate_limiting import RateLimitError 

8from ..search_engine_base import BaseSearchEngine, Exposure, Sensitivity 

9 

10 

11class DuckDuckGoSearchEngine(BaseSearchEngine): 

12 """DuckDuckGo search engine implementation with two-phase retrieval""" 

13 

14 # Mark as public search engine 

15 is_public = True 

16 egress_sensitivity = Sensitivity.NON_SENSITIVE 

17 egress_exposure = Exposure.EXPOSING 

18 # Mark as generic search engine (general web search) 

19 is_generic = True 

20 

21 def __init__( 

22 self, 

23 max_results: int = 10, 

24 region: str = "us", 

25 safe_search: bool = True, 

26 llm: Optional[BaseLLM] = None, 

27 language: str = "English", 

28 include_full_content: bool = False, 

29 max_filtered_results: Optional[int] = None, 

30 settings_snapshot: Optional[Dict[str, Any]] = None, 

31 ): 

32 """ 

33 Initialize the DuckDuckGo search engine. 

34 

35 Args: 

36 max_results: Maximum number of search results 

37 region: Region code for search results 

38 safe_search: Whether to enable safe search 

39 llm: Language model for relevance filtering 

40 language: Language for content processing 

41 include_full_content: Whether to include full webpage content in results 

42 """ 

43 max_results = min(max_results, 50) 

44 # Initialize the BaseSearchEngine with LLM, max_filtered_results, and max_results 

45 super().__init__( 

46 llm=llm, 

47 max_filtered_results=max_filtered_results, 

48 max_results=max_results, 

49 include_full_content=include_full_content, 

50 settings_snapshot=settings_snapshot, 

51 ) 

52 self.region = region 

53 self.safe_search = safe_search 

54 self.language = language 

55 

56 # Initialize the DuckDuckGo wrapper 

57 self.engine = DuckDuckGoSearchAPIWrapper( 

58 region=region, 

59 max_results=max_results, 

60 safesearch="moderate" if safe_search else "off", 

61 ) 

62 

63 # Initialize FullSearchResults if full content is requested 

64 self._init_full_search( 

65 web_search=self.engine, 

66 language=language, 

67 max_results=max_results, 

68 region=region, 

69 time_period="y", 

70 safe_search="Moderate" if safe_search else "Off", 

71 ) 

72 

73 def _get_previews(self, query: str) -> List[Dict[str, Any]]: 

74 """ 

75 Get preview information (titles and snippets) for initial search results. 

76 

77 Args: 

78 query: The search query 

79 

80 Returns: 

81 List of preview dictionaries with 'id', 'title', and 'snippet' keys 

82 """ 

83 try: 

84 # Get search results from DuckDuckGo 

85 results = self.engine.results(query, max_results=self.max_results) 

86 

87 if not isinstance(results, list): 

88 return [] 

89 

90 # Process results to get previews 

91 previews = [] 

92 for i, result in enumerate(results): 

93 preview = { 

94 "id": result.get("link"), # Use URL as ID for DDG 

95 "title": result.get("title", ""), 

96 "snippet": result.get("snippet", ""), 

97 "link": result.get("link", ""), 

98 } 

99 

100 previews.append(preview) 

101 

102 return previews 

103 

104 except Exception as e: 

105 error_msg = str(e) 

106 sanitized = self._sanitize_error_message(error_msg) 

107 logger.exception( 

108 f"Error getting DuckDuckGo previews ({type(e).__name__}): {sanitized}" 

109 ) 

110 

111 # Check for known rate limit patterns 

112 if "202 Ratelimit" in error_msg or "ratelimit" in error_msg.lower(): 

113 raise RateLimitError(f"DuckDuckGo rate limit hit: {sanitized}") 

114 if "403" in error_msg or "forbidden" in error_msg.lower(): 

115 raise RateLimitError( 

116 f"DuckDuckGo access forbidden (possible rate limit): {sanitized}" 

117 ) 

118 if ( 

119 "timeout" in error_msg.lower() 

120 or "timed out" in error_msg.lower() 

121 ): 

122 # Timeouts can sometimes indicate rate limiting 

123 raise RateLimitError( 

124 f"DuckDuckGo timeout (possible rate limit): {sanitized}" 

125 ) 

126 

127 return [] 

128 

129 def _get_full_content( 

130 self, relevant_items: List[Dict[str, Any]] 

131 ) -> List[Dict[str, Any]]: 

132 """ 

133 Get full content for the relevant items by using FullSearchResults. 

134 

135 Args: 

136 relevant_items: List of relevant preview dictionaries 

137 

138 Returns: 

139 List of result dictionaries with full content 

140 """ 

141 # If we have FullSearchResults, use it to get full content 

142 if hasattr(self, "full_search"): 

143 return self.full_search._get_full_content(relevant_items) 

144 

145 # Otherwise, just return the relevant items without full content 

146 return relevant_items