Coverage for src/local_deep_research/research_library/downloaders/pubmed.py: 95%

1"""

2PubMed/PMC PDF Downloader

3"""

5import re

6import time

7from typing import Optional

8from urllib.parse import urlparse

9from loguru import logger

11from ...security import redact_url_for_log

12from .base import BaseDownloader, ContentType, DownloadResult

13from ...security.egress.policy import PolicyDeniedError

16class PubMedDownloader(BaseDownloader):

17 """Downloader for PubMed and PubMed Central articles with PDF and text support."""

19 def __init__(self, timeout: int = 30, rate_limit_delay: float = 1.0):

20 """

21 Initialize PubMed downloader.

23 Args:

24 timeout: Request timeout in seconds

25 rate_limit_delay: Delay between requests to avoid rate limiting

26 """

27 super().__init__(timeout)

28 self.rate_limit_delay = rate_limit_delay

29 self.last_request_time: float = 0.0

31 def can_handle(self, url: str) -> bool:

32 """Check if URL is from PubMed or PMC."""

33 try:

34 parsed = urlparse(url)

35 hostname = parsed.hostname

36 if not hostname:

37 return False

39 # Check for pubmed.ncbi.nlm.nih.gov

40 if hostname == "pubmed.ncbi.nlm.nih.gov":

41 return True

43 # Check for ncbi.nlm.nih.gov with /pmc in path

44 if hostname == "ncbi.nlm.nih.gov" and "/pmc" in parsed.path:

45 return True

47 # Check for europepmc.org and its subdomains

48 if hostname == "europepmc.org" or hostname.endswith(

49 ".europepmc.org"

50 ):

51 return True

53 return False

54 except Exception:

55 return False

57 def download(

58 self, url: str, content_type: ContentType = ContentType.PDF

59 ) -> Optional[bytes]:

60 """Download content from PubMed/PMC."""

61 # Apply rate limiting

62 self._apply_rate_limit()

64 if content_type == ContentType.TEXT:

65 # Try to get full text from API

66 return self._download_text(url)

67 # Download PDF

68 return self._download_pdf_content(url)

70 def download_with_result(

71 self, url: str, content_type: ContentType = ContentType.PDF

72 ) -> DownloadResult:

73 """Download content and return detailed result with skip reason."""

74 # Apply rate limiting

75 self._apply_rate_limit()

77 if content_type == ContentType.TEXT:

78 content = self._download_text(url)

79 if content:

80 return DownloadResult(content=content, is_success=True)

81 # `_download_text` collapses every failure into None: an

82 # unparseable URL, a Europe PMC API error, an unreachable PDF, a

83 # PDF whose text could not be extracted. Naming a subscription

84 # here asserts a paywall for all of them, and the word alone is

85 # what `FailureClassifier.classify_failure` matches on to declare

86 # a permanent `paywall_or_login` failure. A genuine paywall is

87 # established by Europe PMC's `isOpenAccess` flag, which the PDF

88 # path checks and reports on its own.

89 return DownloadResult(

90 skip_reason="Full text not available from the PubMed/PMC APIs"

91 )

92 # Try to download PDF with detailed tracking

93 return self._download_pdf_with_result(url)

95 def _download_pdf_content(self, url: str) -> Optional[bytes]:

96 """Download PDF from PubMed/PMC."""

97 # Handle different URL types

98 parsed = urlparse(url)

99 hostname = parsed.hostname or ""

100 path = parsed.path or ""

101

102 # Check for PMC article direct download

103 if hostname == "ncbi.nlm.nih.gov" and "/pmc/articles/PMC" in path:

104 return self._download_pmc_direct(url)

105 # Check for PubMed main site

106 if hostname == "pubmed.ncbi.nlm.nih.gov":

107 return self._download_pubmed(url)

108 # Check for Europe PMC and subdomains

109 if hostname == "europepmc.org" or hostname.endswith(".europepmc.org"):

