Coverage for src/local_deep_research/web_search_engines/engines/search_engine_nasa_ads.py: 91%
152 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-22 21:20 +0000
1"""NASA Astrophysics Data System (ADS) search engine implementation."""
3from typing import Any, Dict, List, Optional
5from langchain_core.language_models import BaseLLM
6from ...security.secure_logging import logger
8from ...constants import SNIPPET_LENGTH_LONG, USER_AGENT
9from ...security.safe_requests import safe_get
10from ...settings.exceptions import SettingsValueError
11from ..rate_limiting import RateLimitError
12from ..search_engine_base import BaseSearchEngine, Exposure, Sensitivity
15class NasaAdsSearchEngine(BaseSearchEngine):
16 """NASA ADS search engine for physics, astronomy, and astrophysics papers."""
18 # Mark as public search engine
19 is_public = True
20 egress_sensitivity = Sensitivity.NON_SENSITIVE
21 egress_exposure = Exposure.EXPOSING
22 # Scientific/astronomy/astrophysics search engine
23 is_scientific = True
24 is_lexical = True
25 needs_llm_relevance_filter = True
27 def __init__(
28 self,
29 max_results: int = 25,
30 api_key: Optional[str] = None,
31 sort_by: str = "relevance",
32 min_citations: int = 0,
33 from_publication_date: Optional[str] = None,
34 include_arxiv: bool = True,
35 llm: Optional[BaseLLM] = None,
36 max_filtered_results: Optional[int] = None,
37 settings_snapshot: Optional[Dict[str, Any]] = None,
38 **kwargs,
39 ):
40 """
41 Initialize the NASA ADS search engine.
43 Args:
44 max_results: Maximum number of search results
45 api_key: NASA ADS API key (required for higher rate limits)
46 sort_by: Sort order ('relevance', 'citation_count', 'date')
47 min_citations: Minimum citation count filter
48 from_publication_date: Filter papers from this date (YYYY-MM-DD)
49 include_arxiv: Include ArXiv preprints in results
50 llm: Language model for relevance filtering
51 max_filtered_results: Maximum number of results to keep after filtering
52 settings_snapshot: Settings snapshot for configuration
53 **kwargs: Additional parameters to pass to parent class
54 """
55 # Journal filter runs before LLM relevance (Tiers 1-3 are instant)
56 preview_filters = []
57 journal_filter = self._create_journal_filter(
58 "nasa_ads", llm, settings_snapshot
59 )
60 if journal_filter is not None:
61 preview_filters.append(journal_filter)
63 super().__init__(
64 llm=llm,
65 max_filtered_results=max_filtered_results,
66 max_results=max_results,
67 preview_filters=preview_filters, # type: ignore[arg-type]
68 settings_snapshot=settings_snapshot,
69 **kwargs,
70 )
72 self.sort_by = sort_by
73 self.min_citations = min_citations
74 self.include_arxiv = include_arxiv
75 # Handle from_publication_date
76 self.from_publication_date = (
77 from_publication_date
78 if from_publication_date
79 and from_publication_date not in ["False", "false", ""]
80 else None
81 )
83 # Get API key from settings if not provided
84 if not api_key and settings_snapshot:
85 from ...config.search_config import get_setting_from_snapshot
87 try:
88 api_key = get_setting_from_snapshot(
89 "search.engine.web.nasa_ads.api_key",
90 settings_snapshot=settings_snapshot,
91 )
92 except SettingsValueError:
93 raise
94 except Exception:
95 logger.debug(
96 "Failed to read nasa_ads.api_key from settings snapshot",
97 exc_info=True,
98 )
100 # Handle "False" string for api_key
101 self.api_key = (
102 api_key
103 if api_key and api_key not in ["False", "false", ""]
104 else None
105 )
107 # API configuration
108 self.api_base = "https://api.adsabs.harvard.edu/v1"
109 self.headers = {
110 "User-Agent": USER_AGENT,
111 "Accept": "application/json",
112 }
114 if self.api_key:
115 self.headers["Authorization"] = f"Bearer {self.api_key}"
116 logger.info("Using NASA ADS with API key")
117 else:
118 logger.error(
119 "NASA ADS requires an API key to function. Get a free key at: https://ui.adsabs.harvard.edu/user/settings/token"
120 )
122 def _get_previews(self, query: str) -> List[Dict[str, Any]]:
123 """
124 Get preview information for NASA ADS search results.
126 Args:
127 query: The search query (natural language supported)
129 Returns:
130 List of preview dictionaries
131 """
132 logger.info("Searching NASA ADS")
134 # Build the search query - NASA ADS has good natural language support
135 # We can use the query directly or enhance it slightly
136 search_query = query
138 # Build filters
139 filters = []
140 if self.from_publication_date:
141 # Convert YYYY-MM-DD to ADS format
142 try:
143 year = self.from_publication_date.split("-")[0]
144 if year.isdigit(): # Only add if it's a valid year 144 ↛ 152line 144 didn't jump to line 152 because the condition on line 144 was always true
145 filters.append(f"year:{year}-9999")
146 except Exception:
147 logger.debug(
148 "best-effort date parsing, invalid formats skipped",
149 exc_info=True,
150 )
152 if self.min_citations > 0:
153 filters.append(f"citation_count:[{self.min_citations} TO *]")
155 if not self.include_arxiv:
156 filters.append('-bibstem:"arXiv"')
158 # Combine query with filters
159 if filters:
160 full_query = f"{search_query} {' '.join(filters)}"
161 else:
162 full_query = search_query
164 # Build request parameters
165 params = {
166 "q": full_query,
167 "fl": "id,bibcode,title,author,year,pubdate,abstract,citation_count,bibstem,doi,identifier,pub,keyword,aff",
168 "rows": min(
169 self.max_results, 200
170 ), # NASA ADS allows up to 200 per request
171 "start": 0,
172 }
174 # Add sorting
175 sort_map = {
176 "relevance": "score desc",
177 "citation_count": "citation_count desc",
178 "date": "date desc",
179 }
180 params["sort"] = sort_map.get(self.sort_by, "score desc")
182 try:
183 # Apply rate limiting (simple like PubMed)
184 self._last_wait_time = self.rate_tracker.apply_rate_limit(
185 self.engine_type
186 )
187 logger.debug(
188 f"Applied rate limit wait: {self._last_wait_time:.2f}s"
189 )
191 # Make the API request
192 logger.info(
193 f"Making NASA ADS API request "
194 f"({len(str(params['q']))} query chars)"
195 )
196 response = safe_get(
197 f"{self.api_base}/search/query",
198 params=params,
199 headers=self.headers,
200 timeout=30,
201 )
203 # Log rate limit headers if available
204 if "X-RateLimit-Remaining" in response.headers: 204 ↛ 205line 204 didn't jump to line 205 because the condition on line 204 was never true
205 remaining = response.headers.get("X-RateLimit-Remaining")
206 limit = response.headers.get("X-RateLimit-Limit", "unknown")
207 logger.debug(
208 f"NASA ADS rate limit: {remaining}/{limit} requests remaining"
209 )
211 if response.status_code == 200:
212 data = response.json()
213 docs = data.get("response", {}).get("docs", [])
214 num_found = data.get("response", {}).get("numFound", 0)
216 logger.info(
217 f"NASA ADS returned {len(docs)} results (total available: {num_found:,})"
218 )
220 # Format results as previews
221 previews = []
222 for doc in docs:
223 preview = self._format_doc_preview(doc)
224 if preview: 224 ↛ 222line 224 didn't jump to line 222 because the condition on line 224 was always true
225 previews.append(preview)
227 logger.info(f"Successfully formatted {len(previews)} previews")
228 return previews
230 if response.status_code == 429:
231 # Rate limited
232 logger.warning("NASA ADS rate limit reached")
233 raise RateLimitError("NASA ADS rate limit exceeded") # noqa: TRY301 — re-raised by except RateLimitError for base class retry
235 if response.status_code == 401:
236 logger.error("NASA ADS API key is invalid or missing")
237 return []
239 logger.error(
240 f"NASA ADS API error: {response.status_code} - {response.text[:200]}"
241 )
242 return []
244 except RateLimitError:
245 # Re-raise rate limit errors for base class retry handling
246 raise
247 except Exception as e:
248 # logger.warning rather than logger.exception: the traceback
249 # frames hold self.headers (the "Authorization: Bearer <key>"
250 # value) and would render it under loguru diagnose. Redact the
251 # api_key from the message as defense-in-depth.
252 safe_msg = self._scrub_error(e)
253 logger.warning(f"Error searching NASA ADS: {safe_msg}")
254 return []
256 def _format_doc_preview(
257 self, doc: Dict[str, Any]
258 ) -> Optional[Dict[str, Any]]:
259 """
260 Format a NASA ADS document as a preview dictionary.
262 Args:
263 doc: NASA ADS document object
265 Returns:
266 Formatted preview dictionary or None if formatting fails
267 """
268 try:
269 # Extract basic information
270 bibcode = doc.get("bibcode", "")
271 # Get title from list if available
272 title_list = doc.get("title", [])
273 title = title_list[0] if title_list else "No title"
275 # Get abstract or create snippet
276 abstract = doc.get("abstract", "")
277 snippet = (
278 abstract[:SNIPPET_LENGTH_LONG]
279 if abstract
280 else f"Academic paper: {title}"
281 )
283 # Get publication info
284 year = doc.get("year", "unknown")
285 pubdate = doc.get("pubdate", "unknown")
287 # Get journal/source
288 journal = "unknown"
289 if doc.get("pub"):
290 journal = str(doc.get("pub"))
291 elif doc.get("bibstem"):
292 bibstem = doc.get("bibstem", [])
293 if bibstem: 293 ↛ 299line 293 didn't jump to line 299 because the condition on line 293 was always true
294 journal = (
295 bibstem[0] if isinstance(bibstem, list) else bibstem
296 )
298 # Get authors
299 authors = doc.get("author", [])
300 authors_str = ", ".join(authors[:5])
301 if len(authors) > 5:
302 authors_str += " et al."
304 # NASA ADS returns each name as "Last, First" — emit a
305 # structured CSL list so the citation normalizer doesn't have
306 # to re-split the comma-joined display string above and
307 # mangle the family/given pairing in the process.
308 authors_csl: list[dict] = []
309 for raw in authors[:5]:
310 name = (raw or "").strip()
311 if not name: 311 ↛ 312line 311 didn't jump to line 312 because the condition on line 311 was never true
312 continue
313 if "," in name:
314 family, _, given = name.partition(",")
315 authors_csl.append(
316 {"family": family.strip(), "given": given.strip()}
317 )
318 else:
319 authors_csl.append({"literal": name})
321 # Get metrics
322 citation_count = doc.get("citation_count", 0)
324 # Get URL - prefer DOI, fallback to ADS URL
325 url = None
326 if doc.get("doi"):
327 dois = doc.get("doi", [])
328 if dois: 328 ↛ 332line 328 didn't jump to line 332 because the condition on line 328 was always true
329 doi = dois[0] if isinstance(dois, list) else dois
330 url = f"https://doi.org/{doi}"
332 if not url:
333 url = f"https://ui.adsabs.harvard.edu/abs/{bibcode}"
335 # Check if it's ArXiv
336 is_arxiv = "arXiv" in str(doc.get("bibstem", []))
338 # Get keywords
339 keywords = doc.get("keyword", [])
341 # Extract DOI for enrichment layer
342 doi_value = None
343 if doc.get("doi"):
344 dois = doc.get("doi", [])
345 if dois: 345 ↛ 348line 345 didn't jump to line 348 because the condition on line 345 was always true
346 doi_value = dois[0] if isinstance(dois, list) else dois
348 return {
349 "id": bibcode,
350 "title": title,
351 "link": url,
352 "snippet": snippet,
353 "authors": authors_str,
354 "authors_csl": authors_csl or None,
355 "year": year,
356 "date": pubdate,
357 # Both fields emit None (not the "unknown" sentinel) when
358 # no pub/bibstem is available. The "unknown" literal
359 # leaked through the normalizer's container_title fallback
360 # and even matched a real OpenAlex source named "unknown"
361 # (Q1, h_index=5) in the reference DB.
362 "journal": None if journal == "unknown" else journal,
363 # ArXiv preprints have pub="arXiv e-prints" — set journal_ref
364 # to None so the filter's preprint-handling path activates
365 # instead of trying to score "arXiv e-prints" as a journal.
366 "journal_ref": (
367 None if is_arxiv or journal == "unknown" else journal
368 ),
369 "doi": doi_value,
370 "citations": citation_count,
371 "abstract": abstract,
372 "is_arxiv": is_arxiv,
373 "keywords": keywords[:5] if keywords else [],
374 "type": "academic_paper",
375 }
377 except Exception as e:
378 safe_msg = self._scrub_error(e)
379 logger.warning(
380 f"Error formatting NASA ADS document {doc.get('bibcode', 'unknown')}: {safe_msg}"
381 )
382 return None
384 def _get_full_content(
385 self, relevant_items: List[Dict[str, Any]]
386 ) -> List[Dict[str, Any]]:
387 """
388 Get full content for relevant items (NASA ADS provides most content in preview).
390 Args:
391 relevant_items: List of relevant preview dictionaries
393 Returns:
394 List of result dictionaries with full content
395 """
396 # NASA ADS returns comprehensive data in the initial search,
397 # so we don't need a separate full content fetch
398 results = []
399 for item in relevant_items:
400 result = {
401 "title": item.get("title", ""),
402 "link": item.get("link", ""),
403 "snippet": item.get("snippet", ""),
404 "content": item.get("abstract", item.get("snippet", "")),
405 # Forward journal quality fields for content filters
406 "journal_ref": item.get("journal_ref"),
407 "doi": item.get("doi"),
408 "metadata": {
409 "authors": item.get("authors", ""),
410 "year": item.get("year", ""),
411 "journal": item.get("journal", ""),
412 "citations": item.get("citations", 0),
413 "is_arxiv": item.get("is_arxiv", False),
414 "keywords": item.get("keywords", []),
415 "doi": item.get("doi"),
416 },
417 }
418 results.append(result)
420 return results