Coverage for src/local_deep_research/web_search_engines/engines/search_engine_nasa_ads.py: 89%

1"""NASA Astrophysics Data System (ADS) search engine implementation.""" 

2 

3from typing import Any, Dict, List, Optional 

4 

5from langchain_core.language_models import BaseLLM 

6from ...security.secure_logging import logger 

7 

8from ...constants import SNIPPET_LENGTH_LONG, USER_AGENT 

9from ...security.safe_requests import safe_get 

10from ..rate_limiting import RateLimitError 

11from ..search_engine_base import BaseSearchEngine, Exposure, Sensitivity 

12 

13 

14class NasaAdsSearchEngine(BaseSearchEngine): 

15 """NASA ADS search engine for physics, astronomy, and astrophysics papers.""" 

16 

17 # Mark as public search engine 

18 is_public = True 

19 egress_sensitivity = Sensitivity.NON_SENSITIVE 

20 egress_exposure = Exposure.EXPOSING 

21 # Scientific/astronomy/astrophysics search engine 

22 is_scientific = True 

23 is_lexical = True 

24 needs_llm_relevance_filter = True 

25 

26 def __init__( 

27 self, 

28 max_results: int = 25, 

29 api_key: Optional[str] = None, 

30 sort_by: str = "relevance", 

31 min_citations: int = 0, 

32 from_publication_date: Optional[str] = None, 

33 include_arxiv: bool = True, 

34 llm: Optional[BaseLLM] = None, 

35 max_filtered_results: Optional[int] = None, 

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

37 **kwargs, 

38 ): 

39 """ 

40 Initialize the NASA ADS search engine. 

41 

42 Args: 

43 max_results: Maximum number of search results 

44 api_key: NASA ADS API key (required for higher rate limits) 

45 sort_by: Sort order ('relevance', 'citation_count', 'date') 

46 min_citations: Minimum citation count filter 

47 from_publication_date: Filter papers from this date (YYYY-MM-DD) 

48 include_arxiv: Include ArXiv preprints in results 

49 llm: Language model for relevance filtering 

50 max_filtered_results: Maximum number of results to keep after filtering 

51 settings_snapshot: Settings snapshot for configuration 

52 **kwargs: Additional parameters to pass to parent class 

53 """ 

54 # Journal filter runs before LLM relevance (Tiers 1-3 are instant) 

55 preview_filters = [] 

56 journal_filter = self._create_journal_filter( 

57 "nasa_ads", llm, settings_snapshot 

58 ) 

59 if journal_filter is not None: 

60 preview_filters.append(journal_filter) 

61 

62 super().__init__( 

63 llm=llm, 

64 max_filtered_results=max_filtered_results, 

65 max_results=max_results, 

66 preview_filters=preview_filters, # type: ignore[arg-type] 

67 settings_snapshot=settings_snapshot, 

68 **kwargs, 

69 ) 

70 

71 self.sort_by = sort_by 

72 self.min_citations = min_citations 

73 self.include_arxiv = include_arxiv 

74 # Handle from_publication_date 

75 self.from_publication_date = ( 

76 from_publication_date 

77 if from_publication_date 

78 and from_publication_date not in ["False", "false", ""] 

79 else None 

80 ) 

81 

82 # Get API key from settings if not provided 

83 if not api_key and settings_snapshot: 83 ↛ 84line 83 didn't jump to line 84 because the condition on line 83 was never true

84 from ...config.search_config import get_setting_from_snapshot 

85 

86 try: 

87 api_key = get_setting_from_snapshot( 

88 "search.engine.web.nasa_ads.api_key", 

89 settings_snapshot=settings_snapshot, 

90 ) 

91 except Exception: 

92 logger.debug( 

93 "Failed to read nasa_ads.api_key from settings snapshot", 

94 exc_info=True, 

95 ) 

96 

97 # Handle "False" string for api_key 