110 return self._download_europe_pmc(url)

111

112 return None

113

114 def _download_pdf_with_result(self, url: str) -> DownloadResult:

115 """Download PDF and return detailed result with skip reason."""

116 # Handle different URL types

117 if "/pmc/articles/PMC" in url:

118 pmc_match = re.search(r"(PMC\d+)", url)

119 if not pmc_match:

120 return DownloadResult(skip_reason="Invalid PMC URL format")

121

122 pmc_id = pmc_match.group(1)

123 logger.info(f"Downloading PMC article: {pmc_id}")

124

125 # Try Europe PMC first

126 pdf_content = self._download_via_europe_pmc(pmc_id)

127 if pdf_content:

128 return DownloadResult(content=pdf_content, is_success=True)

129

130 # Try NCBI PMC

131 pdf_content = self._download_via_ncbi_pmc(pmc_id)

132 if pdf_content:

133 return DownloadResult(content=pdf_content, is_success=True)

134

135 return DownloadResult(

136 skip_reason=f"PMC article {pmc_id} not accessible - may be retracted or embargoed"

137 )

138

139 if urlparse(url).hostname == "pubmed.ncbi.nlm.nih.gov":

140 # Extract PMID

141 pmid_match = re.search(r"/(\d+)/?", url)

142 if not pmid_match:

143 return DownloadResult(skip_reason="Invalid PubMed URL format")

144

145 pmid = pmid_match.group(1)

146 logger.info(f"Processing PubMed article: {pmid}")

147

148 # Check if article is open access via Europe PMC

149 try:

150 api_url = (

151 "https://www.ebi.ac.uk/europepmc/webservices/rest/search"

152 )

153 params = {"query": f"EXT_ID:{pmid}", "format": "json"}

154

155 response = self.session.get(api_url, params=params, timeout=10)

156

157 if response.status_code == 200: 157 ↛ 199line 157 didn't jump to line 199 because the condition on line 157 was always true

158 data = response.json()

159 results = data.get("resultList", {}).get("result", [])

160

161 if results:

162 article = results[0]

163

164 # Check if article exists but is not open access

165 if article.get("isOpenAccess") != "Y":

166 journal = article.get(

167 "journalTitle", "Unknown journal"

168 )

169 return DownloadResult(

170 skip_reason=f"Article requires subscription to {journal}"

171 )

172

173 # Check if PDF is available

174 if article.get("hasPDF") != "Y":

175 return DownloadResult(

176 skip_reason="No PDF version available for this article"

177 )

178

179 # Try to download

180 pmcid = article.get("pmcid")

181 if pmcid: 181 ↛ 199line 181 didn't jump to line 199 because the condition on line 181 was always true

182 pdf_content = self._download_via_europe_pmc(pmcid)

183 if pdf_content: 183 ↛ 199line 183 didn't jump to line 199 because the condition on line 183 was always true

184 return DownloadResult(

185 content=pdf_content, is_success=True

186 )

187 else:

188 return DownloadResult(

189 skip_reason=f"Article PMID:{pmid} not found in Europe PMC database"

190 )

191 except PolicyDeniedError:

192 raise

193 except Exception as e:

194 logger.debug(

195 f"Error checking article status ({type(e).__name__})"

196 )

197

198 # Try to find PMC ID via NCBI

199 pmc_id = self._get_pmc_id_from_pmid(pmid)

200 if pmc_id:

201 logger.info(f"Found PMC ID: {pmc_id} for PMID: {pmid}")

202

203 # Try downloading via PMC

204 pdf_content = self._download_via_europe_pmc(pmc_id)

205 if pdf_content:

206 return DownloadResult(content=pdf_content, is_success=True)

207

208 pdf_content = self._download_via_ncbi_pmc(pmc_id)

209 if pdf_content: 209 ↛ 210line 209 didn't jump to line 210 because the condition on line 209 was never true

