Coverage for src/local_deep_research/web_search_engines/engines/search_engine_zenodo.py: 96%

1"""Zenodo search engine for open research data and publications.""" 

2 

3import html 

4import re 

5from typing import Any, Dict, List, Optional 

6 

7import requests 

8from langchain_core.language_models import BaseLLM 

9 

10from ...constants import USER_AGENT 

11from ...security.safe_requests import safe_get 

12from ...security.secure_logging import logger 

13from ..rate_limiting import RateLimitError 

14from ..search_engine_base import BaseSearchEngine, Exposure, Sensitivity 

15 

16 

17class ZenodoSearchEngine(BaseSearchEngine): 

18 """ 

19 Zenodo search engine for open research data and publications. 

20 

21 Provides access to millions of research outputs including datasets, 

22 software, publications, and more. No authentication required for search. 

23 """ 

24 

25 is_public = True 

26 egress_sensitivity = Sensitivity.NON_SENSITIVE 

27 egress_exposure = Exposure.EXPOSING 

28 is_generic = False 

29 is_scientific = True 

30 is_code = False 

31 is_lexical = True 

32 needs_llm_relevance_filter = True 

33 

34 def __init__( 

35 self, 

36 max_results: int = 10, 

37 resource_type: Optional[str] = None, 

38 access_right: Optional[str] = None, 

39 communities: Optional[str] = None, 

40 sort: str = "bestmatch", 

41 llm: Optional[BaseLLM] = None, 

42 max_filtered_results: Optional[int] = None, 

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

44 **kwargs, 

45 ): 

46 """ 

47 Initialize the Zenodo search engine. 

48 

49 Args: 

50 max_results: Maximum number of search results 

51 resource_type: Filter by type (dataset, software, publication, etc.) 

52 access_right: Filter by access (open, closed, embargoed, restricted) 

53 communities: Filter by Zenodo community 

54 sort: Sort order (bestmatch, mostrecent, -mostrecent) 

55 llm: Language model for relevance filtering 

56 max_filtered_results: Maximum results after filtering 

57 settings_snapshot: Settings snapshot for thread context 

58 """ 

59 super().__init__( 

60 llm=llm, 

61 max_filtered_results=max_filtered_results, 

62 max_results=max_results, 

63 settings_snapshot=settings_snapshot, 

64 **kwargs, 

65 ) 

66 

67 self.resource_type = resource_type 

68 self.access_right = access_right 

69 self.communities = communities 

70 self.sort = sort 

71 

72 self.base_url = "https://zenodo.org" 

73 self.search_url = f"{self.base_url}/api/records" 

74 

75 # User-Agent header for API requests 

76 self.headers = {"User-Agent": USER_AGENT} 

77 

78 def _build_query_params(self, query: str) -> Dict[str, Any]: 

79 """Build query parameters for the API request.""" 

80 params = { 

81 "q": query, 

82 "size": min(self.max_results, 100), 

83 "sort": self.sort, 

84 } 

85 

86 if self.resource_type: 

87 params["type"] = self.resource_type 

88 

89 if self.access_right: 

90 params["access_right"] = self.access_right 

91 

92 if self.communities: 

93 params["communities"] = self.communities 

94 

95 return params 

96 

97 def _parse_creators(self, creators: List[Dict]) -> List[str]: 

98 """Parse creator/author information.""" 

99 result = [] 

100 for creator in creators[:5]: 

101 name = creator.get("name", "") 

102 if name: 102 ↛ 100line 102 didn't jump to line 100 because the condition on line 102 was always true

103 result.append(name) 

104 return result 

105 

106 def _get_resource_type_label(self, resource_type: Dict) -> str: 

107 """Get human-readable resource type label.""" 

108 if not resource_type: 

109 return "Unknown" 

110 return ( 

111 resource_type.get("title") or resource_type.get("type") or "Unknown" 

112 ) 

113 

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

115 """ 

116 Get preview information for Zenodo records. 

117 

118 Args: 

119 query: The search query 

120 

121 Returns: 

122 List of preview dictionaries 

123 """ 

124 logger.info(f"Getting Zenodo previews for query: {query}") 

125 

126 # Apply rate limiting 

127 self._last_wait_time = self.rate_tracker.apply_rate_limit( 

128 self.engine_type 

129 ) 

