Coverage for src/local_deep_research/web_search_engines/engines/search_engine_mojeek.py: 95%

1from typing import Any, Dict, List, Optional 

2 

3from langchain_core.language_models import BaseLLM 

4 

5from ...security.safe_requests import safe_get 

6from ...security.secure_logging import logger 

7from ..rate_limiting import RateLimitError 

8from ..search_engine_base import BaseSearchEngine, Exposure, Sensitivity 

9 

10 

11class MojeekSearchEngine(BaseSearchEngine): 

12 """ 

13 Mojeek search engine implementation. 

14 

15 Mojeek is a privacy-focused search engine with its own independent 

16 web crawler and index. Requires a paid API key from mojeek.com. 

17 """ 

18 

19 # Mark as public search engine 

20 is_public = True 

21 egress_sensitivity = Sensitivity.NON_SENSITIVE 

22 egress_exposure = Exposure.EXPOSING 

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

24 is_generic = True 

25 is_lexical = True 

26 needs_llm_relevance_filter = True 

27 

28 def _is_valid_search_result(self, url: str) -> bool: 

29 """ 

30 Check if a URL is a valid absolute HTTP(S) URL. 

31 

32 Returns False for relative URLs, empty strings, or non-HTTP schemes. 

33 """ 

34 if not url or not url.lower().startswith(("http://", "https://")): 

35 return False 

36 return True 

37 

38 def __init__( 

39 self, 

40 max_results: int = 10, 

41 language: str = "en", 

42 region: str = "", 

43 safe_search: bool = False, 

44 api_key: Optional[str] = None, 

45 llm: Optional[BaseLLM] = None, 

46 max_filtered_results: Optional[int] = None, 

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

48 include_full_content: bool = True, 

49 **kwargs, 

50 ): 

51 """ 

52 Initialize the Mojeek search engine. 

53 

54 Args: 

55 max_results: Maximum number of search results 

56 language: Language code in ISO 639-1 format (e.g. 'en', 'fr') 

57 region: Country code in ISO 3166-1 alpha-2 format (e.g. 'GB', 'FR') 

58 safe_search: Whether to enable safe search filtering 

59 api_key: Mojeek API key 

60 llm: Language model for relevance filtering 

61 max_filtered_results: Maximum number of results to keep after filtering 

62 settings_snapshot: Settings snapshot for thread context 

63 include_full_content: Whether to include full webpage content 

64 """ 

65 super().__init__( 

66 llm=llm, 

67 max_filtered_results=max_filtered_results, 

68 max_results=max_results, 

69 include_full_content=include_full_content, 

70 settings_snapshot=settings_snapshot, 

71 **kwargs, 

72 ) 

73 

74 # Get API key - check params, settings, or env vars 

75 mojeek_api_key = self._resolve_api_key( 

76 api_key, 

77 "search.engine.web.mojeek.api_key", 

78 engine_name="Mojeek", 

79 settings_snapshot=settings_snapshot, 

80 ) 

81 

82 self.search_url = "https://api.mojeek.com/search" 

83 self.max_results = max_results 

84 self.language = language 

85 self.region = region 

86 self.safe_search = safe_search 

87 self.api_key = mojeek_api_key 

88 

89 # If full content is requested, initialize FullSearchResults 

90 self._init_full_search( 

91 web_search=self, 

92 language=language, 

93 max_results=max_results, 

94 region=region, 

95 safe_search=safe_search, 

96 time_period="y", 

97 ) 

98 

99 def _get_search_results(self, query: str) -> List[Dict[str, Any]]: 

100 """ 

101 Get search results from the Mojeek API. 

102 

103 Args: 

104 query: The search query 

105 

106 Returns: 

107 List of search result dicts 

108 """ 

109 logger.info(f"Mojeek running search for query: {query}") 

110 

111 try: 

112 params = { 

113 "q": query, 

114 "api_key": self.api_key, 

115 "fmt": "json", 

116 "t": self.max_results, 

117 "safe": 1 if self.safe_search else 0, 

118 } 

119 

120 if self.language: 

121 params["lb"] = self.language 

122 params["lbb"] = 100 

123 

124 if self.region: 

125 params["rb"] = self.region 

126 params["rbb"] = 10 

127 

128 logger.info(f"Sending request to Mojeek API at {self.search_url}") 

129 

130 response = safe_get( 

131 self.search_url, 

132 params=params, 

133 timeout=15, 

134 ) 

135 

136 if response.status_code == 403: 

137 raise RateLimitError( # noqa: TRY301 — re-raised by except RateLimitError for base class retry 

138 "Mojeek API rate limit hit (403 Forbidden)" 

139 ) 

140 

141 if response.status_code != 200: 

142 logger.warning( 

143 f"Mojeek API returned status {response.status_code}" 

144 ) 

145 return [] 

146 

147 data = response.json() 

148 

149 response_data = data.get("response", {}) 

150 if response_data.get("status") != "OK": 

151 logger.warning( 

152 f"Mojeek API response status: " 

153 f"{response_data.get('status', 'missing')}" 

154 ) 

155 return [] 

156 

157 raw_results = response_data.get("results", []) 

158 results = [] 

159 for result in raw_results: 

160 url = self._clean_result_url(result.get("url")) 

161 if not self._is_valid_search_result(url): 161 ↛ 162line 161 didn't jump to line 162 because the condition on line 161 was never true

162 continue 

163 results.append( 

164 { 

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

166 "url": url, 

167 "content": result.get("desc", ""), 

168 "engine": "mojeek", 

169 "category": result.get("cats", ""), 

170 } 

171 ) 

172 

173 if results: 

174 logger.info(f"Mojeek returned {len(results)} valid results") 

175 else: 

176 logger.warning( 

177 f"Mojeek returned no valid results for query: {query}" 

178 ) 

179 

180 return results 

181 

182 except RateLimitError: 

183 raise 

184 except Exception as e: 

185 safe_msg = self._scrub_error(e) 

186 logger.warning( 

187 f"Error when searching using Mojeek ({type(e).__name__}): {safe_msg}" 

188 ) 

189 return [] 

190 

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

192 """ 

193 Get preview information for Mojeek search results. 

194 

195 Args: 

196 query: The search query 

197 

198 Returns: 

199 List of preview dictionaries 

200 """ 

201 logger.info(f"Getting Mojeek previews for query: {query}") 

202 

203 results = self._get_search_results(query) 

204 

205 if not results: 

206 logger.warning(f"No Mojeek results found for query: {query}") 

207 return [] 

208 

209 previews = [] 

210 for i, result in enumerate(results): 

211 preview = { 

212 "id": result.get("url", "") or f"mojeek-result-{i}", 

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

214 "link": result.get("url", ""), 

215 "snippet": result.get("content", ""), 

216 "engine": result.get("engine", "mojeek"), 

217 "category": result.get("category", ""), 

218 } 

219 previews.append(preview) 

220 

221 return previews 

222 

223 def _get_full_content( 

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

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

226 """ 

227 Get full content for the relevant search results. 

228 

229 Args: 

230 relevant_items: List of relevant preview dictionaries 

231 

232 Returns: 

233 List of result dictionaries with full content 

234 """ 

235 if self.include_full_content and hasattr(self, "full_search"): 

236 logger.info("Retrieving full webpage content") 

237 try: 

238 return self.full_search._get_full_content(relevant_items) 

239 except Exception as e: 

240 safe_msg = self._scrub_error(e) 

241 logger.exception( 

242 f"Error retrieving full content ({type(e).__name__}): {safe_msg}" 

243 ) 

244 

245 return relevant_items