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

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: Retained for backward compatibility; full-content 

64 retrieval is handled via the factory wrapper. 

65 """ 

66 super().__init__( 

67 llm=llm, 

68 max_filtered_results=max_filtered_results, 

69 max_results=max_results, 

70 include_full_content=include_full_content, 

71 settings_snapshot=settings_snapshot, 

72 **kwargs, 

73 ) 

74 

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

76 mojeek_api_key = self._resolve_api_key( 

77 api_key, 

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

79 engine_name="Mojeek", 

80 settings_snapshot=settings_snapshot, 

81 ) 

82 

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

84 self.max_results = max_results 

85 self.language = language 

86 self.region = region 

87 self.safe_search = safe_search 

88 self.api_key = mojeek_api_key 

89 

90 # Full-content retrieval is the factory wrapper's job — the 

91 # wrapper fetches pages via ``batch_fetch_and_extract`` and 

92 # populates ``full_content`` on each result; the inner engine 

93 # only emits snippets from ``_get_previews``. 

94 

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

96 """ 

97 Get search results from the Mojeek API. 

98 

99 Args: 

100 query: The search query 

101 

102 Returns: 

103 List of search result dicts 

104 """ 

105 logger.info("Mojeek running search") 

106 

107 try: 

108 params = { 

109 "q": query, 

110 "api_key": self.api_key, 

111 "fmt": "json", 

112 "t": self.max_results, 

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

114 } 

115 

116 if self.language: 

117 params["lb"] = self.language 

118 params["lbb"] = 100 

119 

120 if self.region: 

121 params["rb"] = self.region 

122 params["rbb"] = 10 

123 

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

125 

126 response = safe_get( 

127 self.search_url, 

128 params=params, 

129 timeout=15, 

130 ) 

131 

132 if response.status_code == 403: 

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

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

135 ) 

136 

137 if response.status_code != 200: 

138 logger.warning( 

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

140 ) 

141 return [] 

142 

143 data = response.json() 

144 

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

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

147 logger.warning( 

148 f"Mojeek API response status: " 

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

150 ) 

151 return [] 

152 

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

154 results = [] 

155 for result in raw_results: 

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

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

158 continue 

159 results.append( 

160 { 

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

162 "url": url, 

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

164 "engine": "mojeek", 

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

166 } 

167 ) 

168 

169 if results: 

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

171 else: 

172 logger.warning("Mojeek returned no valid results") 

173 

174 return results 

175 

176 except RateLimitError: 

177 raise 

178 except Exception as e: 

179 safe_msg = self._scrub_error(e) 

180 logger.warning( 

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

182 ) 

183 return [] 

184 

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

186 """ 

187 Get preview information for Mojeek search results. 

188 

189 Args: 

190 query: The search query 

191 

192 Returns: 

193 List of preview dictionaries 

194 """ 

195 logger.info("Getting Mojeek previews") 

196 

197 results = self._get_search_results(query) 

198 

199 if not results: 

200 logger.warning("No Mojeek results found") 

201 return [] 

202 

203 previews = [] 

204 for i, result in enumerate(results): 

205 preview = { 

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

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

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

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

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

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

212 } 

213 previews.append(preview) 

214 

215 return previews 

216 

217 def _get_full_content( 

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

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

220 """Return items unchanged. 

221 

222 Full-content retrieval is the factory wrapper's job. The wrapper 

223 fetches pages via ``batch_fetch_and_extract`` and populates 

224 ``full_content`` on each result; the inner engine never fetches 

225 pages itself. 

226 """ 

227 return relevant_items