Coverage for src/local_deep_research/research_library/services/download_service.py: 95%
1265 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-22 21:20 +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"""
11from contextlib import nullcontext
12import hashlib
13import os
14import re
15import time
16import uuid
17from datetime import datetime, UTC
18from pathlib import Path
19from collections.abc import Set as AbstractSet
20from typing import Any, Dict, NamedTuple, Optional, Tuple, assert_never
21from urllib.parse import urlparse
23import requests
24from loguru import logger
25from ...security import redact_url_for_log
26from sqlalchemy import and_, func, or_
27from sqlalchemy.orm import Session, defer, load_only
28import pdfplumber
29from pypdf import PdfReader
31from ...utilities.arxiv import extract_arxiv_id, is_arxiv_paper_url
32from ...constants import DEFAULT_SEARCH_TOOL
33from ...utilities.url_utils import canonical_url_key
34from ...constants import FILE_PATH_SENTINELS, FILE_PATH_TEXT_ONLY
35from ...database.models.download_tracker import (
36 DownloadAttempt,
37 DownloadTracker,
38)
40# SSRF chokepoint: every outbound fetch in this module goes through
41# `safe_get`, which validates via ssrf_validator (raising on failure,
42# re-validating after redirects, always blocking cloud-metadata
43# addresses). `requests` is imported above for its exception types only
44# — do not add a raw requests.get()/post() here. GHSA-9c54 was exactly
45# that bypass, and `is_downloadable_domain` upstream is a relevance
46# filter, not a security control.
47from ...security import (
48 failure_reason_for_log,
49 safe_get,
50 sanitize_error_for_client,
51)
52from ...security.client_safe_errors import (
53 CLIENT_SAFE_DOWNLOAD_MESSAGES,
54 client_safe_download_message,
55)
56from ...security.secure_logging import log_exception_type
57from ...database.models.library import (
58 Collection,
59 Document as Document,
60 DocumentChunk,
61 DocumentCollection,
62 DocumentStatus,
63 DownloadQueue as LibraryDownloadQueue,
64 RAGIndex,
65 RagDocumentStatus,
66)
67from .pdf_storage_manager import (
68 DEFAULT_MAX_PDF_SIZE_MB,
69 PDFStorageManager,
70 resolve_pdf_storage_mode,
71)
72from ...database.models.research import ResearchResource
73from ...database.library_init import get_source_type_id, get_default_library_id
74from ...database.session_context import get_user_db_session, safe_rollback
75from ...utilities.db_utils import get_settings_manager
76from ...library.download_management import RetryManager
77from ...library.download_management.models import (
78 DownloadStatus as RetryStatus,
79)
80from ...config.paths import get_library_directory
81from ..utils import (
82 _TRUSTED_STORAGE_LEAF,
83 apply_user_subdir,
84 document_awaits_pdf,
85 download_claim_is_live,
86 ensure_in_collection,
87 find_tracker_by_url,
88 get_document_for_resource,
89 get_url_hash,
90 get_url_hash_candidates,
91 get_absolute_path_from_settings,
92 is_downloadable_url,
93)
95# Import our modular downloaders
96from ..downloaders import (
97 ContentType,
98 ArxivDownloader,
99 PubMedDownloader,
100 BioRxivDownloader,
101 DirectPDFDownloader,
102 SemanticScholarDownloader,
103 OpenAlexDownloader,
104 GenericDownloader,
105)
106from ..downloaders.arxiv import ArxivTextSource, attribute_arxiv_text
107from ..extraction_catalog import (
108 ExtractionMethodExtra,
109 ExtractionSourceExtra,
110)
112# Extraction ceilings live in ``utilities.pdf_extraction_limits`` so that
113# the base downloader shares one definition without importing this module.
114from ...utilities.pdf_extraction_limits import (
115 MAX_PDF_EXTRACTED_CHARS,
116 MAX_PDF_EXTRACTION_CPU_SECONDS,
117 MAX_PDF_EXTRACTION_PAGES,
118)
121class RefillRank(NamedTuple):
122 """Where a resource stands among the resources that refill a document.
124 A document that awaits its PDF (see ``document_awaits_pdf``) is
125 refilled by a download of its own resource (``Document.resource_id``)
126 or of a resource linked to it by content
127 (``ResearchResource.document_id``), and one download refills it. Unless
128 one of them is already being downloaded or queued, the queueing paths
129 give that download to the resource whose rank sorts first (see
130 ``DownloadService._refilled_by_another_resource``). The fields sort in
131 order, False before True.
132 """
134 #: Its URL cannot be queued: it is not downloadable, or the egress
135 #: policy refuses it.
136 cannot_queue: bool
137 #: The retry manager holds it back: a cooldown, a permanent failure or
138 #: a retry limit.
139 held_back: bool
140 #: Its latest download attempt failed.
141 failed_last: bool
142 #: It is not the document's own resource.
143 linked: bool
144 resource_id: int
146 @property
147 def viable(self) -> bool:
148 """Whether nothing known stands in the way of its download.
150 Its URL can be queued, the retry manager does not hold it back,
151 and its latest download attempt did not fail.
152 """
153 return not (self.cannot_queue or self.held_back or self.failed_last)
156def _text_extraction_quality(method: str | None, source: str | None) -> str:
157 if method in {"native_api", "html_extraction"}:
158 return "high"
159 if method == "pdfplumber" or (
160 method == "pdf_extraction" and source == "pdfplumber"
161 ):
162 return "medium"
163 return "low"
166def _text_extraction_rank(quality: str, method: str, source: str) -> int:
167 # The arXiv API fallback is explicitly metadata/abstract only. Its historic
168 # "high" quality label describes the API, not full-paper coverage.
169 if method == "native_api" and source == "arxiv_api":
170 return -1
171 return {"low": 0, "medium": 1, "high": 2}[quality]
174def _arxiv_text_provenance(source: ArxivTextSource) -> tuple[str, str]:
175 match source:
176 case ArxivTextSource.ARXIV_HTML:
177 return (
178 ExtractionMethodExtra.HTML_EXTRACTION.value,
179 ExtractionSourceExtra.ARXIV_HTML.value,
180 )
181 case ArxivTextSource.LOCAL_PDF:
182 return "pdf_extraction", "local_pdf"
183 case ArxivTextSource.ARXIV_API: 183 ↛ 185line 183 didn't jump to line 185 because the pattern on line 183 always matched
184 return "native_api", "arxiv_api"
185 case unreachable:
186 assert_never(unreachable)
189def _link_retry_to_canonical(
190 session: Session,
191 resource: ResearchResource,
192 failed: Document,
193 canonical: Document,
194) -> None:
195 """Reuse content without cascading away the failed record's history.
197 A FAILED record can still have annotations, stored content or other
198 resource references. Only redirect the resource whose retry established
199 the content match; retain the original record and its dependents.
200 Collection membership is copied, never its stale indexing state.
201 """
202 resource.document_id = canonical.id
203 collection_ids = {link.collection_id for link in failed.collections}
204 library = session.query(Collection).filter_by(name="Library").first()
205 if library: 205 ↛ 207line 205 didn't jump to line 207 because the condition on line 205 was always true
206 collection_ids.add(library.id)
207 for collection_id in collection_ids:
208 ensure_in_collection(session, canonical.id, collection_id)
211def _has_reusable_text(document: Document) -> bool:
212 """Reuse successful text, including recovered failed-attempt provenance.
214 An upload with no extraction method still needs its stored PDF extracted.
215 """
216 return bool(
217 document.text_content
218 and (
219 (
220 document.extraction_method
221 and document.extraction_method != "failed"
222 )
223 or (
224 document.extraction_method == "failed"
225 and document.status == DocumentStatus.COMPLETED
226 and document.error_message is None
227 )
228 )
229 )
232# Single source of truth for caller-safe skip-reason tokens lives in
233# ``security.client_safe_errors`` (PR #6564 review: the services and HTML
234# downloader copies drifted). Aliases below preserve the historic import
235# locations for back-compat; new code should import from ``security``.
236_CLIENT_SAFE_DOWNLOAD_MESSAGES = CLIENT_SAFE_DOWNLOAD_MESSAGES
239def _client_safe_download_message(exc: BaseException) -> str:
240 """Back-compat alias for :func:`client_safe_download_message`."""
241 return client_safe_download_message(exc)
244class DownloadService:
245 """Service for downloading and managing research PDFs."""
247 def __init__(
248 self,
249 username: str,
250 password: Optional[str] = None,
251 settings_snapshot: Optional[Dict[str, Any]] = None,
252 ):
253 """Initialize download service for a user.
255 Args:
256 username: The username to download for
257 password: Optional password for encrypted database access
258 settings_snapshot: Optional snapshot used to build the
259 EgressContext for per-URL scope gating. When None,
260 downloads run without scope enforcement (back-compat
261 for the existing caller; new callers must pass it).
262 """
263 self.username = username
264 self.password = password
265 self.settings = get_settings_manager(username=username)
266 self._closed = False
267 # When no snapshot is passed (e.g. the library download routes, which
268 # construct DownloadService(username, password) directly), build one
269 # from this user's settings so egress policy is still enforced.
270 # Without this, _egress_context stays None and _check_url_against_policy
271 # returns (True, "no_context") — every download ungated. Best-effort:
272 # on failure, fall through to the pre-policy behavior.
273 # A real backend error (get_settings_snapshot raised) is different
274 # from a test double returning a non-dict: under a private scope we
275 # must NOT silently fall through to ungated downloads when we simply
276 # failed to READ the user's settings. Track that case and fail closed.
277 self._snapshot_build_failed = False
278 if settings_snapshot is None:
279 try:
280 built = self.settings.get_settings_snapshot()
281 # Only adopt a real dict — a test double or unavailable
282 # backend hands back something else; treat that as "no
283 # snapshot" (ungated back-compat) rather than letting the
284 # policy evaluation fail closed on a non-dict.
285 if isinstance(built, dict):
286 settings_snapshot = built
287 except Exception:
288 # The user HAS a settings backend (self.settings exists) but
289 # reading it raised — a private scope could be configured and
290 # we just couldn't see it. Fail closed rather than ungate.
291 logger.bind(policy_audit=True).warning(
292 "DownloadService: settings snapshot build raised; "
293 "locking downloads closed (fail-safe)"
294 )
295 self._snapshot_build_failed = True
296 self._settings_snapshot = settings_snapshot
297 self._egress_context = self._build_egress_context(settings_snapshot)
299 # Debug settings manager and user context
300 logger.info(
301 f"[DOWNLOAD_SERVICE] Settings manager initialized: {type(self.settings)}, username: {self.username}"
302 )
304 # Get library path from settings (uses centralized path, respects LDR_DATA_DIR)
305 storage_path_setting = self.settings.get_setting(
306 "research_library.storage_path",
307 str(get_library_directory()),
308 )
309 logger.debug(
310 f"[DOWNLOAD_SERVICE_INIT] Storage path setting retrieved: {storage_path_setting} (type: {type(storage_path_setting)})"
311 )
313 if storage_path_setting is None:
314 logger.error(
315 "[DOWNLOAD_SERVICE_INIT] CRITICAL: storage_path_setting is None!"
316 )
317 raise ValueError("Storage path setting cannot be None")
319 # Per-user library root (issue #5521). PDFs are written to a per-user
320 # subdirectory (<base>/<username>) instead of the shared base, so two
321 # users' per-user autoincrement resource ids can't collide on the same
322 # `pdfs/<resource_id>.pdf` path in one shared directory. This mirrors
323 # get_library_storage_path()'s layout but derives the base from this
324 # user's own settings manager (self.settings) so it stays correct in
325 # background/scheduler threads that have no request context.
326 base_library_root = (
327 Path(os.path.expandvars(storage_path_setting))
328 .expanduser()
329 .resolve()
330 )
331 shared_library = self.settings.get_setting(
332 "research_library.shared_library", False
333 )
334 self.library_root = str(
335 apply_user_subdir(base_library_root, self.username, shared_library)
336 )
337 logger.debug(
338 f"[DOWNLOAD_SERVICE_INIT] Library root resolved to: {self.library_root}"
339 )
341 # Create directory structure
342 self._setup_directories()
344 # Initialize modular downloaders
346 # Get Semantic Scholar API key from settings
347 semantic_scholar_api_key = self.settings.get_setting(
348 "search.engine.web.semantic_scholar.api_key", ""
349 )
351 # Same for OpenAlex: the downloader hits api.openalex.org for every
352 # work lookup, so it needs the key too. It is passed through raw —
353 # ``OpenAlexDownloader.__init__`` runs ``normalize_openalex_api_key``
354 # itself, so placeholders/sentinels/non-strings become None
355 # (keyless free tier) there, in one place.
356 openalex_api_key = self.settings.get_setting(
357 "search.engine.web.openalex.api_key", ""
358 )
360 # ArxivDownloader precedes DirectPDFDownloader: for a paper URL it
361 # builds the canonical PDF URL rather than fetching whatever the
362 # input pointed at. It can only do that for URLs that name a paper,
363 # and can_handle now says so, so arXiv's legacy /ftp/ PDF layout --
364 # which yields no identifier -- still reaches DirectPDFDownloader
365 # instead of being claimed and skipped.
366 self.downloaders = [
367 ArxivDownloader(timeout=30),
368 DirectPDFDownloader(timeout=30),
369 SemanticScholarDownloader(
370 timeout=30,
371 api_key=semantic_scholar_api_key
372 if semantic_scholar_api_key
373 else None,
374 ),
375 # OpenAlex work lookup; the key is optional (it raises the
376 # daily budget 10x) and None means the keyless free tier.
377 OpenAlexDownloader(timeout=30, api_key=openalex_api_key),
378 PubMedDownloader(timeout=30, rate_limit_delay=1.0),
379 BioRxivDownloader(timeout=30),
380 GenericDownloader(timeout=30), # Generic should be last (fallback)
381 ]
383 # Initialize retry manager for smart failure tracking
384 self.retry_manager = RetryManager(username, password)
385 logger.info(
386 f"[DOWNLOAD_SERVICE] Initialized retry manager for user: {username}"
387 )
389 # PubMed rate limiting state
390 self._pubmed_delay = 1.0 # 1 second delay for PubMed
391 self._last_pubmed_request = 0.0 # Track last request time
393 def close(self):
394 """Close all downloader resources."""
395 if self._closed:
396 return
397 self._closed = True
399 from ...utilities.resource_utils import safe_close
401 for downloader in self.downloaders:
402 safe_close(downloader, "downloader")
404 # Close the settings manager's DB session to return the connection
405 # to the pool. SettingsManager.close() is idempotent.
406 safe_close(self.settings, "settings manager", allow_none=True)
408 # Clear references to allow garbage collection
409 self.downloaders = []
410 self.retry_manager = None
411 self.settings = None
413 def __enter__(self):
414 """Enter context manager."""
415 return self
417 def __exit__(self, exc_type, exc_val, exc_tb):
418 """Exit context manager, ensuring cleanup."""
419 self.close()
420 return False
422 def _setup_directories(self):
423 """Create library directory structure."""
424 from ...security.directory_creation import create_directory
426 # Only create the root and pdfs folder - flat structure.
427 paths = [
428 self.library_root,
429 str(Path(self.library_root) / _TRUSTED_STORAGE_LEAF),
430 ]
431 for path in paths:
432 create_directory(
433 path,
434 context="download service library directory",
435 )
437 def _normalize_url(self, url: str) -> str:
438 """Normalize URL for consistent hashing.
440 Legacy spelling (pre-B2 #6758): retained for back-compat callers
441 and unit tests, but NOT used by :meth:`_get_url_hash` anymore —
442 that path goes through the shared ``normalize_url_for_hash``
443 (lowercase-before-sort) so writer and readers agree (issue #6760
444 follow-up).
445 """
446 # Remove protocol variations
447 url = re.sub(r"^https?://", "", url)
448 # Remove www
449 url = re.sub(r"^www\.", "", url)
450 # Remove trailing slashes
451 url = url.rstrip("/")
452 # Sort query parameters
453 if "?" in url:
454 base, query = url.split("?", 1)
455 params = sorted(query.split("&"))
456 url = f"{base}?{'&'.join(params)}"
457 return url.lower()
459 def _get_url_hash(self, url: str) -> str:
460 """Generate SHA256 hash of normalized URL.
462 B2 (issue #6758): canonicalize the raw URL first so tracking
463 params, fragments, default ports, and userinfo hash alike — the
464 legacy ``_normalize_url`` strip alone strips the scheme before the
465 canonical form can be parsed, which would silently keep the gap.
467 The writer must hash exactly what the readers hash: ``get_url_hash``
468 already applies :func:`normalize_url_for_hash` (canonicalize +
469 strip + lowercase-before-sort), so composing ``_normalize_url`` here
470 double-normalizes — ``sort(raw-case).lower()`` followed by a second
471 ``sort(lower)`` — and diverges from the single-normalized reader on
472 mixed-case query URLs (issue #6760 follow-up: writer hash in none
473 of the reader candidates → sync deletes the document, re-download
474 then hits ``uq_download_tracker_url_hash``). Pass the canonical
475 form straight into ``get_url_hash``.
476 """
477 return get_url_hash(canonical_url_key(url))
479 def _build_egress_context(self, settings_snapshot):
480 """Build an EgressContext from the supplied snapshot. Returns
481 None when no snapshot is available — callers fall through to
482 the pre-policy behavior (back-compat with non-scheduler callers).
484 Sets ``self._policy_locked = True`` when a snapshot WAS supplied
485 but the policy itself cannot be evaluated (corrupt scope or
486 missing primary). The check_url method honors
487 this flag and fails closed — the previous code returned None on
488 PolicyDeniedError and check_url then returned ``(True,
489 "no_context")``, which silently allowed every download under a
490 misconfigured policy.
491 """
492 self._policy_locked = False
493 # Use an explicit `is None` check, not a truthiness test: an empty
494 # dict {} is a real (if minimal) snapshot, and `if not {}` would
495 # silently skip policy gating for it — fail-open. Match the
496 # `is None` guard used by the LLM/embeddings gates.
497 if settings_snapshot is None:
498 return None
499 try:
500 from ...security.egress.policy import (
501 PolicyDeniedError,
502 context_from_snapshot,
503 resolve_run_primary_engine,
504 )
505 except ImportError:
506 logger.debug(
507 "egress_policy unavailable in DownloadService; "
508 "downloads will not be scope-gated"
509 )
510 return None
512 try:
513 return context_from_snapshot(
514 settings_snapshot,
515 resolve_run_primary_engine(
516 settings_snapshot, default=DEFAULT_SEARCH_TOOL
517 ),
518 username=self.username,
519 )
520 except (PolicyDeniedError, ValueError) as exc:
521 logger.bind(policy_audit=True).warning(
522 "DownloadService policy unavailable; locking out "
523 "every URL check",
524 reason=str(exc),
525 )
526 self._policy_locked = True
527 return None
529 def _check_url_against_policy(
530 self, url: str, *, record_denial: bool = True
531 ) -> Tuple[bool, str]:
532 """Run evaluate_url against self._egress_context. Returns
533 (allowed, reason). When no context is configured (legacy
534 callers OR tests that mock __init__), returns (True, "no_context")
535 so back-compat callers aren't broken — but new callers passing
536 a snapshot DO get gated. Logged to policy_audit on denial.
538 ``record_denial=False`` returns the same decision without recording
539 a denial: it is not logged, here or by ``evaluate_url``, and does
540 not count toward the run's denied-fetch quota. Only ``refill_rank``
541 passes it, since it fetches nothing on the decision. Every other
542 caller checks a URL before it is queued or fetched, and records its
543 denials.
544 """
545 # ``_policy_locked`` is set by _build_egress_context when a
546 # snapshot was supplied but the policy itself raised. Without
547 # this short-circuit, _egress_context is None and the legacy
548 # fall-through below would silently allow every URL.
549 if getattr(self, "_policy_locked", False): 549 ↛ 550line 549 didn't jump to line 550 because the condition on line 549 was never true
550 return False, "policy_unavailable"
552 # Reading the user's settings raised during __init__ — we could not
553 # determine the scope, so fail closed instead of ungating downloads.
554 if getattr(self, "_snapshot_build_failed", False): 554 ↛ 555line 554 didn't jump to line 555 because the condition on line 554 was never true
555 return False, "settings_unavailable"
557 # Use getattr so tests that mock DownloadService.__init__ (and
558 # therefore never set _egress_context) still work — they
559 # operated under the pre-policy contract.
560 egress_context = getattr(self, "_egress_context", None)
561 if egress_context is None:
562 return True, "no_context"
563 from ...security.egress.policy import evaluate_url
564 from ...security.ssrf_validator import redact_url_for_log
566 decision = evaluate_url(
567 url, egress_context, record_denial=record_denial
568 )
569 if not decision.allowed and record_denial:
570 logger.bind(policy_audit=True).warning(
571 "DownloadService URL denied by egress policy",
572 # Redact userinfo creds / query-string tokens from the log.
573 url=redact_url_for_log(url),
574 scope=egress_context.scope.value,
575 reason=decision.reason,
576 )
577 return decision.allowed, decision.reason
579 def is_already_downloaded(self, url: str) -> Tuple[bool, Optional[str]]:
580 """
581 Check if URL is already downloaded.
583 Returns:
584 Tuple of (is_downloaded, file_path)
585 """
586 # B2 compat (#6758): pre-upgrade rows still carry a legacy url_hash;
587 # match canonical first, then legacy spellings (sequential
588 # filter_by so MagicMock-forced misses keep working).
589 with get_user_db_session(self.username, self.password) as session:
590 candidates = get_url_hash_candidates(url)
591 tracker = None
592 for candidate in candidates:
593 tracker = (
594 session.query(DownloadTracker)
595 .filter_by(url_hash=candidate, is_downloaded=True)
596 .first()
597 )
598 if tracker is not None:
599 break
601 if tracker and tracker.file_path:
602 # Compute absolute path and verify file still exists. Pass the
603 # username so per-user files resolve, with legacy-shared
604 # fallback for pre-isolation downloads (issue #5521). Pass
605 # self.settings so storage_path/shared_library are read from
606 # this user's captured manager — the ambient one is db-less in
607 # background/scheduler threads and would resolve the default
608 # root, missing a user's previously saved storage_path. Use
609 # getattr: __init__ always sets self.settings in production
610 # (incl. scheduler threads), but test doubles that construct
611 # a DownloadService via __new__/partial mocks may not — fall
612 # back to None (the ambient get_settings_manager()) rather
613 # than raising AttributeError.
614 absolute_path = get_absolute_path_from_settings(
615 tracker.file_path,
616 self.username,
617 settings_manager=getattr(self, "settings", None),
618 )
619 if absolute_path and absolute_path.is_file():
620 return True, str(absolute_path)
621 if absolute_path:
622 # File was deleted, mark as not downloaded
623 tracker.is_downloaded = False
624 session.commit()
625 # If absolute_path is None, path was blocked - treat as not downloaded
627 return False, None
629 def get_text_content(self, resource_id: int) -> Optional[str]:
630 """
631 Get text content for a research resource.
633 This will try to:
634 1. Produce text through a source-specific downloader if available
635 2. Extract text from downloaded PDF if exists
636 3. Download PDF and extract text if not yet downloaded
638 Args:
639 resource_id: ID of the research resource
641 Returns:
642 Text content as string, or None if extraction failed
643 """
644 with get_user_db_session(self.username, self.password) as session:
645 resource = session.get(ResearchResource, resource_id)
646 if not resource:
647 logger.error(f"Resource {resource_id} not found")
648 return None
650 url = resource.url
652 # Egress policy gate — denied URLs short-circuit before any
653 # downloader fires a request.
654 allowed, reason = self._check_url_against_policy(url)
655 if not allowed: 655 ↛ 656line 655 didn't jump to line 656 because the condition on line 655 was never true
656 logger.warning(
657 f"Skipping text extraction for {redact_url_for_log(url)}: refused by "
658 f"egress policy ({reason})"
659 )
660 return None
662 # Find appropriate downloader
663 for downloader in self.downloaders:
664 if downloader.can_handle(url):
665 logger.info(
666 f"Using {downloader.__class__.__name__} for text extraction from {redact_url_for_log(url)}"
667 )
668 try:
669 # Try to get text content
670 text = downloader.download_text(url)
671 if text:
672 logger.info(
673 f"Successfully extracted text for: {resource.title[:50]}"
674 )
675 return text
676 except Exception as e:
677 # Type name only: downloader exception texts can
678 # embed the raw resource URL.
679 logger.warning(
680 f"Failed to extract text ({type(e).__name__}) "
681 f"for {redact_url_for_log(url)}"
682 )
683 break
685 logger.warning(
686 f"Could not extract text for {redact_url_for_log(url)}"
687 )
688 return None
690 def queue_research_downloads(
691 self,
692 research_id: str,
693 collection_id: Optional[str] = None,
694 *,
695 text_only: bool = False,
696 run_research_ids: Optional[AbstractSet[str]] = None,
697 ) -> int:
698 """
699 Queue all downloadable PDFs from a research session.
701 Args:
702 research_id: The research session ID
703 collection_id: Optional target collection ID (defaults to Library if not provided)
704 text_only: True when the queued resources only get their text
705 extracted (``download_bulk``'s "text_only" mode). A completed
706 document that awaits its PDF (see ``document_awaits_pdf``)
707 already holds its text, so it then counts as done, like any
708 completed document. A text extraction would not fetch its PDF
709 either, so it would be queued and reported again on every run.
710 run_research_ids: The research sessions whose queued downloads
711 the same run downloads, as a set (``download_bulk`` builds
712 one from its ``research_ids`` for the whole run and passes it
713 for each research), or None when nothing downloads the queue
714 right away. It is only looked up, never copied, unless it
715 lacks ``research_id``. Decides which resource refills a
716 document that awaits its PDF, see the notes.
718 Returns:
719 Number of items queued
721 Notes:
722 Resets existing FAILED/COMPLETED queue entries for the research back
723 to PENDING, effectively retrying them. Resources that already have a
724 PENDING queue entry or a COMPLETED Document are skipped, except a
725 COMPLETED Document that awaits its PDF when PDFs are queued. As of #4685,
726 ``download_bulk`` calls this unconditionally on every bulk run, so
727 previously-failed downloads are automatically retried each time.
728 A PROCESSING entry is in flight in another ``download_bulk`` stream,
729 so it is left alone rather than reset: resetting it would let this
730 run re-claim and re-download it (#4691).
732 A document that awaits its PDF can be refilled through its own
733 resource or any resource linked to it by content
734 (``resource.document_id``), and one download refills it. So a
735 run leaves the document to one of them (see
736 ``_refilled_by_another_resource``): to one that a download holds,
737 else to one already queued for the run, unless that one is not
738 viable and another ranks before it, else to the one that ranks
739 first in ``refill_rank``. That is the document's own resource
740 unless it cannot succeed (its URL cannot be queued, the retry
741 manager holds it back, or its latest attempt failed) while a
742 linked one can. Once the document holds its PDF again,
743 ``download_resource`` reports the document's own resource as
744 already downloaded, and for a linked one whose URL the egress
745 policy allows, ``_download_pdf`` reuses that stored PDF instead
746 of fetching it. Of this research's resources the document's own
747 one is considered first.
748 """
749 queued = 0
750 # The research sessions whose queue the same run downloads (see
751 # _refilled_by_another_resource). download_bulk passes one set for
752 # the whole run, so it is only looked up here: copying it for each
753 # research made the bulk pre-pass quadratic in the number of
754 # research sessions.
755 run = run_research_ids
756 if run is not None and research_id not in run: 756 ↛ 757line 756 didn't jump to line 757 because the condition on line 756 was never true
757 run = run | {research_id}
758 # The refill_rank of each resource compared below, by resource and
759 # document owner, so that the egress policy and the retry manager
760 # are asked once per resource in this call.
761 ranks: Dict[Tuple[int, Optional[int]], RefillRank] = {}
763 # Get default library collection if no collection_id provided
764 if not collection_id:
765 from ...database.library_init import get_default_library_id
767 collection_id = get_default_library_id(self.username, self.password)
769 with get_user_db_session(self.username, self.password) as session:
770 # Get all resources for this research
771 resources = (
772 session.query(ResearchResource)
773 .filter_by(research_id=research_id)
774 .all()
775 )
777 # A document's own resource before the resources linked to it
778 # by content, so that the own resource of a document awaiting
779 # its PDF is considered first when this research has both. A
780 # resource can also point to its own document through
781 # document_id. The sort is stable and leaves the other
782 # resources in order. It reads the owner of each document these
783 # resources point to, in one query of bare columns: loading the
784 # documents would load their text (#4560).
785 owner_of_linked_document = (
786 dict(
787 session.query(ResearchResource.id, Document.resource_id)
788 .join(Document, ResearchResource.document_id == Document.id)
789 .filter(ResearchResource.research_id == research_id)
790 .all()
791 )
792 if any(res.document_id for res in resources)
793 else {}
794 )
796 def linked_to_another_resources_document(res) -> bool:
797 return bool(res.document_id) and (
798 owner_of_linked_document.get(res.id) != res.id
799 )
801 for resource in sorted(
802 resources, key=linked_to_another_resources_document
803 ):
804 if self._is_downloadable(resource):
805 # The document awaiting its PDF that this resource would
806 # refill, if any (see document_awaits_pdf).
807 awaiting = None
808 # Library resources linked via document_id are already
809 # done, unless PDFs are queued and the linked document
810 # still needs its PDF (see document_awaits_pdf).
811 if resource.document_id:
812 # Not its text, which nothing here reads (#4560).
813 linked = (
814 None
815 if text_only
816 else session.query(Document)
817 .options(defer(Document.text_content))
818 .filter(Document.id == resource.document_id)
819 .first()
820 )
821 if linked is None or not document_awaits_pdf(linked):
822 continue
823 awaiting = linked
825 # Egress policy gate before queueing.
826 allowed, reason = self._check_url_against_policy(
827 resource.url
828 )
829 if not allowed:
830 logger.warning(
831 f"Skipping queue for {redact_url_for_log(resource.url)}: refused by "
832 f"egress policy ({reason})"
833 )
834 continue
836 # Check if already queued
837 existing_queue = (
838 session.query(LibraryDownloadQueue)
839 .filter_by(
840 resource_id=resource.id,
841 status=DocumentStatus.PENDING,
842 )
843 .first()
844 )
846 # Check if already downloaded (trust the database status).
847 # When PDFs are queued, a completed document that still
848 # needs its PDF does not count (see document_awaits_pdf).
849 # An explicit link is authoritative; an older retained
850 # document can still name this resource as its owner.
851 existing_doc = (
852 awaiting
853 if resource.document_id
854 else session.query(Document)
855 .filter_by(
856 resource_id=resource.id,
857 status=DocumentStatus.COMPLETED,
858 )
859 .first()
860 )
861 if (
862 not text_only
863 and existing_doc is not None
864 and document_awaits_pdf(existing_doc)
865 ):
866 awaiting = existing_doc
867 existing_doc = None
869 # Queue if not already queued and not marked as completed
870 if not existing_queue and not existing_doc:
871 # One download refills a document awaiting its PDF:
872 # leave it to the resource that does.
873 if (
874 awaiting is not None
875 and self._refilled_by_another_resource(
876 session,
877 awaiting,
878 resource,
879 run,
880 research_id,
881 ranks,
882 )
883 ):
884 continue
885 # Check one more time if ANY queue entry exists (regardless of status)
886 any_queue = (
887 session.query(LibraryDownloadQueue)
888 .filter_by(resource_id=resource.id)
889 .first()
890 )
892 if any_queue:
893 # Reset the existing queue entry, but never one a
894 # concurrent download_bulk stream has claimed: a
895 # PROCESSING row is in flight, and resetting it to
896 # PENDING lets this stream re-claim and re-download
897 # a resource the other stream is still downloading
898 # (issue #4691). The status guard lives inside the
899 # UPDATE, not in an `if` above it, so the check and
900 # the write cannot straddle another stream's claim.
901 reset = (
902 session.query(LibraryDownloadQueue)
903 .filter(
904 LibraryDownloadQueue.id == any_queue.id,
905 LibraryDownloadQueue.status
906 != DocumentStatus.PROCESSING,
907 )
908 .update(
909 {
910 LibraryDownloadQueue.status: DocumentStatus.PENDING,
911 LibraryDownloadQueue.research_id: research_id,
912 LibraryDownloadQueue.collection_id: collection_id,
913 },
914 synchronize_session=False,
915 )
916 )
917 if reset:
918 queued += 1
919 else:
920 # Add new queue entry
921 queue_entry = LibraryDownloadQueue(
922 resource_id=resource.id,
923 research_id=research_id,
924 collection_id=collection_id,
925 priority=0,
926 status=DocumentStatus.PENDING,
927 )
928 session.add(queue_entry)
929 queued += 1
931 session.commit()
932 logger.info(
933 f"Queued {queued} downloads for research {research_id} to collection {collection_id}"
934 )
936 return queued
938 def _is_downloadable(self, resource: ResearchResource) -> bool:
939 """Check if a resource is likely downloadable as PDF.
941 Delegates to the consolidated is_downloadable_url() from utils.
942 """
943 return is_downloadable_url(resource.url)
945 def refill_rank(
946 self, resource_id: int, url: Optional[str], owner_id: Optional[int]
947 ) -> RefillRank:
948 """The rank of a resource that would refill a document awaiting its PDF.
950 ``owner_id`` is the document's own resource (``Document.resource_id``).
951 See ``RefillRank``: a viable resource ranks before one that is not,
952 and among viable resources the document's own one ranks first. Of
953 the others, one that can be queued ranks before one that cannot, and
954 one the retry manager would retry now before one it holds back. The
955 rank asks the egress policy and the retry manager, whose record of
956 attempts the "Download All New" retry-policy filter reads too.
957 ``queue_research_downloads`` and the "Download All New" route both
958 choose the resource that refills a document by this rank.
960 Ranking fetches nothing, so it records no denial of the egress
961 policy: a refused URL is neither written to the audit log nor
962 counted toward the run's denied-fetch quota (see
963 ``_check_url_against_policy``). Fetching the URL still checks it and
964 records the denial, and so does queueing it in
965 ``queue_research_downloads``.
966 """
967 cannot_queue = not (
968 url
969 and is_downloadable_url(url)
970 and self._check_url_against_policy(url, record_denial=False)[0]
971 )
972 held_back = failed_last = False
973 if self.retry_manager: 973 ↛ 984line 973 didn't jump to line 984 because the condition on line 973 was always true
974 held_back = not self.retry_manager.should_retry_resource(
975 resource_id
976 ).can_retry
977 status = self.retry_manager.status_tracker.get_resource_status(
978 resource_id
979 )
980 failed_last = status is not None and status["status"] in (
981 RetryStatus.TEMPORARILY_FAILED.value,
982 RetryStatus.PERMANENTLY_FAILED.value,
983 )
984 return RefillRank(
985 cannot_queue=cannot_queue,
986 held_back=held_back,
987 failed_last=failed_last,
988 linked=resource_id != owner_id,
989 resource_id=resource_id,
990 )
992 def _refilled_by_another_resource(
993 self,
994 session: Session,
995 document: Document,
996 resource: ResearchResource,
997 run: Optional[AbstractSet[str]],
998 research_id: str,
999 ranks: Dict[Tuple[int, Optional[int]], RefillRank],
1000 ) -> bool:
1001 """Whether a resource other than ``resource`` refills ``document``.
1003 A document that awaits its PDF (see ``document_awaits_pdf``) is
1004 refilled by a download of its own resource (``Document.resource_id``,
1005 unless that resource points to another document through
1006 ``document_id``) or of any resource linked to it by content
1007 (``ResearchResource.document_id``): each downloads into it, see
1008 ``get_document_for_resource``. One such download refills it, so
1009 ``queue_research_downloads`` queues ``resource`` only when this is
1010 False. ``research_id`` is the research being queued, ``run`` the
1011 research sessions of the run (see ``queue_research_downloads``),
1012 and ``ranks`` holds the ``refill_rank`` of the resources compared
1013 so far in that call. True when another of these resources
1015 * has a PROCESSING queue entry that a download still holds (see
1016 ``download_claim_is_live``): another stream is downloading it.
1017 An entry that a restart left PROCESSING does not count, and that
1018 resource is not queued either, since no reset touches a
1019 PROCESSING entry;
1020 * has a PENDING queue entry that the same run downloads, i.e. of a
1021 research session in ``run``, and is viable or ranks before
1022 ``resource`` (see ``RefillRank``). Without a run (``run`` None,
1023 when nothing downloads the queue right away) any PENDING entry
1024 counts: the run that downloads the queue later checks again with
1025 its own research sessions. Within a run, a PENDING entry of a
1026 research session outside it does not count, since the run would
1027 not download it;
1028 * ranks before ``resource`` and would be queued by the same call or
1029 run: its research session is ``research_id`` or one in ``run``,
1030 and it has no queue entry PENDING or PROCESSING. Ranking before
1031 ``resource``, which can be queued, it can be queued too.
1033 So, unless one of them is being downloaded, a resource that cannot
1034 succeed does not keep a viable one from refilling the document, and
1035 of the viable ones in the run the document's own resource refills
1036 it, whichever research the run queues first, unless another viable
1037 one is already queued for the run.
1038 """
1039 # Bare columns: the full rows would load each resource's
1040 # content_preview (#4560).
1041 others = (
1042 session.query(
1043 ResearchResource.id,
1044 ResearchResource.url,
1045 ResearchResource.research_id,
1046 )
1047 .filter(
1048 ResearchResource.id != resource.id,
1049 or_(
1050 ResearchResource.document_id == document.id,
1051 and_(
1052 ResearchResource.id == document.resource_id,
1053 ResearchResource.document_id.is_(None),
1054 ),
1055 ),
1056 )
1057 .all()
1058 )
1059 if not others:
1060 return False
1061 entries = {
1062 entry.resource_id: entry
1063 for entry in session.query(
1064 LibraryDownloadQueue.resource_id,
1065 LibraryDownloadQueue.status,
1066 LibraryDownloadQueue.research_id,
1067 LibraryDownloadQueue.last_attempt_at,
1068 ).filter(
1069 LibraryDownloadQueue.resource_id.in_(
1070 [other.id for other in others]
1071 ),
1072 LibraryDownloadQueue.status.in_(
1073 (DocumentStatus.PENDING, DocumentStatus.PROCESSING)
1074 ),
1075 )
1076 }
1078 def rank_of(res) -> RefillRank:
1079 key = (res.id, document.resource_id)
1080 if key not in ranks:
1081 ranks[key] = self.refill_rank(
1082 res.id, res.url, document.resource_id
1083 )
1084 return ranks[key]
1086 own_rank = rank_of(resource)
1087 queued_now = run if run is not None else {research_id}
1088 for other in others:
1089 entry = entries.get(other.id)
1090 if entry is not None and entry.status == DocumentStatus.PROCESSING:
1091 if download_claim_is_live(entry.last_attempt_at):
1092 return True
1093 continue
1094 rank = rank_of(other)
1095 if entry is not None:
1096 if (run is None or entry.research_id in run) and (
1097 rank.viable or rank < own_rank
1098 ):
1099 return True
1100 continue
1101 if rank < own_rank and other.research_id in queued_now:
1102 return True
1103 return False
1105 def download_resource(self, resource_id: int) -> Tuple[bool, Optional[str]]:
1106 """
1107 Download a specific resource.
1109 Returns:
1110 Tuple of (success: bool, skip_reason: str or None)
1111 """
1112 with get_user_db_session(self.username, self.password) as session:
1113 resource = session.get(ResearchResource, resource_id)
1114 if not resource:
1115 logger.error(f"Resource {resource_id} not found")
1116 return False, "Resource not found"
1118 # Check if already downloaded (trust the database after sync).
1119 # A completed document that the sync made text-only because its
1120 # PDF file was missing still needs its PDF, so it is downloaded
1121 # again into that document (see document_awaits_pdf).
1122 existing_query = session.query(Document).filter_by(
1123 resource_id=resource_id, status=DocumentStatus.COMPLETED
1124 )
1125 if resource.document_id:
1126 # Never short-circuit on a retained older document. Keep
1127 # linked-document retries on the normal path so they repair
1128 # queue, tracker, and collection bookkeeping.
1129 existing_query = existing_query.filter(
1130 Document.id == resource.document_id
1131 )
1132 existing_doc = existing_query.first()
1133 if existing_doc and not document_awaits_pdf(existing_doc):
1134 logger.info(
1135 "Resource already downloaded (according to database)"
1136 )
1137 return True, None
1139 # Get collection_id from queue entry if it exists
1140 queue_entry = (
1141 session.query(LibraryDownloadQueue)
1142 .filter_by(resource_id=resource_id)
1143 .first()
1144 )
1145 collection_id = (
1146 queue_entry.collection_id
1147 if queue_entry and queue_entry.collection_id
1148 else None
1149 )
1151 # Create download tracker entry. B2 compat (#6758): reuse a
1152 # pre-upgrade legacy-hash row when present instead of inserting
1153 # a second tracker for the same URL.
1154 url_hash = self._get_url_hash(resource.url)
1155 tracker = find_tracker_by_url(session, resource.url)
1157 if not tracker:
1158 # Wrap the new-tracker INSERT in a SAVEPOINT so it is
1159 # structurally separated from the shared thread-local
1160 # session's outer transaction. Three reasons:
1161 # 1. The savepoint commit makes the new tracker row
1162 # durable even if a later step on this thread rolls
1163 # back the outer transaction.
1164 # 2. If the INSERT itself raises (e.g. UNIQUE conflict
1165 # on a concurrent same-URL insert), only the savepoint
1166 # rolls back — the caller's outer transaction state
1167 # is untouched.
1168 # 3. The savepoint scopes ``session.expunge(tracker)``
1169 # below: the pending INSERT is part of the savepoint's
1170 # write-set, so the expunge does not cancel a write
1171 # that was about to promote.
1172 # Note: the savepoint does NOT prevent the unconditional
1173 # ``session.commit()`` at the bottom of this block from
1174 # promoting caller-staged rows already in the outer
1175 # transaction — SQLAlchemy's savepoint commit only absorbs
1176 # the inner writes; the outer commit still flushes
1177 # everything. None of the three current call sites stage
1178 # rows across the call (see callers in web/routers/library.py);
1179 # the discipline is documented at the call sites.
1180 # Honesty note on durability: with nothing else staged, the
1181 # SAVEPOINT is the first statement in autocommit so its
1182 # RELEASE commits immediately (a second connection already
1183 # sees the row; a later rollback() is a no-op) — durability
1184 # here rests on that driver behavior, not on savepoint
1185 # nesting alone.
1186 with session.begin_nested():
1187 tracker = DownloadTracker(
1188 url=resource.url,
1189 url_hash=url_hash,
1190 first_resource_id=resource.id,
1191 is_downloaded=False,
1192 )
1193 session.add(tracker)
1194 # Explicit flush inside the savepoint — the savepoint's
1195 # __exit__ does auto-flush its writes into the outer
1196 # transaction, but a future SQLAlchemy change or an
1197 # autoflush=False caller would silently drop the INSERT
1198 # before ``session.expunge(tracker)`` cancels it. The
1199 # flush makes the FK-on-first-download invariant explicit
1200 # at this site rather than relying on savepoint-exit
1201 # side effects. See PR #6564 P1 review.
1202 session.flush()
1204 # Snapshot plain values for the post-download block (#6560, #6564)
1205 resource_id_val = resource.id
1206 resource_url = resource.url
1208 # Expunge instances so their loaded attributes do not expire on commit
1209 # and accessing them outside the session will not trigger lazy reload.
1210 if hasattr(session, "expunge"): 1210 ↛ 1218line 1210 didn't jump to line 1218 because the condition on line 1210 was always true
1211 session.expunge(resource)
1212 session.expunge(tracker)
1214 # Unconditionally commit to release the connection back to QueuePool (#6564).
1215 # The new-tracker INSERT is enclosed in the savepoint above and is
1216 # therefore already durable; this commit only closes the outer
1217 # transaction and does not promote any caller-staged row.
1218 session.commit()
1220 # Attempt download outside database session (#6560)
1221 success, skip_reason, status_code = self._download_pdf(
1222 resource, tracker, None, collection_id
1223 )
1225 # Open short transaction for post-download queue and retry updates
1226 with get_user_db_session(self.username, self.password) as session:
1227 # Record attempt with retry manager for smart failure tracking
1228 self.retry_manager.record_attempt(
1229 resource_id=resource_id_val,
1230 result=(success, skip_reason),
1231 status_code=status_code,
1232 url=resource_url,
1233 details=skip_reason
1234 or (
1235 "Successfully downloaded" if success else "Download failed"
1236 ),
1237 session=session,
1238 )
1240 # Update queue status if exists
1241 queue_entry = (
1242 session.query(LibraryDownloadQueue)
1243 .filter_by(resource_id=resource_id_val)
1244 .first()
1245 )
1247 if queue_entry:
1248 queue_entry.status = (
1249 DocumentStatus.COMPLETED
1250 if success
1251 else DocumentStatus.FAILED
1252 )
1253 queue_entry.completed_at = datetime.now(UTC)
1255 session.commit()
1257 # Trigger auto-indexing for successfully downloaded documents
1258 if success and self.password:
1259 try:
1260 from ...web.routers.rag import trigger_auto_index
1261 from ...database.library_init import get_default_library_id
1263 # Resolve via get_document_for_resource so a dedup-linked
1264 # resource (resource.document_id -> canonical Document
1265 # owned by another resource) still triggers indexing —
1266 # the old resource_id-only lookup missed it and skipped
1267 # auto-indexing in the requested collection (#6758).
1268 resource_for_index = session.get(
1269 ResearchResource, resource_id
1270 )
1271 doc = (
1272 get_document_for_resource(session, resource_for_index)
1273 if resource_for_index is not None
1274 else None
1275 )
1276 if doc is None:
1277 doc = (
1278 session.query(Document)
1279 .filter_by(resource_id=resource_id)
1280 .order_by(Document.created_at.desc())
1281 .first()
1282 )
1283 if doc:
1284 # Use collection_id from queue entry or default Library
1285 # NB: pass username string, not the SQLAlchemy session
1286 target_collection = (
1287 collection_id
1288 or get_default_library_id(
1289 self.username, self.password
1290 )
1291 )
1292 if target_collection: 1292 ↛ 1314line 1292 didn't jump to line 1314 because the condition on line 1292 was always true
1293 trigger_auto_index(
1294 [doc.id],
1295 target_collection,
1296 self.username,
1297 self.password,
1298 )
1299 except Exception as exc:
1300 # The Document SELECT above runs on the shared session; a
1301 # connection-level failure there leaves it needing a
1302 # rollback. Auto-indexing is best-effort so we swallow and
1303 # return — the with-block exits normally and
1304 # get_user_db_session's rollback never fires. Recover the
1305 # session here so a later op on it (same request/thread)
1306 # doesn't cascade. (No-op for the other failures reachable
1307 # here: get_default_library_id self-heals via its own inner
1308 # get_user_db_session block, and trigger_auto_index does its
1309 # DB work off-thread, so neither dirties this session.)
1310 safe_rollback(session, "download_resource auto-index")
1311 log_exception_type("Failed to trigger auto-indexing", exc)
1312 finally:
1313 # Release any checked-out connection from the auto-index query
1314 session.commit()
1316 return success, skip_reason
1318 def _download_pdf(
1319 self,
1320 resource: ResearchResource,
1321 tracker: DownloadTracker,
1322 session: Optional[Session] = None,
1323 collection_id: Optional[str] = None,
1324 ) -> Tuple[bool, Optional[str], Optional[int]]:
1325 """
1326 Perform the actual PDF download.
1328 Args:
1329 resource: The research resource to download
1330 tracker: Download tracker for this URL
1331 session: Database session (optional; if None, short transaction opened after download)
1332 collection_id: Optional target collection ID (defaults to Library if not provided)
1334 Returns:
1335 Tuple of (success: bool, skip_reason: Optional[str], status_code: Optional[int])
1336 """
1337 url = resource.url
1338 # Snapshot the title before any session work. When the caller passes
1339 # session=None this instance is detached, and its columns are expired
1340 # by the caller's own commit, so the value has to be read while it is
1341 # still loaded. Canonical text production below uses it for its log
1342 # lines and runs with no session open.
1343 resource_title = resource.title
1345 # Egress policy gate at the network-fire point. The entry caller
1346 # (download_resource) does not gate, so this is the true PEP — a
1347 # denied URL must never reach a downloader regardless of route.
1348 allowed, reason = self._check_url_against_policy(url)
1349 if not allowed:
1350 logger.warning(
1351 f"Skipping PDF download for {redact_url_for_log(url)}: refused by egress "
1352 f"policy ({reason})"
1353 )
1354 return False, f"egress_policy_denied:{reason}", None
1356 attempted_at = datetime.now(UTC)
1357 try:
1358 # Use modular downloaders with detailed skip reasons.
1359 # Download occurs outside the database session (#6560).
1360 pdf_content = None
1361 selected_downloader = None
1362 skip_reason = None
1363 status_code = None
1365 if resource.document_id:
1366 # A duplicate resource points to another resource's saved
1367 # snapshot. Reuse its bytes while still repairing this
1368 # resource's tracker, collection membership and queue state.
1369 # Missing storage falls through to the normal retry path.
1370 cache_session_cm = (
1371 nullcontext(session)
1372 if session is not None
1373 else get_user_db_session(self.username, self.password)
1374 )
1375 with cache_session_cm as cache_session:
1376 cached_pdf = self._load_cached_pdf(cache_session, resource)
1377 if session is None:
1378 # This read belongs to the download. Release it even
1379 # when an outer route still owns a session scope.
1380 cache_session.rollback()
1381 if cached_pdf is not None:
1382 pdf_content = cached_pdf[0]
1384 for downloader in self.downloaders:
1385 if pdf_content is not None:
1386 break
1387 if downloader.can_handle(url):
1388 logger.info(
1389 f"Using {downloader.__class__.__name__} for {redact_url_for_log(url)}"
1390 )
1391 result = downloader.download_with_result(
1392 url, ContentType.PDF
1393 )
1394 selected_downloader = downloader
1396 if result.is_success and result.content:
1397 pdf_content = result.content
1398 status_code = result.status_code
1399 break
1400 if result.skip_reason:
1401 skip_reason = result.skip_reason
1402 status_code = result.status_code
1403 logger.info(
1404 f"Download skipped: {failure_reason_for_log(skip_reason)}"
1405 )
1406 if isinstance(
1407 downloader, (ArxivDownloader, GenericDownloader)
1408 ):
1409 break
1411 if selected_downloader is None and pdf_content is None:
1412 logger.error(
1413 f"No downloader found for {redact_url_for_log(url)}"
1414 )
1415 skip_reason = "No compatible downloader available"
1417 # Produce the canonical text BEFORE the persistence block.
1418 # download_text_with_source may issue a network (HTML) request,
1419 # and holding an uncommitted write transaction (Document row +
1420 # save_pdf blob) open across that round-trip extends the write
1421 # lock for the full network latency. No session is open here at
1422 # all, so that request also cannot sit on a checked-out
1423 # connection (#6560). Production and persistence both stay
1424 # best-effort: a failure in either must not fail the PDF
1425 # download itself.
1426 text_details: tuple[str, str, str] | None = None
1427 if pdf_content:
1428 try:
1429 if isinstance(selected_downloader, ArxivDownloader):
1430 logger.info(
1431 f"Producing canonical arXiv text for: {resource_title[:50]}"
1432 )
1433 arxiv_text = (
1434 selected_downloader.download_text_with_source(
1435 url,
1436 pdf_content=pdf_content,
1437 )
1438 )
1439 if arxiv_text is not None: 1439 ↛ 1456line 1439 didn't jump to line 1456 because the condition on line 1439 was always true
1440 extraction_method, extraction_source = (
1441 _arxiv_text_provenance(arxiv_text.source)
1442 )
1443 text_details = (
1444 arxiv_text.text,
1445 extraction_method,
1446 extraction_source,
1447 )
1448 else:
1449 logger.info(
1450 f"Extracting text from downloaded PDF for: {resource_title[:50]}"
1451 )
1452 text = self._extract_text_from_pdf(pdf_content)
1453 if text:
1454 text_details = (text, "pdf_extraction", "local_pdf")
1456 if text_details is None:
1457 logger.warning(
1458 f"Text production returned empty text for: {resource_title[:50]}"
1459 )
1460 except Exception as exc:
1461 logger.opt(exception=False).warning(
1462 "Failed to produce text after PDF download "
1463 "({error_type}), but PDF download succeeded",
1464 error_type=type(exc).__name__,
1465 )
1466 text_details = None
1468 # Open short transaction only after download completion to stage
1469 # the DownloadAttempt and update document records (#6560).
1470 session_cm = (
1471 nullcontext(session)
1472 if session is not None
1473 else get_user_db_session(self.username, self.password)
1474 )
1476 with session_cm as sess:
1477 if session is None:
1478 db_resource = sess.get(ResearchResource, resource.id)
1479 db_tracker = (
1480 sess.query(DownloadTracker)
1481 .filter_by(url_hash=tracker.url_hash)
1482 .first()
1483 )
1484 target_resource = (
1485 db_resource if db_resource is not None else resource
1486 )
1487 target_tracker = (
1488 db_tracker if db_tracker is not None else tracker
1489 )
1490 else:
1491 target_resource = resource
1492 target_tracker = tracker
1494 # Snapshot scalar resource values used AFTER the staging commit
1495 # so attribute access does not lazy-refresh a checked-out
1496 # connection through the staging session immediately before
1497 # text production's network/CPU work. PR #6564 P2 review.
1498 # Only the title is read in log lines after the staging
1499 # commit; the freshly fetched `db_resource` below already has
1500 # its scalar columns eagerly loaded, so passing it to
1501 # `_save_text_with_db` / `get_document_for_resource` is
1502 # safe (those read `id` and `document_id`, never lazy-loaded).
1503 staging_resource_title = target_resource.title
1505 # Helper for attempt number calculation
1506 if hasattr(target_tracker, "download_attempts") and hasattr(
1507 target_tracker.download_attempts, "count"
1508 ):
1509 attempt_num = target_tracker.download_attempts.count() + 1
1510 elif hasattr(tracker, "download_attempts") and hasattr( 1510 ↛ 1513line 1510 didn't jump to line 1513 because the condition on line 1510 was never true
1511 tracker.download_attempts, "count"
1512 ):
1513 attempt_num = tracker.download_attempts.count() + 1
1514 else:
1515 attempt_num = 1
1516 if not pdf_content:
1517 error_msg = skip_reason or "Failed to download PDF content"
1518 attempt = DownloadAttempt(
1519 url_hash=tracker.url_hash,
1520 attempt_number=attempt_num,
1521 attempted_at=attempted_at,
1522 succeeded=False,
1523 error_message=error_msg,
1524 )
1525 sess.add(attempt)
1526 # Record the failure in its own transaction so the row
1527 # survives whatever the caller does with the session
1528 # after this early return.
1529 sess.commit()
1530 logger.info(
1531 "Download failed with reason: "
1532 f"{failure_reason_for_log(error_msg)}"
1533 )
1534 return False, error_msg, status_code
1536 # Get PDF storage mode setting. Resolve through the operator gate
1537 # BEFORE any PDF is written: a stored value or an
1538 # LDR_RESEARCH_LIBRARY_PDF_STORAGE_MODE env override could still
1539 # read "filesystem" even though the settings UI withholds that
1540 # option, so the unencrypted mode is coerced to the encrypted
1541 # "database" default here unless the operator opted in. This is
1542 # the true enforcement point — hiding the dropdown alone is not
1543 # enough.
1544 pdf_storage_mode = resolve_pdf_storage_mode(
1545 self.settings.get_setting(
1546 "research_library.pdf_storage_mode", "none"
1547 )
1548 )
1549 max_pdf_size_mb = int(
1550 self.settings.get_setting(
1551 "research_library.max_pdf_size_mb",
1552 DEFAULT_MAX_PDF_SIZE_MB,
1553 )
1554 )
1555 logger.info(
1556 f"[DOWNLOAD_SERVICE] PDF storage mode: {pdf_storage_mode}"
1557 )
1559 # Update tracker
1560 import hashlib
1562 file_hash_val = hashlib.sha256(pdf_content).hexdigest()
1563 file_size_val = len(pdf_content)
1564 downloaded_at_val = datetime.now(UTC)
1566 existing_doc = get_document_for_resource(sess, target_resource)
1567 if (
1568 existing_doc
1569 and document_awaits_pdf(existing_doc)
1570 and existing_doc.document_hash != file_hash_val
1571 and sess.query(Document.id)
1572 .filter(Document.document_hash == file_hash_val)
1573 .first()
1574 is not None
1575 ):
1576 # Keeping the old hash while storing different bytes
1577 # poisons later content deduplication: a download of
1578 # the original PDF would then serve the replacement.
1579 # Preserve both documents until the caller can relink
1580 # to the canonical owner, before changing any storage,
1581 # text, or tracker state.
1582 error_msg = (
1583 "PDF refill conflicts with another library document"
1584 )
1585 sess.add(
1586 DownloadAttempt(
1587 url_hash=target_tracker.url_hash,
1588 attempt_number=attempt_num,
1589 attempted_at=attempted_at,
1590 succeeded=False,
1591 error_message=error_msg,
1592 )
1593 )
1594 sess.commit()
1595 return False, error_msg, status_code
1597 target_tracker.file_hash = file_hash_val
1598 target_tracker.file_size = file_size_val
1599 target_tracker.is_downloaded = True
1600 target_tracker.downloaded_at = downloaded_at_val
1601 if hasattr(tracker, "file_hash"): 1601 ↛ 1608line 1601 didn't jump to line 1608 because the condition on line 1601 was always true
1602 tracker.file_hash = file_hash_val
1603 tracker.file_size = file_size_val
1604 tracker.is_downloaded = True
1605 tracker.downloaded_at = downloaded_at_val
1607 # Read and write PDFs only within the per-user root.
1608 pdf_storage_manager = PDFStorageManager(
1609 library_root=self.library_root,
1610 storage_mode=pdf_storage_mode,
1611 max_pdf_size_mb=max_pdf_size_mb,
1612 )
1614 # Update attempt with success info
1615 attempt = DownloadAttempt(
1616 url_hash=tracker.url_hash,
1617 attempt_number=attempt_num,
1618 attempted_at=attempted_at,
1619 succeeded=True,
1620 bytes_downloaded=len(pdf_content),
1621 )
1622 sess.add(attempt)
1624 canonical_retry = False
1625 # Older placeholder recovery could leave a nonempty PDF
1626 # labelled "unknown"; that content still needs preservation.
1627 if (
1628 existing_doc
1629 and existing_doc.document_hash != file_hash_val
1630 and not document_awaits_pdf(existing_doc)
1631 and (
1632 existing_doc.status == DocumentStatus.COMPLETED
1633 or (
1634 existing_doc.file_type in ("pdf", "unknown")
1635 and existing_doc.file_size > 0
1636 )
1637 or pdf_storage_manager.has_pdf(existing_doc, sess)
1638 )
1639 ):
1640 # Preserve a saved snapshot even if later extraction
1641 # failed, or the link appeared while the download was in
1642 # flight. Resolve changed bytes by hash without rewriting
1643 # the document used by other resources.
1644 # A sync-marked missing PDF may refill the same record;
1645 # the collision guard above already protects other owners.
1646 existing_doc = None
1647 canonical_retry = True
1648 if (
1649 existing_doc
1650 and existing_doc.status == DocumentStatus.FAILED
1651 ):
1652 # A retry can discover bytes already owned by another
1653 # document. Resolve that owner BEFORE replacing the failed
1654 # hash, otherwise the next autoflush violates uniqueness.
1655 owner = (
1656 sess.query(Document)
1657 .filter_by(document_hash=file_hash_val)
1658 .first()
1659 )
1660 if owner and owner.id != existing_doc.id:
1661 _link_retry_to_canonical(
1662 sess, target_resource, existing_doc, owner
1663 )
1664 existing_doc = None
1665 canonical_retry = True
1666 if (
1667 existing_doc
1668 and existing_doc.resource_id != target_resource.id
1669 and existing_doc.document_hash == file_hash_val
1670 ):
1671 # A repeated retry now resolves through document_id.
1672 # It may itself be FAILED after another extraction attempt.
1673 # Use the dedup storage path again so a none-mode retry
1674 # cannot downgrade the canonical PDF's stored bytes.
1675 existing_doc = None
1676 canonical_retry = True
1678 if existing_doc:
1679 pdf_document_id = existing_doc.id
1680 refills_missing_pdf = document_awaits_pdf(existing_doc)
1681 # Update existing document. Only replace document_hash when
1682 # transitioning from FAILED — that's the placeholder hash from
1683 # _record_failed_text_extraction. For any other prior state
1684 # the hash is already a real content hash and clobbering it
1685 # risks UNIQUE-constraint collisions (issue #3827).
1686 was_failed = existing_doc.status == DocumentStatus.FAILED
1687 if was_failed:
1688 existing_doc.document_hash = target_tracker.file_hash
1689 existing_doc.error_message = None
1690 existing_doc.file_size = len(pdf_content)
1691 existing_doc.status = DocumentStatus.COMPLETED
1692 existing_doc.processed_at = datetime.now(UTC)
1694 # A document that the library sync made text-only after
1695 # its PDF file went missing gets its PDF back here, in
1696 # the same document, whatever the bytes hash to (see
1697 # document_awaits_pdf). A publisher or arXiv can serve
1698 # a regenerated PDF, so the bytes can differ from the
1699 # lost ones. The document then takes their hash, as a
1700 # new download would. A hash already owned by another
1701 # document was refused before any writes above.
1702 if (
1703 refills_missing_pdf
1704 and existing_doc.document_hash != file_hash_val
1705 ):
1706 existing_doc.document_hash = file_hash_val
1708 if refills_missing_pdf and collection_id:
1709 # A linked resource may request a different collection.
1710 # Keep its membership when changed bytes refill this
1711 # record instead of taking the canonical dedup path.
1712 ensure_in_collection(
1713 sess, existing_doc.id, collection_id
1714 )
1716 # Save PDF using storage manager (updates storage_mode and file_path)
1717 file_path_result, _ = pdf_storage_manager.save_pdf(
1718 pdf_content=pdf_content,
1719 document=existing_doc,
1720 session=sess,
1721 filename=f"{target_resource.id}.pdf",
1722 url=url,
1723 resource_id=target_resource.id,
1724 )
1725 if file_path_result:
1726 # The PDF is now stored in this document, so label
1727 # it a PDF, as a new download is. A failed-download
1728 # placeholder is labelled "unknown" and a text
1729 # extraction "text". Either label hides the stored
1730 # PDF from PDFStorageManager.has_pdf. Once the sync
1731 # has made the document text-only, it also keeps
1732 # document_awaits_pdf from asking for the PDF again.
1733 existing_doc.file_type = "pdf"
1734 existing_doc.mime_type = "application/pdf"
1735 elif refills_missing_pdf:
1736 # The PDF of a document that still needed one was
1737 # downloaded but not stored, as in storage mode
1738 # "none" or for a PDF over the size limit. Leave the
1739 # document as a "none"-mode download leaves one, so
1740 # that the download paths stop fetching it again.
1741 existing_doc.file_path = None
1743 # Update tracker
1744 target_tracker.file_path = (
1745 file_path_result if file_path_result else None
1746 )
1747 target_tracker.file_name = (
1748 Path(file_path_result).name
1749 if file_path_result and file_path_result != "database"
1750 else None
1751 )
1752 if hasattr(tracker, "file_path"): 1752 ↛ 2030line 1752 didn't jump to line 2030 because the condition on line 1752 was always true
1753 tracker.file_path = target_tracker.file_path
1754 tracker.file_name = target_tracker.file_name
1755 else:
1756 # B1 (issue #6758): content-hash dedup across resources.
1757 # Two different URLs can yield identical PDF bytes. Without
1758 # a pre-check the INSERT below violates UNIQUE(document_hash)
1759 # and the second resource ends up document-less after the
1760 # IntegrityError rollback. Mirror the text-path pre-check in
1761 # _save_text_with_db: link to the canonical Document instead.
1762 existing_by_hash = (
1763 sess.query(Document)
1764 .filter_by(document_hash=target_tracker.file_hash)
1765 .first()
1766 )
1767 if existing_by_hash:
1768 pdf_document_id = existing_by_hash.id
1769 target_resource.document_id = existing_by_hash.id
1770 if canonical_retry:
1771 if (
1772 existing_by_hash.status
1773 in (
1774 DocumentStatus.COMPLETED,
1775 DocumentStatus.FAILED,
1776 )
1777 and existing_by_hash.text_content
1778 ):
1779 # Preserve the canonical extraction and its
1780 # indexed chunks even when a later extraction
1781 # failed. FAILED does not mean text is absent.
1782 text_details = None
1783 if existing_by_hash.status == DocumentStatus.FAILED:
1784 existing_by_hash.status = (
1785 DocumentStatus.COMPLETED
1786 )
1787 existing_by_hash.error_message = None
1788 existing_by_hash.processed_at = datetime.now(
1789 UTC
1790 )
1791 try:
1792 dedup_collection_id = (
1793 collection_id
1794 or get_default_library_id(
1795 self.username, self.password
1796 )
1797 )
1798 except Exception as exc:
1799 log_exception_type(
1800 "Failed to get library collection for dedup link",
1801 exc,
1802 )
1803 dedup_collection_id = None
1804 if dedup_collection_id:
1805 ensure_in_collection(
1806 sess,
1807 existing_by_hash.id,
1808 dedup_collection_id,
1809 )
1810 else:
1811 logger.warning(
1812 f"Library collection not found - deduped document {existing_by_hash.id} will not be linked to default collection"
1813 )
1814 # Preserve/upgrade PDF storage on the dedup hit path.
1815 # Without this, a first download with storage "none"
1816 # followed by identical bytes with storage "database"
1817 # leaves the canonical document with no saved PDF while
1818 # the second tracker claims success. Only upgrade (never
1819 # downgrade): when the canonical already stores the PDF,
1820 # mirror its location onto this tracker for consistency.
1821 try:
1822 missing_pdf = canonical_retry and not (
1823 pdf_storage_manager.has_pdf(
1824 existing_by_hash, sess
1825 )
1826 )
1827 previous_paths = {existing_by_hash.file_path}
1828 if existing_by_hash.storage_mode == "database":
1829 previous_paths.add("database")
1830 needs_upgrade = (
1831 existing_by_hash.storage_mode in (None, "none")
1832 or missing_pdf
1833 ) and (
1834 pdf_storage_mode in ("database", "filesystem")
1835 )
1836 if needs_upgrade:
1837 file_path_result, _ = (
1838 pdf_storage_manager.save_pdf(
1839 pdf_content=pdf_content,
1840 document=existing_by_hash,
1841 session=sess,
1842 filename=f"{target_resource.id}.pdf",
1843 url=url,
1844 resource_id=target_resource.id,
1845 )
1846 )
1847 target_tracker.file_path = (
1848 file_path_result
1849 if file_path_result
1850 else None
1851 )
1852 target_tracker.file_name = (
1853 Path(file_path_result).name
1854 if file_path_result
1855 and file_path_result != "database"
1856 else None
1857 )
1858 if file_path_result:
1859 # The PDF is now stored in this document:
1860 # label it a PDF, as the existing-document
1861 # branch above does.
1862 existing_by_hash.file_type = "pdf"
1863 existing_by_hash.mime_type = (
1864 "application/pdf"
1865 )
1866 # Keep sibling trackers for the same bytes
1867 # consistent: the canonical URL's tracker still
1868 # has file_path=None from its none-mode download,
1869 # which would otherwise make sync treat the now
1870 # upgraded document as missing.
1871 try:
1872 siblings = (
1873 sess.query(DownloadTracker)
1874 .filter_by(
1875 file_hash=target_tracker.file_hash
1876 )
1877 .all()
1878 )
1879 for sibling in siblings:
1880 if not sibling.file_path or (
1881 missing_pdf
1882 and sibling.file_path
1883 in previous_paths
1884 ):
1885 sibling.file_path = (
1886 target_tracker.file_path
1887 )
1888 sibling.file_name = (
1889 target_tracker.file_name
1890 )
1891 except Exception as exc:
1892 log_exception_type(
1893 "Failed to sync sibling trackers after storage upgrade",
1894 exc,
1895 )
1896 if canonical_retry:
1897 raise
1898 elif missing_pdf:
1899 # none-mode retries may recover text without
1900 # storing bytes. Do not advertise a lost file.
1901 target_tracker.file_path = None
1902 target_tracker.file_name = None
1903 elif existing_by_hash.storage_mode == "database":
1904 target_tracker.file_path = "database"
1905 target_tracker.file_name = None
1906 elif (
1907 existing_by_hash.file_path
1908 in FILE_PATH_SENTINELS
1909 ):
1910 # Never mirror file_path sentinels
1911 # (``blob_deleted``, ``text_only_not_stored``,
1912 # ``metadata_only``) into the tracker: they name
1913 # no file, and the sync mid-branch would resolve
1914 # them as a relative path, miss, and run
1915 # ``delete_document_completely`` on the
1916 # text/chunks/vectors the sentinel was protecting
1917 # (issue #6760 follow-up). Leave the duplicate
1918 # tracker path-less, the same shape as a fresh
1919 # none-mode download.
1920 target_tracker.file_path = None
1921 target_tracker.file_name = None
1922 elif existing_by_hash.file_path:
1923 target_tracker.file_path = (
1924 existing_by_hash.file_path
1925 )
1926 target_tracker.file_name = (
1927 Path(existing_by_hash.file_path).name
1928 if existing_by_hash.file_path != "database"
1929 else None
1930 )
1931 else:
1932 # Canonical has no stored file (none-mode): same
1933 # shape as a fresh none-mode download
1934 # (is_downloaded=True, no path). Bounded
1935 # re-download, not data loss.
1936 target_tracker.file_path = None
1937 target_tracker.file_name = None
1938 if document_awaits_pdf(existing_by_hash):
1939 # Storage was disabled or refused this PDF's
1940 # size. Settle a linked refill as a text-only
1941 # download, just like the own-document path,
1942 # so every bulk run does not fetch it again.
1943 existing_by_hash.file_path = None
1944 except Exception as exc:
1945 log_exception_type(
1946 "Failed to reuse/upgrade deduped PDF storage",
1947 exc,
1948 )
1949 if canonical_retry: 1949 ↛ 1951line 1949 didn't jump to line 1951 because the condition on line 1949 was always true
1950 raise
1951 if hasattr(tracker, "file_path"): 1951 ↛ 1954line 1951 didn't jump to line 1954 because the condition on line 1951 was always true
1952 tracker.file_path = target_tracker.file_path
1953 tracker.file_name = target_tracker.file_name
1954 logger.info(
1955 f"Linked resource {target_resource.id} to existing Document "
1956 f"{existing_by_hash.id} (matched on content hash)"
1957 )
1958 else:
1959 # Get source type ID for research downloads
1960 try:
1961 source_type_id = get_source_type_id(
1962 self.username,
1963 "research_download",
1964 self.password,
1965 )
1966 # Use provided collection_id or default to Library
1967 library_collection_id = (
1968 collection_id
1969 or get_default_library_id(
1970 self.username, self.password
1971 )
1972 )
1973 except Exception as exc:
1974 log_exception_type(
1975 "Failed to get source type or library collection",
1976 exc,
1977 )
1978 raise
1980 # Create new unified document entry
1981 doc_id = str(uuid.uuid4())
1982 pdf_document_id = doc_id
1983 doc = Document(
1984 id=doc_id,
1985 source_type_id=source_type_id,
1986 resource_id=target_resource.id,
1987 research_id=target_resource.research_id,
1988 document_hash=target_tracker.file_hash,
1989 original_url=url,
1990 file_size=len(pdf_content),
1991 file_type="pdf",
1992 mime_type="application/pdf",
1993 title=target_resource.title,
1994 status=DocumentStatus.COMPLETED,
1995 processed_at=datetime.now(UTC),
1996 storage_mode=pdf_storage_mode,
1997 )
1998 sess.add(doc)
1999 sess.flush() # Ensure doc.id is available for blob storage
2000 target_resource.document_id = doc.id
2001 # Save PDF using storage manager (updates storage_mode and file_path)
2002 file_path_result, _ = pdf_storage_manager.save_pdf(
2003 pdf_content=pdf_content,
2004 document=doc,
2005 session=sess,
2006 filename=f"{target_resource.id}.pdf",
2007 url=url,
2008 resource_id=target_resource.id,
2009 )
2011 # Update tracker
2012 target_tracker.file_path = (
2013 file_path_result if file_path_result else None
2014 )
2015 target_tracker.file_name = (
2016 Path(file_path_result).name
2017 if file_path_result
2018 and file_path_result != "database"
2019 else None
2020 )
2021 if hasattr(tracker, "file_path"): 2021 ↛ 2026line 2021 didn't jump to line 2026 because the condition on line 2021 was always true
2022 tracker.file_path = target_tracker.file_path
2023 tracker.file_name = target_tracker.file_name
2025 # Link document to default Library collection
2026 ensure_in_collection(
2027 sess, doc_id, library_collection_id
2028 )
2030 if pdf_storage_mode == "database":
2031 logger.info(
2032 f"Successfully stored PDF in database: {redact_url_for_log(resource.url)}"
2033 )
2034 elif pdf_storage_mode == "filesystem":
2035 logger.info(f"Successfully downloaded: {tracker.file_path}")
2036 else:
2037 logger.info(
2038 f"Successfully extracted text from: {redact_url_for_log(resource.url)}"
2039 )
2041 # Commit the PDF persistence as its own transaction BEFORE
2042 # the best-effort text save below. Until this point the
2043 # attempt row, the Document row, the PDF blob and the
2044 # collection link are all uncommitted. _save_text_with_db
2045 # rolls its session back before re-raising (issue #3827),
2046 # so without this commit a DB error in the text save would
2047 # discard every one of those writes while the caller went on
2048 # to mark the queue COMPLETED and record a retry success —
2049 # the user is told success and the database keeps nothing.
2050 # The caller's later commit stays correct: it opens a fresh
2051 # transaction for the retry/queue bookkeeping that follows.
2052 sess.commit()
2054 # Save the produced text only after the PDF has been stored and
2055 # committed, in its own transaction.
2056 if text_details is not None:
2057 text, extraction_method, extraction_source = text_details
2058 text_session_cm = (
2059 nullcontext(session)
2060 if session is not None
2061 else get_user_db_session(self.username, self.password)
2062 )
2063 try:
2064 with text_session_cm as text_sess:
2065 # The resource link can change after the PDF commit.
2066 # Claim the exact stored snapshot before saving its
2067 # extracted text; a no-op write also prevents deletion
2068 # or hash replacement until this transaction ends.
2069 claimed = (
2070 text_sess.query(Document)
2071 .filter_by(
2072 id=pdf_document_id,
2073 document_hash=file_hash_val,
2074 )
2075 .update(
2076 {Document.id: pdf_document_id},
2077 synchronize_session=False,
2078 )
2079 )
2080 if claimed != 1:
2081 if session is None:
2082 text_sess.rollback()
2083 logger.info(
2084 "Skipping text because its PDF snapshot changed"
2085 )
2086 return True, None, status_code
2088 # Save text to encrypted database
2089 self._save_text_with_db(
2090 resource=target_resource,
2091 text=text,
2092 session=text_sess,
2093 extraction_method=extraction_method,
2094 extraction_source=extraction_source,
2095 pdf_document_id=pdf_document_id,
2096 )
2097 if session is None:
2098 text_sess.commit()
2099 logger.info(
2100 f"Successfully produced and saved text for: {staging_resource_title[:50]}"
2101 )
2102 except Exception as exc:
2103 log_exception_type(
2104 "Failed to save text after PDF download, but PDF download succeeded",
2105 exc,
2106 )
2108 return True, None, status_code
2110 except Exception as e:
2111 logger.opt(exception=False).warning(
2112 "Download failed ({error_type}) for {url}",
2113 error_type=type(e).__name__,
2114 url=redact_url_for_log(url),
2115 )
2116 # Two distinct outputs:
2117 # * server-side diagnostic (persisted in DownloadAttempt.error_message)
2118 # — full str(e), with credential scrubbing applied before storage.
2119 # * caller-safe skip_reason (the second tuple element, and what
2120 # flows into retry-manager / bulk-stream SSE messages) — must
2121 # never carry SQL text, target URLs, or file paths. The Zotero
2122 # sync_service uses the same shape: look up str(exc) in a fixed
2123 # map of author-written constants and otherwise return just the
2124 # exception class name. sanitize_error_for_client() is not a
2125 # substitute — its docstring says it scrubs credentials only.
2126 server_error = sanitize_error_for_client(str(e))
2127 skip_reason = _client_safe_download_message(e)
2128 tracker.is_accessible = False
2129 if session is not None:
2130 safe_rollback(session, "_download_pdf")
2131 session.add(
2132 DownloadAttempt(
2133 url_hash=tracker.url_hash,
2134 attempt_number=tracker.download_attempts.count() + 1
2135 if hasattr(tracker, "download_attempts")
2136 else 1,
2137 attempted_at=datetime.now(UTC),
2138 succeeded=False,
2139 error_type=type(e).__name__,
2140 error_message=server_error,
2141 )
2142 )
2143 else:
2144 try:
2145 with get_user_db_session(
2146 self.username, self.password
2147 ) as err_sess:
2148 db_tracker = (
2149 err_sess.query(DownloadTracker)
2150 .filter_by(url_hash=tracker.url_hash)
2151 .first()
2152 )
2153 if db_tracker:
2154 db_tracker.is_accessible = False
2155 err_sess.add(
2156 DownloadAttempt(
2157 url_hash=tracker.url_hash,
2158 attempt_number=tracker.download_attempts.count()
2159 + 1
2160 if hasattr(tracker, "download_attempts")
2161 else 1,
2162 attempted_at=datetime.now(UTC),
2163 succeeded=False,
2164 error_type=type(e).__name__,
2165 error_message=server_error,
2166 )
2167 )
2168 err_sess.commit()
2169 except Exception as exc:
2170 log_exception_type(
2171 "Failed to record failed attempt in database", exc
2172 )
2173 return False, skip_reason, None
2175 def _extract_text_from_pdf(self, pdf_content: bytes) -> Optional[str]:
2176 """
2177 Extract text from PDF content using multiple methods for best results.
2179 pdfplumber is tried first; PyPDF runs only when pdfplumber produced
2180 no text. What is bounded (``utilities/pdf_extraction_limits``): at
2181 most ``MAX_PDF_EXTRACTION_PAGES`` pages are extracted per attempt,
2182 at most ``MAX_PDF_EXTRACTED_CHARS`` characters are returned
2183 (separators included; the page that crosses the ceiling is
2184 truncated to the budget left), and ONE CPU-time budget of
2185 ``MAX_PDF_EXTRACTION_CPU_SECONDS`` covers the whole extraction,
2186 both attempts included. It is measured with
2187 ``time.thread_time()``, the calling thread's own CPU time, so
2188 waiting for the GIL or for other extractions does not spend it.
2189 If the budget is spent when pdfplumber finishes without text,
2190 the PyPDF fallback is not started. Every pdfplumber page the walk
2191 reaches is closed once done with, the break paths included, which
2192 releases its per-page caches (parsed layout objects, text map).
2193 Closing does not release what pdfminer decoded and parsed for the
2194 page -- decoded streams, parsed objects, fonts -- which it keeps
2195 on the document, nor does pypdf release its own; both walks hand
2196 each page to a releaser (``utilities/pdf_stream_release``) that
2197 releases those caches once they reach
2198 ``MAX_PDF_RETAINED_DECODED_BYTES``, so what the walk holds
2199 between pages stays near that threshold.
2201 What is NOT bounded: the budget is checked between pages only,
2202 so a single page's ``extract_text()`` cannot be interrupted. The
2203 pdfplumber path walks pages lazily, but the PyPDF fallback may
2204 eagerly materialize the page tree during ``PdfReader`` construction
2205 or its first access to ``pages``. That operation can take O(all
2206 pages) time and memory before the page ceiling can stop the walk.
2207 What one page decodes and parses is held until the release after it.
2209 Args:
2210 pdf_content: Raw PDF bytes
2212 Returns:
2213 Extracted text or None if extraction fails
2214 """
2215 # One CPU budget for the whole extraction, shared by both attempts:
2216 # a per-attempt budget would let a document that exhausts the
2217 # pdfplumber budget buy a second full budget (and a second
2218 # page-tree parse) in the PyPDF fallback. thread_time() counts
2219 # only this thread's CPU, not wall time, so GIL waits and other
2220 # requests' extractions do not spend it (GIL-switch overhead
2221 # does; the budget is sized for that, see the constant).
2222 deadline = time.thread_time() + MAX_PDF_EXTRACTION_CPU_SECONDS
2223 try:
2224 # First try with pdfplumber (better for complex layouts)
2225 import io
2227 from ...utilities.pdf_stream_release import (
2228 PdfplumberWalkReleaser,
2229 PypdfWalkReleaser,
2230 )
2231 from ...utilities.resource_utils import safe_close
2232 from ...utilities.pdf_page_walk import iter_pdfplumber_pages
2234 # Extraction ceilings: the storage cap bounds file SIZE, not
2235 # extraction COMPLEXITY. These limits bound work and retained
2236 # text during the lazy page walk. The page ceiling is checked
2237 # before extracting text from the first page beyond the limit;
2238 # character and CPU budgets are enforced during the walk.
2239 pdf_stream = io.BytesIO(pdf_content)
2240 pdf = pdfplumber.open(pdf_stream)
2241 try:
2242 text_parts = []
2243 extracted_chars = 0
2244 releaser = PdfplumberWalkReleaser(pdf)
2245 for page_number, page in enumerate(
2246 iter_pdfplumber_pages(pdf, releaser)
2247 ):
2248 # pdfplumber keeps each page's parsed layout objects
2249 # and text map cached on the Page until close(): about
2250 # 25 MB per dense page, so walking to the page ceiling
2251 # without closing would hold gigabytes. Close every
2252 # page this loop reaches, on the break paths too.
2253 try:
2254 if page_number >= MAX_PDF_EXTRACTION_PAGES:
2255 logger.warning(
2256 "PDF extraction stopped at page ceiling "
2257 f"({MAX_PDF_EXTRACTION_PAGES}); document exceeds "
2258 "the page ceiling"
2259 )
2260 break
2261 if time.thread_time() >= deadline:
2262 logger.warning(
2263 "PDF extraction stopped at the CPU-time ceiling "
2264 f"({MAX_PDF_EXTRACTION_CPU_SECONDS}s) after "
2265 f"{page_number} pages"
2266 )
2267 break
2268 # The count includes the next join separator. Stop
2269 # before extraction when no text from this page fits.
2270 remaining = MAX_PDF_EXTRACTED_CHARS - extracted_chars
2271 if remaining <= 0:
2272 logger.warning(
2273 "PDF extraction stopped at character ceiling "
2274 f"({MAX_PDF_EXTRACTED_CHARS})"
2275 )
2276 break
2277 page_text = page.extract_text()
2278 if page_text:
2279 if len(page_text) > remaining:
2280 # Keep the slice that fits instead of dropping
2281 # the tripping page: a first page over the
2282 # ceiling must still yield the budget's worth
2283 # of text, not nothing.
2284 if remaining > 0: 2284 ↛ 2287line 2284 didn't jump to line 2287 because the condition on line 2284 was always true
2285 text_parts.append(page_text[:remaining])
2286 extracted_chars += remaining
2287 logger.warning(
2288 "PDF extraction truncated at character "
2289 f"ceiling ({MAX_PDF_EXTRACTED_CHARS})"
2290 )
2291 break
2292 # + 2 for the "\n\n" separator the join adds.
2293 extracted_chars += len(page_text) + 2
2294 text_parts.append(page_text)
2295 finally:
2296 # safe_close: a failing close() must not mask the
2297 # page's own exception or discard text kept so far.
2298 safe_close(page, "pdfplumber page")
2299 # close() leaves what pdfminer decoded and parsed
2300 # for the page cached on the document; release it
2301 # once it reaches MAX_PDF_RETAINED_DECODED_BYTES.
2302 releaser.after_page()
2304 if text_parts:
2305 return "\n\n".join(text_parts)
2306 finally:
2307 safe_close(pdf_stream, "PDF stream")
2309 # Fallback to PyPDF if pdfplumber produced no text -- but only
2310 # within the same budget: do not start a second parse once the
2311 # CPU budget is spent.
2312 if time.thread_time() >= deadline:
2313 logger.warning(
2314 "PDF extraction CPU-time ceiling "
2315 f"({MAX_PDF_EXTRACTION_CPU_SECONDS}s) reached before any "
2316 "text was extracted; not starting the PyPDF fallback"
2317 )
2318 return None
2320 reader = PdfReader(io.BytesIO(pdf_content))
2321 text_parts = []
2322 extracted_chars = 0
2323 releaser = PypdfWalkReleaser(reader)
2324 for page_number, page in enumerate(reader.pages):
2325 if page_number >= MAX_PDF_EXTRACTION_PAGES:
2326 logger.warning(
2327 "PDF extraction (PyPDF fallback) stopped at page "
2328 f"ceiling ({MAX_PDF_EXTRACTION_PAGES})"
2329 )
2330 break
2331 if time.thread_time() >= deadline:
2332 logger.warning(
2333 "PDF extraction (PyPDF fallback) stopped at the "
2334 f"CPU-time ceiling ({MAX_PDF_EXTRACTION_CPU_SECONDS}s) "
2335 f"after {page_number} pages"
2336 )
2337 break
2338 remaining = MAX_PDF_EXTRACTED_CHARS - extracted_chars
2339 if remaining <= 0:
2340 logger.warning(
2341 "PDF extraction (PyPDF fallback) stopped at "
2342 f"character ceiling ({MAX_PDF_EXTRACTED_CHARS})"
2343 )
2344 break
2345 releaser.before_page()
2346 text = page.extract_text()
2347 # pypdf caches what it decodes and parses on the reader too.
2348 releaser.after_page()
2349 if text:
2350 if len(text) > remaining:
2351 if remaining > 0: 2351 ↛ 2354line 2351 didn't jump to line 2354 because the condition on line 2351 was always true
2352 text_parts.append(text[:remaining])
2353 extracted_chars += remaining
2354 logger.warning(
2355 "PDF extraction (PyPDF fallback) truncated at "
2356 f"character ceiling ({MAX_PDF_EXTRACTED_CHARS})"
2357 )
2358 break
2359 # + 2 for the "\n\n" separator the join adds.
2360 extracted_chars += len(text) + 2
2361 text_parts.append(text)
2363 if text_parts:
2364 return "\n\n".join(text_parts)
2366 logger.warning("No text could be extracted from PDF")
2367 return None
2369 except Exception as exc:
2370 log_exception_type("Failed to extract text from PDF", exc)
2371 return None
2373 def download_as_text(self, resource_id: int) -> Tuple[bool, Optional[str]]:
2374 """
2375 Download resource and extract text to encrypted database.
2377 Args:
2378 resource_id: ID of the resource to download
2380 Returns:
2381 Tuple of (success, error_message)
2382 """
2383 with get_user_db_session(self.username, self.password) as session:
2384 # Get the resource
2385 resource = (
2386 session.query(ResearchResource)
2387 .filter_by(id=resource_id)
2388 .first()
2389 )
2390 if not resource:
2391 return False, "Resource not found"
2393 # Handle library resources — content already in the database
2394 # (local-only, no network — skip retry checks)
2395 if resource.source_type == "library" or (
2396 resource.url and resource.url.startswith("/library/document/")
2397 ):
2398 return self._try_library_text_extraction(session, resource)
2400 # Try existing text in database
2401 result = self._try_existing_text(session, resource_id)
2402 if result is not None:
2403 return result
2405 # Try legacy text files on disk
2406 result = self._try_legacy_text_file(session, resource, resource_id)
2407 if result is not None: 2407 ↛ 2408line 2407 didn't jump to line 2408 because the condition on line 2407 was never true
2408 return result
2410 # Canonical arXiv sources, HTML first. This is network-bearing,
2411 # so it applies the retry gate itself; when that denies it, it
2412 # only reads the stored PDF and otherwise returns None rather
2413 # than a verdict, so the local-only strategies below still run.
2414 result = self._try_arxiv_text_extraction(session, resource)
2415 if result is not None:
2416 self._record_retry_attempt(resource, result, session)
2417 return result
2419 # Try extracting from existing PDF (pure filesystem, no network)
2420 result = self._try_existing_pdf_extraction(
2421 session, resource, resource_id
2422 )
2423 if result is not None:
2424 return result
2426 # Check retry eligibility before network-dependent extraction
2427 if self.retry_manager: 2427 ↛ 2443line 2427 didn't jump to line 2443 because the condition on line 2427 was always true
2428 retry_decision = self.retry_manager.should_retry_resource(
2429 resource_id
2430 )
2431 if not retry_decision.can_retry:
2432 # The reason can embed a stored failure_message; rows
2433 # written before the domain fix may hold URL userinfo.
2434 # The caller still gets the reason unchanged (the
2435 # user's own data).
2436 logger.info(
2437 f"Skipping resource {resource_id}: "
2438 f"{failure_reason_for_log(retry_decision.reason)}"
2439 )
2440 return False, retry_decision.reason
2442 # Try source-specific direct text extraction
2443 result = self._try_api_text_extraction(session, resource)
2444 if result is not None:
2445 self._record_retry_attempt(resource, result, session)
2446 return result
2448 # Fallback: Download PDF and extract
2449 result = self._fallback_pdf_extraction(session, resource)
2450 self._record_retry_attempt(resource, result, session)
2451 return result
2453 def _record_retry_attempt(
2454 self, resource, result: Tuple[bool, Optional[str]], session=None
2455 ) -> None:
2456 """Record a download attempt with the retry manager."""
2457 if not self.retry_manager: 2457 ↛ 2458line 2457 didn't jump to line 2458 because the condition on line 2457 was never true
2458 return
2459 self.retry_manager.record_attempt(
2460 resource_id=resource.id,
2461 result=result,
2462 url=resource.url or "",
2463 details=result[1]
2464 or (
2465 "Successfully extracted text"
2466 if result[0]
2467 else "Text extraction failed"
2468 ),
2469 session=session,
2470 )
2472 def _try_library_text_extraction(
2473 self, session, resource
2474 ) -> Tuple[bool, Optional[str]]:
2475 """Handle library resources — content already exists in local database.
2477 Returns:
2478 Tuple of (success, error_message). On success: (True, None).
2479 On failure: (False, description of what went wrong).
2480 """
2481 # 1. Extract document UUID from metadata (authoritative) or URL (fallback)
2482 doc_id = None
2483 metadata = (resource.resource_metadata or {}).get("original_data", {})
2484 if isinstance(metadata, dict):
2485 meta_inner = metadata.get("metadata", {})
2486 if isinstance(meta_inner, dict):
2487 doc_id = meta_inner.get("source_id") or meta_inner.get(
2488 "document_id"
2489 )
2491 if not doc_id and resource.url:
2492 # Parse from /library/document/{uuid} or /library/document/{uuid}/pdf
2493 match = re.match(r"^/library/document/([^/]+)", resource.url)
2494 if match: 2494 ↛ 2497line 2494 didn't jump to line 2497 because the condition on line 2494 was always true
2495 doc_id = match.group(1)
2497 if not doc_id:
2498 return False, "Could not extract library document ID"
2500 # 2. Query the existing Document by its primary key (UUID string)
2501 doc = session.query(Document).filter_by(id=doc_id).first()
2502 if not doc:
2503 return False, f"Library document {doc_id} not found in database"
2505 # 3. If text already exists and extraction succeeded, return success
2506 if _has_reusable_text(doc):
2507 logger.info(f"Library document {doc_id} already has text content")
2508 resource.document_id = doc.id
2509 session.commit()
2510 return True, None
2512 # 4. Try extracting text from stored PDF
2513 pdf_storage_mode = self.settings.get_setting(
2514 "research_library.pdf_storage_mode", "none"
2515 )
2516 pdf_manager = PDFStorageManager(
2517 library_root=self.library_root,
2518 storage_mode=pdf_storage_mode,
2519 )
2520 pdf_content = pdf_manager.load_pdf(doc, session)
2522 if not pdf_content:
2523 return (
2524 False,
2525 f"Library document {doc_id} has no text or PDF content",
2526 )
2528 text = self._extract_text_from_pdf(pdf_content)
2529 if not text:
2530 return (
2531 False,
2532 f"Failed to extract text from library document {doc_id} PDF",
2533 )
2535 # 5. Update Document directly (don't use _save_text_with_db — it
2536 # queries by resource_id which mismatches for library docs)
2537 doc.text_content = text
2538 doc.character_count = len(text)
2539 doc.word_count = len(text.split())
2540 doc.extraction_method = "pdf_extraction"
2541 doc.extraction_source = "pdfplumber"
2542 doc.extraction_quality = "medium"
2543 # A truncated extraction (page, character or CPU-time ceiling) is
2544 # only signalled by the warnings logged in
2545 # ``_extract_text_from_pdf``: this row records the text but not
2546 # that it is partial, so a 700-page book yields the text of its
2547 # first 500 pages and is indexed as if that were the whole book.
2548 # The ``Document`` model (``database/models/library.py``) has no
2549 # free-form metadata column to hang ``extraction_truncated`` on —
2550 # its JSON columns are ``authors`` and ``tags``, and
2551 # ``document_metadata`` belongs to ``DocumentChunk``. Persisting
2552 # the flag needs a schema change (a nullable ``Boolean`` beside
2553 # ``has_formatting_issues``/``has_encoding_issues``, or a JSON
2554 # ``extraction_metadata``), which is left to the owner. It would
2555 # be set here and wherever else ``_extract_text_from_pdf`` output
2556 # is saved: the three ``_save_text_with_db`` calls fed by it (in
2557 # ``_download_pdf``, ``_try_existing_pdf_extraction`` and
2558 # ``_fallback_pdf_extraction``), plus the ``_save_text_with_db``
2559 # calls in ``_try_arxiv_text_extraction``,
2560 # ``_text_from_cached_pdf`` (the stored-PDF read on a retry or
2561 # egress denial) and ``_try_api_text_extraction`` for downloader
2562 # text (arXiv PDF text among it) that
2563 # ``BaseDownloader.extract_text_from_pdf`` bounds the same way.
2565 resource.document_id = doc.id
2566 session.commit()
2568 logger.info(
2569 f"Extracted text from library document {doc_id} "
2570 f"({doc.word_count} words)"
2571 )
2572 return True, None
2574 def _try_existing_text(
2575 self, session, resource_id: int
2576 ) -> Optional[Tuple[bool, Optional[str]]]:
2577 """Check if text already exists in database (in Document.text_content)."""
2578 resource = session.get(ResearchResource, resource_id)
2579 existing_doc = (
2580 get_document_for_resource(session, resource)
2581 if resource is not None
2582 else session.query(Document)
2583 .filter_by(resource_id=resource_id)
2584 .first()
2585 )
2587 if not existing_doc:
2588 return None
2590 # Check if text content exists and extraction was successful
2591 if _has_reusable_text(existing_doc):
2592 logger.info(
2593 f"Text content already exists in Document for resource_id={resource_id}, extraction_method={existing_doc.extraction_method}"
2594 )
2595 return True, None
2597 # No text content or failed extraction
2598 logger.debug(
2599 f"Document exists but no valid text content: resource_id={resource_id}, extraction_method={existing_doc.extraction_method}"
2600 )
2601 return None # Fall through to re-extraction
2603 def _try_legacy_text_file(
2604 self, session, resource, resource_id: int
2605 ) -> Optional[Tuple[bool, Optional[str]]]:
2606 """Check for legacy text files on disk."""
2607 txt_path = Path(self.library_root) / "txt"
2608 existing_files = (
2609 # nosemgrep: semgrep.rules.path-traversal-risk, reason: Per-user library text directory and integer resource ID from the user DB; this only enumerates legacy names
2610 list(txt_path.glob(f"*_{resource_id}.txt"))
2611 if txt_path.exists()
2612 else []
2613 )
2615 if not existing_files:
2616 return None
2618 logger.info(f"Text file already exists on disk: {existing_files[0]}")
2619 self._create_text_document_record(
2620 session,
2621 resource,
2622 existing_files[0],
2623 extraction_method="unknown",
2624 extraction_source="legacy_file",
2625 )
2626 session.commit()
2627 return True, None
2629 def _try_arxiv_text_extraction(
2630 self, session: Session, resource
2631 ) -> Optional[Tuple[bool, Optional[str]]]:
2632 """Produce text from canonical arXiv sources, or None to fall through.
2634 None means "this path does not apply", never "this resource has no
2635 text": the caller must go on to the local-only strategies. A URL that
2636 names no paper returns None. So does an exhausted retry budget once
2637 the stored PDF has been tried, because this path is the only
2638 network-bearing one that runs ahead of _try_existing_pdf_extraction.
2639 Deciding the whole call here would have denied a resource with a
2640 stored PDF the local read it used to get (download_as_text applies
2641 the same gate itself, after that read, so the budget still holds).
2643 Both network denials read the stored PDF first, entirely locally,
2644 through _text_from_cached_pdf. That reader loads the PDF through
2645 PDFStorageManager, so it covers the default database storage, where
2646 the document has no file path and _try_existing_pdf_extraction
2647 finds nothing. Text produced that way satisfies the call. Otherwise
2648 (no stored PDF, no text, or a failed read or save) a retry denial
2649 returns None and an egress denial stands with its own reason.
2650 """
2651 url = resource.url or ""
2652 if not is_arxiv_paper_url(url):
2653 return None
2655 # Retry gate before any network access. A denial still reads the
2656 # stored PDF, which makes no network request and so is outside what
2657 # the retry gate limits. Only a success is returned from here;
2658 # anything else returns None rather than a verdict, so the
2659 # local-only strategies below still run and download_as_text's own
2660 # gate, after them, returns the denial (so the budget still holds).
2661 if self.retry_manager: 2661 ↛ 2687line 2661 didn't jump to line 2687 because the condition on line 2661 was always true
2662 retry_decision = self.retry_manager.should_retry_resource(
2663 resource.id
2664 )
2665 if not retry_decision.can_retry:
2666 # The reason can embed a stored failure_message (see
2667 # download_as_text).
2668 logger.info(
2669 "Skipping canonical arXiv text for resource {}: {}",
2670 resource.id,
2671 failure_reason_for_log(retry_decision.reason),
2672 )
2673 try:
2674 cached_outcome = self._text_from_cached_pdf(
2675 session, resource
2676 )
2677 except Exception as exc:
2678 safe_rollback(session, "_try_arxiv_text_extraction")
2679 log_exception_type(
2680 "Stored arXiv PDF text extraction failed", exc
2681 )
2682 return None
2683 if cached_outcome is not None and cached_outcome[0]:
2684 return cached_outcome
2685 return None
2687 try:
2688 downloader = self._get_downloader(url)
2689 if not isinstance(downloader, ArxivDownloader): 2689 ↛ 2690line 2689 didn't jump to line 2690 because the condition on line 2689 was never true
2690 return False, "No arXiv downloader available"
2692 allowed, reason = self._check_url_against_policy(url)
2693 if not allowed:
2694 # Egress policy refuses the network, but a stored PDF can
2695 # still produce text entirely locally before the denial
2696 # stands. Only a success replaces the denial: a failed
2697 # read or save must not bury the egress_policy_denied
2698 # reason the caller records.
2699 try:
2700 cached_outcome = self._text_from_cached_pdf(
2701 session, resource
2702 )
2703 except Exception as exc:
2704 safe_rollback(session, "_try_arxiv_text_extraction")
2705 log_exception_type(
2706 "Stored arXiv PDF text extraction failed", exc
2707 )
2708 cached_outcome = None
2709 if cached_outcome is not None and cached_outcome[0]:
2710 return cached_outcome
2711 return False, f"egress_policy_denied:{reason}"
2713 pdf_content: bytes | None = None
2714 pdf_document_id: str | None = None
2715 cached_pdf = self._load_cached_arxiv_pdf(session, resource)
2716 if cached_pdf is not None:
2717 pdf_content, pdf_document_id = cached_pdf
2719 if pdf_content is None:
2720 arxiv_text = downloader.download_text_with_source(url)
2721 else:
2722 arxiv_text = downloader.download_text_with_source(
2723 url,
2724 pdf_content=pdf_content,
2725 )
2726 except Exception as exc:
2727 safe_rollback(session, "_try_arxiv_text_extraction")
2728 logger.opt(exception=False).warning(
2729 "Canonical arXiv text extraction failed ({error_type})",
2730 error_type=type(exc).__name__,
2731 )
2732 # Caller-visible: fixed token only. sanitize_error_for_client()
2733 # scrubs credentials but leaves SQL text / URLs / paths intact,
2734 # and this string flows into download_as_text's return and the
2735 # bulk-stream SSE `error` field — route through the same mapping
2736 # as _download_pdf (PR #6564 follow-up).
2737 safe_error = _client_safe_download_message(exc)
2738 return False, safe_error
2740 if arxiv_text is None:
2741 return False, "Could not retrieve arXiv text"
2743 extraction_method, extraction_source = _arxiv_text_provenance(
2744 arxiv_text.source
2745 )
2746 try:
2747 _ = self._save_text_with_db(
2748 resource,
2749 arxiv_text.text,
2750 session,
2751 extraction_method=extraction_method,
2752 extraction_source=extraction_source,
2753 pdf_document_id=pdf_document_id,
2754 )
2755 session.commit()
2756 except Exception as exc:
2757 safe_rollback(session, "_try_arxiv_text_extraction")
2758 log_exception_type("Failed to save canonical arXiv text", exc)
2759 # Same caller-safe mapping as above — never str(exc) (see comment
2760 # on the fetch-failure path).
2761 safe_error = _client_safe_download_message(exc)
2762 return False, safe_error
2764 return True, None
2766 def _load_cached_arxiv_pdf(
2767 self, session: Session, resource
2768 ) -> Tuple[bytes, str] | None:
2769 return self._load_cached_pdf(session, resource)
2771 def _load_cached_pdf(
2772 self, session: Session, resource
2773 ) -> Tuple[bytes, str] | None:
2774 """Return (pdf_bytes, document_id) for a stored completed PDF."""
2775 pdf_document = get_document_for_resource(session, resource)
2776 if (
2777 pdf_document is None
2778 or getattr(pdf_document, "status", None)
2779 not in {DocumentStatus.COMPLETED, DocumentStatus.COMPLETED.value}
2780 or getattr(pdf_document, "file_type", None) != "pdf"
2781 ):
2782 return None
2783 settings = self.settings
2784 if settings is None: 2784 ↛ 2785line 2784 didn't jump to line 2785 because the condition on line 2784 was never true
2785 return None
2786 pdf_storage_mode = resolve_pdf_storage_mode(
2787 settings.get_setting("research_library.pdf_storage_mode", "none")
2788 )
2789 pdf_manager = PDFStorageManager(
2790 library_root=Path(self.library_root),
2791 storage_mode=pdf_storage_mode,
2792 )
2793 pdf_content = pdf_manager.load_pdf(pdf_document, session)
2794 if pdf_content is None:
2795 return None
2796 return pdf_content, getattr(pdf_document, "id")
2798 def _text_from_cached_pdf(
2799 self, session: Session, resource
2800 ) -> Optional[Tuple[bool, Optional[str]]]:
2801 """Produce text from the stored PDF without any network access.
2803 Used when the egress policy or the retry gate denies the network
2804 but the PDF downloaded by an earlier attempt is still on hand. Returns None when there is
2805 no usable stored PDF or it yields no text, so the caller's verdict
2806 stands; a completed extraction returns (True, None).
2807 """
2808 cached_pdf = self._load_cached_arxiv_pdf(session, resource)
2809 if cached_pdf is None:
2810 return None
2811 pdf_content, pdf_document_id = cached_pdf
2812 text = ArxivDownloader.extract_text_from_pdf(pdf_content)
2813 if not text:
2814 return None
2815 # Same attribution download_text_with_source gives PDF text saved
2816 # with these labels. Callers only reach here for an arXiv paper
2817 # URL, so the id is always present.
2818 arxiv_id = extract_arxiv_id(resource.url or "")
2819 if arxiv_id: 2819 ↛ 2821line 2819 didn't jump to line 2821 because the condition on line 2819 was always true
2820 text = attribute_arxiv_text(text, arxiv_id)
2821 try:
2822 _ = self._save_text_with_db(
2823 resource,
2824 text,
2825 session,
2826 extraction_method="pdf_extraction",
2827 extraction_source="local_pdf",
2828 pdf_document_id=pdf_document_id,
2829 )
2830 session.commit()
2831 except Exception as exc:
2832 safe_rollback(session, "_try_arxiv_text_extraction")
2833 log_exception_type("Failed to save local arXiv PDF text", exc)
2834 safe_error = _client_safe_download_message(exc)
2835 return False, safe_error
2836 return True, None
2838 def _try_existing_pdf_extraction(
2839 self, session, resource, resource_id: int
2840 ) -> Optional[Tuple[bool, Optional[str]]]:
2841 """Extract text from the PDF file of the resource's document.
2843 Reads the file that the completed document's own ``file_path``
2844 names (filesystem storage), without a request, and saves its text
2845 into that document. A PDF stored in the database is not read here
2846 (``_text_from_cached_pdf`` reads one for an arXiv paper). Returns
2847 None, so that the caller tries its other sources, when the resource
2848 has no document, the document is not completed or names no file,
2849 its file is missing or yields no text, or the read or the save
2850 fails.
2851 """
2852 # An older retained snapshot may still name this resource as its
2853 # owner after the resource links to a newer document.
2854 pdf_document = get_document_for_resource(session, resource)
2856 # A loaded status is a DocumentStatus member, which never equals
2857 # the string "completed".
2858 if not pdf_document or pdf_document.status != DocumentStatus.COMPLETED:
2859 return None
2861 # Validate path to prevent path traversal attacks. Resolve against the
2862 # per-user root only; shared legacy filenames cannot prove ownership. Pass
2863 # self.settings so storage_path/shared_library come from this user's
2864 # captured manager — the ambient one is db-less in background/scheduler
2865 # threads (this runs under the document scheduler) and would resolve
2866 # the default root, missing a user's previously saved storage_path. Use
2867 # getattr: __init__ always sets self.settings in production (incl.
2868 # scheduler threads), but test doubles that construct a
2869 # DownloadService via __new__/partial mocks may not — fall back to
2870 # None (the ambient get_settings_manager()) rather than raising
2871 # AttributeError.
2872 if (
2873 not pdf_document.file_path
2874 or pdf_document.file_path in FILE_PATH_SENTINELS
2875 ):
2876 return None
2877 pdf_path = get_absolute_path_from_settings(
2878 pdf_document.file_path,
2879 self.username,
2880 settings_manager=getattr(self, "settings", None),
2881 )
2882 if not pdf_path or not pdf_path.is_file():
2883 return None
2885 logger.info(f"Found existing PDF, extracting text from: {pdf_path}")
2886 try:
2887 # nosemgrep: semgrep.rules.path-traversal-risk, reason: get_absolute_path_from_settings requires user context and confines this read to the selected library root; missing PDFs never fall back to another root
2888 with open(pdf_path, "rb") as f:
2889 pdf_content = f.read()
2890 text = self._extract_text_from_pdf(pdf_content)
2892 if not text:
2893 return None
2895 self._save_text_with_db(
2896 resource,
2897 text,
2898 session,
2899 extraction_method="pdf_extraction",
2900 extraction_source="pdfplumber",
2901 pdf_document_id=pdf_document.id,
2902 )
2903 session.commit()
2904 return True, None
2906 except Exception as exc:
2907 log_exception_type("Failed to extract text from existing PDF", exc)
2908 # The commit above can fail and poison the shared session.
2909 # download_as_text swallows this None and may then return cleanly
2910 # (e.g. a non-retry decision), so the with-block exits without an
2911 # exception — get_user_db_session's rollback never fires. Recover
2912 # the session here or the next operation on this thread cascades.
2913 safe_rollback(session, "_try_existing_pdf_extraction")
2914 return None # Fall through to other methods
2916 def _try_api_text_extraction(
2917 self, session, resource
2918 ) -> Optional[Tuple[bool, Optional[str]]]:
2919 """Try source-specific direct text extraction."""
2920 logger.info(
2921 f"Attempting direct API text extraction from: {redact_url_for_log(resource.url)}"
2922 )
2924 downloader = self._get_downloader(resource.url)
2925 if not downloader:
2926 return None
2928 # Egress policy gate. Return a non-None tuple (not None) on denial
2929 # so download_as_text hard-stops here instead of falling through
2930 # to _fallback_pdf_extraction and firing a second request.
2931 allowed, reason = self._check_url_against_policy(resource.url)
2932 if not allowed:
2933 logger.warning(
2934 f"Skipping API text extraction for {redact_url_for_log(resource.url)}: refused "
2935 f"by egress policy ({reason})"
2936 )
2937 return False, f"egress_policy_denied:{reason}"
2939 if isinstance(downloader, ArxivDownloader):
2940 arxiv_text = downloader.download_text_with_source(resource.url)
2941 if arxiv_text is None: 2941 ↛ 2942line 2941 didn't jump to line 2942 because the condition on line 2941 was never true
2942 return None
2943 text = arxiv_text.text
2944 extraction_method, extraction_source = _arxiv_text_provenance(
2945 arxiv_text.source
2946 )
2947 else:
2948 result = downloader.download_with_result(
2949 resource.url, ContentType.TEXT
2950 )
2951 if not result.is_success or not result.content:
2952 return None
2954 text = (
2955 result.content.decode("utf-8", errors="ignore")
2956 if isinstance(result.content, bytes)
2957 else result.content
2958 )
2959 extraction_method = "native_api"
2960 extraction_source = (
2961 "pubmed_api"
2962 if isinstance(downloader, PubMedDownloader)
2963 else "unknown"
2964 )
2966 try:
2967 self._save_text_with_db(
2968 resource,
2969 text,
2970 session,
2971 extraction_method=extraction_method,
2972 extraction_source=extraction_source,
2973 )
2974 session.commit()
2975 logger.info(
2976 f"✓ SUCCESS: Got text from {extraction_source.upper()} for '{resource.title[:50]}...'"
2977 )
2978 return True, None
2979 except Exception as e:
2980 # Roll back FIRST: the failed commit/flush poisoned the shared
2981 # session, and even dereferencing resource.id in the log line below
2982 # can trigger a refresh that re-raises PendingRollbackError.
2983 # download_as_text reuses this session (_record_retry_attempt)
2984 # after we return, so it must be clean.
2985 safe_rollback(session, "_try_api_text_extraction")
2986 log_exception_type("Failed to save text for resource", e)
2987 # Caller-visible: fixed token only (CWE-209). sanitize_error_for_client
2988 # leaves SQL text intact and this flows into download_as_text's
2989 # return and the bulk-stream SSE `error` field.
2990 safe_error = _client_safe_download_message(e)
2991 return False, safe_error
2993 def _fallback_pdf_extraction(
2994 self, session, resource
2995 ) -> Tuple[bool, Optional[str]]:
2996 """Fallback: Download PDF to memory and extract text."""
2997 logger.info(
2998 f"Direct text extraction unavailable; trying in-memory PDF for: {redact_url_for_log(resource.url)}"
2999 )
3001 downloader = self._get_downloader(resource.url)
3002 if not downloader:
3003 error_msg = "No compatible downloader found"
3004 logger.warning(
3005 f"✗ FAILED: {failure_reason_for_log(error_msg)} for '{resource.title[:50]}...'"
3006 )
3007 self._record_failed_text_extraction(
3008 session, resource, error=error_msg
3009 )
3010 session.commit()
3011 return False, error_msg
3013 # Egress policy gate at the network-fire point.
3014 allowed, reason = self._check_url_against_policy(resource.url)
3015 if not allowed:
3016 error_msg = f"egress_policy_denied:{reason}"
3017 logger.warning(
3018 f"Skipping fallback PDF download for {redact_url_for_log(resource.url)}: "
3019 f"refused by egress policy ({reason})"
3020 )
3021 self._record_failed_text_extraction(
3022 session, resource, error=error_msg
3023 )
3024 session.commit()
3025 return False, error_msg
3027 result = downloader.download_with_result(resource.url, ContentType.PDF)
3029 if not result.is_success or not result.content:
3030 error_msg = result.skip_reason or "Failed to download PDF"
3031 logger.warning(
3032 f"✗ FAILED: Could not download PDF for '{resource.title[:50]}...' | Error: {failure_reason_for_log(error_msg)}"
3033 )
3034 self._record_failed_text_extraction(
3035 session, resource, error=f"PDF download failed: {error_msg}"
3036 )
3037 session.commit()
3038 return False, f"PDF extraction failed: {error_msg}"
3040 # Extract text from PDF
3041 text = self._extract_text_from_pdf(result.content)
3042 if not text:
3043 error_msg = "PDF text extraction returned empty text"
3044 logger.warning(
3045 f"Failed to extract text from PDF for: {redact_url_for_log(resource.url)}"
3046 )
3047 self._record_failed_text_extraction(
3048 session, resource, error=error_msg
3049 )
3050 session.commit()
3051 return False, error_msg
3053 try:
3054 self._save_text_with_db(
3055 resource,
3056 text,
3057 session,
3058 extraction_method="pdf_extraction",
3059 extraction_source="pdfplumber_fallback",
3060 )
3061 session.commit()
3062 logger.info(
3063 f"✓ SUCCESS: Extracted text from '{resource.title[:50]}...'"
3064 )
3065 return True, None
3066 except Exception as e:
3067 # Roll back FIRST (see _try_api_text_extraction): dereferencing
3068 # resource.id while the session is poisoned would itself re-raise
3069 # PendingRollbackError. download_as_text reuses this session after
3070 # we return.
3071 safe_rollback(session, "_fallback_pdf_extraction")
3072 log_exception_type("Failed to save text for resource", e)
3073 # Caller-visible: fixed token only (CWE-209) — same mapping as
3074 # _download_pdf; never str(e).
3075 safe_error = _client_safe_download_message(e)
3076 return False, safe_error
3078 def _get_downloader(self, url: str):
3079 """
3080 Get the appropriate downloader for a URL.
3082 Args:
3083 url: The URL to download from
3085 Returns:
3086 The appropriate downloader instance or None
3087 """
3088 for downloader in self.downloaders:
3089 if downloader.can_handle(url):
3090 return downloader
3091 return None
3093 def _download_generic(self, url: str) -> Optional[bytes]:
3094 """Generic PDF download method."""
3095 try:
3096 headers = {
3097 "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
3098 }
3099 response = safe_get(
3100 url, headers=headers, timeout=30, allow_redirects=True
3101 )
3102 response.raise_for_status()
3104 # Verify it's a PDF
3105 content_type = response.headers.get("Content-Type", "")
3106 if (
3107 "pdf" not in content_type.lower()
3108 and not response.content.startswith(b"%PDF")
3109 ):
3110 logger.warning(f"Response is not a PDF: {content_type}")
3111 return None
3113 return response.content
3115 except Exception as e:
3116 # raise_for_status()/SafeSession texts embed the raw URL —
3117 # log the exception type and the redacted URL only.
3118 logger.warning(
3119 f"Generic download failed ({type(e).__name__}) for "
3120 f"{redact_url_for_log(url)}"
3121 )
3122 return None
3124 def _download_arxiv(self, url: str) -> Optional[bytes]:
3125 """Download from arXiv."""
3126 try:
3127 # Convert abstract URL to PDF URL
3128 pdf_url = url.replace("abs", "pdf")
3129 if not pdf_url.endswith(".pdf"):
3130 pdf_url += ".pdf"
3132 return self._download_generic(pdf_url)
3133 except Exception as e:
3134 logger.warning(
3135 f"arXiv download failed ({type(e).__name__}) for "
3136 f"{redact_url_for_log(url)}"
3137 )
3138 return None
3140 def _try_europe_pmc(self, pmid: str) -> Optional[bytes]:
3141 """Try downloading from Europe PMC which often has better PDF availability."""
3142 try:
3143 # Europe PMC API is more reliable for PDFs
3144 # Check if PDF is available
3145 api_url = f"https://www.ebi.ac.uk/europepmc/webservices/rest/search?query=EXT_ID:{pmid}&format=json"
3146 response = safe_get(api_url, timeout=10)
3148 if response.status_code == 200:
3149 data = response.json()
3150 results = data.get("resultList", {}).get("result", [])
3152 if results:
3153 article = results[0]
3154 # Check if article has open access PDF
3155 if (
3156 article.get("isOpenAccess") == "Y"
3157 and article.get("hasPDF") == "Y"
3158 ):
3159 pmcid = article.get("pmcid")
3160 if pmcid:
3161 # Europe PMC PDF URL
3162 pdf_url = f"https://europepmc.org/backend/ptpmcrender.fcgi?accid={pmcid}&blobtype=pdf"
3163 logger.info(
3164 f"Found Europe PMC PDF for PMID {pmid}: {pmcid}"
3165 )
3167 headers = {
3168 "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
3169 }
3171 pdf_response = safe_get(
3172 pdf_url,
3173 headers=headers,
3174 timeout=30,
3175 allow_redirects=True,
3176 )
3177 if pdf_response.status_code == 200: 3177 ↛ 3189line 3177 didn't jump to line 3189 because the condition on line 3177 was always true
3178 content_type = pdf_response.headers.get(
3179 "content-type", ""
3180 )
3181 if ( 3181 ↛ 3189line 3181 didn't jump to line 3189 because the condition on line 3181 was always true
3182 "pdf" in content_type.lower()
3183 or len(pdf_response.content) > 1000
3184 ):
3185 return pdf_response.content
3186 except Exception as e:
3187 logger.debug(f"Europe PMC download failed ({type(e).__name__})")
3189 return None
3191 def _download_pubmed(self, url: str) -> Optional[bytes]:
3192 """Download from PubMed/PubMed Central with rate limiting."""
3193 try:
3194 # Apply rate limiting for PubMed requests
3195 current_time = time.time()
3196 time_since_last = current_time - self._last_pubmed_request
3197 if time_since_last < self._pubmed_delay:
3198 sleep_time = self._pubmed_delay - time_since_last
3199 logger.debug(
3200 f"Rate limiting: sleeping {sleep_time:.2f}s before PubMed request"
3201 )
3202 time.sleep(sleep_time)
3203 self._last_pubmed_request = time.time()
3205 # If it's already a PMC article, download directly
3206 if "/articles/PMC" in url:
3207 pmc_match = re.search(r"(PMC\d+)", url)
3208 if pmc_match: 3208 ↛ 3417line 3208 didn't jump to line 3417 because the condition on line 3208 was always true
3209 pmc_id = pmc_match.group(1)
3211 # Try Europe PMC (more reliable)
3212 europe_url = f"https://europepmc.org/backend/ptpmcrender.fcgi?accid={pmc_id}&blobtype=pdf"
3213 headers = {
3214 "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
3215 }
3217 try:
3218 response = safe_get(
3219 europe_url,
3220 headers=headers,
3221 timeout=30,
3222 allow_redirects=True,
3223 )
3224 if response.status_code == 200: 3224 ↛ 3417line 3224 didn't jump to line 3417 because the condition on line 3224 was always true
3225 content_type = response.headers.get(
3226 "content-type", ""
3227 )
3228 if ( 3228 ↛ 3417line 3228 didn't jump to line 3417 because the condition on line 3228 was always true
3229 "pdf" in content_type.lower()
3230 or len(response.content) > 1000
3231 ):
3232 logger.info(
3233 f"Downloaded PDF via Europe PMC for {pmc_id}"
3234 )
3235 return response.content
3236 except Exception as e:
3237 logger.debug(
3238 f"Direct Europe PMC download failed ({type(e).__name__})"
3239 )
3240 return None
3242 # If it's a regular PubMed URL, try to find PMC version
3243 elif urlparse(url).hostname == "pubmed.ncbi.nlm.nih.gov":
3244 # Extract PMID from URL
3245 pmid_match = re.search(r"/(\d+)/?", url)
3246 if pmid_match:
3247 pmid = pmid_match.group(1)
3248 logger.info(f"Attempting to download PDF for PMID: {pmid}")
3250 # Try Europe PMC first (more reliable)
3251 pdf_content = self._try_europe_pmc(pmid)
3252 if pdf_content:
3253 return pdf_content
3255 # First try using NCBI E-utilities API to find PMC ID
3256 try:
3257 # Use elink to convert PMID to PMCID
3258 elink_url = "https://eutils.ncbi.nlm.nih.gov/entrez/eutils/elink.fcgi"
3259 params = {
3260 "dbfrom": "pubmed",
3261 "db": "pmc",
3262 "id": pmid,
3263 "retmode": "json",
3264 }
3266 api_response = safe_get(
3267 elink_url, params=params, timeout=10
3268 )
3269 if api_response.status_code == 200: 3269 ↛ 3352line 3269 didn't jump to line 3352 because the condition on line 3269 was always true
3270 data = api_response.json()
3271 # Parse the response to find PMC ID
3272 link_sets = data.get("linksets", [])
3273 if link_sets and "linksetdbs" in link_sets[0]:
3274 for linksetdb in link_sets[0]["linksetdbs"]: 3274 ↛ 3352line 3274 didn't jump to line 3352 because the loop on line 3274 didn't complete
3275 if linksetdb.get( 3275 ↛ 3274line 3275 didn't jump to line 3274 because the condition on line 3275 was always true
3276 "dbto"
3277 ) == "pmc" and linksetdb.get("links"):
3278 pmc_id_num = linksetdb["links"][0]
3279 # Now fetch PMC details to get the correct PMC ID format
3280 esummary_url = "https://eutils.ncbi.nlm.nih.gov/entrez/eutils/esummary.fcgi"
3281 summary_params = {
3282 "db": "pmc",
3283 "id": pmc_id_num,
3284 "retmode": "json",
3285 }
3286 summary_response = safe_get(
3287 esummary_url,
3288 params=summary_params,
3289 timeout=10,
3290 )
3291 if summary_response.status_code == 200: 3291 ↛ 3274line 3291 didn't jump to line 3274 because the condition on line 3291 was always true
3292 summary_data = (
3293 summary_response.json()
3294 )
3295 result = summary_data.get(
3296 "result", {}
3297 ).get(str(pmc_id_num), {})
3298 if result: 3298 ↛ 3274line 3298 didn't jump to line 3274 because the condition on line 3298 was always true
3299 # PMC IDs in the API don't have the "PMC" prefix
3300 pmc_id = f"PMC{pmc_id_num}"
3301 logger.info(
3302 f"Found PMC ID via API: {pmc_id} for PMID: {pmid}"
3303 )
3305 # Try Europe PMC with the PMC ID
3306 europe_url = f"https://europepmc.org/backend/ptpmcrender.fcgi?accid={pmc_id}&blobtype=pdf"
3308 time.sleep(self._pubmed_delay)
3310 headers = {
3311 "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
3312 }
3314 try:
3315 response = safe_get(
3316 europe_url,
3317 headers=headers,
3318 timeout=30,
3319 allow_redirects=True,
3320 )
3321 if ( 3321 ↛ 3274line 3321 didn't jump to line 3274 because the condition on line 3321 was always true
3322 response.status_code
3323 == 200
3324 ):
3325 content_type = response.headers.get(
3326 "content-type", ""
3327 )
3328 if ( 3328 ↛ 3274line 3328 didn't jump to line 3274 because the condition on line 3328 was always true
3329 "pdf"
3330 in content_type.lower()
3331 or len(
3332 response.content
3333 )
3334 > 1000
3335 ):
3336 logger.info(
3337 f"Downloaded PDF via Europe PMC for {pmc_id}"
3338 )
3339 return (
3340 response.content
3341 )
3342 except Exception as e:
3343 logger.debug(
3344 f"Europe PMC download with PMC ID failed ({type(e).__name__})"
3345 )
3346 except Exception as e:
3347 logger.debug(
3348 f"API lookup failed, trying webpage scraping ({type(e).__name__})"
3349 )
3351 # Fallback to webpage scraping if API fails
3352 try:
3353 headers = {
3354 "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
3355 }
3356 response = safe_get(url, headers=headers, timeout=10)
3357 if response.status_code == 200: 3357 ↛ 3417line 3357 didn't jump to line 3417 because the condition on line 3357 was always true
3358 # Look for PMC ID in the page
3359 pmc_match = re.search(r"PMC\d+", response.text)
3360 if pmc_match: 3360 ↛ 3399line 3360 didn't jump to line 3399 because the condition on line 3360 was always true
3361 pmc_id = pmc_match.group(0)
3362 logger.info(
3363 f"Found PMC ID via webpage: {pmc_id} for PMID: {pmid}"
3364 )
3366 # Add delay before downloading PDF
3367 time.sleep(self._pubmed_delay)
3369 # Try Europe PMC with the PMC ID (more reliable)
3370 europe_url = f"https://europepmc.org/backend/ptpmcrender.fcgi?accid={pmc_id}&blobtype=pdf"
3371 headers = {
3372 "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
3373 }
3375 try:
3376 response = safe_get(
3377 europe_url,
3378 headers=headers,
3379 timeout=30,
3380 allow_redirects=True,
3381 )
3382 if response.status_code == 200: 3382 ↛ 3417line 3382 didn't jump to line 3417 because the condition on line 3382 was always true
3383 content_type = response.headers.get(
3384 "content-type", ""
3385 )
3386 if ( 3386 ↛ 3417line 3386 didn't jump to line 3417 because the condition on line 3386 was always true
3387 "pdf" in content_type.lower()
3388 or len(response.content) > 1000
3389 ):
3390 logger.info(
3391 f"Downloaded PDF via Europe PMC for {pmc_id}"
3392 )
3393 return response.content
3394 except Exception as e:
3395 logger.debug(
3396 f"Europe PMC download failed ({type(e).__name__})"
3397 )
3398 else:
3399 logger.info(
3400 f"No PMC version found for PMID: {pmid}"
3401 )
3402 except requests.exceptions.HTTPError as e:
3403 if e.response.status_code == 429: 3403 ↛ 3410line 3403 didn't jump to line 3410 because the condition on line 3403 was always true
3404 logger.warning(
3405 "Rate limited by PubMed, increasing delay"
3406 )
3407 self._pubmed_delay = min(
3408 self._pubmed_delay * 2, 5.0
3409 ) # Max 5 seconds
3410 raise
3411 except Exception as e:
3412 logger.debug(
3413 f"Could not check for PMC version "
3414 f"({type(e).__name__})"
3415 )
3417 return self._download_generic(url)
3418 except Exception as e:
3419 # A re-raised HTTPError's text embeds the raw resource URL.
3420 logger.warning(
3421 f"PubMed download failed ({type(e).__name__}) for "
3422 f"{redact_url_for_log(url)}"
3423 )
3424 return None
3426 def _download_semantic_scholar(self, url: str) -> Optional[bytes]:
3427 """Download from Semantic Scholar."""
3428 # Semantic Scholar doesn't host PDFs directly
3429 # Would need to extract actual PDF URL from page
3430 return None
3432 def _download_biorxiv(self, url: str) -> Optional[bytes]:
3433 """Download from bioRxiv."""
3434 try:
3435 # Convert to PDF URL
3436 pdf_url = url.replace(".org/", ".org/content/")
3437 pdf_url = re.sub(r"v\d+$", "", pdf_url) # Remove version
3438 pdf_url += ".full.pdf"
3440 return self._download_generic(pdf_url)
3441 except Exception as e:
3442 logger.warning(
3443 f"bioRxiv download failed ({type(e).__name__}) for "
3444 f"{redact_url_for_log(url)}"
3445 )
3446 return None
3448 def _save_text_with_db(
3449 self,
3450 resource: ResearchResource,
3451 text: str,
3452 session: Session,
3453 extraction_method: str,
3454 extraction_source: str,
3455 pdf_document_id: str | None = None,
3456 ) -> Optional[str]:
3457 """
3458 Save extracted text to encrypted database.
3460 Args:
3461 resource: The research resource
3462 text: Extracted text content
3463 session: Database session
3464 extraction_method: How the text was extracted
3465 extraction_source: Specific tool/API used
3466 pdf_document_id: ID of PDF document if extracted from PDF
3468 Returns:
3469 None (previously returned text file path, now removed)
3470 """
3471 try:
3472 if not text.strip():
3473 return None
3475 # Calculate text metadata for database
3476 word_count = len(text.split())
3477 character_count = len(text)
3478 extraction_quality = _text_extraction_quality(
3479 extraction_method, extraction_source
3480 )
3482 # Find the document by pdf_document_id or resource_id. Both reads
3483 # touch scalar columns (`document_id`, `id`) that are eagerly
3484 # loaded when `resource` was queried, so passing a detached
3485 # `resource` from a snapshot block (post-staging-commit) is safe
3486 # and does not lazy-refresh through this session. PR #6564 P2.
3487 doc = None
3488 if pdf_document_id:
3489 doc = (
3490 session.query(Document)
3491 .filter_by(id=pdf_document_id)
3492 .first()
3493 )
3494 else:
3495 doc = get_document_for_resource(session, resource)
3497 if doc and (
3498 doc.status == DocumentStatus.FAILED
3499 or doc.resource_id != resource.id
3500 ):
3501 text_hash = hashlib.sha256(text.encode()).hexdigest()
3502 if (
3503 doc.status == DocumentStatus.FAILED
3504 and pdf_document_id is None
3505 ):
3506 # Text-only retries can resolve a different hash owner;
3507 # an explicit PDF target must keep its snapshot identity.
3508 owner = (
3509 session.query(Document)
3510 .filter_by(document_hash=text_hash)
3511 .first()
3512 )
3513 if owner and owner.id != doc.id:
3514 _link_retry_to_canonical(session, resource, doc, owner)
3515 doc = owner
3516 if (
3517 doc.document_hash == text_hash
3518 and doc.status
3519 in (DocumentStatus.COMPLETED, DocumentStatus.FAILED)
3520 and doc.text_content
3521 ):
3522 # The hash can identify original bytes, not extracted
3523 # text. Reusing those bytes must preserve the owner's
3524 # extraction, provenance and existing collection indexes.
3525 if doc.status == DocumentStatus.FAILED:
3526 doc.status = DocumentStatus.COMPLETED
3527 doc.error_message = None
3528 doc.processed_at = datetime.now(UTC)
3529 return None
3531 if doc:
3532 old_text = doc.text_content or ""
3533 if old_text.strip():
3534 # Duplicate PDFs share one extraction. Automatically replace
3535 # it only with nondecreasing quality AND text coverage. A
3536 # native API abstract must not displace a longer full paper;
3537 # whitespace padding is not additional coverage. Ties keep
3538 # the original text and provenance stable.
3539 ranks = {"low": 0, "medium": 1, "high": 2}
3540 old_quality = doc.extraction_quality
3541 if old_quality not in ranks:
3542 old_quality = _text_extraction_quality(
3543 doc.extraction_method, doc.extraction_source
3544 )
3545 old_rank = _text_extraction_rank(
3546 old_quality,
3547 doc.extraction_method,
3548 doc.extraction_source,
3549 )
3550 new_rank = _text_extraction_rank(
3551 extraction_quality, extraction_method, extraction_source
3552 )
3553 old_size = len("".join(old_text.split()))
3554 new_size = len("".join(text.split()))
3555 if (
3556 new_rank < old_rank
3557 or new_size < old_size
3558 or (new_rank == old_rank and new_size == old_size)
3559 ):
3560 if doc.status == DocumentStatus.FAILED:
3561 doc.status = DocumentStatus.COMPLETED
3562 doc.error_message = None
3563 return None
3565 # Claim the observed extraction before changing it. SQLite
3566 # ignores SELECT FOR UPDATE: a conditional no-op write instead
3567 # holds its write lock until the caller commits. If another
3568 # worker already changed the extraction, preserve that result.
3569 claimed = (
3570 session.query(Document)
3571 .filter(
3572 Document.id == doc.id,
3573 Document.text_content == doc.text_content,
3574 Document.extraction_quality == doc.extraction_quality,
3575 Document.extraction_method == doc.extraction_method,
3576 Document.extraction_source == doc.extraction_source,
3577 Document.document_hash == doc.document_hash,
3578 Document.status == doc.status,
3579 )
3580 .update({Document.id: doc.id}, synchronize_session=False)
3581 )
3582 if claimed == 0:
3583 session.expire(doc)
3584 return None
3586 if old_text != text:
3587 # Keep the vector IDs for replacement, but persist why
3588 # these chunks are pending. Live vectors alone must not
3589 # let reconciliation mark an older extraction current.
3590 for chunk in (
3591 session.query(DocumentChunk)
3592 .options(load_only(DocumentChunk.document_metadata))
3593 .filter_by(source_type="document", source_id=doc.id)
3594 .all()
3595 ):
3596 chunk.document_metadata = {
3597 **(chunk.document_metadata or {}),
3598 "_ldr_text_invalidated": True,
3599 }
3600 # The legacy flag and canonical per-index marker must both
3601 # move with the text, in this transaction. Leave chunk rows
3602 # available for replace-on-reindex to remove their vectors.
3603 affected_indexes = (
3604 session.query(RagDocumentStatus.rag_index_id)
3605 .filter_by(document_id=doc.id)
3606 .distinct()
3607 .all()
3608 )
3609 session.query(DocumentCollection).filter_by(
3610 document_id=doc.id
3611 ).update(
3612 {DocumentCollection.indexed: False},
3613 synchronize_session="fetch",
3614 )
3615 session.query(RagDocumentStatus).filter_by(
3616 document_id=doc.id
3617 ).delete(synchronize_session="fetch")
3618 # Reindex treats a removed status as a new contribution.
3619 # Remove its old contribution now, in the same transaction,
3620 # or each text improvement inflates the index totals.
3621 for (index_id,) in affected_indexes:
3622 chunk_count, document_count = (
3623 session.query(
3624 func.sum(RagDocumentStatus.chunk_count),
3625 func.count(),
3626 )
3627 .filter_by(rag_index_id=index_id)
3628 .one()
3629 )
3630 session.query(RAGIndex).filter_by(id=index_id).update(
3631 {
3632 RAGIndex.chunk_count: chunk_count or 0,
3633 RAGIndex.total_documents: document_count,
3634 },
3635 synchronize_session="fetch",
3636 )
3638 # Update existing document with extracted text. Only replace
3639 # a failed placeholder's hash on a text-only retry (#3827).
3640 # An explicit PDF document already has the downloaded byte
3641 # hash, including a failed canonical document reused by dedup.
3642 was_failed = doc.status == DocumentStatus.FAILED
3643 doc.text_content = text
3644 doc.character_count = character_count
3645 doc.word_count = word_count
3646 doc.extraction_method = extraction_method
3647 doc.extraction_source = extraction_source
3648 doc.status = DocumentStatus.COMPLETED
3649 if was_failed and pdf_document_id is None:
3650 doc.document_hash = hashlib.sha256(
3651 text.encode()
3652 ).hexdigest()
3653 if was_failed:
3654 doc.error_message = None
3655 doc.processed_at = datetime.now(UTC)
3657 doc.extraction_quality = extraction_quality
3659 logger.debug(
3660 f"Updated document {doc.id} with extracted text ({word_count} words)"
3661 )
3662 else:
3663 # Create a new Document for text-only extraction
3664 # Generate hash from text content
3665 text_hash = hashlib.sha256(text.encode()).hexdigest()
3667 # Dedup against existing content hash. If another resource
3668 # already produced identical extracted text, link this
3669 # resource to the canonical Document instead of inserting
3670 # a duplicate that would violate the UNIQUE constraint
3671 # (issue #3827). Mirrors research_history_indexer.py:322.
3672 existing_by_hash = (
3673 session.query(Document)
3674 .filter_by(document_hash=text_hash)
3675 .first()
3676 )
3677 if existing_by_hash:
3678 resource.document_id = existing_by_hash.id
3679 library_collection = (
3680 session.query(Collection)
3681 .filter_by(name="Library")
3682 .first()
3683 )
3684 if library_collection: 3684 ↛ 3691line 3684 didn't jump to line 3691 because the condition on line 3684 was always true
3685 ensure_in_collection(
3686 session,
3687 existing_by_hash.id,
3688 library_collection.id,
3689 )
3690 else:
3691 logger.warning(
3692 f"Library collection not found - deduped document {existing_by_hash.id} will not be linked to default collection"
3693 )
3694 logger.info(
3695 f"Linked resource {resource.id} to existing Document "
3696 f"{existing_by_hash.id} (matched on content hash)"
3697 )
3698 return None
3700 # Get source type for research downloads
3701 try:
3702 source_type_id = get_source_type_id(
3703 self.username, "research_download", self.password
3704 )
3705 except Exception as exc:
3706 log_exception_type(
3707 "Failed to get source type for text document", exc
3708 )
3709 raise
3711 # Create new document
3712 doc_id = str(uuid.uuid4())
3713 doc = Document(
3714 id=doc_id,
3715 source_type_id=source_type_id,
3716 resource_id=resource.id,
3717 research_id=resource.research_id,
3718 document_hash=text_hash,
3719 original_url=resource.url,
3720 file_path=FILE_PATH_TEXT_ONLY,
3721 file_size=character_count, # Use character count as file size for text-only
3722 file_type="text",
3723 mime_type="text/plain",
3724 title=resource.title,
3725 text_content=text,
3726 character_count=character_count,
3727 word_count=word_count,
3728 extraction_method=extraction_method,
3729 extraction_source=extraction_source,
3730 extraction_quality=extraction_quality,
3731 status=DocumentStatus.COMPLETED,
3732 processed_at=datetime.now(UTC),
3733 )
3734 session.add(doc)
3736 # Link to default Library collection
3737 library_collection = (
3738 session.query(Collection).filter_by(name="Library").first()
3739 )
3740 if library_collection:
3741 ensure_in_collection(session, doc_id, library_collection.id)
3742 else:
3743 logger.warning(
3744 f"Library collection not found - document {doc_id} will not be linked to default collection"
3745 )
3747 logger.info(
3748 f"Created new document {doc_id} for text-only extraction ({word_count} words)"
3749 )
3751 logger.info(
3752 f"Saved text to encrypted database ({word_count} words)"
3753 )
3754 return None
3756 except Exception as exc:
3757 # Rollback BEFORE re-raising so the shared thread-local session
3758 # is clean by the time the caller's loop reaches its next
3759 # iteration. Without this, an IntegrityError leaves the session
3760 # in PendingRollbackError state and every subsequent ORM access
3761 # cascades (issue #3827).
3762 safe_rollback(session, "_save_text_with_db")
3763 log_exception_type("Error saving text to encrypted database", exc)
3764 raise # Re-raise so caller can handle the error
3766 def _create_text_document_record(
3767 self,
3768 session: Session,
3769 resource: ResearchResource,
3770 file_path: Path,
3771 extraction_method: str,
3772 extraction_source: str,
3773 ):
3774 """Update existing Document with text from file (for legacy text files)."""
3775 try:
3776 # Read file to get metadata
3777 text = file_path.read_text(encoding="utf-8", errors="ignore")
3778 word_count = len(text.split())
3779 character_count = len(text)
3781 # Find the Document for this resource
3782 doc = get_document_for_resource(session, resource)
3784 if doc:
3785 # Update existing document with text content
3786 doc.text_content = text
3787 doc.character_count = character_count
3788 doc.word_count = word_count
3789 doc.extraction_method = extraction_method
3790 doc.extraction_source = extraction_source
3791 doc.extraction_quality = (
3792 "low" # Unknown quality for legacy files
3793 )
3794 logger.info(
3795 f"Updated document {doc.id} with text from file: {file_path.name}"
3796 )
3797 else:
3798 logger.warning(
3799 f"No document found to update for resource {resource.id}"
3800 )
3802 except Exception as exc:
3803 log_exception_type(
3804 "Error updating document with text from file", exc
3805 )
3807 def _record_failed_text_extraction(
3808 self, session: Session, resource: ResearchResource, error: str
3809 ):
3810 """Record a failed text extraction attempt in the Document."""
3811 try:
3812 # Find the Document for this resource
3813 doc = get_document_for_resource(session, resource)
3815 if doc:
3816 # Update document with extraction error
3817 doc.error_message = error
3818 doc.extraction_method = "failed"
3819 doc.extraction_quality = "low"
3820 doc.status = DocumentStatus.FAILED
3821 logger.info(
3822 f"Recorded failed text extraction for document {doc.id}: "
3823 f"{failure_reason_for_log(error)}"
3824 )
3825 else:
3826 # Create a new Document for failed extraction
3827 # This enables tracking failures and retry capability
3828 source_type_id = get_source_type_id(
3829 self.username, "research_download", self.password
3830 )
3832 # Deterministic hash so retries update the same record
3833 failed_hash = hashlib.sha256(
3834 f"failed:{resource.url}:{resource.id}".encode()
3835 ).hexdigest()
3837 doc_id = str(uuid.uuid4())
3838 doc = Document(
3839 id=doc_id,
3840 source_type_id=source_type_id,
3841 resource_id=resource.id,
3842 research_id=resource.research_id,
3843 document_hash=failed_hash,
3844 original_url=resource.url,
3845 file_path=None,
3846 file_size=0,
3847 file_type="unknown",
3848 title=resource.title,
3849 status=DocumentStatus.FAILED,
3850 error_message=error,
3851 extraction_method="failed",
3852 extraction_quality="low",
3853 processed_at=datetime.now(UTC),
3854 )
3855 session.add(doc)
3857 logger.info(
3858 f"Created failed document {doc_id} for resource "
3859 f"{resource.id}: {failure_reason_for_log(error)}"
3860 )
3862 except Exception as exc:
3863 log_exception_type("Error recording failed text extraction", exc)