Coverage for src/local_deep_research/research_library/downloaders/openalex.py: 98%
105 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-22 21:20 +0000
1"""
2OpenAlex PDF Downloader
4Downloads PDFs from OpenAlex using their API to find open access PDFs.
5OpenAlex aggregates open access information from multiple sources.
6"""
8import re
9from typing import Optional
10from urllib.parse import urlparse
12import requests
13from loguru import logger
15from ...security import redact_url_for_log
16from ...security.log_sanitizer import scrub_error
17from ...utilities.openalex_enrichment import (
18 normalize_openalex_api_key,
19 send_with_key_fallback,
20)
21from .base import BaseDownloader, ContentType, DownloadResult
24class OpenAlexDownloader(BaseDownloader):
25 """Downloader for OpenAlex papers with open access PDF support."""
27 def __init__(
28 self,
29 timeout: int = 30,
30 polite_pool_email: Optional[str] = None,
31 api_key: Optional[str] = None,
32 ):
33 """
34 Initialize OpenAlex downloader.
36 Args:
37 timeout: Request timeout in seconds
38 polite_pool_email: Optional email included in the User-Agent for
39 identification; retained under its existing name for compatibility
40 api_key: Optional OpenAlex API key for a higher daily request budget.
41 Placeholders and sentinel values normalize to None so the
42 downloader falls back to the keyless free tier instead of
43 sending a Bearer token OpenAlex will reject. A real key
44 OpenAlex *does* reject is dropped on first use and every
45 later lookup goes out keyless, so a bad key never turns
46 an available PDF into "no PDF available".
47 """
48 super().__init__(timeout)
49 self.polite_pool_email = polite_pool_email
50 self.api_key: Optional[str] = normalize_openalex_api_key(api_key)
51 # Set only once OpenAlex refuses the key; kept so ``_scrub``
52 # still has the literal to redact after ``api_key`` is cleared.
53 self._rejected_api_key: Optional[str] = None
54 self.base_api_url = "https://api.openalex.org"
56 def _scrub(self, error: Exception) -> str:
57 """Return a log-safe rendering of *error*.
59 logger.warning + this helper replaces logger.exception throughout
60 ``_get_pdf_url``: that frame's locals hold ``headers`` (the
61 "Authorization: Bearer <key>" value), which loguru's diagnose
62 would render into the traceback. Same documented trade-off as
63 ``search_engine_nasa_ads``. This class is not a
64 ``BaseSearchEngine``, so it does its own literal-key redaction
65 instead of relying on ``_scrub_error``/``_secret_attrs``.
67 ``getattr`` with a default, like ``BaseSearchEngine._scrub_error``:
68 this runs inside ``except`` handlers, and a partially-constructed
69 instance (``__new__``, or a ``super().__init__`` that raised)
70 must degrade to a shape-only scrub rather than raise
71 ``AttributeError`` from inside the handler.
72 """
73 return scrub_error(
74 error,
75 getattr(self, "api_key", None),
76 getattr(self, "_rejected_api_key", None),
77 )
79 def can_handle(self, url: str) -> bool:
80 """Check if URL is from OpenAlex."""
81 try:
82 hostname = urlparse(url).hostname
83 return bool(
84 hostname
85 and (
86 hostname == "openalex.org"
87 or hostname.endswith(".openalex.org")
88 )
89 )
90 except (ValueError, AttributeError, TypeError):
91 return False
93 def download(
94 self, url: str, content_type: ContentType = ContentType.PDF
95 ) -> Optional[bytes]:
96 """Download content from OpenAlex."""
97 result = self.download_with_result(url, content_type)
98 return result.content if result.is_success else None
100 def download_with_result(
101 self, url: str, content_type: ContentType = ContentType.PDF
102 ) -> DownloadResult:
103 """Download PDF and return detailed result with skip reason."""
104 # Only support PDF downloads for now
105 if content_type != ContentType.PDF:
106 return DownloadResult(
107 skip_reason="Text extraction not yet supported for OpenAlex"
108 )
110 # Extract work ID from URL
111 work_id = self._extract_work_id(url)
112 if not work_id:
113 return DownloadResult(
114 skip_reason="Invalid OpenAlex URL - could not extract work ID"
115 )
117 logger.info(f"Looking up OpenAlex work: {work_id}")
119 # Get work details from API to find PDF URL
120 pdf_url = self._get_pdf_url(work_id)
122 if not pdf_url:
123 return DownloadResult(
124 skip_reason="Not open access - no free PDF available"
125 )
127 # Download the PDF from the open access URL
128 logger.info(
129 f"Downloading open access PDF from: {redact_url_for_log(pdf_url)}"
130 )
131 pdf_content = super()._download_pdf(pdf_url)
133 if pdf_content:
134 return DownloadResult(content=pdf_content, is_success=True)
135 return DownloadResult(
136 skip_reason="Open access PDF URL found but download failed"
137 )
139 def _extract_work_id(self, url: str) -> Optional[str]:
140 """
141 Extract OpenAlex work ID from URL.
143 Handles formats like:
144 - https://openalex.org/W123456789
145 - https://openalex.org/works/W123456789
147 Returns:
148 Work ID (e.g., W123456789) or None if not found
149 """
150 # Use urlparse for more robust URL handling (handles query strings, fragments)
151 parsed = urlparse(url)
152 if not parsed.netloc or "openalex.org" not in parsed.netloc:
153 return None
155 # Extract work ID from path (W followed by digits)
156 # Handles /works/W123 or /W123
157 path = parsed.path
158 match = re.search(r"(?:/works/)?(W\d+)", path)
159 return match.group(1) if match else None
161 def _get_pdf_url(self, work_id: str) -> Optional[str]:
162 """
163 Get open access PDF URL from OpenAlex API.
165 Args:
166 work_id: OpenAlex work ID (e.g., W123456789)
168 Returns:
169 PDF URL if available, None otherwise
170 """
171 try:
172 # Construct API request
173 api_url = f"{self.base_api_url}/works/{work_id}"
174 params = {"select": "id,open_access,best_oa_location"}
176 # Identify the client by email when configured.
177 headers = {}
178 if self.polite_pool_email:
179 headers["User-Agent"] = f"mailto:{self.polite_pool_email}"
181 def _send(with_api_key: bool) -> requests.Response:
182 request_headers = dict(headers)
183 if with_api_key and self.api_key:
184 # Bearer header rather than the equally-supported
185 # ?api_key= query parameter, so the key stays out of
186 # URLs. https://help.openalex.org/api/authentication/
187 request_headers["Authorization"] = f"Bearer {self.api_key}"
188 return self.session.get(
189 api_url,
190 params=params,
191 headers=request_headers,
192 timeout=self.timeout,
193 )
195 def _drop_rejected_key() -> None:
196 # The single mutation point: this runs *before* the
197 # keyless retry, so even if that retry raises, the
198 # refused key is already gone and cannot be sent again.
199 # ``or self._rejected_api_key``: a second call must not
200 # store None over the literal and drop it out of
201 # ``_scrub``'s redaction set.
202 self._rejected_api_key = self.api_key or self._rejected_api_key
203 self.api_key = None
205 # Make API request. The shared helper retries keylessly once
206 # if OpenAlex refuses the key and drops the key for this
207 # downloader's lifetime, so a stale key costs one extra
208 # request rather than every open-access PDF. If the keyless
209 # retry fails too, the status branches below apply the
210 # pre-existing behaviour (warn, return None).
211 response, _ = send_with_key_fallback(
212 _send,
213 api_key=self.api_key,
214 context="OpenAlex download",
215 on_key_rejected=_drop_rejected_key,
216 )
218 if response.status_code == 200:
219 data = response.json()
221 # Check if it's open access
222 open_access_info = data.get("open_access", {})
223 is_oa = open_access_info.get("is_oa", False)
225 if not is_oa:
226 logger.info(f"Work {work_id} is not open access")
227 return None
229 # Get PDF URL from best open access location
230 best_oa_location = data.get("best_oa_location", {})
231 if best_oa_location:
232 # Try pdf_url first, fall back to landing_page_url
233 pdf_url = best_oa_location.get("pdf_url")
234 if pdf_url:
235 logger.info(
236 f"Found open access PDF for work {work_id}: {redact_url_for_log(pdf_url)}"
237 )
238 return str(pdf_url)
240 # Some works have landing page but no direct PDF
241 landing_url = best_oa_location.get("landing_page_url")
242 if landing_url: 242 ↛ 273line 242 didn't jump to line 273 because the condition on line 242 was always true
243 logger.info(
244 f"Found landing page for work {work_id}: {redact_url_for_log(landing_url)}"
245 )
246 # Validate that landing page is actually a PDF before returning
247 try:
248 head_response = self.session.head(
249 landing_url,
250 timeout=self.timeout,
251 allow_redirects=True,
252 )
253 content_type = head_response.headers.get(
254 "Content-Type", ""
255 ).lower()
256 if "application/pdf" in content_type:
257 logger.info(
258 f"Landing page is a direct PDF link for work {work_id}"
259 )
260 return str(landing_url)
261 logger.info(
262 f"Landing page is not a PDF (Content-Type: {content_type}), skipping"
263 )
264 except Exception as exc:
265 # Type only: urllib3 errors embed the scheme-less
266 # path and query, which the scrubber misses.
267 logger.warning(
268 f"Failed to validate landing page URL "
269 f"({type(exc).__name__}) for work {work_id}: "
270 f"{redact_url_for_log(landing_url)}"
271 )
273 logger.info(
274 f"No PDF URL available for open access work {work_id}"
275 )
276 return None
278 if response.status_code == 404:
279 logger.warning(f"Work not found in OpenAlex: {work_id}")
280 return None
281 logger.warning(f"OpenAlex API error: {response.status_code}")
282 return None
284 except requests.exceptions.RequestException as exc:
285 safe_msg = self._scrub(exc)
286 logger.warning(f"Failed to query OpenAlex API: {safe_msg}")
287 return None
288 except ValueError as exc:
289 # JSON decode errors are expected runtime errors
290 safe_msg = self._scrub(exc)
291 logger.warning(f"Failed to parse OpenAlex API response: {safe_msg}")
292 return None
293 # Note: KeyError and TypeError are not caught - they indicate programming
294 # bugs that should propagate for debugging