98 self.api_key = ( 

99 api_key 

100 if api_key and api_key not in ["False", "false", ""] 

101 else None 

102 ) 

103 

104 # API configuration 

105 self.api_base = "https://api.adsabs.harvard.edu/v1" 

106 self.headers = { 

107 "User-Agent": USER_AGENT, 

108 "Accept": "application/json", 

109 } 

110 

111 if self.api_key: 

112 self.headers["Authorization"] = f"Bearer {self.api_key}" 

113 logger.info("Using NASA ADS with API key") 

114 else: 

115 logger.error( 

116 "NASA ADS requires an API key to function. Get a free key at: https://ui.adsabs.harvard.edu/user/settings/token" 

117 ) 

118 

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

120 """ 

121 Get preview information for NASA ADS search results. 

122 

123 Args: 

124 query: The search query (natural language supported) 

125 

126 Returns: 

127 List of preview dictionaries 

128 """ 

129 logger.info(f"Searching NASA ADS for: {query}") 

130 

131 # Build the search query - NASA ADS has good natural language support 

132 # We can use the query directly or enhance it slightly 

133 search_query = query 

134 

135 # Build filters 

136 filters = [] 

137 if self.from_publication_date: 

138 # Convert YYYY-MM-DD to ADS format 

139 try: 

140 year = self.from_publication_date.split("-")[0] 

141 if year.isdigit(): # Only add if it's a valid year 141 ↛ 149line 141 didn't jump to line 149 because the condition on line 141 was always true

142 filters.append(f"year:{year}-9999") 

143 except Exception: 

144 logger.debug( 

145 "best-effort date parsing, invalid formats skipped", 

146 exc_info=True, 

147 ) 

148 

149 if self.min_citations > 0: 

150 filters.append(f"citation_count:[{self.min_citations} TO *]") 

151 

152 if not self.include_arxiv: 

153 filters.append('-bibstem:"arXiv"') 

154 

155 # Combine query with filters 

156 if filters: 

157 full_query = f"{search_query} {' '.join(filters)}" 

158 else: 

159 full_query = search_query 

160 

161 # Build request parameters 

162 params = { 

163 "q": full_query, 

164 "fl": "id,bibcode,title,author,year,pubdate,abstract,citation_count,bibstem,doi,identifier,pub,keyword,aff", 

165 "rows": min( 

166 self.max_results, 200 

167 ), # NASA ADS allows up to 200 per request 

168 "start": 0, 

169 } 

170 

171 # Add sorting 

172 sort_map = { 

173 "relevance": "score desc", 

174 "citation_count": "citation_count desc", 

175 "date": "date desc", 

176 } 

177 params["sort"] = sort_map.get(self.sort_by, "score desc") 

178 

179 try: 

180 # Apply rate limiting (simple like PubMed) 

181 self._last_wait_time = self.rate_tracker.apply_rate_limit( 

182 self.engine_type 

183 ) 

184 logger.debug( 

185 f"Applied rate limit wait: {self._last_wait_time:.2f}s" 

186 ) 

187 

188 # Make the API request 

189 logger.info( 

190 f"Making NASA ADS API request with query: {str(params['q'])[:100]}..." 

191 ) 

192 response = safe_get( 

193 f"{self.api_base}/search/query", 

194 params=params, 

195 headers=self.headers, 

196 timeout=30, 

197 ) 

198 

199 # Log rate limit headers if available 

200 if "X-RateLimit-Remaining" in response.headers: 200 ↛ 201line 200 didn't jump to line 201 because the condition on line 200 was never true

201 remaining = response.headers.get("X-RateLimit-Remaining") 

202 limit = response.headers.get("X-RateLimit-Limit", "unknown") 

203 logger.debug( 

204 f"NASA ADS rate limit: {remaining}/{limit} requests remaining" 

205 ) 

206 

207 if response.status_code == 200: 

208 data = response.json() 

