Coverage for src/local_deep_research/web_search_engines/engines/search_engine_openalex.py: 93%
181 statements
« prev ^ index » next coverage.py v7.15.1, created at 2026-07-20 01:24 +0000
1"""OpenAlex search engine implementation for academic papers and research."""
3from typing import Any, Dict, List, Optional
5from langchain_core.language_models import BaseLLM
7from ...constants import SNIPPET_LENGTH_LONG, USER_AGENT
8from ...security.safe_requests import safe_get
9from ...security.secure_logging import logger
10from ..rate_limiting import RateLimitError
11from ..search_engine_base import BaseSearchEngine, Exposure, Sensitivity
14class OpenAlexSearchEngine(BaseSearchEngine):
15 """OpenAlex search engine implementation with natural language query support."""
17 # Mark as public search engine
18 is_public = True
19 egress_sensitivity = Sensitivity.NON_SENSITIVE
20 egress_exposure = Exposure.EXPOSING
21 # Scientific/academic search engine
22 is_scientific = True
23 is_lexical = True
24 needs_llm_relevance_filter = True
26 def __init__(
27 self,
28 max_results: int = 25,
29 email: Optional[str] = None,
30 sort_by: str = "relevance",
31 filter_open_access: bool = False,
32 min_citations: int = 0,
33 from_publication_date: Optional[str] = None,
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 OpenAlex search engine.
42 Args:
43 max_results: Maximum number of search results
44 email: Email for polite pool (gets faster response) - optional
45 sort_by: Sort order ('relevance', 'cited_by_count', 'publication_date')
46 filter_open_access: Only return open access papers
47 min_citations: Minimum citation count filter
48 from_publication_date: Filter papers from this date (YYYY-MM-DD)
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 "openalex", llm, settings_snapshot
58 )
59 if journal_filter is not None:
60 preview_filters.append(journal_filter)
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 )
71 self.sort_by = sort_by
72 self.filter_open_access = filter_open_access
73 self.min_citations = min_citations
74 # Only set from_publication_date if it's not empty or "False"
75 self.from_publication_date = (
76 from_publication_date
77 if from_publication_date and from_publication_date != "False"
78 else None
79 )
81 # Get email from settings if not provided
82 if not email and settings_snapshot:
83 from ...config.search_config import get_setting_from_snapshot
85 try:
86 email = get_setting_from_snapshot(
87 "search.engine.web.openalex.email",
88 settings_snapshot=settings_snapshot,
89 )
90 except Exception:
91 logger.debug(
92 "Failed to read openalex.email from settings snapshot",
93 exc_info=True,
94 )
96 # Handle "False" string for email
97 self.email = email if email and email != "False" else None
99 # API configuration
100 self.api_base = "https://api.openalex.org"
101 self.headers = {
102 "User-Agent": f"{USER_AGENT} ({email})" if email else USER_AGENT,
103 "Accept": "application/json",
104 }
106 if email:
107 # Email allows access to polite pool with faster response times
108 logger.info(f"Using OpenAlex polite pool with email: {email}")
109 else:
110 logger.info(
111 "Using OpenAlex without email (consider adding email for faster responses)"
112 )
114 def _get_previews(self, query: str) -> List[Dict[str, Any]]:
115 """
116 Get preview information for OpenAlex search results.
118 Args:
119 query: The search query (natural language supported!)
121 Returns:
122 List of preview dictionaries
123 """
124 logger.info(f"Searching OpenAlex for: {query}")
126 # Build the search URL with parameters
127 params = {
128 "search": query, # OpenAlex handles natural language beautifully
129 "per_page": min(self.max_results, 200), # OpenAlex allows up to 200
130 "page": 1,
131 # Request specific fields including abstract for snippets
132 "select": "id,display_name,publication_year,publication_date,doi,primary_location,authorships,cited_by_count,open_access,best_oa_location,abstract_inverted_index",
133 }
135 # Add optional filters
136 filters = []
138 if self.filter_open_access:
139 filters.append("is_oa:true")
141 if self.min_citations > 0:
142 filters.append(f"cited_by_count:>{self.min_citations}")
144 if self.from_publication_date and self.from_publication_date != "False":
145 filters.append(
146 f"from_publication_date:{self.from_publication_date}"
147 )
149 if filters:
150 params["filter"] = ",".join(filters)
152 # Add sorting
153 sort_map = {
154 "relevance": "relevance_score:desc",
155 "cited_by_count": "cited_by_count:desc",
156 "publication_date": "publication_date:desc",
157 }
158 params["sort"] = sort_map.get(self.sort_by, "relevance_score:desc")
160 # Add email to params for polite pool
161 if self.email and self.email != "False":
162 params["mailto"] = self.email
164 try:
165 # Apply rate limiting before making the request (simple like PubMed)
166 self._last_wait_time = self.rate_tracker.apply_rate_limit(
167 self.engine_type
168 )
169 logger.debug(
170 f"Applied rate limit wait: {self._last_wait_time:.2f}s"
171 )
173 # Make the API request
174 logger.info(f"Making OpenAlex API request with params: {params}")
175 response = safe_get(
176 f"{self.api_base}/works",
177 params=params,
178 headers=self.headers,
179 timeout=30,
180 )
181 logger.info(f"OpenAlex API response status: {response.status_code}")
183 # Log rate limit info if available
184 if "x-ratelimit-remaining" in response.headers:
185 remaining = response.headers.get("x-ratelimit-remaining")
186 limit = response.headers.get("x-ratelimit-limit", "unknown")
187 logger.debug(
188 f"OpenAlex rate limit: {remaining}/{limit} requests remaining"
189 )
191 if response.status_code == 200:
192 data = response.json()
193 results = data.get("results", [])
194 meta = data.get("meta", {})
195 total_count = meta.get("count", 0)
197 logger.info(
198 f"OpenAlex returned {len(results)} results (total available: {total_count:,})"
199 )
201 # Log first result structure for debugging
202 if results:
203 first_result = results[0]
204 logger.debug(
205 f"First result keys: {list(first_result.keys())}"
206 )
207 logger.debug(
208 f"First result has abstract: {'abstract_inverted_index' in first_result}"
209 )
210 if "open_access" in first_result:
211 logger.debug(
212 f"Open access structure: {first_result['open_access']}"
213 )
215 # Format results as previews
216 previews = []
217 for i, work in enumerate(results):
218 logger.debug(
219 f"Formatting work {i + 1}/{len(results)}: {(work.get('display_name') or 'Unknown')[:50]}"
220 )
221 preview = self._format_work_preview(work)
222 if preview:
223 previews.append(preview)
224 logger.debug(
225 f"Preview created with snippet: {preview.get('snippet', '')[:100]}..."
226 )
227 else:
228 logger.warning(f"Failed to format work {i + 1}")
230 logger.info(
231 f"Successfully formatted {len(previews)} previews from {len(results)} results"
232 )
233 return previews
235 if response.status_code == 429:
236 # Rate limited (very rare with OpenAlex)
237 logger.warning("OpenAlex rate limit reached")
238 raise RateLimitError("OpenAlex rate limit exceeded") # noqa: TRY301 — re-raised by except RateLimitError for base class retry
240 logger.error(
241 f"OpenAlex API error: {response.status_code} - {response.text[:200]}"
242 )
243 return []
245 except RateLimitError:
246 # Re-raise rate limit errors for base class retry handling
247 raise
248 except Exception as e:
249 safe_msg = self._scrub_error(e)
250 logger.exception(
251 f"Error searching OpenAlex ({type(e).__name__}): {safe_msg}"
252 )
253 return []
255 def _format_work_preview(
256 self, work: Dict[str, Any]
257 ) -> Optional[Dict[str, Any]]:
258 """
259 Format an OpenAlex work as a preview dictionary.
261 Args:
262 work: OpenAlex work object
264 Returns:
265 Formatted preview dictionary or None if formatting fails
266 """
267 try:
268 # Extract basic information
269 # Use `or` instead of dict.get default — OpenAlex routinely
270 # returns these keys with explicit None values, which would
271 # bypass the default and crash on downstream string ops.
272 work_id = work.get("id") or ""
273 title = work.get("display_name") or "No title"
274 logger.debug(f"Formatting work: {title[:50]}")
276 # Build snippet from abstract or first part of title
277 abstract = None
278 if work.get("abstract_inverted_index"):
279 logger.debug(
280 f"Found abstract_inverted_index with {len(work['abstract_inverted_index'])} words"
281 )
282 # Reconstruct abstract from inverted index
283 abstract = self._reconstruct_abstract(
284 work["abstract_inverted_index"]
285 )
286 logger.debug(
287 f"Reconstructed abstract length: {len(abstract) if abstract else 0}"
288 )
289 else:
290 logger.debug("No abstract_inverted_index found")
292 snippet = (
293 abstract[:SNIPPET_LENGTH_LONG]
294 if abstract
295 else f"Academic paper: {title}"
296 )
297 logger.debug(f"Created snippet: {snippet[:100]}...")
299 # Get publication info
300 publication_year = work.get("publication_year", "unknown")
301 publication_date = work.get("publication_date", "unknown")
303 # Get venue/journal info
304 venue = work.get("primary_location", {})
305 journal_name = "unknown"
306 openalex_source_id = None
307 source_type = None
308 issn = None
309 if venue:
310 source = venue.get("source", {})
311 if source:
312 journal_name = source.get("display_name") or "unknown"
313 # Extract source ID for journal quality lookups
314 raw_sid = source.get("id") or ""
315 if raw_sid:
316 openalex_source_id = raw_sid.split("/")[-1]
317 source_type = source.get("type")
318 # Forward the linking ISSN so the reputation filter's
319 # Tier 2/3 lookups can use it instead of falling back
320 # to fuzzy name matching.
321 issn = source.get("issn_l") or None
323 # Get authors
324 authors = []
325 for authorship in work.get("authorships", [])[
326 :5
327 ]: # Limit to 5 authors
328 author = authorship.get("author", {})
329 if author: 329 ↛ 325line 329 didn't jump to line 325 because the condition on line 329 was always true
330 authors.append(author.get("display_name", ""))
332 authors_str = ", ".join(authors)
333 if len(work.get("authorships", [])) > 5:
334 authors_str += " et al."
336 # Extract author affiliations for the institution-tier scoring.
337 # Each entry is a dict with the OpenAlex institution id, ROR id,
338 # and display name — the lookup_institution() helper accepts any
339 # of those three.
340 affiliations: list[dict] = []
341 seen_inst_ids: set[str] = set()
342 for authorship in work.get("authorships", []):
343 for inst in authorship.get("institutions", []) or []: 343 ↛ 344line 343 didn't jump to line 344 because the loop on line 343 never started
344 raw_id = inst.get("id") or ""
345 short_id = raw_id.split("/")[-1] if raw_id else ""
346 if short_id and short_id in seen_inst_ids:
347 continue
348 if short_id:
349 seen_inst_ids.add(short_id)
350 affiliations.append(
351 {
352 "openalex_id": short_id or None,
353 "ror": (inst.get("ror") or "")
354 .rstrip("/")
355 .split("/")[-1]
356 or None,
357 "name": inst.get("display_name"),
358 }
359 )
361 # Get metrics
362 cited_by_count = work.get("cited_by_count", 0)
364 # Get URL - prefer DOI, fallback to OpenAlex URL.
365 # `.get("doi", work_id)` is wrong: when the key exists with value
366 # None (common for non-DOI works) it returns None, not the
367 # default. Use `or` so a None DOI falls through to work_id.
368 url = work.get("doi") or work_id
369 if not url.startswith("http"):
370 if url.startswith("https://doi.org/"): 370 ↛ 371line 370 didn't jump to line 371 because the condition on line 370 was never true
371 pass # Already a full DOI URL
372 elif url.startswith("10."):
373 url = f"https://doi.org/{url}"
374 else:
375 url = work_id # OpenAlex URL
377 # Check if open access
378 open_access_info = work.get("open_access", {})
379 is_oa = (
380 open_access_info.get("is_oa", False)
381 if open_access_info
382 else False
383 )
384 oa_url = None
385 if is_oa:
386 best_location = work.get("best_oa_location", {})
387 if best_location: 387 ↛ 392line 387 didn't jump to line 392 because the condition on line 387 was always true
388 oa_url = best_location.get("pdf_url") or best_location.get(
389 "landing_page_url"
390 )
392 return {
393 "id": work_id,
394 "title": title,
395 "link": url,
396 "snippet": snippet,
397 "authors": authors_str,
398 "year": publication_year,
399 "date": publication_date,
400 # Both fields emit None (not the "unknown" sentinel) when
401 # OpenAlex has no venue for this work. Downstream consumers
402 # (citation normalizer, journal reputation filter) treat
403 # missing venue as "no scoring signal", which is accurate;
404 # the old "unknown" sentinel leaked through the normalizer
405 # as a literal container_title and even matched a real
406 # OpenAlex source named "unknown" (h_index=5, Q1) in the
407 # reference DB.
408 "journal": journal_name if journal_name != "unknown" else None,
409 "journal_ref": journal_name
410 if journal_name != "unknown"
411 else None,
412 "issn": issn,
413 "affiliations": affiliations or None,
414 "openalex_source_id": openalex_source_id,
415 "source_type": source_type,
416 "citations": cited_by_count,
417 "is_open_access": is_oa,
418 "oa_url": oa_url,
419 "abstract": abstract,
420 "type": "academic_paper",
421 }
423 except Exception as e:
424 safe_msg = self._scrub_error(e)
425 logger.exception(
426 f"Error formatting OpenAlex work: {work.get('id', 'unknown')} ({type(e).__name__}): {safe_msg}"
427 )
428 return None
430 def _reconstruct_abstract(
431 self, inverted_index: Dict[str, List[int]]
432 ) -> str:
433 """
434 Reconstruct abstract text from OpenAlex inverted index format.
436 Args:
437 inverted_index: Dictionary mapping words to their positions
439 Returns:
440 Reconstructed abstract text
441 """
442 try:
443 # Create position-word mapping
444 position_word = {}
445 for word, positions in inverted_index.items():
446 for pos in positions:
447 position_word[pos] = word
449 # Sort by position and reconstruct
450 sorted_positions = sorted(position_word.keys())
451 words = [position_word[pos] for pos in sorted_positions]
453 return " ".join(words)
455 except Exception:
456 logger.debug("Could not reconstruct abstract from inverted index")
457 return ""
459 def _get_full_content(
460 self, relevant_items: List[Dict[str, Any]]
461 ) -> List[Dict[str, Any]]:
462 """
463 Get full content for relevant items (OpenAlex provides most content in preview).
465 Args:
466 relevant_items: List of relevant preview dictionaries
468 Returns:
469 List of result dictionaries with full content
470 """
471 # OpenAlex returns comprehensive data in the initial search,
472 # so we don't need a separate full content fetch
473 results = []
474 for item in relevant_items:
475 result = {
476 "title": item.get("title", ""),
477 "link": item.get("link", ""),
478 "snippet": item.get("snippet", ""),
479 "content": item.get("abstract", item.get("snippet", "")),
480 # Forward journal quality fields for content filters
481 "journal_ref": item.get("journal_ref"),
482 "openalex_source_id": item.get("openalex_source_id"),
483 "source_type": item.get("source_type"),
484 "affiliations": item.get("affiliations"),
485 "metadata": {
486 "authors": item.get("authors", ""),
487 "year": item.get("year", ""),
488 "journal": item.get("journal", ""),
489 "citations": item.get("citations", 0),
490 "is_open_access": item.get("is_open_access", False),
491 "oa_url": item.get("oa_url"),
492 "affiliations": item.get("affiliations"),
493 },
494 }
495 results.append(result)
497 return results