130 

131 try: 

132 params = self._build_query_params(query) 

133 response = safe_get( 

134 self.search_url, 

135 params=params, 

136 headers=self.headers, 

137 timeout=30, 

138 ) 

139 

140 self._raise_if_rate_limit(response.status_code) 

141 

142 response.raise_for_status() 

143 data = response.json() 

144 

145 hits = data.get("hits", {}) 

146 results = hits.get("hits", []) 

147 total = hits.get("total", 0) 

148 logger.info( 

149 f"Found {total} Zenodo results, returning {len(results)}" 

150 ) 

151 

152 previews = [] 

153 for record in results[: self.max_results]: 

154 try: 

155 record_id = record.get("id") 

156 metadata = record.get("metadata", {}) 

157 

158 title = metadata.get("title", "Untitled") 

159 

160 # Get creators 

161 creators = self._parse_creators( 

162 metadata.get("creators", []) 

163 ) 

164 

165 # Get description/abstract 

166 description = metadata.get("description", "") 

167 # Strip HTML tags and decode entities for snippet 

168 if description: 

169 description = html.unescape( 

170 re.sub(r"<[^>]+>", "", description) 

171 ) 

172 description = description[:500] 

173 

174 # Get DOI 

175 doi = metadata.get("doi", "") 

176 

177 # Get publication date 

178 pub_date = metadata.get("publication_date", "") 

179 

180 # Get resource type 

181 resource_type = metadata.get("resource_type", {}) 

182 type_label = self._get_resource_type_label(resource_type) 

183 

184 # Get access right 

185 access = metadata.get("access_right", "open") 

186 

187 # Get keywords 

188 keywords = metadata.get("keywords", [])[:10] 

189 

190 # Get license 

191 license_info = metadata.get("license", {}) 

192 license_id = ( 

193 license_info.get("id", "") if license_info else "" 

194 ) 

195 

196 # Get links 

197 links = record.get("links", {}) 

198 record_url = links.get( 

199 "self_html", f"{self.base_url}/records/{record_id}" 

200 ) 

201 doi_url = links.get("doi", "") 

202 

203 # Build snippet 

204 snippet_parts = [] 

205 if creators: 

206 snippet_parts.append(f"By {', '.join(creators[:2])}") 

207 if type_label: 207 ↛ 220line 207 didn't jump to line 220 because the condition on line 207 was always true

208 type_str = f"Type: {type_label}" 

209 # Add access status and license inline 

210 access_license = [] 

211 if access: 211 ↛ 215line 211 didn't jump to line 215 because the condition on line 211 was always true

212 access_license.append( 

213 access.replace("_", " ").title() 

214 ) 

215 if license_id: 

216 access_license.append(license_id.upper()) 

217 if access_license: 217 ↛ 219line 217 didn't jump to line 219 because the condition on line 217 was always true

218 type_str += f" ({', '.join(access_license)})" 

219 snippet_parts.append(type_str) 

220 if pub_date: 

221 snippet_parts.append(f"Published: {pub_date}") 

222 if description: 

223 snippet_parts.append(description[:200]) 

224 snippet = ". ".join(snippet_parts) 

225 

226 preview = { 

227 "id": str(record_id), 

228 "title": title, 

229 "link": record_url, 

230 "snippet": snippet, 

231 "authors": creators, 

232 "doi": doi, 

233 "doi_url": doi_url, 

234 "publication_date": pub_date, 

235 "resource_type": type_label, 

236 "access_right": access, 

237 "keywords": keywords, 

238 "license": license_id, 

239 "description": description, 

240 "source": "Zenodo", 

241 "_raw": record, 

242 } 

243 

244 previews.append(preview) 

245 

246 except Exception as e: 

247 safe_msg = self._scrub_error(e) 

248 logger.exception( 

249 f"Error parsing Zenodo record ({type(e).__name__}): {safe_msg}" 

250 ) 

251 continue 

252 

253 return previews 

254 

255 except (requests.RequestException, ValueError) as e: 

256 safe_msg = self._scrub_error(e) 

257 logger.exception( 

258 f"Zenodo API request failed ({type(e).__name__}): {safe_msg}" 

259 ) 

260 self._raise_if_rate_limit(e) 

261 return [] 

262 

