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