Coverage for src/local_deep_research/research_library/services/download_service.py: 94%
783 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-06 15:42 +0000
1"""
2PDF Download Service for Research Library
4Handles downloading PDFs from various academic sources with:
5- Deduplication using download tracker
6- Source-specific download strategies (arXiv, PubMed, etc.)
7- Progress tracking and error handling
8- File organization and storage
9"""
11import hashlib
12import os
13import re
14import time
15import uuid
16from datetime import datetime, UTC
17from pathlib import Path
18from typing import Any, Dict, Optional, Tuple
19from urllib.parse import urlparse
21import requests
22from loguru import logger
23from sqlalchemy.orm import Session
24import pdfplumber
25from pypdf import PdfReader
27from ...utilities.type_utils import unwrap_setting
28from ...constants import FILE_PATH_SENTINELS, FILE_PATH_TEXT_ONLY
29from ...database.models.download_tracker import (
30 DownloadAttempt,
31 DownloadTracker,
32)
34# SSRF chokepoint: every outbound fetch in this module goes through
35# `safe_get`, which validates via ssrf_validator (raising on failure,
36# re-validating after redirects, always blocking cloud-metadata
37# addresses). `requests` is imported above for its exception types only
38# — do not add a raw requests.get()/post() here. GHSA-9c54 was exactly
39# that bypass, and `is_downloadable_domain` upstream is a relevance
40# filter, not a security control.
41from ...security import safe_get, sanitize_error_for_client
42from ...database.models.library import (
43 Collection,
44 Document as Document,
45 DocumentStatus,
46 DownloadQueue as LibraryDownloadQueue,
47)
48from .pdf_storage_manager import (
49 DEFAULT_MAX_PDF_SIZE_MB,
50 PDFStorageManager,
51 resolve_pdf_storage_mode,
52)
53from ...database.models.research import ResearchResource
54from ...database.library_init import get_source_type_id, get_default_library_id
55from ...database.session_context import get_user_db_session, safe_rollback
56from ...utilities.db_utils import get_settings_manager
57from ...library.download_management import RetryManager
58from ...config.paths import get_library_directory
59from ..utils import (
60 apply_user_subdir,
61 ensure_in_collection,
62 get_document_for_resource,
63 get_url_hash,
64 get_absolute_path_from_settings,
65 is_downloadable_url,
66)
68# Import our modular downloaders
69from ..downloaders import (
70 ContentType,
71 ArxivDownloader,
72 PubMedDownloader,
73 BioRxivDownloader,
74 DirectPDFDownloader,
75 SemanticScholarDownloader,
76 OpenAlexDownloader,
77 GenericDownloader,
78)
79from ...constants import DEFAULT_SEARCH_TOOL
82class DownloadService:
83 """Service for downloading and managing research PDFs."""
85 def __init__(
86 self,
87 username: str,
88 password: Optional[str] = None,
89 settings_snapshot: Optional[Dict[str, Any]] = None,
90 ):
91 """Initialize download service for a user.
93 Args:
94 username: The username to download for
95 password: Optional password for encrypted database access
96 settings_snapshot: Optional snapshot used to build the
97 EgressContext for per-URL scope gating. When None,
98 downloads run without scope enforcement (back-compat
99 for the existing caller; new callers must pass it).
100 """
101 self.username = username
102 self.password = password
103 self.settings = get_settings_manager(username=username)
104 self._closed = False
105 # When no snapshot is passed (e.g. the library download routes, which
106 # construct DownloadService(username, password) directly), build one
107 # from this user's settings so egress policy is still enforced.
108 # Without this, _egress_context stays None and _check_url_against_policy
109 # returns (True, "no_context") — every download ungated. Best-effort:
110 # on failure, fall through to the pre-policy behavior.
111 # A real backend error (get_settings_snapshot raised) is different
112 # from a test double returning a non-dict: under a private scope we
113 # must NOT silently fall through to ungated downloads when we simply
114 # failed to READ the user's settings. Track that case and fail closed.
115 self._snapshot_build_failed = False
116 if settings_snapshot is None:
117 try:
118 built = self.settings.get_settings_snapshot()
119 # Only adopt a real dict — a test double or unavailable
120 # backend hands back something else; treat that as "no
121 # snapshot" (ungated back-compat) rather than letting the
122 # policy evaluation fail closed on a non-dict.
123 if isinstance(built, dict):
124 settings_snapshot = built
125 except Exception:
126 # The user HAS a settings backend (self.settings exists) but
127 # reading it raised — a private scope could be configured and
128 # we just couldn't see it. Fail closed rather than ungate.
129 logger.bind(policy_audit=True).warning(
130 "DownloadService: settings snapshot build raised; "
131 "locking downloads closed (fail-safe)"
132 )
133 self._snapshot_build_failed = True
134 self._settings_snapshot = settings_snapshot
135 self._egress_context = self._build_egress_context(settings_snapshot)
137 # Debug settings manager and user context
138 logger.info(
139 f"[DOWNLOAD_SERVICE] Settings manager initialized: {type(self.settings)}, username: {self.username}"
140 )
142 # Get library path from settings (uses centralized path, respects LDR_DATA_DIR)
143 storage_path_setting = self.settings.get_setting(
144 "research_library.storage_path",
145 str(get_library_directory()),
146 )
147 logger.debug(
148 f"[DOWNLOAD_SERVICE_INIT] Storage path setting retrieved: {storage_path_setting} (type: {type(storage_path_setting)})"
149 )
151 if storage_path_setting is None:
152 logger.error(
153 "[DOWNLOAD_SERVICE_INIT] CRITICAL: storage_path_setting is None!"
154 )
155 raise ValueError("Storage path setting cannot be None")
157 # Per-user library root (issue #5521). PDFs are written to a per-user
158 # subdirectory (<base>/<username>) instead of the shared base, so two
159 # users' per-user autoincrement resource ids can't collide on the same
160 # `pdfs/<resource_id>.pdf` path in one shared directory. This mirrors
161 # get_library_storage_path()'s layout but derives the base from this
162 # user's own settings manager (self.settings) so it stays correct in
163 # background/scheduler threads that have no request context.
164 # `legacy_library_root` is the old shared base: it is never written to,
165 # only used as a READ fallback so PDFs downloaded before per-user
166 # isolation still load (no blind bulk file move — see PR notes).
167 base_library_root = (
168 Path(os.path.expandvars(storage_path_setting))
169 .expanduser()
170 .resolve()
171 )
172 shared_library = self.settings.get_setting(
173 "research_library.shared_library", False
174 )
175 self.legacy_library_root = str(base_library_root)
176 self.library_root = str(
177 apply_user_subdir(base_library_root, self.username, shared_library)
178 )
179 logger.debug(
180 f"[DOWNLOAD_SERVICE_INIT] Library root resolved to: {self.library_root} "
181 f"(legacy shared root: {self.legacy_library_root})"
182 )
184 # Create directory structure
185 self._setup_directories()
187 # Initialize modular downloaders
188 # DirectPDFDownloader first for efficiency with direct PDF links
190 # Get Semantic Scholar API key from settings
191 semantic_scholar_api_key = self.settings.get_setting(
192 "search.engine.web.semantic_scholar.api_key", ""
193 )
195 self.downloaders = [
196 DirectPDFDownloader(timeout=30), # Handle direct PDF links first
197 SemanticScholarDownloader(
198 timeout=30,
199 api_key=semantic_scholar_api_key
200 if semantic_scholar_api_key
201 else None,
202 ),
203 OpenAlexDownloader(
204 timeout=30
205 ), # OpenAlex with API lookup (no key needed)
206 ArxivDownloader(timeout=30),
207 PubMedDownloader(timeout=30, rate_limit_delay=1.0),
208 BioRxivDownloader(timeout=30),
209 GenericDownloader(timeout=30), # Generic should be last (fallback)
210 ]
212 # Initialize retry manager for smart failure tracking
213 self.retry_manager = RetryManager(username, password)
214 logger.info(
215 f"[DOWNLOAD_SERVICE] Initialized retry manager for user: {username}"
216 )
218 # PubMed rate limiting state
219 self._pubmed_delay = 1.0 # 1 second delay for PubMed
220 self._last_pubmed_request = 0.0 # Track last request time
222 def close(self):
223 """Close all downloader resources."""
224 if self._closed:
225 return
226 self._closed = True
228 from ...utilities.resource_utils import safe_close
230 for downloader in self.downloaders:
231 safe_close(downloader, "downloader")
233 # Close the settings manager's DB session to return the connection
234 # to the pool. SettingsManager.close() is idempotent.
235 safe_close(self.settings, "settings manager", allow_none=True)
237 # Clear references to allow garbage collection
238 self.downloaders = []
239 self.retry_manager = None
240 self.settings = None
242 def __enter__(self):
243 """Enter context manager."""
244 return self
246 def __exit__(self, exc_type, exc_val, exc_tb):
247 """Exit context manager, ensuring cleanup."""
248 self.close()
249 return False
251 def _setup_directories(self):
252 """Create library directory structure."""
253 from ...security.directory_creation import create_directory
255 # Only create the root and pdfs folder - flat structure.
256 paths = [
257 self.library_root,
258 str(Path(self.library_root) / "pdfs"),
259 ]
260 for path in paths:
261 create_directory(
262 path,
263 context="download service library directory",
264 )
266 def _normalize_url(self, url: str) -> str:
267 """Normalize URL for consistent hashing."""
268 # Remove protocol variations
269 url = re.sub(r"^https?://", "", url)
270 # Remove www
271 url = re.sub(r"^www\.", "", url)
272 # Remove trailing slashes
273 url = url.rstrip("/")
274 # Sort query parameters
275 if "?" in url:
276 base, query = url.split("?", 1)
277 params = sorted(query.split("&"))
278 url = f"{base}?{'&'.join(params)}"
279 return url.lower()
281 def _get_url_hash(self, url: str) -> str:
282 """Generate SHA256 hash of normalized URL."""
283 normalized = self._normalize_url(url)
284 return get_url_hash(normalized)
286 def _build_egress_context(self, settings_snapshot):
287 """Build an EgressContext from the supplied snapshot. Returns
288 None when no snapshot is available — callers fall through to
289 the pre-policy behavior (back-compat with non-scheduler callers).
291 Sets ``self._policy_locked = True`` when a snapshot WAS supplied
292 but the policy itself cannot be evaluated (corrupt scope
293 value). The check_url method honors
294 this flag and fails closed — the previous code returned None on
295 PolicyDeniedError and check_url then returned ``(True,
296 "no_context")``, which silently allowed every download under a
297 misconfigured policy.
298 """
299 self._policy_locked = False
300 # Use an explicit `is None` check, not a truthiness test: an empty
301 # dict {} is a real (if minimal) snapshot, and `if not {}` would
302 # silently skip policy gating for it — fail-open. Match the
303 # `is None` guard used by the LLM/embeddings gates.
304 if settings_snapshot is None:
305 return None
306 try:
307 from ...security.egress.policy import (
308 PolicyDeniedError,
309 context_from_snapshot,
310 )
311 except ImportError:
312 logger.debug(
313 "egress_policy unavailable in DownloadService; "
314 "downloads will not be scope-gated"
315 )
316 return None
318 primary_raw = unwrap_setting(
319 settings_snapshot.get("search.tool", DEFAULT_SEARCH_TOOL)
320 )
321 try:
322 return context_from_snapshot(
323 settings_snapshot,
324 primary_raw or DEFAULT_SEARCH_TOOL,
325 username=self.username,
326 )
327 except (PolicyDeniedError, ValueError) as exc:
328 logger.bind(policy_audit=True).warning(
329 "DownloadService policy unavailable; locking out "
330 "every URL check",
331 reason=str(exc),
332 )
333 self._policy_locked = True
334 return None
336 def _check_url_against_policy(self, url: str) -> Tuple[bool, str]:
337 """Run evaluate_url against self._egress_context. Returns
338 (allowed, reason). When no context is configured (legacy
339 callers OR tests that mock __init__), returns (True, "no_context")
340 so back-compat callers aren't broken — but new callers passing
341 a snapshot DO get gated. Logged to policy_audit on denial.
342 """
343 # ``_policy_locked`` is set by _build_egress_context when a
344 # snapshot was supplied but the policy itself raised. Without
345 # this short-circuit, _egress_context is None and the legacy
346 # fall-through below would silently allow every URL.
347 if getattr(self, "_policy_locked", False): 347 ↛ 348line 347 didn't jump to line 348 because the condition on line 347 was never true
348 return False, "policy_unavailable"
350 # Reading the user's settings raised during __init__ — we could not
351 # determine the scope, so fail closed instead of ungating downloads.
352 if getattr(self, "_snapshot_build_failed", False): 352 ↛ 353line 352 didn't jump to line 353 because the condition on line 352 was never true
353 return False, "settings_unavailable"
355 # Use getattr so tests that mock DownloadService.__init__ (and
356 # therefore never set _egress_context) still work — they
357 # operated under the pre-policy contract.
358 egress_context = getattr(self, "_egress_context", None)
359 if egress_context is None:
360 return True, "no_context"
361 from ...security.egress.policy import evaluate_url
362 from ...security.ssrf_validator import redact_url_for_log
364 decision = evaluate_url(url, egress_context)
365 if not decision.allowed:
366 logger.bind(policy_audit=True).warning(
367 "DownloadService URL denied by egress policy",
368 # Redact userinfo creds / query-string tokens from the log.
369 url=redact_url_for_log(url),
370 scope=egress_context.scope.value,
371 reason=decision.reason,
372 )
373 return decision.allowed, decision.reason
375 def is_already_downloaded(self, url: str) -> Tuple[bool, Optional[str]]:
376 """
377 Check if URL is already downloaded.
379 Returns:
380 Tuple of (is_downloaded, file_path)
381 """
382 url_hash = self._get_url_hash(url)
384 with get_user_db_session(self.username, self.password) as session:
385 tracker = (
386 session.query(DownloadTracker)
387 .filter_by(url_hash=url_hash, is_downloaded=True)
388 .first()
389 )
391 if tracker and tracker.file_path:
392 # Compute absolute path and verify file still exists. Pass the
393 # username so per-user files resolve, with legacy-shared
394 # fallback for pre-isolation downloads (issue #5521). Pass
395 # self.settings so storage_path/shared_library are read from
396 # this user's captured manager — the ambient one is db-less in
397 # background/scheduler threads and would resolve the default
398 # root, missing a user's UI-customized storage_path. Use
399 # getattr: __init__ always sets self.settings in production
400 # (incl. scheduler threads), but test doubles that construct
401 # a DownloadService via __new__/partial mocks may not — fall
402 # back to None (the ambient get_settings_manager()) rather
403 # than raising AttributeError.
404 absolute_path = get_absolute_path_from_settings(
405 tracker.file_path,
406 self.username,
407 settings_manager=getattr(self, "settings", None),
408 )
409 if absolute_path and absolute_path.is_file():
410 return True, str(absolute_path)
411 if absolute_path:
412 # File was deleted, mark as not downloaded
413 tracker.is_downloaded = False
414 session.commit()
415 # If absolute_path is None, path was blocked - treat as not downloaded
417 return False, None
419 def get_text_content(self, resource_id: int) -> Optional[str]:
420 """
421 Get text content for a research resource.
423 This will try to:
424 1. Fetch text directly from APIs if available
425 2. Extract text from downloaded PDF if exists
426 3. Download PDF and extract text if not yet downloaded
428 Args:
429 resource_id: ID of the research resource
431 Returns:
432 Text content as string, or None if extraction failed
433 """
434 with get_user_db_session(self.username, self.password) as session:
435 resource = session.get(ResearchResource, resource_id)
436 if not resource:
437 logger.error(f"Resource {resource_id} not found")
438 return None
440 url = resource.url
442 # Egress policy gate — denied URLs short-circuit before any
443 # downloader fires a request.
444 allowed, reason = self._check_url_against_policy(url)
445 if not allowed: 445 ↛ 446line 445 didn't jump to line 446 because the condition on line 445 was never true
446 logger.warning(
447 f"Skipping text extraction for {url}: refused by "
448 f"egress policy ({reason})"
449 )
450 return None
452 # Find appropriate downloader
453 for downloader in self.downloaders:
454 if downloader.can_handle(url):
455 logger.info(
456 f"Using {downloader.__class__.__name__} for text extraction from {url}"
457 )
458 try:
459 # Try to get text content
460 text = downloader.download_text(url)
461 if text:
462 logger.info(
463 f"Successfully extracted text for: {resource.title[:50]}"
464 )
465 return text
466 except Exception:
467 logger.exception("Failed to extract text")
468 break
470 logger.warning(f"Could not extract text for {url}")
471 return None
473 def queue_research_downloads(
474 self, research_id: str, collection_id: Optional[str] = None
475 ) -> int:
476 """
477 Queue all downloadable PDFs from a research session.
479 Args:
480 research_id: The research session ID
481 collection_id: Optional target collection ID (defaults to Library if not provided)
483 Returns:
484 Number of items queued
486 Notes:
487 Resets existing FAILED/COMPLETED queue entries for the research back
488 to PENDING, effectively retrying them. Resources that already have a
489 PENDING queue entry or a COMPLETED Document are skipped. As of #4685,
490 ``download_bulk`` calls this unconditionally on every bulk run, so
491 previously-failed downloads are automatically retried each time.
492 A PROCESSING entry is in flight in another ``download_bulk`` stream,
493 so it is left alone rather than reset: resetting it would let this
494 run re-claim and re-download it (#4691).
495 """
496 queued = 0
498 # Get default library collection if no collection_id provided
499 if not collection_id:
500 from ...database.library_init import get_default_library_id
502 collection_id = get_default_library_id(self.username, self.password)
504 with get_user_db_session(self.username, self.password) as session:
505 # Get all resources for this research
506 resources = (
507 session.query(ResearchResource)
508 .filter_by(research_id=research_id)
509 .all()
510 )
512 for resource in resources:
513 if self._is_downloadable(resource):
514 # Library resources linked via document_id are already done
515 if resource.document_id:
516 continue
518 # Egress policy gate before queueing.
519 allowed, reason = self._check_url_against_policy(
520 resource.url
521 )
522 if not allowed: 522 ↛ 523line 522 didn't jump to line 523 because the condition on line 522 was never true
523 logger.warning(
524 f"Skipping queue for {resource.url}: refused by "
525 f"egress policy ({reason})"
526 )
527 continue
529 # Check if already queued
530 existing_queue = (
531 session.query(LibraryDownloadQueue)
532 .filter_by(
533 resource_id=resource.id,
534 status=DocumentStatus.PENDING,
535 )
536 .first()
537 )
539 # Check if already downloaded (trust the database status)
540 existing_doc = (
541 session.query(Document)
542 .filter_by(
543 resource_id=resource.id,
544 status=DocumentStatus.COMPLETED,
545 )
546 .first()
547 )
549 # Queue if not already queued and not marked as completed
550 if not existing_queue and not existing_doc: 550 ↛ 512line 550 didn't jump to line 512 because the condition on line 550 was always true
551 # Check one more time if ANY queue entry exists (regardless of status)
552 any_queue = (
553 session.query(LibraryDownloadQueue)
554 .filter_by(resource_id=resource.id)
555 .first()
556 )
558 if any_queue:
559 # Reset the existing queue entry, but never one a
560 # concurrent download_bulk stream has claimed: a
561 # PROCESSING row is in flight, and resetting it to
562 # PENDING lets this stream re-claim and re-download
563 # a resource the other stream is still downloading
564 # (issue #4691). The status guard lives inside the
565 # UPDATE, not in an `if` above it, so the check and
566 # the write cannot straddle another stream's claim.
567 reset = (
568 session.query(LibraryDownloadQueue)
569 .filter(
570 LibraryDownloadQueue.id == any_queue.id,
571 LibraryDownloadQueue.status
572 != DocumentStatus.PROCESSING,
573 )
574 .update(
575 {
576 LibraryDownloadQueue.status: DocumentStatus.PENDING,
577 LibraryDownloadQueue.research_id: research_id,
578 LibraryDownloadQueue.collection_id: collection_id,
579 },
580 synchronize_session=False,
581 )
582 )
583 if reset:
584 queued += 1
585 else:
586 # Add new queue entry
587 queue_entry = LibraryDownloadQueue(
588 resource_id=resource.id,
589 research_id=research_id,
590 collection_id=collection_id,
591 priority=0,
592 status=DocumentStatus.PENDING,
593 )
594 session.add(queue_entry)
595 queued += 1
597 session.commit()
598 logger.info(
599 f"Queued {queued} downloads for research {research_id} to collection {collection_id}"
600 )
602 return queued
604 def _is_downloadable(self, resource: ResearchResource) -> bool:
605 """Check if a resource is likely downloadable as PDF.
607 Delegates to the consolidated is_downloadable_url() from utils.
608 """
609 return is_downloadable_url(resource.url)
611 def download_resource(self, resource_id: int) -> Tuple[bool, Optional[str]]:
612 """
613 Download a specific resource.
615 Returns:
616 Tuple of (success: bool, skip_reason: str or None)
617 """
618 with get_user_db_session(self.username, self.password) as session:
619 resource = session.get(ResearchResource, resource_id)
620 if not resource:
621 logger.error(f"Resource {resource_id} not found")
622 return False, "Resource not found"
624 # Check if already downloaded (trust the database after sync)
625 existing_doc = (
626 session.query(Document)
627 .filter_by(
628 resource_id=resource_id, status=DocumentStatus.COMPLETED
629 )
630 .first()
631 )
633 if existing_doc:
634 logger.info(
635 "Resource already downloaded (according to database)"
636 )
637 return True, None
639 # Get collection_id from queue entry if it exists
640 queue_entry = (
641 session.query(LibraryDownloadQueue)
642 .filter_by(resource_id=resource_id)
643 .first()
644 )
645 collection_id = (
646 queue_entry.collection_id
647 if queue_entry and queue_entry.collection_id
648 else None
649 )
651 # Create download tracker entry
652 url_hash = self._get_url_hash(resource.url)
653 tracker = (
654 session.query(DownloadTracker)
655 .filter_by(url_hash=url_hash)
656 .first()
657 )
659 if not tracker:
660 tracker = DownloadTracker(
661 url=resource.url,
662 url_hash=url_hash,
663 first_resource_id=resource.id,
664 is_downloaded=False,
665 )
666 session.add(tracker)
667 session.commit()
669 # Attempt download
670 success, skip_reason, status_code = self._download_pdf(
671 resource, tracker, session, collection_id
672 )
674 # Record attempt with retry manager for smart failure tracking
675 self.retry_manager.record_attempt(
676 resource_id=resource.id,
677 result=(success, skip_reason),
678 status_code=status_code,
679 url=resource.url,
680 details=skip_reason
681 or (
682 "Successfully downloaded" if success else "Download failed"
683 ),
684 session=session,
685 )
687 # Update queue status if exists
688 queue_entry = (
689 session.query(LibraryDownloadQueue)
690 .filter_by(resource_id=resource_id)
691 .first()
692 )
694 if queue_entry:
695 queue_entry.status = (
696 DocumentStatus.COMPLETED
697 if success
698 else DocumentStatus.FAILED
699 )
700 queue_entry.completed_at = datetime.now(UTC)
702 session.commit()
704 # Trigger auto-indexing for successfully downloaded documents
705 if success and self.password:
706 try:
707 from ...web.routers.rag import trigger_auto_index
708 from ...database.library_init import get_default_library_id
710 # Get the document that was just created
711 doc = (
712 session.query(Document)
713 .filter_by(resource_id=resource_id)
714 .order_by(Document.created_at.desc())
715 .first()
716 )
717 if doc:
718 # Use collection_id from queue entry or default Library
719 # NB: pass username string, not the SQLAlchemy session
720 target_collection = (
721 collection_id
722 or get_default_library_id(
723 self.username, self.password
724 )
725 )
726 if target_collection: 726 ↛ anywhereline 726 didn't jump anywhere: it always raised an exception.
727 trigger_auto_index(
728 [doc.id],
729 target_collection,
730 self.username,
731 self.password,
732 )
733 except Exception:
734 # The Document SELECT above runs on the shared session; a
735 # connection-level failure there leaves it needing a
736 # rollback. Auto-indexing is best-effort so we swallow and
737 # return — the with-block exits normally and
738 # get_user_db_session's rollback never fires. Recover the
739 # session here so a later op on it (same request/thread)
740 # doesn't cascade. (No-op for the other failures reachable
741 # here: get_default_library_id self-heals via its own inner
742 # get_user_db_session block, and trigger_auto_index does its
743 # DB work off-thread, so neither dirties this session.)
744 safe_rollback(session, "download_resource auto-index")
745 logger.exception("Failed to trigger auto-indexing")
747 return success, skip_reason
749 def _download_pdf(
750 self,
751 resource: ResearchResource,
752 tracker: DownloadTracker,
753 session: Session,
754 collection_id: Optional[str] = None,
755 ) -> Tuple[bool, Optional[str], Optional[int]]:
756 """
757 Perform the actual PDF download.
759 Args:
760 resource: The research resource to download
761 tracker: Download tracker for this URL
762 session: Database session
763 collection_id: Optional target collection ID (defaults to Library if not provided)
765 Returns:
766 Tuple of (success: bool, skip_reason: Optional[str], status_code: Optional[int])
767 """
768 url = resource.url
770 # Egress policy gate at the network-fire point. The entry caller
771 # (download_resource) does not gate, so this is the true PEP — a
772 # denied URL must never reach a downloader regardless of route.
773 allowed, reason = self._check_url_against_policy(url)
774 if not allowed:
775 logger.warning(
776 f"Skipping PDF download for {url}: refused by egress "
777 f"policy ({reason})"
778 )
779 return False, f"egress_policy_denied:{reason}", None
781 # Log attempt
782 attempt = DownloadAttempt(
783 url_hash=tracker.url_hash,
784 attempt_number=tracker.download_attempts.count() + 1
785 if hasattr(tracker, "download_attempts")
786 else 1,
787 attempted_at=datetime.now(UTC),
788 )
789 session.add(attempt)
791 try:
792 # Use modular downloaders with detailed skip reasons
793 pdf_content = None
794 downloader_used = None
795 skip_reason = None
796 status_code = None
798 for downloader in self.downloaders:
799 if downloader.can_handle(url):
800 logger.info(
801 f"Using {downloader.__class__.__name__} for {url}"
802 )
803 result = downloader.download_with_result(
804 url, ContentType.PDF
805 )
806 downloader_used = downloader.__class__.__name__
808 if result.is_success and result.content:
809 pdf_content = result.content
810 status_code = result.status_code
811 break
812 if result.skip_reason: 812 ↛ 798line 812 didn't jump to line 798 because the condition on line 812 was always true
813 skip_reason = result.skip_reason
814 status_code = result.status_code
815 logger.info(f"Download skipped: {skip_reason}")
816 # Keep trying other downloaders unless it's the GenericDownloader
817 if isinstance(downloader, GenericDownloader):
818 break
820 if not downloader_used:
821 logger.error(f"No downloader found for {url}")
822 skip_reason = "No compatible downloader available"
824 if not pdf_content:
825 error_msg = skip_reason or "Failed to download PDF content"
826 # Store skip reason in attempt for retrieval
827 attempt.error_message = error_msg
828 attempt.succeeded = False
829 session.commit()
830 logger.info(f"Download failed with reason: {error_msg}")
831 return False, error_msg, status_code
833 # Get PDF storage mode setting. Resolve through the operator gate
834 # BEFORE any PDF is written: a stored value or an
835 # LDR_RESEARCH_LIBRARY_PDF_STORAGE_MODE env override could still
836 # read "filesystem" even though the settings UI withholds that
837 # option, so the unencrypted mode is coerced to the encrypted
838 # "database" default here unless the operator opted in. This is
839 # the true enforcement point — hiding the dropdown alone is not
840 # enough.
841 pdf_storage_mode = resolve_pdf_storage_mode(
842 self.settings.get_setting(
843 "research_library.pdf_storage_mode", "none"
844 )
845 )
846 max_pdf_size_mb = int(
847 self.settings.get_setting(
848 "research_library.max_pdf_size_mb",
849 DEFAULT_MAX_PDF_SIZE_MB,
850 )
851 )
852 logger.info(
853 f"[DOWNLOAD_SERVICE] PDF storage mode: {pdf_storage_mode}"
854 )
856 # Update tracker
857 import hashlib
859 tracker.file_hash = hashlib.sha256(pdf_content).hexdigest()
860 tracker.file_size = len(pdf_content)
861 tracker.is_downloaded = True
862 tracker.downloaded_at = datetime.now(UTC)
864 # Initialize PDF storage manager. Writes go to the per-user root;
865 # legacy_root gives read fallback for pre-isolation files (#5521).
866 pdf_storage_manager = PDFStorageManager(
867 library_root=self.library_root,
868 storage_mode=pdf_storage_mode,
869 max_pdf_size_mb=max_pdf_size_mb,
870 legacy_root=self.legacy_library_root,
871 )
873 # Update attempt with success info
874 attempt.succeeded = True
876 # Check if library document already exists
877 existing_doc = get_document_for_resource(session, resource)
879 if existing_doc:
880 # Update existing document. Only replace document_hash when
881 # transitioning from FAILED — that's the placeholder hash from
882 # _record_failed_text_extraction. For any other prior state
883 # the hash is already a real content hash and clobbering it
884 # risks UNIQUE-constraint collisions (issue #3827).
885 was_failed = existing_doc.status == DocumentStatus.FAILED
886 if was_failed:
887 existing_doc.document_hash = tracker.file_hash
888 existing_doc.file_size = len(pdf_content)
889 existing_doc.status = DocumentStatus.COMPLETED
890 existing_doc.processed_at = datetime.now(UTC)
892 # Save PDF using storage manager (updates storage_mode and file_path)
893 file_path_result, _ = pdf_storage_manager.save_pdf(
894 pdf_content=pdf_content,
895 document=existing_doc,
896 session=session,
897 filename=f"{resource.id}.pdf",
898 url=url,
899 resource_id=resource.id,
900 )
902 # Update tracker
903 tracker.file_path = (
904 file_path_result if file_path_result else None
905 )
906 tracker.file_name = (
907 Path(file_path_result).name
908 if file_path_result and file_path_result != "database"
909 else None
910 )
911 else:
912 # Get source type ID for research downloads
913 try:
914 source_type_id = get_source_type_id(
915 self.username, "research_download", self.password
916 )
917 # Use provided collection_id or default to Library
918 library_collection_id = (
919 collection_id
920 or get_default_library_id(self.username, self.password)
921 )
922 except Exception:
923 logger.exception(
924 "Failed to get source type or library collection"
925 )
926 raise
928 # Create new unified document entry
929 doc_id = str(uuid.uuid4())
930 doc = Document(
931 id=doc_id,
932 source_type_id=source_type_id,
933 resource_id=resource.id,
934 research_id=resource.research_id,
935 document_hash=tracker.file_hash,
936 original_url=url,
937 file_size=len(pdf_content),
938 file_type="pdf",
939 mime_type="application/pdf",
940 title=resource.title,
941 status=DocumentStatus.COMPLETED,
942 processed_at=datetime.now(UTC),
943 storage_mode=pdf_storage_mode,
944 )
945 session.add(doc)
946 session.flush() # Ensure doc.id is available for blob storage
948 # Save PDF using storage manager (updates storage_mode and file_path)
949 file_path_result, _ = pdf_storage_manager.save_pdf(
950 pdf_content=pdf_content,
951 document=doc,
952 session=session,
953 filename=f"{resource.id}.pdf",
954 url=url,
955 resource_id=resource.id,
956 )
958 # Update tracker
959 tracker.file_path = (
960 file_path_result if file_path_result else None
961 )
962 tracker.file_name = (
963 Path(file_path_result).name
964 if file_path_result and file_path_result != "database"
965 else None
966 )
968 # Link document to default Library collection
969 ensure_in_collection(session, doc_id, library_collection_id)
971 # Update attempt
972 attempt.succeeded = True
973 attempt.bytes_downloaded = len(pdf_content)
975 if pdf_storage_mode == "database":
976 logger.info(
977 f"Successfully stored PDF in database: {resource.url}"
978 )
979 elif pdf_storage_mode == "filesystem":
980 logger.info(f"Successfully downloaded: {tracker.file_path}")
981 else:
982 logger.info(f"Successfully extracted text from: {resource.url}")
984 # Automatically extract and save text after successful PDF download
985 try:
986 logger.info(
987 f"Extracting text from downloaded PDF for: {resource.title[:50]}"
988 )
989 text = self._extract_text_from_pdf(pdf_content)
991 if text:
992 # Get the document ID we just created/updated
993 pdf_doc = get_document_for_resource(session, resource)
994 pdf_document_id = pdf_doc.id if pdf_doc else None
996 # Save text to encrypted database
997 self._save_text_with_db(
998 resource=resource,
999 text=text,
1000 session=session,
1001 extraction_method="pdf_extraction",
1002 extraction_source="local_pdf",
1003 pdf_document_id=pdf_document_id,
1004 )
1005 logger.info(
1006 f"Successfully extracted and saved text for: {resource.title[:50]}"
1007 )
1008 else:
1009 logger.warning(
1010 f"Text extraction returned empty text for: {resource.title[:50]}"
1011 )
1012 except Exception:
1013 logger.exception(
1014 "Failed to extract text from PDF, but PDF download succeeded"
1015 )
1016 # Don't fail the entire download if text extraction fails
1018 return True, None, status_code
1020 except Exception as e:
1021 logger.exception(f"Download failed for {url}")
1022 # The error may have come from a failed flush/commit in the
1023 # success path above (e.g. save_pdf / ensure_in_collection),
1024 # leaving the *shared* session in PendingRollbackError. The caller
1025 # (download_resource) keeps using this same session and commits
1026 # again, so roll back here or that commit cascades. This swallow-
1027 # and-return path is invisible to get_user_db_session's own
1028 # rollback (we don't propagate), hence the explicit recovery.
1029 safe_rollback(session, "_download_pdf")
1030 tracker.is_accessible = False
1031 # request/network errors can echo a resource URL carrying an
1032 # api_key/token or user:pass@host — scrub before returning. The
1033 # message is surfaced to the browser via the download SSE stream.
1034 safe_error = sanitize_error_for_client(str(e))
1035 # The rollback discarded the pending ``attempt`` row added above,
1036 # so re-record the failed attempt on the now-clean session for
1037 # download_resource to commit.
1038 session.add(
1039 DownloadAttempt(
1040 url_hash=tracker.url_hash,
1041 attempt_number=tracker.download_attempts.count() + 1
1042 if hasattr(tracker, "download_attempts")
1043 else 1,
1044 attempted_at=datetime.now(UTC),
1045 succeeded=False,
1046 error_type=type(e).__name__,
1047 error_message=safe_error,
1048 )
1049 )
1050 return False, safe_error, None
1052 def _extract_text_from_pdf(self, pdf_content: bytes) -> Optional[str]:
1053 """
1054 Extract text from PDF content using multiple methods for best results.
1056 Args:
1057 pdf_content: Raw PDF bytes
1059 Returns:
1060 Extracted text or None if extraction fails
1061 """
1062 try:
1063 # First try with pdfplumber (better for complex layouts)
1064 import io
1066 with pdfplumber.open(io.BytesIO(pdf_content)) as pdf:
1067 text_parts = []
1068 for page in pdf.pages:
1069 page_text = page.extract_text()
1070 if page_text:
1071 text_parts.append(page_text)
1073 if text_parts:
1074 return "\n\n".join(text_parts)
1076 # Fallback to PyPDF if pdfplumber fails
1077 reader = PdfReader(io.BytesIO(pdf_content))
1078 text_parts = []
1079 for page in reader.pages:
1080 text = page.extract_text()
1081 if text:
1082 text_parts.append(text)
1084 if text_parts:
1085 return "\n\n".join(text_parts)
1087 logger.warning("No text could be extracted from PDF")
1088 return None
1090 except Exception:
1091 logger.exception("Failed to extract text from PDF")
1092 return None
1094 def download_as_text(self, resource_id: int) -> Tuple[bool, Optional[str]]:
1095 """
1096 Download resource and extract text to encrypted database.
1098 Args:
1099 resource_id: ID of the resource to download
1101 Returns:
1102 Tuple of (success, error_message)
1103 """
1104 with get_user_db_session(self.username, self.password) as session:
1105 # Get the resource
1106 resource = (
1107 session.query(ResearchResource)
1108 .filter_by(id=resource_id)
1109 .first()
1110 )
1111 if not resource:
1112 return False, "Resource not found"
1114 # Handle library resources — content already in the database
1115 # (local-only, no network — skip retry checks)
1116 if resource.source_type == "library" or (
1117 resource.url and resource.url.startswith("/library/document/")
1118 ):
1119 return self._try_library_text_extraction(session, resource)
1121 # Try existing text in database
1122 result = self._try_existing_text(session, resource_id)
1123 if result is not None:
1124 return result
1126 # Try legacy text files on disk
1127 result = self._try_legacy_text_file(session, resource, resource_id)
1128 if result is not None: 1128 ↛ 1129line 1128 didn't jump to line 1129 because the condition on line 1128 was never true
1129 return result
1131 # Try extracting from existing PDF
1132 result = self._try_existing_pdf_extraction(
1133 session, resource, resource_id
1134 )
1135 if result is not None: 1135 ↛ 1136line 1135 didn't jump to line 1136 because the condition on line 1135 was never true
1136 return result
1138 # Check retry eligibility before network-dependent extraction
1139 if self.retry_manager: 1139 ↛ 1148line 1139 didn't jump to line 1148 because the condition on line 1139 was always true
1140 decision = self.retry_manager.should_retry_resource(resource_id)
1141 if not decision.can_retry: 1141 ↛ 1142line 1141 didn't jump to line 1142 because the condition on line 1141 was never true
1142 logger.info(
1143 f"Skipping resource {resource_id}: {decision.reason}"
1144 )
1145 return False, decision.reason
1147 # Try API text extraction
1148 result = self._try_api_text_extraction(session, resource)
1149 if result is not None:
1150 self._record_retry_attempt(resource, result, session)
1151 return result
1153 # Fallback: Download PDF and extract
1154 result = self._fallback_pdf_extraction(session, resource)
1155 self._record_retry_attempt(resource, result, session)
1156 return result
1158 def _record_retry_attempt(
1159 self, resource, result: Tuple[bool, Optional[str]], session=None
1160 ) -> None:
1161 """Record a download attempt with the retry manager."""
1162 if not self.retry_manager: 1162 ↛ 1163line 1162 didn't jump to line 1163 because the condition on line 1162 was never true
1163 return
1164 self.retry_manager.record_attempt(
1165 resource_id=resource.id,
1166 result=result,
1167 url=resource.url or "",
1168 details=result[1]
1169 or (
1170 "Successfully extracted text"
1171 if result[0]
1172 else "Text extraction failed"
1173 ),
1174 session=session,
1175 )
1177 def _try_library_text_extraction(
1178 self, session, resource
1179 ) -> Tuple[bool, Optional[str]]:
1180 """Handle library resources — content already exists in local database.
1182 Returns:
1183 Tuple of (success, error_message). On success: (True, None).
1184 On failure: (False, description of what went wrong).
1185 """
1186 # 1. Extract document UUID from metadata (authoritative) or URL (fallback)
1187 doc_id = None
1188 metadata = (resource.resource_metadata or {}).get("original_data", {})
1189 if isinstance(metadata, dict):
1190 meta_inner = metadata.get("metadata", {})
1191 if isinstance(meta_inner, dict):
1192 doc_id = meta_inner.get("source_id") or meta_inner.get(
1193 "document_id"
1194 )
1196 if not doc_id and resource.url:
1197 # Parse from /library/document/{uuid} or /library/document/{uuid}/pdf
1198 match = re.match(r"^/library/document/([^/]+)", resource.url)
1199 if match: 1199 ↛ 1202line 1199 didn't jump to line 1202 because the condition on line 1199 was always true
1200 doc_id = match.group(1)
1202 if not doc_id:
1203 return False, "Could not extract library document ID"
1205 # 2. Query the existing Document by its primary key (UUID string)
1206 doc = session.query(Document).filter_by(id=doc_id).first()
1207 if not doc:
1208 return False, f"Library document {doc_id} not found in database"
1210 # 3. If text already exists and extraction succeeded, return success
1211 if (
1212 doc.text_content
1213 and doc.extraction_method
1214 and doc.extraction_method != "failed"
1215 ):
1216 logger.info(f"Library document {doc_id} already has text content")
1217 resource.document_id = doc.id
1218 session.commit()
1219 return True, None
1221 # 4. Try extracting text from stored PDF
1222 pdf_storage_mode = self.settings.get_setting(
1223 "research_library.pdf_storage_mode", "none"
1224 )
1225 pdf_manager = PDFStorageManager(
1226 library_root=self.library_root,
1227 storage_mode=pdf_storage_mode,
1228 legacy_root=self.legacy_library_root,
1229 )
1230 pdf_content = pdf_manager.load_pdf(doc, session)
1232 if not pdf_content:
1233 return (
1234 False,
1235 f"Library document {doc_id} has no text or PDF content",
1236 )
1238 text = self._extract_text_from_pdf(pdf_content)
1239 if not text:
1240 return (
1241 False,
1242 f"Failed to extract text from library document {doc_id} PDF",
1243 )
1245 # 5. Update Document directly (don't use _save_text_with_db — it
1246 # queries by resource_id which mismatches for library docs)
1247 doc.text_content = text
1248 doc.character_count = len(text)
1249 doc.word_count = len(text.split())
1250 doc.extraction_method = "pdf_extraction"
1251 doc.extraction_source = "pdfplumber"
1252 doc.extraction_quality = "medium"
1254 resource.document_id = doc.id
1255 session.commit()
1257 logger.info(
1258 f"Extracted text from library document {doc_id} "
1259 f"({doc.word_count} words)"
1260 )
1261 return True, None
1263 def _try_existing_text(
1264 self, session, resource_id: int
1265 ) -> Optional[Tuple[bool, Optional[str]]]:
1266 """Check if text already exists in database (in Document.text_content)."""
1267 existing_doc = (
1268 session.query(Document).filter_by(resource_id=resource_id).first()
1269 )
1271 if not existing_doc:
1272 return None
1274 # Check if text content exists and extraction was successful
1275 if (
1276 existing_doc.text_content
1277 and existing_doc.extraction_method
1278 and existing_doc.extraction_method != "failed"
1279 ):
1280 logger.info(
1281 f"Text content already exists in Document for resource_id={resource_id}, extraction_method={existing_doc.extraction_method}"
1282 )
1283 return True, None
1285 # No text content or failed extraction
1286 logger.debug(
1287 f"Document exists but no valid text content: resource_id={resource_id}, extraction_method={existing_doc.extraction_method}"
1288 )
1289 return None # Fall through to re-extraction
1291 def _try_legacy_text_file(
1292 self, session, resource, resource_id: int
1293 ) -> Optional[Tuple[bool, Optional[str]]]:
1294 """Check for legacy text files on disk."""
1295 txt_path = Path(self.library_root) / "txt"
1296 existing_files = (
1297 list(txt_path.glob(f"*_{resource_id}.txt"))
1298 if txt_path.exists()
1299 else []
1300 )
1302 if not existing_files:
1303 return None
1305 logger.info(f"Text file already exists on disk: {existing_files[0]}")
1306 self._create_text_document_record(
1307 session,
1308 resource,
1309 existing_files[0],
1310 extraction_method="unknown",
1311 extraction_source="legacy_file",
1312 )
1313 session.commit()
1314 return True, None
1316 def _try_existing_pdf_extraction(
1317 self, session, resource, resource_id: int
1318 ) -> Optional[Tuple[bool, Optional[str]]]:
1319 """Try extracting text from existing PDF in database."""
1320 pdf_document = (
1321 session.query(Document).filter_by(resource_id=resource_id).first()
1322 )
1324 if not pdf_document or pdf_document.status != "completed":
1325 return None
1327 # Validate path to prevent path traversal attacks. Resolve against the
1328 # per-user root with legacy-shared fallback (issue #5521) so a PDF
1329 # downloaded before per-user isolation is still found. Pass
1330 # self.settings so storage_path/shared_library come from this user's
1331 # captured manager — the ambient one is db-less in background/scheduler
1332 # threads (this runs under the document scheduler) and would resolve
1333 # the default root, missing a user's UI-customized storage_path. Use
1334 # getattr: __init__ always sets self.settings in production (incl.
1335 # scheduler threads), but test doubles that construct a
1336 # DownloadService via __new__/partial mocks may not — fall back to
1337 # None (the ambient get_settings_manager()) rather than raising
1338 # AttributeError.
1339 if (
1340 not pdf_document.file_path
1341 or pdf_document.file_path in FILE_PATH_SENTINELS
1342 ):
1343 return None
1344 pdf_path = get_absolute_path_from_settings(
1345 pdf_document.file_path,
1346 self.username,
1347 settings_manager=getattr(self, "settings", None),
1348 )
1349 if not pdf_path or not pdf_path.is_file():
1350 return None
1352 logger.info(f"Found existing PDF, extracting text from: {pdf_path}")
1353 try:
1354 with open(pdf_path, "rb") as f:
1355 pdf_content = f.read()
1356 text = self._extract_text_from_pdf(pdf_content)
1358 if not text:
1359 return None
1361 self._save_text_with_db(
1362 resource,
1363 text,
1364 session,
1365 extraction_method="pdf_extraction",
1366 extraction_source="pdfplumber",
1367 pdf_document_id=pdf_document.id,
1368 )
1369 session.commit()
1370 return True, None
1372 except Exception:
1373 logger.exception(
1374 f"Failed to extract text from existing PDF: {pdf_path}"
1375 )
1376 # The commit above can fail and poison the shared session.
1377 # download_as_text swallows this None and may then return cleanly
1378 # (e.g. a non-retry decision), so the with-block exits without an
1379 # exception — get_user_db_session's rollback never fires. Recover
1380 # the session here or the next operation on this thread cascades.
1381 safe_rollback(session, "_try_existing_pdf_extraction")
1382 return None # Fall through to other methods
1384 def _try_api_text_extraction(
1385 self, session, resource
1386 ) -> Optional[Tuple[bool, Optional[str]]]:
1387 """Try direct API text extraction."""
1388 logger.info(
1389 f"Attempting direct API text extraction from: {resource.url}"
1390 )
1392 downloader = self._get_downloader(resource.url)
1393 if not downloader:
1394 return None
1396 # Egress policy gate. Return a non-None tuple (not None) on denial
1397 # so download_as_text hard-stops here instead of falling through
1398 # to _fallback_pdf_extraction and firing a second request.
1399 allowed, reason = self._check_url_against_policy(resource.url)
1400 if not allowed:
1401 logger.warning(
1402 f"Skipping API text extraction for {resource.url}: refused "
1403 f"by egress policy ({reason})"
1404 )
1405 return False, f"egress_policy_denied:{reason}"
1407 result = downloader.download_with_result(resource.url, ContentType.TEXT)
1409 if not result.is_success or not result.content:
1410 return None
1412 # Decode text content
1413 text = (
1414 result.content.decode("utf-8", errors="ignore")
1415 if isinstance(result.content, bytes)
1416 else result.content
1417 )
1419 # Determine extraction source
1420 extraction_source = "unknown"
1421 if isinstance(downloader, ArxivDownloader):
1422 extraction_source = "arxiv_api"
1423 elif isinstance(downloader, PubMedDownloader): 1423 ↛ 1424line 1423 didn't jump to line 1424 because the condition on line 1423 was never true
1424 extraction_source = "pubmed_api"
1426 try:
1427 self._save_text_with_db(
1428 resource,
1429 text,
1430 session,
1431 extraction_method="native_api",
1432 extraction_source=extraction_source,
1433 )
1434 session.commit()
1435 logger.info(
1436 f"✓ SUCCESS: Got text from {extraction_source.upper()} API for '{resource.title[:50]}...'"
1437 )
1438 return True, None
1439 except Exception as e:
1440 # Roll back FIRST: the failed commit/flush poisoned the shared
1441 # session, and even dereferencing resource.id in the log line below
1442 # can trigger a refresh that re-raises PendingRollbackError.
1443 # download_as_text reuses this session (_record_retry_attempt)
1444 # after we return, so it must be clean.
1445 safe_rollback(session, "_try_api_text_extraction")
1446 logger.exception(f"Failed to save text for resource {resource.id}")
1447 # Sanitize error message before returning to API
1448 safe_error = sanitize_error_for_client(
1449 f"Failed to save text: {str(e)}"
1450 )
1451 return False, safe_error
1453 def _fallback_pdf_extraction(
1454 self, session, resource
1455 ) -> Tuple[bool, Optional[str]]:
1456 """Fallback: Download PDF to memory and extract text."""
1457 logger.info(
1458 f"API text extraction failed, falling back to in-memory PDF download for: {resource.url}"
1459 )
1461 downloader = self._get_downloader(resource.url)
1462 if not downloader:
1463 error_msg = "No compatible downloader found"
1464 logger.warning(
1465 f"✗ FAILED: {error_msg} for '{resource.title[:50]}...'"
1466 )
1467 self._record_failed_text_extraction(
1468 session, resource, error=error_msg
1469 )
1470 session.commit()
1471 return False, error_msg
1473 # Egress policy gate at the network-fire point.
1474 allowed, reason = self._check_url_against_policy(resource.url)
1475 if not allowed:
1476 error_msg = f"egress_policy_denied:{reason}"
1477 logger.warning(
1478 f"Skipping fallback PDF download for {resource.url}: "
1479 f"refused by egress policy ({reason})"
1480 )
1481 self._record_failed_text_extraction(
1482 session, resource, error=error_msg
1483 )
1484 session.commit()
1485 return False, error_msg
1487 result = downloader.download_with_result(resource.url, ContentType.PDF)
1489 if not result.is_success or not result.content:
1490 error_msg = result.skip_reason or "Failed to download PDF"
1491 logger.warning(
1492 f"✗ FAILED: Could not download PDF for '{resource.title[:50]}...' | Error: {error_msg}"
1493 )
1494 self._record_failed_text_extraction(
1495 session, resource, error=f"PDF download failed: {error_msg}"
1496 )
1497 session.commit()
1498 return False, f"PDF extraction failed: {error_msg}"
1500 # Extract text from PDF
1501 text = self._extract_text_from_pdf(result.content)
1502 if not text:
1503 error_msg = "PDF text extraction returned empty text"
1504 logger.warning(
1505 f"Failed to extract text from PDF for: {resource.url}"
1506 )
1507 self._record_failed_text_extraction(
1508 session, resource, error=error_msg
1509 )
1510 session.commit()
1511 return False, error_msg
1513 try:
1514 self._save_text_with_db(
1515 resource,
1516 text,
1517 session,
1518 extraction_method="pdf_extraction",
1519 extraction_source="pdfplumber_fallback",
1520 )
1521 session.commit()
1522 logger.info(
1523 f"✓ SUCCESS: Extracted text from '{resource.title[:50]}...'"
1524 )
1525 return True, None
1526 except Exception as e:
1527 # Roll back FIRST (see _try_api_text_extraction): dereferencing
1528 # resource.id while the session is poisoned would itself re-raise
1529 # PendingRollbackError. download_as_text reuses this session after
1530 # we return.
1531 safe_rollback(session, "_fallback_pdf_extraction")
1532 logger.exception(f"Failed to save text for resource {resource.id}")
1533 # Sanitize error message before returning to API
1534 safe_error = sanitize_error_for_client(
1535 f"Failed to save text: {str(e)}"
1536 )
1537 return False, safe_error
1539 def _get_downloader(self, url: str):
1540 """
1541 Get the appropriate downloader for a URL.
1543 Args:
1544 url: The URL to download from
1546 Returns:
1547 The appropriate downloader instance or None
1548 """
1549 for downloader in self.downloaders:
1550 if downloader.can_handle(url):
1551 return downloader
1552 return None
1554 def _download_generic(self, url: str) -> Optional[bytes]:
1555 """Generic PDF download method."""
1556 try:
1557 headers = {
1558 "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
1559 }
1560 response = safe_get(
1561 url, headers=headers, timeout=30, allow_redirects=True
1562 )
1563 response.raise_for_status()
1565 # Verify it's a PDF
1566 content_type = response.headers.get("Content-Type", "")
1567 if (
1568 "pdf" not in content_type.lower()
1569 and not response.content.startswith(b"%PDF")
1570 ):
1571 logger.warning(f"Response is not a PDF: {content_type}")
1572 return None
1574 return response.content
1576 except Exception:
1577 logger.exception("Generic download failed")
1578 return None
1580 def _download_arxiv(self, url: str) -> Optional[bytes]:
1581 """Download from arXiv."""
1582 try:
1583 # Convert abstract URL to PDF URL
1584 pdf_url = url.replace("abs", "pdf")
1585 if not pdf_url.endswith(".pdf"):
1586 pdf_url += ".pdf"
1588 return self._download_generic(pdf_url)
1589 except Exception:
1590 logger.exception("arXiv download failed")
1591 return None
1593 def _try_europe_pmc(self, pmid: str) -> Optional[bytes]:
1594 """Try downloading from Europe PMC which often has better PDF availability."""
1595 try:
1596 # Europe PMC API is more reliable for PDFs
1597 # Check if PDF is available
1598 api_url = f"https://www.ebi.ac.uk/europepmc/webservices/rest/search?query=EXT_ID:{pmid}&format=json"
1599 response = safe_get(api_url, timeout=10)
1601 if response.status_code == 200:
1602 data = response.json()
1603 results = data.get("resultList", {}).get("result", [])
1605 if results:
1606 article = results[0]
1607 # Check if article has open access PDF
1608 if (
1609 article.get("isOpenAccess") == "Y"
1610 and article.get("hasPDF") == "Y"
1611 ):
1612 pmcid = article.get("pmcid")
1613 if pmcid:
1614 # Europe PMC PDF URL
1615 pdf_url = f"https://europepmc.org/backend/ptpmcrender.fcgi?accid={pmcid}&blobtype=pdf"
1616 logger.info(
1617 f"Found Europe PMC PDF for PMID {pmid}: {pmcid}"
1618 )
1620 headers = {
1621 "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
1622 }
1624 pdf_response = safe_get(
1625 pdf_url,
1626 headers=headers,
1627 timeout=30,
1628 allow_redirects=True,
1629 )
1630 if pdf_response.status_code == 200: 1630 ↛ 1642line 1630 didn't jump to line 1642 because the condition on line 1630 was always true
1631 content_type = pdf_response.headers.get(
1632 "content-type", ""
1633 )
1634 if ( 1634 ↛ 1642line 1634 didn't jump to line 1642 because the condition on line 1634 was always true
1635 "pdf" in content_type.lower()
1636 or len(pdf_response.content) > 1000
1637 ):
1638 return pdf_response.content
1639 except Exception as e:
1640 logger.debug(f"Europe PMC download failed: {e}")
1642 return None
1644 def _download_pubmed(self, url: str) -> Optional[bytes]:
1645 """Download from PubMed/PubMed Central with rate limiting."""
1646 try:
1647 # Apply rate limiting for PubMed requests
1648 current_time = time.time()
1649 time_since_last = current_time - self._last_pubmed_request
1650 if time_since_last < self._pubmed_delay:
1651 sleep_time = self._pubmed_delay - time_since_last
1652 logger.debug(
1653 f"Rate limiting: sleeping {sleep_time:.2f}s before PubMed request"
1654 )
1655 time.sleep(sleep_time)
1656 self._last_pubmed_request = time.time()
1658 # If it's already a PMC article, download directly
1659 if "/articles/PMC" in url:
1660 pmc_match = re.search(r"(PMC\d+)", url)
1661 if pmc_match: 1661 ↛ 1865line 1661 didn't jump to line 1865 because the condition on line 1661 was always true
1662 pmc_id = pmc_match.group(1)
1664 # Try Europe PMC (more reliable)
1665 europe_url = f"https://europepmc.org/backend/ptpmcrender.fcgi?accid={pmc_id}&blobtype=pdf"
1666 headers = {
1667 "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
1668 }
1670 try:
1671 response = safe_get(
1672 europe_url,
1673 headers=headers,
1674 timeout=30,
1675 allow_redirects=True,
1676 )
1677 if response.status_code == 200: 1677 ↛ 1865line 1677 didn't jump to line 1865 because the condition on line 1677 was always true
1678 content_type = response.headers.get(
1679 "content-type", ""
1680 )
1681 if ( 1681 ↛ 1865line 1681 didn't jump to line 1865 because the condition on line 1681 was always true
1682 "pdf" in content_type.lower()
1683 or len(response.content) > 1000
1684 ):
1685 logger.info(
1686 f"Downloaded PDF via Europe PMC for {pmc_id}"
1687 )
1688 return response.content
1689 except Exception as e:
1690 logger.debug(f"Direct Europe PMC download failed: {e}")
1691 return None
1693 # If it's a regular PubMed URL, try to find PMC version
1694 elif urlparse(url).hostname == "pubmed.ncbi.nlm.nih.gov":
1695 # Extract PMID from URL
1696 pmid_match = re.search(r"/(\d+)/?", url)
1697 if pmid_match:
1698 pmid = pmid_match.group(1)
1699 logger.info(f"Attempting to download PDF for PMID: {pmid}")
1701 # Try Europe PMC first (more reliable)
1702 pdf_content = self._try_europe_pmc(pmid)
1703 if pdf_content:
1704 return pdf_content
1706 # First try using NCBI E-utilities API to find PMC ID
1707 try:
1708 # Use elink to convert PMID to PMCID
1709 elink_url = "https://eutils.ncbi.nlm.nih.gov/entrez/eutils/elink.fcgi"
1710 params = {
1711 "dbfrom": "pubmed",
1712 "db": "pmc",
1713 "id": pmid,
1714 "retmode": "json",
1715 }
1717 api_response = safe_get(
1718 elink_url, params=params, timeout=10
1719 )
1720 if api_response.status_code == 200: 1720 ↛ 1803line 1720 didn't jump to line 1803 because the condition on line 1720 was always true
1721 data = api_response.json()
1722 # Parse the response to find PMC ID
1723 link_sets = data.get("linksets", [])
1724 if link_sets and "linksetdbs" in link_sets[0]:
1725 for linksetdb in link_sets[0]["linksetdbs"]: 1725 ↛ 1803line 1725 didn't jump to line 1803 because the loop on line 1725 didn't complete
1726 if linksetdb.get( 1726 ↛ 1725line 1726 didn't jump to line 1725 because the condition on line 1726 was always true
1727 "dbto"
1728 ) == "pmc" and linksetdb.get("links"):
1729 pmc_id_num = linksetdb["links"][0]
1730 # Now fetch PMC details to get the correct PMC ID format
1731 esummary_url = "https://eutils.ncbi.nlm.nih.gov/entrez/eutils/esummary.fcgi"
1732 summary_params = {
1733 "db": "pmc",
1734 "id": pmc_id_num,
1735 "retmode": "json",
1736 }
1737 summary_response = safe_get(
1738 esummary_url,
1739 params=summary_params,
1740 timeout=10,
1741 )
1742 if summary_response.status_code == 200: 1742 ↛ 1725line 1742 didn't jump to line 1725 because the condition on line 1742 was always true
1743 summary_data = (
1744 summary_response.json()
1745 )
1746 result = summary_data.get(
1747 "result", {}
1748 ).get(str(pmc_id_num), {})
1749 if result: 1749 ↛ 1725line 1749 didn't jump to line 1725 because the condition on line 1749 was always true
1750 # PMC IDs in the API don't have the "PMC" prefix
1751 pmc_id = f"PMC{pmc_id_num}"
1752 logger.info(
1753 f"Found PMC ID via API: {pmc_id} for PMID: {pmid}"
1754 )
1756 # Try Europe PMC with the PMC ID
1757 europe_url = f"https://europepmc.org/backend/ptpmcrender.fcgi?accid={pmc_id}&blobtype=pdf"
1759 time.sleep(self._pubmed_delay)
1761 headers = {
1762 "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
1763 }
1765 try:
1766 response = safe_get(
1767 europe_url,
1768 headers=headers,
1769 timeout=30,
1770 allow_redirects=True,
1771 )
1772 if ( 1772 ↛ 1725line 1772 didn't jump to line 1725 because the condition on line 1772 was always true
1773 response.status_code
1774 == 200
1775 ):
1776 content_type = response.headers.get(
1777 "content-type", ""
1778 )
1779 if ( 1779 ↛ 1725line 1779 didn't jump to line 1725 because the condition on line 1779 was always true
1780 "pdf"
1781 in content_type.lower()
1782 or len(
1783 response.content
1784 )
1785 > 1000
1786 ):
1787 logger.info(
1788 f"Downloaded PDF via Europe PMC for {pmc_id}"
1789 )
1790 return (
1791 response.content
1792 )
1793 except Exception as e:
1794 logger.debug(
1795 f"Europe PMC download with PMC ID failed: {e}"
1796 )
1797 except Exception as e:
1798 logger.debug(
1799 f"API lookup failed, trying webpage scraping: {e}"
1800 )
1802 # Fallback to webpage scraping if API fails
1803 try:
1804 headers = {
1805 "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
1806 }
1807 response = safe_get(url, headers=headers, timeout=10)
1808 if response.status_code == 200: 1808 ↛ 1865line 1808 didn't jump to line 1865 because the condition on line 1808 was always true
1809 # Look for PMC ID in the page
1810 pmc_match = re.search(r"PMC\d+", response.text)
1811 if pmc_match: 1811 ↛ 1850line 1811 didn't jump to line 1850 because the condition on line 1811 was always true
1812 pmc_id = pmc_match.group(0)
1813 logger.info(
1814 f"Found PMC ID via webpage: {pmc_id} for PMID: {pmid}"
1815 )
1817 # Add delay before downloading PDF
1818 time.sleep(self._pubmed_delay)
1820 # Try Europe PMC with the PMC ID (more reliable)
1821 europe_url = f"https://europepmc.org/backend/ptpmcrender.fcgi?accid={pmc_id}&blobtype=pdf"
1822 headers = {
1823 "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
1824 }
1826 try:
1827 response = safe_get(
1828 europe_url,
1829 headers=headers,
1830 timeout=30,
1831 allow_redirects=True,
1832 )
1833 if response.status_code == 200: 1833 ↛ 1865line 1833 didn't jump to line 1865 because the condition on line 1833 was always true
1834 content_type = response.headers.get(
1835 "content-type", ""
1836 )
1837 if ( 1837 ↛ 1865line 1837 didn't jump to line 1865 because the condition on line 1837 was always true
1838 "pdf" in content_type.lower()
1839 or len(response.content) > 1000
1840 ):
1841 logger.info(
1842 f"Downloaded PDF via Europe PMC for {pmc_id}"
1843 )
1844 return response.content
1845 except Exception as e:
1846 logger.debug(
1847 f"Europe PMC download failed: {e}"
1848 )
1849 else:
1850 logger.info(
1851 f"No PMC version found for PMID: {pmid}"
1852 )
1853 except requests.exceptions.HTTPError as e:
1854 if e.response.status_code == 429: 1854 ↛ 1861line 1854 didn't jump to line 1861 because the condition on line 1854 was always true
1855 logger.warning(
1856 "Rate limited by PubMed, increasing delay"
1857 )
1858 self._pubmed_delay = min(
1859 self._pubmed_delay * 2, 5.0
1860 ) # Max 5 seconds
1861 raise
1862 except Exception as e:
1863 logger.debug(f"Could not check for PMC version: {e}")
1865 return self._download_generic(url)
1866 except Exception:
1867 logger.exception("PubMed download failed")
1868 return None
1870 def _download_semantic_scholar(self, url: str) -> Optional[bytes]:
1871 """Download from Semantic Scholar."""
1872 # Semantic Scholar doesn't host PDFs directly
1873 # Would need to extract actual PDF URL from page
1874 return None
1876 def _download_biorxiv(self, url: str) -> Optional[bytes]:
1877 """Download from bioRxiv."""
1878 try:
1879 # Convert to PDF URL
1880 pdf_url = url.replace(".org/", ".org/content/")
1881 pdf_url = re.sub(r"v\d+$", "", pdf_url) # Remove version
1882 pdf_url += ".full.pdf"
1884 return self._download_generic(pdf_url)
1885 except Exception:
1886 logger.exception("bioRxiv download failed")
1887 return None
1889 def _save_text_with_db(
1890 self,
1891 resource: ResearchResource,
1892 text: str,
1893 session: Session,
1894 extraction_method: str,
1895 extraction_source: str,
1896 pdf_document_id: Optional[int] = None,
1897 ) -> Optional[str]:
1898 """
1899 Save extracted text to encrypted database.
1901 Args:
1902 resource: The research resource
1903 text: Extracted text content
1904 session: Database session
1905 extraction_method: How the text was extracted
1906 extraction_source: Specific tool/API used
1907 pdf_document_id: ID of PDF document if extracted from PDF
1909 Returns:
1910 None (previously returned text file path, now removed)
1911 """
1912 try:
1913 # Calculate text metadata for database
1914 word_count = len(text.split())
1915 character_count = len(text)
1917 # Find the document by pdf_document_id or resource_id
1918 doc = None
1919 if pdf_document_id:
1920 doc = (
1921 session.query(Document)
1922 .filter_by(id=pdf_document_id)
1923 .first()
1924 )
1925 else:
1926 doc = get_document_for_resource(session, resource)
1928 if doc:
1929 # Update existing document with extracted text. Only replace
1930 # document_hash when transitioning from FAILED — see issue
1931 # #3827. PDF-bytes hashes set at creation must remain stable;
1932 # text-content hashes collide far more often (identical
1933 # extracted text from different PDFs).
1934 was_failed = doc.status == DocumentStatus.FAILED
1935 doc.text_content = text
1936 doc.character_count = character_count
1937 doc.word_count = word_count
1938 doc.extraction_method = extraction_method
1939 doc.extraction_source = extraction_source
1940 doc.status = DocumentStatus.COMPLETED
1941 if was_failed:
1942 doc.document_hash = hashlib.sha256(
1943 text.encode()
1944 ).hexdigest()
1945 doc.processed_at = datetime.now(UTC)
1947 # Set quality based on method
1948 if extraction_method == "native_api":
1949 doc.extraction_quality = "high"
1950 elif (
1951 extraction_method == "pdf_extraction"
1952 and extraction_source == "pdfplumber"
1953 ):
1954 doc.extraction_quality = "medium"
1955 else:
1956 doc.extraction_quality = "low"
1958 logger.debug(
1959 f"Updated document {doc.id} with extracted text ({word_count} words)"
1960 )
1961 else:
1962 # Create a new Document for text-only extraction
1963 # Generate hash from text content
1964 text_hash = hashlib.sha256(text.encode()).hexdigest()
1966 # Dedup against existing content hash. If another resource
1967 # already produced identical extracted text, link this
1968 # resource to the canonical Document instead of inserting
1969 # a duplicate that would violate the UNIQUE constraint
1970 # (issue #3827). Mirrors research_history_indexer.py:322.
1971 existing_by_hash = (
1972 session.query(Document)
1973 .filter_by(document_hash=text_hash)
1974 .first()
1975 )
1976 if existing_by_hash:
1977 resource.document_id = existing_by_hash.id
1978 library_collection = (
1979 session.query(Collection)
1980 .filter_by(name="Library")
1981 .first()
1982 )
1983 if library_collection: 1983 ↛ 1990line 1983 didn't jump to line 1990 because the condition on line 1983 was always true
1984 ensure_in_collection(
1985 session,
1986 existing_by_hash.id,
1987 library_collection.id,
1988 )
1989 else:
1990 logger.warning(
1991 f"Library collection not found - deduped document {existing_by_hash.id} will not be linked to default collection"
1992 )
1993 logger.info(
1994 f"Linked resource {resource.id} to existing Document "
1995 f"{existing_by_hash.id} (matched on content hash)"
1996 )
1997 return None
1999 # Get source type for research downloads
2000 try:
2001 source_type_id = get_source_type_id(
2002 self.username, "research_download", self.password
2003 )
2004 except Exception:
2005 logger.exception(
2006 "Failed to get source type for text document"
2007 )
2008 raise
2010 # Create new document
2011 doc_id = str(uuid.uuid4())
2012 doc = Document(
2013 id=doc_id,
2014 source_type_id=source_type_id,
2015 resource_id=resource.id,
2016 research_id=resource.research_id,
2017 document_hash=text_hash,
2018 original_url=resource.url,
2019 file_path=FILE_PATH_TEXT_ONLY,
2020 file_size=character_count, # Use character count as file size for text-only
2021 file_type="text",
2022 mime_type="text/plain",
2023 title=resource.title,
2024 text_content=text,
2025 character_count=character_count,
2026 word_count=word_count,
2027 extraction_method=extraction_method,
2028 extraction_source=extraction_source,
2029 extraction_quality="high"
2030 if extraction_method == "native_api"
2031 else "medium",
2032 status=DocumentStatus.COMPLETED,
2033 processed_at=datetime.now(UTC),
2034 )
2035 session.add(doc)
2037 # Link to default Library collection
2038 library_collection = (
2039 session.query(Collection).filter_by(name="Library").first()
2040 )
2041 if library_collection:
2042 ensure_in_collection(session, doc_id, library_collection.id)
2043 else:
2044 logger.warning(
2045 f"Library collection not found - document {doc_id} will not be linked to default collection"
2046 )
2048 logger.info(
2049 f"Created new document {doc_id} for text-only extraction ({word_count} words)"
2050 )
2052 logger.info(
2053 f"Saved text to encrypted database ({word_count} words)"
2054 )
2055 return None
2057 except Exception:
2058 # Rollback BEFORE re-raising so the shared thread-local session
2059 # is clean by the time the caller's loop reaches its next
2060 # iteration. Without this, an IntegrityError leaves the session
2061 # in PendingRollbackError state and every subsequent ORM access
2062 # cascades (issue #3827).
2063 safe_rollback(session, "_save_text_with_db")
2064 logger.exception("Error saving text to encrypted database")
2065 raise # Re-raise so caller can handle the error
2067 def _create_text_document_record(
2068 self,
2069 session: Session,
2070 resource: ResearchResource,
2071 file_path: Path,
2072 extraction_method: str,
2073 extraction_source: str,
2074 ):
2075 """Update existing Document with text from file (for legacy text files)."""
2076 try:
2077 # Read file to get metadata
2078 text = file_path.read_text(encoding="utf-8", errors="ignore")
2079 word_count = len(text.split())
2080 character_count = len(text)
2082 # Find the Document for this resource
2083 doc = get_document_for_resource(session, resource)
2085 if doc:
2086 # Update existing document with text content
2087 doc.text_content = text
2088 doc.character_count = character_count
2089 doc.word_count = word_count
2090 doc.extraction_method = extraction_method
2091 doc.extraction_source = extraction_source
2092 doc.extraction_quality = (
2093 "low" # Unknown quality for legacy files
2094 )
2095 logger.info(
2096 f"Updated document {doc.id} with text from file: {file_path.name}"
2097 )
2098 else:
2099 logger.warning(
2100 f"No document found to update for resource {resource.id}"
2101 )
2103 except Exception:
2104 logger.exception("Error updating document with text from file")
2106 def _record_failed_text_extraction(
2107 self, session: Session, resource: ResearchResource, error: str
2108 ):
2109 """Record a failed text extraction attempt in the Document."""
2110 try:
2111 # Find the Document for this resource
2112 doc = get_document_for_resource(session, resource)
2114 if doc:
2115 # Update document with extraction error
2116 doc.error_message = error
2117 doc.extraction_method = "failed"
2118 doc.extraction_quality = "low"
2119 doc.status = DocumentStatus.FAILED
2120 logger.info(
2121 f"Recorded failed text extraction for document {doc.id}: {error}"
2122 )
2123 else:
2124 # Create a new Document for failed extraction
2125 # This enables tracking failures and retry capability
2126 source_type_id = get_source_type_id(
2127 self.username, "research_download", self.password
2128 )
2130 # Deterministic hash so retries update the same record
2131 failed_hash = hashlib.sha256(
2132 f"failed:{resource.url}:{resource.id}".encode()
2133 ).hexdigest()
2135 doc_id = str(uuid.uuid4())
2136 doc = Document(
2137 id=doc_id,
2138 source_type_id=source_type_id,
2139 resource_id=resource.id,
2140 research_id=resource.research_id,
2141 document_hash=failed_hash,
2142 original_url=resource.url,
2143 file_path=None,
2144 file_size=0,
2145 file_type="unknown",
2146 title=resource.title,
2147 status=DocumentStatus.FAILED,
2148 error_message=error,
2149 extraction_method="failed",
2150 extraction_quality="low",
2151 processed_at=datetime.now(UTC),
2152 )
2153 session.add(doc)
2155 logger.info(
2156 f"Created failed document {doc_id} for resource {resource.id}: {error}"
2157 )
2159 except Exception:
2160 logger.exception("Error recording failed text extraction")