210 return DownloadResult(content=pdf_content, is_success=True)

211

212 return DownloadResult(

213 skip_reason=f"PMC version exists but PDF not accessible (PMC ID: {pmc_id})"

214 )

215

216 return DownloadResult(

217 skip_reason="No free full-text available - article may be paywalled"

218 )

219

220 parsed = urlparse(url)

221 hostname = parsed.hostname or ""

222 if hostname == "europepmc.org" or hostname.endswith(".europepmc.org"):

223 pmc_match = re.search(r"(PMC\d+)", url)

224 if pmc_match:

225 pmc_id = pmc_match.group(1)

226 pdf_content = self._download_via_europe_pmc(pmc_id)

227 if pdf_content:

228 return DownloadResult(content=pdf_content, is_success=True)

229 return DownloadResult(

230 skip_reason=f"Europe PMC article {pmc_id} not accessible"

231 )

232 return DownloadResult(skip_reason="Invalid Europe PMC URL format")

233 return DownloadResult(skip_reason="Unsupported PubMed/PMC URL format")

234

235 def _download_text(self, url: str) -> Optional[bytes]:

236 """Download full text content from PubMed/PMC APIs."""

237 # Extract PMID or PMC ID

238 pmid = None

239 pmc_id = None

240

241 parsed_url = urlparse(url)

242 if parsed_url.hostname == "pubmed.ncbi.nlm.nih.gov":

243 pmid_match = re.search(r"/(\d+)/?", url)

244 if pmid_match: 244 ↛ 252line 244 didn't jump to line 252 because the condition on line 244 was always true

245 pmid = pmid_match.group(1)

246 elif "/pmc/articles/PMC" in url:

247 pmc_match = re.search(r"(PMC\d+)", url)

248 if pmc_match: 248 ↛ 252line 248 didn't jump to line 252 because the condition on line 248 was always true

249 pmc_id = pmc_match.group(1)

250

251 # Try Europe PMC API for full text

252 if pmid or pmc_id:

253 text = self._fetch_text_from_europe_pmc(pmid, pmc_id)

254 if text:

255 return text.encode("utf-8")

256

257 # Fallback: Download PDF and extract text

258 pdf_content = self._download_pdf_content(url)

259 if pdf_content:

260 text = self.extract_text_from_pdf(pdf_content)

261 if text:

262 return text.encode("utf-8")

263

264 return None

265

266 def _fetch_text_from_europe_pmc(

267 self, pmid: Optional[str], pmc_id: Optional[str]

268 ) -> Optional[str]:

269 """Fetch full text from Europe PMC API."""

270 try:

271 # Construct query

272 if pmc_id:

273 query = f"PMC:{pmc_id.replace('PMC', '')}"

274 elif pmid:

275 query = f"EXT_ID:{pmid}"

276 else:

277 return None

278

279 # Get article metadata first

280 api_url = "https://www.ebi.ac.uk/europepmc/webservices/rest/search"

281 params = {

282 "query": query,

283 "format": "json",

284 "resultType": "core", # Get more detailed results

285 }

286

287 response = self.session.get(api_url, params=params, timeout=10)

288

289 if response.status_code == 200: 289 ↛ 324line 289 didn't jump to line 324 because the condition on line 289 was always true

290 data = response.json()

291 results = data.get("resultList", {}).get("result", [])

292

293 if results and results[0].get("isOpenAccess") == "Y":

294 article = results[0]

295 # Try to get full text XML

296 if article.get("pmcid"):

297 fulltext_url = f"https://www.ebi.ac.uk/europepmc/webservices/rest/{article['pmcid']}/fullTextXML"

298 text_response = self.session.get(

299 fulltext_url, timeout=30

300 )

301

302 if text_response.status_code == 200:

303 # Extract text from XML (simple approach - just get text content)

304 import re

305

306 xml_content = text_response.text

307 # Remove XML tags to get plain text

