Coverage for src/local_deep_research/research_library/downloaders/base.py: 96%
241 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-22 21:20 +0000
1"""
2Base Academic Content Downloader Abstract Class
3"""
5from abc import ABC, abstractmethod
6from typing import Any, Callable, Dict, NamedTuple, Optional
7from enum import Enum
8import time
9import requests
10from urllib.parse import urlparse
11from loguru import logger
13# Import our adaptive rate limiting system
14from ...web_search_engines.rate_limiting import (
15 AdaptiveRateLimitTracker,
16)
17from ...security import SafeSession, safe_requests
18from ...security.egress.policy import PolicyDeniedError
19from ...utilities.pdf_extraction_limits import (
20 MAX_PDF_EXTRACTED_CHARS,
21 MAX_PDF_EXTRACTION_CPU_SECONDS,
22 MAX_PDF_EXTRACTION_PAGES,
23)
24from ...utilities.pdf_stream_release import PypdfWalkReleaser
25from ...security.log_sanitizer import scrub_error
26from ...security.ssrf_validator import (
27 authority_may_be_userinfo,
28 redact_url_for_log,
29)
30from ...utilities.resource_utils import safe_close
32# Import centralized User-Agent from constants
33from ...constants import USER_AGENT # noqa: F401 - re-exported for backward compatibility
36def rate_limit_authority(url: str) -> str:
37 """Return *url*'s authority (``host`` or ``host:port``) for rate-limit keys.
39 RFC 3986 userinfo, path, query and fragment are dropped and an IPv6
40 literal keeps its brackets. A URL with no authority at all yields
41 ``""``; one whose authority will not parse (an invalid port, an
42 unterminated ``[``) or whose host carries whitespace or a control
43 character yields ``"invalid_authority"``.
45 All four downloader fetch paths derive their per-host adaptive
46 rate-limit bucket through this one helper, so the keys cannot drift
47 apart. It parses the *original* URL rather than re-parsing
48 ``redact_url_for_log(url)``: that helper is a display formatter, and its
49 rendering of an input it cannot parse has no authority left to recover
50 -- ``" https://slow-host.example/paper"`` renders as ``"?:// https"``,
51 while ``requests`` lstrips that same leading whitespace and fetches the
52 URL anyway. Re-parsing the rendering therefore dropped every
53 whitespace-prefixed host into one shared bucket, where a single
54 429-heavy host raises the wait for all the others.
56 ``AdaptiveRateLimitTracker`` renders the key into its own log messages,
57 so the value must stay one printable token. A host carrying whitespace
58 or a control character is not a legal RFC 3986 reg-name, and is not the
59 host contacted either -- ``requests`` percent-encodes it, preparing
60 ``"https://exa mple.com/p"`` as ``https://exa%20mple.com/p`` -- so it
61 buckets as ``"invalid_authority"`` rather than being echoed into a log
62 line.
64 An authority that may really be a userinfo prefix
65 (``https://tok/en@host/`` parses to the host ``tok``,
66 ``https://admin:P@ss.example#1@host/`` to ``ss.example``; see
67 ``authority_may_be_userinfo``) also yields ``"invalid_authority"``, so
68 the tracker never logs a fragment of a credential as a key.
69 """
70 try:
71 parsed = urlparse(url)
72 hostname = parsed.hostname or ""
73 host = f"[{hostname}]" if ":" in hostname else hostname
74 authority = host
75 if parsed.port is not None:
76 authority = f"{authority}:{parsed.port}"
77 # ``authority``, not ``host``: ``https://u:@:1#x@host/`` parses to
78 # an empty host and the port ``1``, a fragment of the password.
79 if authority and authority_may_be_userinfo(
80 host,
81 parsed.port is not None,
82 url,
83 has_userinfo="@" in parsed.netloc,
84 ):
85 return "invalid_authority"
86 except ValueError:
87 return "invalid_authority"
88 if any(ch.isspace() or not ch.isprintable() for ch in authority):
89 return "invalid_authority"
90 return authority
93class ContentType(Enum):
94 """Supported content types for download."""
96 PDF = "pdf"
97 TEXT = "text"
100class DownloadResult(NamedTuple):
101 """Result of a download attempt."""
103 content: Optional[bytes] = None
104 skip_reason: Optional[str] = None
105 is_success: bool = False
106 status_code: Optional[int] = None
107 # Abstract/metadata fallbacks are useful display text, but cannot
108 # satisfy a later request for the complete source from a shared cache.
109 content_is_complete: bool = True
112class _ResponseBodyTooLarge(Exception):
113 """A response body exceeded the size cap while being decoded.
115 Internal to this module: it separates the body-cap abort (raised by
116 ``safe_requests._install_body_guard``'s bounded reader during
117 ``response.content``/``response.text`` access) from the
118 ``ValueError``s ``SafeSession`` raises for SSRF validation, so the
119 retry loop below can label it precisely instead of mislabeling an
120 SSRF rejection as a size problem (or vice versa).
121 """
124# HTTP statuses that answer "no such document" rather than signalling a
125# failing host: a fetch ending on one of these is not reported through the
126# ``on_transport_failure`` callbacks of ``_download_pdf`` and
127# ``HTMLDownloader._fetch_html_with_final_url``.
128_DOCUMENT_ABSENT_STATUSES: frozenset[int] = frozenset({404, 410})
131class BaseDownloader(ABC):
132 """Abstract base class for academic content downloaders."""
134 def __init__(self, timeout: int = 30):
135 """
136 Initialize the downloader.
138 Args:
139 timeout: Request timeout in seconds
140 """
141 self.timeout = timeout
142 self.session = SafeSession(block_https_downgrade=True)
143 self.session.headers.update({"User-Agent": USER_AGENT})
145 # Initialize rate limiter for PDF downloads
146 # We'll use domain-specific rate limiting
147 self.rate_tracker = AdaptiveRateLimitTracker(
148 programmatic_mode=False # We want to persist rate limit data
149 )
151 def close(self):
152 """
153 Close the HTTP session and clean up resources.
155 Call this method when done using the downloader to prevent
156 connection/file descriptor leaks.
157 """
158 if hasattr(self, "session") and self.session:
159 try:
160 self.session.close()
161 except Exception as exc:
162 # Keep the scrubbed message -- an event saying only that
163 # *something* failed is not a usable diagnostic -- but not
164 # the traceback, whose frame-locals loguru's ``diagnose``
165 # renders. That is the middle ground ``log_sanitizer``
166 # documents ("log without the traceback instead --
167 # ``logger.warning(scrub_error(exc, *known_secrets))``") and
168 # the ``safe_msg`` shape the sibling ``openalex`` downloader
169 # already uses. ERROR is kept (the level the
170 # ``logger.exception`` this replaced logged at): this
171 # handler's frames hold no URL and no credential.
172 safe_msg = scrub_error(exc)
173 logger.opt(exception=False).error(
174 f"Error closing downloader session: {safe_msg}"
175 )
176 finally:
177 self.session = None # type: ignore[assignment]
179 def __del__(self):
180 """Destructor to ensure session is closed."""
181 self.close()
183 def __enter__(self):
184 """Context manager entry."""
185 return self
187 def __exit__(self, exc_type, exc_val, exc_tb):
188 """Context manager exit - ensures session cleanup."""
189 self.close()
190 return False
192 @abstractmethod
193 def can_handle(self, url: str) -> bool:
194 """
195 Check if this downloader can handle the given URL.
197 Args:
198 url: The URL to check
200 Returns:
201 True if this downloader can handle the URL
202 """
203 pass
205 @abstractmethod
206 def download(
207 self, url: str, content_type: ContentType = ContentType.PDF
208 ) -> Optional[bytes]:
209 """
210 Download content from the given URL.
212 Args:
213 url: The URL to download from
214 content_type: Type of content to download (PDF or TEXT)
216 Returns:
217 Content as bytes, or None if download failed
218 For TEXT type, returns UTF-8 encoded text as bytes
219 """
220 pass
222 def download_pdf(self, url: str) -> Optional[bytes]:
223 """
224 Convenience method to download PDF.
226 Args:
227 url: The URL to download from
229 Returns:
230 PDF content as bytes, or None if download failed
231 """
232 return self.download(url, ContentType.PDF)
234 def download_with_result(
235 self, url: str, content_type: ContentType = ContentType.PDF
236 ) -> DownloadResult:
237 """
238 Download content and return detailed result with skip reason.
240 Args:
241 url: The URL to download from
242 content_type: Type of content to download
244 Returns:
245 DownloadResult with content and/or skip reason
246 """
247 # Default implementation - derived classes should override for specific reasons
248 content = self.download(url, content_type)
249 if content:
250 return DownloadResult(content=content, is_success=True)
251 return DownloadResult(
252 skip_reason="Download failed - content not available"
253 )
255 def download_text(self, url: str) -> Optional[str]:
256 """
257 Convenience method to download and return text content.
259 Args:
260 url: The URL to download from
262 Returns:
263 Text content as string, or None if download failed
264 """
265 content = self.download(url, ContentType.TEXT)
266 if content:
267 try:
268 return content.decode("utf-8")
269 except UnicodeDecodeError:
270 logger.opt(exception=False).error(
271 f"Failed to decode text content from "
272 f"{redact_url_for_log(url)}"
273 )
274 return None
276 def _is_pdf_content(
277 self, response: requests.Response, content: bytes | None = None
278 ) -> bool:
279 """
280 Check if response contains PDF content.
282 Args:
283 response: The response to check
284 content: Body bytes the caller has already read (and bounded)
285 under its own error accounting. Sniffed instead of
286 ``response.content`` so a caller that has read the body
287 itself never has a second, silently swallowed read here.
288 See ``HTMLDownloader._fetch_html_with_final_url``, which
289 needs an aborted body read to reach its own transport-
290 failure path.
292 Returns:
293 True if response appears to contain PDF content
294 """
295 headers = getattr(response, "headers", {}) or {}
296 try:
297 content_type = headers.get("content-type", "").lower()
298 except Exception:
299 content_type = ""
301 # Check content type
302 if "pdf" in content_type:
303 return True
305 # Check if content starts with PDF magic bytes
306 if content is not None:
307 # Caller-supplied bytes: the read already happened and
308 # succeeded, so there is nothing left to swallow here.
309 return len(content) > 4 and content[:4] == b"%PDF"
310 try:
311 body = getattr(response, "content", b"") or b""
312 if len(body) > 4: 312 ↛ 317line 312 didn't jump to line 317 because the condition on line 312 was always true
313 return body[:4] == b"%PDF"
314 except Exception:
315 logger.debug("base.pdf_magic_check_failed")
317 return False
319 def _ensure_decoded_body_cap(self, response: requests.Response) -> None:
320 """Install a second decoded-byte body guard for a valid, under-cap
321 ``Content-Length``.
323 ``SafeSession.send`` installs ``_install_body_guard`` (which
324 bounds ``raw.read()``/``read_chunked()`` at
325 ``MAX_RESPONSE_SIZE``, counting already-decoded bytes) on every
326 response it accepts, including one whose *valid, under-cap*
327 ``Content-Length`` says nothing about how far a gzip body can
328 decode while ``response.content`` / ``response.text`` is read.
329 Install the same guard again for that case, so the decoded body
330 of *this* response stays bounded without relying on the session's
331 guard. For a ``SafeSession`` response the two counters count the
332 same bytes against the same threshold, so the second is harmless.
334 This only bounds the body of the final response that callers of
335 this method go on to read (``response.content`` in
336 ``_download_pdf``, the streamed body in
337 ``_fetch_html_with_final_url``). With ``allow_redirects=True``,
338 intermediate redirect responses never reach this guard;
339 ``SafeSession.resolve_redirects`` size-checks their
340 ``Content-Length`` and discards their bodies unread.
342 The request must have been made with ``stream=True`` for this to
343 matter: with ``stream=False``, ``requests.Session.send`` consumes
344 the body before returning, i.e. before this guard could wrap
345 anything.
346 """
347 if getattr(response, "raw", None) is None:
348 # No raw stream to wrap (the body is already materialised, e.g.
349 # a Response built in memory): nothing left to bound here.
350 return
351 content_length = response.headers.get("Content-Length")
352 if not content_length:
353 # Absent/empty: SafeSession.send already installed the guard.
354 return
355 try:
356 parts = [
357 p.strip() for p in str(content_length).split(",") if p.strip()
358 ]
359 sizes = {int(p) for p in parts}
360 except ValueError:
361 # Unparseable: _check_response_size treated it as absent and
362 # already installed the guard (differing duplicate values were
363 # rejected inside SafeSession.send before we got here).
364 return
365 if ( 365 ↛ exitline 365 didn't return from function '_ensure_decoded_body_cap' because the condition on line 365 was always true
366 len(sizes) == 1
367 and next(iter(sizes)) <= safe_requests.MAX_RESPONSE_SIZE
368 ):
369 # A single valid Content-Length within the cap, where the
370 # decoded body can still exceed the header (e.g. gzip).
371 # SafeSession.send has guarded this response already; stacking
372 # a second guard here (or on the negative or duplicate CL
373 # corner cases) is harmless — both counters share the same
374 # threshold.
375 safe_requests._install_body_guard(response)
377 def _download_pdf(
378 self,
379 url: str,
380 headers: Optional[Dict[str, str]] = None,
381 *,
382 max_attempts: int = 3,
383 on_rate_limited: Optional[Callable[[], None]] = None,
384 on_transport_failure: Optional[Callable[[], None]] = None,
385 ) -> Optional[bytes]:
386 """
387 Helper method to download PDF with error handling and retry logic.
388 Uses our optimized adaptive rate limiting/retry system.
390 Args:
391 url: The URL to download
392 headers: Optional additional headers
393 max_attempts: Attempts made before giving up on HTTP 429/503,
394 timeouts and connection errors (default 3). A caller that
395 must not retry into a rate-limited host passes 1.
396 on_rate_limited: Called once when the download gives up on
397 HTTP 429/503, so a caller can stop sending further
398 requests to that host. When given, that give-up is logged
399 as a warning rather than an error: the caller handles it.
400 on_transport_failure: Called once when the download fails
401 without the host answering "no such document" (HTTP 404
402 or 410) or "not a PDF", and without a 429/503 give-up:
403 timeouts and connection errors after the last attempt,
404 any other request error, any other non-200 status, the
405 decoded-body cap, or an unexpected error. A caller fetching
406 several documents uses it to stop instead of paying a
407 timeout per document. When given, every one of these
408 failures is logged as a warning rather than an error.
410 Returns:
411 PDF content as bytes, or None if download failed
412 """
413 engine_type = f"pdf_download_{rate_limit_authority(url)}"
415 logger.debug(
416 f"Downloading PDF from {redact_url_for_log(url)} with adaptive "
417 f"rate limiting (max {max_attempts} attempts)"
418 )
420 for attempt in range(1, max_attempts + 1): 420 ↛ 643line 420 didn't jump to line 643 because the loop on line 420 didn't complete
421 # Apply adaptive rate limiting before the request
422 wait_time = self.rate_tracker.apply_rate_limit(engine_type)
424 try:
425 # Prepare headers
426 if headers:
427 request_headers = dict(self.session.headers)
428 request_headers.update(headers)
429 else:
430 request_headers = dict(self.session.headers)
432 # Make the request. stream=True leaves the body unread
433 # until the status check below, so a non-200 body is closed
434 # unread and _ensure_decoded_body_cap can wrap the reads
435 # before response.content. SafeSession.send bounds the
436 # decoded read at MAX_RESPONSE_SIZE either way; with
437 # stream=False it would read the body before returning,
438 # before that second guard could wrap anything. That covers
439 # the final response only; SafeSession discards
440 # intermediate redirect bodies unread.
441 response = self.session.get(
442 url,
443 headers=request_headers,
444 timeout=self.timeout,
445 allow_redirects=True,
446 stream=True,
447 )
449 try:
450 # Check response
451 if response.status_code == 200:
452 self._ensure_decoded_body_cap(response)
453 try:
454 body = response.content
455 except ValueError as e:
456 raise _ResponseBodyTooLarge(str(e)) from e
457 if self._is_pdf_content(response):
458 logger.debug(
459 f"Successfully downloaded PDF from "
460 f"{redact_url_for_log(url)} on attempt {attempt}"
461 )
462 # Record successful outcome
463 self.rate_tracker.record_outcome(
464 engine_type=engine_type,
465 wait_time=wait_time,
466 success=True,
467 retry_count=attempt,
468 search_result_count=1, # We got the PDF
469 )
470 return body
471 logger.warning(
472 f"Response is not a PDF: {response.headers.get('content-type', 'unknown')}"
473 )
474 # Record failure but don't retry for wrong content type
475 self.rate_tracker.record_outcome(
476 engine_type=engine_type,
477 wait_time=wait_time,
478 success=False,
479 retry_count=attempt,
480 error_type="NotPDF",
481 )
482 return None
483 if response.status_code in [
484 429,
485 503,
486 ]: # Rate limit or service unavailable
487 logger.warning(
488 f"Attempt {attempt}/{max_attempts} - HTTP "
489 f"{response.status_code} from {redact_url_for_log(url)}"
490 )
491 # Record rate limit failure
492 self.rate_tracker.record_outcome(
493 engine_type=engine_type,
494 wait_time=wait_time,
495 success=False,
496 retry_count=attempt,
497 error_type=f"HTTP_{response.status_code}",
498 )
499 if attempt == max_attempts:
500 # A caller that passed on_rate_limited handles
501 # the give-up itself (the arXiv full-text step
502 # stops and keeps summaries), so it is not an
503 # error here; ERROR records also reach the
504 # user's browser via frontend_progress_sink.
505 give_up = (
506 f"Failed to download from "
507 f"{redact_url_for_log(url)}: HTTP "
508 f"{response.status_code} after {max_attempts} attempts"
509 )
510 if on_rate_limited is not None:
511 logger.warning(give_up)
512 on_rate_limited()
513 else:
514 logger.error(give_up)
515 return None
516 # Continue retry loop with adaptive wait
517 continue
518 logger.warning(
519 f"Failed to download from {redact_url_for_log(url)}: "
520 f"HTTP {response.status_code}"
521 )
522 # Record failure but don't retry for other status codes
523 self.rate_tracker.record_outcome(
524 engine_type=engine_type,
525 wait_time=wait_time,
526 success=False,
527 retry_count=attempt,
528 error_type=f"HTTP_{response.status_code}",
529 )
530 if (
531 on_transport_failure is not None
532 and response.status_code
533 not in _DOCUMENT_ABSENT_STATUSES
534 ):
535 on_transport_failure()
536 return None
537 finally:
538 # stream=True leaves the connection unreleased until the
539 # body is consumed or the response is closed; close on
540 # every path so non-200/NotPDF responses return their
541 # connection to the pool instead of leaking it.
542 safe_close(response, "downloader response")
544 except PolicyDeniedError:
545 raise
546 except (
547 requests.exceptions.Timeout,
548 requests.exceptions.ConnectionError,
549 ) as e:
550 # Record network failure
551 self.rate_tracker.record_outcome(
552 engine_type=engine_type,
553 wait_time=wait_time,
554 success=False,
555 retry_count=attempt,
556 error_type=type(e).__name__,
557 )
558 if attempt == max_attempts:
559 give_up = (
560 f"{type(e).__name__} downloading from "
561 f"{redact_url_for_log(url)} after {max_attempts} attempts"
562 )
563 if on_transport_failure is not None:
564 # The caller handles the give-up (see
565 # on_transport_failure), so it is not an error here.
566 logger.warning(give_up)
567 on_transport_failure()
568 else:
569 logger.opt(exception=False).error(give_up)
570 return None
571 logger.warning(
572 f"Attempt {attempt}/{max_attempts} - {type(e).__name__} "
573 f"downloading from {redact_url_for_log(url)}"
574 )
575 continue # Retry with adaptive wait
576 except requests.exceptions.RequestException as e:
577 request_error = (
578 f"Request error downloading from {redact_url_for_log(url)} "
579 f"({type(e).__name__})"
580 )
581 if on_transport_failure is not None:
582 logger.warning(request_error)
583 else:
584 logger.opt(exception=False).error(request_error)
585 # Record failure but don't retry
586 self.rate_tracker.record_outcome(
587 engine_type=engine_type,
588 wait_time=wait_time,
589 success=False,
590 retry_count=attempt,
591 error_type=type(e).__name__,
592 )
593 if on_transport_failure is not None:
594 on_transport_failure()
595 return None
596 except _ResponseBodyTooLarge:
597 # The decoded body passed MAX_RESPONSE_SIZE mid-read (guard
598 # installed by _ensure_decoded_body_cap). Not retryable: the
599 # body is oversized regardless of attempt count. The
600 # exception carries only the byte counts; its detail is
601 # kept out of the log line per the sensitive-logging hook.
602 logger.warning(
603 f"Response body exceeded size cap downloading from "
604 f"{redact_url_for_log(url)}"
605 )
606 self.rate_tracker.record_outcome(
607 engine_type=engine_type,
608 wait_time=wait_time,
609 success=False,
610 retry_count=attempt,
611 error_type="ResponseBodyTooLarge",
612 )
613 if on_transport_failure is not None:
614 on_transport_failure()
615 return None
616 except Exception as e:
617 # Includes the ValueError SafeSession raises when the URL
618 # fails validation, which is also how a failed DNS lookup
619 # surfaces, so an offline machine lands here. A caller that
620 # passed on_transport_failure handles it (see the docstring),
621 # so it is not an error here; ERROR records reach the
622 # user's browser via frontend_progress_sink.
623 unexpected_error = (
624 f"Unexpected error downloading from "
625 f"{redact_url_for_log(url)} ({type(e).__name__})"
626 )
627 if on_transport_failure is not None:
628 logger.warning(unexpected_error)
629 else:
630 logger.opt(exception=False).error(unexpected_error)
631 # Record failure but don't retry
632 self.rate_tracker.record_outcome(
633 engine_type=engine_type,
634 wait_time=wait_time,
635 success=False,
636 retry_count=attempt,
637 error_type="UnexpectedError",
638 )
639 if on_transport_failure is not None:
640 on_transport_failure()
641 return None
643 return None
645 @staticmethod
646 def extract_text_from_pdf(
647 pdf_content: bytes, max_pages: Optional[int] = None
648 ) -> Optional[str]:
649 """
650 Extract text from PDF content using in-memory processing.
652 This is part of the public API and can be used by other modules.
654 What is bounded (``utilities/pdf_extraction_limits``): at most
655 ``MAX_PDF_EXTRACTION_PAGES`` pages are extracted, at most
656 ``MAX_PDF_EXTRACTED_CHARS`` characters are returned (separators
657 included; the page that crosses the ceiling is truncated to the
658 budget left), and no new page is started once the calling thread
659 has spent ``MAX_PDF_EXTRACTION_CPU_SECONDS`` of its own CPU time
660 (``time.thread_time()``, so GIL waits and other requests' load do
661 not count) since entry. pypdf keeps every stream it decodes and
662 every object it parses on the reader; a releaser
663 (``utilities/pdf_stream_release``) releases them once they reach
664 ``MAX_PDF_RETAINED_DECODED_BYTES``, so what the walk holds
665 between pages stays near that threshold.
667 What is NOT bounded: the budget is checked between pages only,
668 so neither a single page's ``extract_text()`` nor pypdf's
669 page-tree construction (``PdfReader`` and the first access to its
670 pages, O(all pages) in time and memory for a file with many pages)
671 can be interrupted, and what one page decodes and parses is held
672 until the release after it.
674 Args:
675 pdf_content: PDF file content as bytes
676 max_pages: Maximum number of pages to extract (oldest-first).
677 None (default) extracts all pages, preserving the historic
678 behavior for dedicated PDF paths. Bounded callers (e.g. the
679 HTML-downloader recovery path, reachable from any
680 HTML-classified URL) pass a small cap so a crafted PDF with
681 many pages cannot turn one ingest into unbounded CPU/memory
682 work -- pypdf caps each stream but not the document total.
684 Returns:
685 Extracted text, or None if extraction failed
686 """
687 # Started at entry so the budget covers the whole call, parse
688 # included, as in the download service's extractor. Per-thread CPU
689 # time, not wall time, so time spent waiting on other requests is
690 # not charged (GIL-switch overhead still is; see the constant).
691 deadline = time.thread_time() + MAX_PDF_EXTRACTION_CPU_SECONDS
692 try:
693 import io
695 # Use pypdf for in-memory PDF text extraction (no disk writes)
696 from pypdf import PdfReader
698 pdf_file = io.BytesIO(pdf_content)
699 pdf_reader = PdfReader(pdf_file)
701 text_content = []
702 extracted_chars = 0
703 releaser = PypdfWalkReleaser(pdf_reader)
704 # The CPU budget is checked BETWEEN pages only — the sole check
705 # point available without running the extractor in its own
706 # process — so a single slow page still runs to completion
707 # inside pypdf.
708 for page_number, page in enumerate(pdf_reader.pages):
709 if max_pages is not None and page_number >= max_pages:
710 logger.warning(
711 "PDF extraction stopped at caller page cap "
712 f"({max_pages}); document has "
713 f"{len(pdf_reader.pages)} pages"
714 )
715 break
716 if page_number >= MAX_PDF_EXTRACTION_PAGES:
717 logger.warning(
718 "PDF extraction stopped at page ceiling "
719 f"({MAX_PDF_EXTRACTION_PAGES}); document has "
720 f"{len(pdf_reader.pages)} pages"
721 )
722 break
723 if time.thread_time() >= deadline:
724 logger.warning(
725 "PDF extraction stopped at the CPU-time ceiling "
726 f"({MAX_PDF_EXTRACTION_CPU_SECONDS}s) after "
727 f"{page_number} pages"
728 )
729 break
730 # The running count reserves the next join separator too.
731 # Do not risk a failing page when none of its text can fit.
732 remaining = MAX_PDF_EXTRACTED_CHARS - extracted_chars
733 if remaining <= 0:
734 logger.warning(
735 "PDF extraction stopped at character ceiling "
736 f"({MAX_PDF_EXTRACTED_CHARS})"
737 )
738 break
739 releaser.before_page()
740 text = page.extract_text()
741 # pypdf keeps what it decodes and parses cached on the
742 # reader; release it once it reaches
743 # MAX_PDF_RETAINED_DECODED_BYTES.
744 releaser.after_page()
745 if text:
746 if len(text) > remaining:
747 # Keep the slice that fits rather than dropping
748 # the tripping page whole.
749 if remaining > 0: 749 ↛ 752line 749 didn't jump to line 752 because the condition on line 749 was always true
750 text_content.append(text[:remaining])
751 extracted_chars += remaining
752 logger.warning(
753 "PDF extraction truncated at character ceiling "
754 f"({MAX_PDF_EXTRACTED_CHARS})"
755 )
756 break
757 # + 1 for the "\n" separator the join adds.
758 extracted_chars += len(text) + 1
759 text_content.append(text)
761 full_text = "\n".join(text_content)
762 return full_text if full_text.strip() else None
764 except Exception as exc:
765 # Scrubbed message, no traceback -- see ``close`` above. pypdf can
766 # raise ``PdfReadError`` with a message that embeds a slice of
767 # the raw stream (e.g. a malformed header byte sequence). The
768 # message is kept here not because it is always the whole
769 # diagnostic, but because this frame holds only the PDF bytes --
770 # no URL or credential -- so there is nothing for it to leak.
771 safe_msg = scrub_error(exc)
772 logger.opt(exception=False).error(
773 f"Failed to extract text from PDF: {safe_msg}"
774 )
775 return None
777 def _fetch_text_from_api(self, url: str) -> Optional[str]:
778 """
779 Fetch full text directly from API.
781 This is a placeholder - derived classes should implement
782 API-specific text fetching logic.
784 Args:
785 url: The URL or identifier
787 Returns:
788 Full text content, or None if not available
789 """
790 return None
792 def get_metadata(self, url: str) -> Dict[str, Any]:
793 """
794 Get metadata about the resource (optional override).
796 Args:
797 url: The URL to get metadata for
799 Returns:
800 Dictionary with metadata
801 """
802 return {}