263 def _get_full_content( 

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

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

266 """ 

267 Get full content for the relevant Zenodo records. 

268 

269 Args: 

270 relevant_items: List of relevant preview dictionaries 

271 

272 Returns: 

273 List of result dictionaries with full content 

274 """ 

275 logger.info( 

276 f"Getting full content for {len(relevant_items)} Zenodo records" 

277 ) 

278 

279 results = [] 

280 for item in relevant_items: 

281 result = item.copy() 

282 

283 raw = item.get("_raw", {}) 

284 if raw: 

285 metadata = raw.get("metadata", {}) 

286 

287 # Get full description (strip HTML tags and decode entities) 

288 desc = metadata.get("description", "") 

289 if desc: 

290 desc = html.unescape(re.sub(r"<[^>]+>", "", desc)) 

291 result["description"] = desc 

292 

293 # Get all keywords 

294 result["keywords"] = metadata.get("keywords", []) 

295 

296 # Get related identifiers 

297 result["related_identifiers"] = metadata.get( 

298 "related_identifiers", [] 

299 ) 

300 

301 # Get files info 

302 files = raw.get("files") or [] 

303 result["files"] = [ 

304 { 

305 "filename": f.get("key", ""), 

306 "size": f.get("size", 0), 

307 "checksum": f.get("checksum", ""), 

308 } 

309 for f in files[:10] 

310 ] 

311 

312 # Get references 

313 result["references"] = metadata.get("references", []) 

314 

315 # Build content summary 

316 content_parts = [] 

317 if result.get("authors"): 

318 content_parts.append( 

319 f"Authors: {', '.join(result['authors'])}" 

320 ) 

321 if result.get("resource_type"): 

322 content_parts.append(f"Type: {result['resource_type']}") 

323 if result.get("publication_date"): 

324 content_parts.append( 

325 f"Published: {result['publication_date']}" 

326 ) 

327 if result.get("doi"): 

328 content_parts.append(f"DOI: {result['doi']}") 

329 if result.get("keywords"): 

330 content_parts.append( 

331 f"Keywords: {', '.join(str(k) for k in result['keywords'][:5])}" 

332 ) 

333 if result.get("license"): 

334 content_parts.append(f"License: {result['license']}") 

335 if result.get("description"): 

336 content_parts.append( 

337 f"\nDescription: {result['description'][:1000]}" 

338 ) 

339 

340 result["content"] = "\n".join(content_parts) 

341 

342 # Clean up internal fields 

343 if "_raw" in result: 

344 del result["_raw"] 

345 

346 results.append(result) 

347 

348 return results 

349 

350 def get_record(self, record_id: int) -> Optional[Dict[str, Any]]: 

351 """ 

352 Get a specific record by Zenodo ID. 

353 

354 Args: 

355 record_id: The Zenodo record ID 

356 

357 Returns: 

358 Record dictionary or None 

359 """ 

360 try: 

361 url = f"{self.search_url}/{record_id}" 

362 response = safe_get(url, headers=self.headers, timeout=30) 

363 self._raise_if_rate_limit(response.status_code) 

364 response.raise_for_status() 

365 return response.json() # type: ignore[no-any-return] 

366 except RateLimitError: 

367 raise 

368 except Exception as e: 

369 safe_msg = self._scrub_error(e) 

370 logger.exception( 

371 f"Error fetching Zenodo record {record_id} ({type(e).__name__}): {safe_msg}" 

372 ) 

373 return None 

374 

375 def search_datasets(self, query: str) -> List[Dict[str, Any]]: 

376 """ 

377 Search specifically for datasets. 

378 

379 Args: 

380 query: The search query 

381 

382 Returns: 

383 List of matching datasets 

384 """ 

385 original_type = self.resource_type 

386 try: 

387 self.resource_type = "dataset" 

388 return self.run(query) 

389 finally: 

390 self.resource_type = original_type 

391 

392 def search_software(self, query: str) -> List[Dict[str, Any]]: 

393 """ 

394 Search specifically for software. 

395 

396 Args: 

397 query: The search query 

398 

399 Returns: 

400 List of matching software records 

401 """ 

402 original_type = self.resource_type 

403 try: 

404 self.resource_type = "software" 

405 return self.run(query) 

406 finally: 

407 self.resource_type = original_type