308 text = re.sub(r"<[^>]+>", " ", xml_content)

309 text = " ".join(text.split())

310

311 if text:

312 logger.info(

313 "Retrieved full text from Europe PMC API"

314 )

315 return text

316

317 except PolicyDeniedError:

318 raise

319 except Exception as e:

320 logger.debug(

321 f"Failed to fetch text from Europe PMC ({type(e).__name__})"

322 )

323

324 return None

325

326 def _apply_rate_limit(self):

327 """Apply rate limiting between requests."""

328 current_time = time.time()

329 time_since_last = current_time - self.last_request_time

330

331 if time_since_last < self.rate_limit_delay:

332 sleep_time = self.rate_limit_delay - time_since_last

333 logger.debug(f"Rate limiting: sleeping {sleep_time:.2f}s")

334 time.sleep(sleep_time)

335

336 self.last_request_time = time.time()

337

338 def _download_pmc_direct(self, url: str) -> Optional[bytes]:

339 """Download directly from PMC URL."""

340 pmc_match = re.search(r"(PMC\d+)", url)

341 if not pmc_match:

342 return None

343

344 pmc_id = pmc_match.group(1)

345 logger.info(f"Downloading PMC article: {pmc_id}")

346

347 # Try Europe PMC first (more reliable)

348 pdf_content = self._download_via_europe_pmc(pmc_id)

349 if pdf_content:

350 return pdf_content

351

352 # Fallback to NCBI PMC

353 return self._download_via_ncbi_pmc(pmc_id)

354

355 def _download_pubmed(self, url: str) -> Optional[bytes]:

356 """Download from PubMed URL."""

357 # Extract PMID

358 pmid_match = re.search(r"/(\d+)/?", url)

359 if not pmid_match:

360 return None

361

362 pmid = pmid_match.group(1)

363 logger.info(f"Processing PubMed article: {pmid}")

364

365 # Try Europe PMC API first

366 pdf_content = self._try_europe_pmc_api(pmid)

367 if pdf_content:

368 return pdf_content

369

370 # Try to find PMC ID via NCBI API

371 pmc_id = self._get_pmc_id_from_pmid(pmid)

372 if pmc_id:

373 logger.info(f"Found PMC ID: {pmc_id} for PMID: {pmid}")

374

375 # Try Europe PMC with PMC ID

376 pdf_content = self._download_via_europe_pmc(pmc_id)

377 if pdf_content: 377 ↛ 381line 377 didn't jump to line 381 because the condition on line 377 was always true

378 return pdf_content

379

380 # Try NCBI PMC

381 pdf_content = self._download_via_ncbi_pmc(pmc_id)

382 if pdf_content:

383 return pdf_content

384

385 logger.info(f"No PMC version available for PMID: {pmid}")

386 return None

387

388 def _download_europe_pmc(self, url: str) -> Optional[bytes]:

389 """Download from Europe PMC URL."""

390 # Extract PMC ID from URL

391 pmc_match = re.search(r"(PMC\d+)", url)

392 if pmc_match:

393 pmc_id = pmc_match.group(1)

394 return self._download_via_europe_pmc(pmc_id)

395 return None

396

397 def _try_europe_pmc_api(self, pmid: str) -> Optional[bytes]:

398 """Try downloading via Europe PMC API using PMID."""

399 try:

400 # Query Europe PMC API

401 api_url = "https://www.ebi.ac.uk/europepmc/webservices/rest/search"

402 params = {"query": f"EXT_ID:{pmid}", "format": "json"}

403

404 response = self.session.get(api_url, params=params, timeout=10)

405

406 if response.status_code == 200:

407 data = response.json()

408 results = data.get("resultList", {}).get("result", [])

409

410 if results:

411 article = results[0]

412 # Check if article has open access PDF

413 if (

414 article.get("isOpenAccess") == "Y"

415 and article.get("hasPDF") == "Y"

416 ):