209 docs = data.get("response", {}).get("docs", []) 

210 num_found = data.get("response", {}).get("numFound", 0) 

211 

212 logger.info( 

213 f"NASA ADS returned {len(docs)} results (total available: {num_found:,})" 

214 ) 

215 

216 # Format results as previews 

217 previews = [] 

218 for doc in docs: 

219 preview = self._format_doc_preview(doc) 

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

221 previews.append(preview) 

222 

223 logger.info(f"Successfully formatted {len(previews)} previews") 

224 return previews 

225 

226 if response.status_code == 429: 

227 # Rate limited 

228 logger.warning("NASA ADS rate limit reached") 

229 raise RateLimitError("NASA ADS rate limit exceeded") # noqa: TRY301 — re-raised by except RateLimitError for base class retry 

230 

231 if response.status_code == 401: 

232 logger.error("NASA ADS API key is invalid or missing") 

233 return [] 

234 

235 logger.error( 

236 f"NASA ADS API error: {response.status_code} - {response.text[:200]}" 

237 ) 

238 return [] 

239 

240 except RateLimitError: 

241 # Re-raise rate limit errors for base class retry handling 

242 raise 

243 except Exception as e: 

244 # logger.warning rather than logger.exception: the traceback 

245 # frames hold self.headers (the "Authorization: Bearer <key>" 

246 # value) and would render it under loguru diagnose. Redact the 

247 # api_key from the message as defense-in-depth. 

248 safe_msg = self._scrub_error(e) 

249 logger.warning(f"Error searching NASA ADS: {safe_msg}") 

250 return [] 

251 

252 def _format_doc_preview( 

253 self, doc: Dict[str, Any] 

254 ) -> Optional[Dict[str, Any]]: 

255 """ 

256 Format a NASA ADS document as a preview dictionary. 

257 

258 Args: 

259 doc: NASA ADS document object 

260 

261 Returns: 

262 Formatted preview dictionary or None if formatting fails 

263 """ 

264 try: 

265 # Extract basic information 

266 bibcode = doc.get("bibcode", "") 

267 # Get title from list if available 

268 title_list = doc.get("title", []) 

269 title = title_list[0] if title_list else "No title" 

270 

271 # Get abstract or create snippet 

272 abstract = doc.get("abstract", "") 

273 snippet = ( 

274 abstract[:SNIPPET_LENGTH_LONG] 

275 if abstract 

276 else f"Academic paper: {title}" 

277 ) 

278 

279 # Get publication info 

280 year = doc.get("year", "unknown") 

281 pubdate = doc.get("pubdate", "unknown") 

282 

283 # Get journal/source 

284 journal = "unknown" 

285 if doc.get("pub"): 

286 journal = str(doc.get("pub")) 

287 elif doc.get("bibstem"): 

288 bibstem = doc.get("bibstem", []) 

289 if bibstem: 289 ↛ 295line 289 didn't jump to line 295 because the condition on line 289 was always true

290 journal = ( 

291 bibstem[0] if isinstance(bibstem, list) else bibstem 

292 ) 

293 

294 # Get authors 

295 authors = doc.get("author", []) 

296 authors_str = ", ".join(authors[:5]) 

297 if len(authors) > 5: 

298 authors_str += " et al." 

299 

300 # NASA ADS returns each name as "Last, First" — emit a 

301 # structured CSL list so the citation normalizer doesn't have 

302 # to re-split the comma-joined display string above and 

303 # mangle the family/given pairing in the process. 

304 authors_csl: list[dict] = [] 

305 for raw in authors[:5]: 

306 name = (raw or "").strip() 

307 if not name: 307 ↛ 308line 307 didn't jump to line 308 because the condition on line 307 was never true

308 continue 

309 if "," in name: 

310 family, _, given = name.partition(",") 

311 authors_csl.append( 

312 {"family": family.strip(), "given": given.strip()} 

313 ) 

314 else: 

315 authors_csl.append({"literal": name}) 

316 

317 # Get metrics 

318 citation_count = doc.get("citation_count", 0) 

319 

320 # Get URL - prefer DOI, fallback to ADS URL 

321 url = None 

322 if doc.get("doi"): 

323 dois = doc.get("doi", []) 

324 if dois: 324 ↛ 328line 324 didn't jump to line 328 because the condition on line 324 was always true

325 doi = dois[0] if isinstance(dois, list) else dois 

326 url = f"https://doi.org/{doi}" 

327 

328 if not url: 

329 url = f"https://ui.adsabs.harvard.edu/abs/{bibcode}" 

330 

331 # Check if it's ArXiv 

332 is_arxiv = "arXiv" in str(doc.get("bibstem", [])) 

333 

334 # Get keywords 

335 keywords = doc.get("keyword", []) 

336 

337 # Extract DOI for enrichment layer 

338 doi_value = None 

339 if doc.get("doi"): 

340 dois = doc.get("doi", []) 

341 if dois: 341 ↛ 344line 341 didn't jump to line 344 because the condition on line 341 was always true

342 doi_value = dois[0] if isinstance(dois, list) else dois 

343 

344 return { 

345 "id": bibcode, 

346 "title": title, 

347 "link": url, 

348 "snippet": snippet, 

349 "authors": authors_str, 

350 "authors_csl": authors_csl or None, 

351 "year": year, 

352 "date": pubdate, 

353 # Both fields emit None (not the "unknown" sentinel) when 

354 # no pub/bibstem is available. The "unknown" literal 

355 # leaked through the normalizer's container_title fallback 

356 # and even matched a real OpenAlex source named "unknown" 

357 # (Q1, h_index=5) in the reference DB. 

358 "journal": None if journal == "unknown" else journal, 

359 # ArXiv preprints have pub="arXiv e-prints" — set journal_ref 

360 # to None so the filter's preprint-handling path activates 

361 # instead of trying to score "arXiv e-prints" as a journal. 

362 "journal_ref": ( 

363 None if is_arxiv or journal == "unknown" else journal 

364 ), 

365 "doi": doi_value, 

366 "citations": citation_count, 

367 "abstract": abstract, 

368 "is_arxiv": is_arxiv, 

369 "keywords": keywords[:5] if keywords else [], 

370 "type": "academic_paper", 

371 } 

372 

373 except Exception as e: 

374 safe_msg = self._scrub_error(e) 

375 logger.warning( 

376 f"Error formatting NASA ADS document {doc.get('bibcode', 'unknown')}: {safe_msg}" 

377 ) 

378 return None 

379 

380 def _get_full_content( 

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

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

383 """ 

384 Get full content for relevant items (NASA ADS provides most content in preview). 

385 

386 Args: 

387 relevant_items: List of relevant preview dictionaries 

388 

389 Returns: 

390 List of result dictionaries with full content 

391 """ 

392 # NASA ADS returns comprehensive data in the initial search, 

393 # so we don't need a separate full content fetch 

394 results = [] 

395 for item in relevant_items: 

396 result = { 

397 "title": item.get("title", ""), 

398 "link": item.get("link", ""), 

399 "snippet": item.get("snippet", ""), 

400 "content": item.get("abstract", item.get("snippet", "")), 

401 # Forward journal quality fields for content filters 

402 "journal_ref": item.get("journal_ref"), 

403 "doi": item.get("doi"), 

404 "metadata": { 

405 "authors": item.get("authors", ""), 

406 "year": item.get("year", ""), 

407 "journal": item.get("journal", ""), 

408 "citations": item.get("citations", 0), 

409 "is_arxiv": item.get("is_arxiv", False), 

410 "keywords": item.get("keywords", []), 

411 "doi": item.get("doi"), 

412 }, 

413 } 

414 results.append(result) 

415 

416 return results