417 pmcid = article.get("pmcid")

418 if pmcid:

419 logger.info(

420 f"Found open access PDF via Europe PMC API: {pmcid}"

421 )

422 return self._download_via_europe_pmc(pmcid)

423

424 except PolicyDeniedError:

425 raise

426 except Exception as e:

427 logger.debug(f"Europe PMC API query failed ({type(e).__name__})")

428

429 return None

430

431 def _get_pmc_id_from_pmid(self, pmid: str) -> Optional[str]:

432 """Convert PMID to PMC ID using NCBI E-utilities."""

433 try:

434 # Use NCBI E-utilities to find PMC ID

435 elink_url = (

436 "https://eutils.ncbi.nlm.nih.gov/entrez/eutils/elink.fcgi"

437 )

438 params = {

439 "dbfrom": "pubmed",

440 "db": "pmc",

441 "id": pmid,

442 "retmode": "json",

443 }

444

445 response = self.session.get(elink_url, params=params, timeout=10)

446

447 if response.status_code == 200:

448 data = response.json()

449 link_sets = data.get("linksets", [])

450

451 if link_sets and "linksetdbs" in link_sets[0]:

452 for linksetdb in link_sets[0]["linksetdbs"]:

453 if linksetdb.get("dbto") == "pmc" and linksetdb.get(

454 "links"

455 ):

456 pmc_id_num = linksetdb["links"][0]

457 return f"PMC{pmc_id_num}"

458

459 except PolicyDeniedError:

460 raise

461 except Exception as e:

462 logger.debug(f"NCBI E-utilities lookup failed ({type(e).__name__})")

463

464 # Fallback: Try scraping the PubMed page

465 try:

466 response = self.session.get(

467 f"https://pubmed.ncbi.nlm.nih.gov/{pmid}/", timeout=10

468 )

469

470 if response.status_code == 200:

471 pmc_match = re.search(r"PMC\d+", response.text)

472 if pmc_match: 472 ↛ 480line 472 didn't jump to line 480 because the condition on line 472 was always true

473 return pmc_match.group(0)

474

475 except PolicyDeniedError:

476 raise

477 except Exception as e:

478 logger.debug(f"PubMed page scraping failed ({type(e).__name__})")

479

480 return None

481

482 def _download_via_europe_pmc(self, pmc_id: str) -> Optional[bytes]:

483 """Download PDF via Europe PMC."""

484 # Europe PMC PDF URL

485 pdf_url = f"https://europepmc.org/backend/ptpmcrender.fcgi?accid={pmc_id}&blobtype=pdf"

486

487 logger.debug(f"Trying Europe PMC: {redact_url_for_log(pdf_url)}")

488 pdf_content = self._download_pdf(pdf_url)

489

490 if pdf_content:

491 logger.info(f"Successfully downloaded from Europe PMC: {pmc_id}")

492

493 return pdf_content

494

495 def _download_via_ncbi_pmc(self, pmc_id: str) -> Optional[bytes]:

496 """Download PDF via NCBI PMC."""

497 # Try different NCBI PMC URL patterns

498 url_patterns = [

499 f"https://www.ncbi.nlm.nih.gov/pmc/articles/{pmc_id}/pdf/",

500 f"https://www.ncbi.nlm.nih.gov/pmc/articles/{pmc_id}/pdf/main.pdf",

501 ]

502

503 for pdf_url in url_patterns:

504 logger.debug(f"Trying NCBI PMC: {redact_url_for_log(pdf_url)}")

505

506 # Add referer header for NCBI

507 headers = {

508 "Referer": f"https://www.ncbi.nlm.nih.gov/pmc/articles/{pmc_id}/"

509 }

510

511 pdf_content = self._download_pdf(pdf_url, headers)

512 if pdf_content:

513 logger.info(f"Successfully downloaded from NCBI PMC: {pmc_id}")

514 return pdf_content

515

516 return None