Coverage for src/local_deep_research/web/services/pdf_extraction_service.py: 98%
83 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-22 21:20 +0000
1"""
2PDF text extraction service.
4Provides efficient PDF text extraction with single-pass processing.
5Complements pdf_service.py which handles PDF generation.
6"""
8import io
9import time
10from typing import Any, Dict, List
12import pdfplumber
13from loguru import logger
15from local_deep_research.security.filename_sanitizer import sanitize_filename
16from local_deep_research.utilities.pdf_extraction_limits import (
17 MAX_PDF_EXTRACTED_CHARS,
18 MAX_PDF_EXTRACTION_CPU_SECONDS,
19 MAX_PDF_EXTRACTION_PAGES,
20)
21from local_deep_research.utilities.pdf_page_walk import iter_pdfplumber_pages
22from local_deep_research.utilities.pdf_stream_release import (
23 PdfplumberWalkReleaser,
24)
25from local_deep_research.utilities.resource_utils import safe_close
28class PDFExtractionService:
29 """Service for extracting text and metadata from PDF files."""
31 @staticmethod
32 def extract_text_and_metadata(
33 pdf_content: bytes, filename: str
34 ) -> Dict[str, Any]:
35 """
36 Extract text and metadata from PDF in a single pass.
38 This method opens the PDF only once and extracts both text content
39 and metadata (page count) in the same operation, avoiding the
40 performance issue of opening the file multiple times.
42 What is bounded (``utilities/pdf_extraction_limits``): text is
43 extracted from at most ``MAX_PDF_EXTRACTION_PAGES`` pages, at most
44 ``MAX_PDF_EXTRACTED_CHARS`` characters are kept (separators
45 included; the page that crosses the ceiling is truncated to the
46 budget left), and no new page is started once the calling thread
47 has spent ``MAX_PDF_EXTRACTION_CPU_SECONDS`` of its own CPU time
48 (``time.thread_time()``) since entry. The page tree is walked
49 lazily (``utilities/pdf_page_walk``), so a document declaring more
50 pages than the ceiling has at most the ceiling's pages (plus the
51 one that shows there are more) built, not all of them. Every page
52 the walk reaches is closed once done with, which releases its
53 per-page layout caches; what pdfminer decodes and parses for a
54 page stays cached on the document after that, so the walk hands
55 each page to a releaser (``utilities/pdf_stream_release``) that
56 releases those caches once they reach
57 ``MAX_PDF_RETAINED_DECODED_BYTES``.
59 What is NOT bounded: the budget is checked between pages only,
60 so neither ``pdfplumber.open`` (the cross-reference parse) nor a
61 single page's ``extract_text()`` can be interrupted, and what one
62 page decodes and parses is held until the release after it.
64 Args:
65 pdf_content: Raw PDF file bytes
66 filename: Original filename (for logging)
68 Returns:
69 Dictionary with keys:
70 - 'text': Extracted text content
71 - 'pages': Number of pages the walk reached. When the walk
72 stops early (a ceiling, or the CPU budget) the pages after
73 it are never built, so this is a lower bound and
74 'truncated' is True. The page tree's declared ``/Count``
75 is deliberately not read: resolving it can loop inside
76 pdfminer (a self-referencing object, or an object-stream
77 or ``/Length`` reference cycle) after the CPU budget has
78 stopped applying
79 - 'size': File size in bytes
80 - 'filename': Original filename
81 - 'truncated': True if a page or character ceiling, or the
82 CPU budget, stopped the extraction early (False otherwise, including failures)
83 - 'success': Boolean indicating success
84 - 'error': Error message if failed (None if successful)
86 Raises:
87 No exceptions - errors are captured in return dict
88 """
89 # Defense in depth: re-sanitize even though callers should
90 # have sanitized already — guards against new callers that forget
91 try:
92 filename = sanitize_filename(filename)
93 except Exception:
94 filename = "unnamed.pdf"
96 # Started at entry so the budget covers the whole call, the open
97 # included, as in the library download paths. Per-thread CPU time,
98 # not wall time: waiting on other requests is not charged.
99 deadline = time.thread_time() + MAX_PDF_EXTRACTION_CPU_SECONDS
100 try:
101 # Not a ``with`` block: ``PDF.close()`` accesses ``PDF.pages``,
102 # which builds every page of the document. The stream is an
103 # in-memory buffer, so nothing else needs closing; each page
104 # the walk reaches is closed below.
105 pdf = pdfplumber.open(io.BytesIO(pdf_content))
106 text_parts = []
107 extracted_chars = 0
108 truncated = False
109 pages_seen = 0
110 releaser = PdfplumberWalkReleaser(pdf)
111 # The walk calls releaser.before_page() before building each
112 # page, so what building it resolves and caches (its
113 # /Resources, /Contents, page-tree nodes) is counted too.
114 for page_number, page in enumerate(
115 iter_pdfplumber_pages(pdf, releaser)
116 ):
117 pages_seen = page_number + 1
118 # Close every page this loop reaches, on the break
119 # paths too: pdfplumber keeps a page's parsed layout
120 # objects and text map cached on it until close().
121 try:
122 if page_number >= MAX_PDF_EXTRACTION_PAGES:
123 truncated = True
124 logger.warning(
125 f"PDF extraction stopped at page ceiling "
126 f"({MAX_PDF_EXTRACTION_PAGES}) for {filename}"
127 )
128 break
129 if time.thread_time() >= deadline:
130 truncated = True
131 logger.warning(
132 "PDF extraction stopped at the CPU-time "
133 f"ceiling ({MAX_PDF_EXTRACTION_CPU_SECONDS}s) "
134 f"after {page_number} pages for {filename}"
135 )
136 break
137 separator_chars = 1 if text_parts else 0
138 remaining = (
139 MAX_PDF_EXTRACTED_CHARS
140 - extracted_chars
141 - separator_chars
142 )
143 if remaining <= 0:
144 truncated = True
145 logger.warning(
146 f"PDF extraction stopped at character "
147 f"ceiling ({MAX_PDF_EXTRACTED_CHARS}) "
148 f"for {filename}"
149 )
150 break
151 page_text = page.extract_text()
152 if page_text:
153 if len(page_text) > remaining:
154 # Keep the slice that fits rather than
155 # dropping the tripping page whole: a first
156 # page over the ceiling still yields the
157 # budget's text.
158 text_parts.append(page_text[:remaining])
159 extracted_chars += separator_chars + remaining
160 truncated = True
161 logger.warning(
162 f"PDF extraction truncated at character "
163 f"ceiling ({MAX_PDF_EXTRACTED_CHARS}) "
164 f"for {filename}"
165 )
166 break
167 extracted_chars += separator_chars + len(page_text)
168 text_parts.append(page_text)
169 finally:
170 # safe_close: a failing close() must not mask the
171 # page's own exception.
172 safe_close(page, "pdfplumber page")
173 # close() leaves what pdfminer decoded and parsed
174 # for the page cached on the document; release it
175 # once it reaches MAX_PDF_RETAINED_DECODED_BYTES.
176 releaser.after_page()
178 # A lower bound when truncated: the pages after the stop
179 # were never built, and /Count is not consulted (see the
180 # docstring) because no reference is resolved after the walk.
181 page_count = pages_seen
183 # Combine all text
184 full_text = "\n".join(text_parts)
186 # Check if any text was extracted
187 if not full_text.strip():
188 logger.warning(f"No extractable text found in {filename}")
189 return {
190 "text": "",
191 "pages": page_count,
192 "size": len(pdf_content),
193 "filename": filename,
194 "truncated": truncated,
195 "success": False,
196 "error": "No extractable text found",
197 }
199 logger.info(
200 f"Successfully extracted text from {filename} "
201 f"({len(full_text)} chars, {page_count} pages)"
202 )
204 return {
205 "text": full_text.strip(),
206 "pages": page_count,
207 "size": len(pdf_content),
208 "filename": filename,
209 "truncated": truncated,
210 "success": True,
211 "error": None,
212 }
214 except Exception:
215 # Log full exception details server-side for debugging
216 logger.exception(f"Error extracting text from {filename}")
217 # Return generic error message to avoid exposing internal details
218 return {
219 "text": "",
220 "pages": 0,
221 "size": len(pdf_content),
222 "filename": filename,
223 "truncated": False,
224 "success": False,
225 "error": "Failed to extract text from PDF",
226 }
228 @staticmethod
229 def extract_batch(files_data: List[Dict[str, Any]]) -> Dict[str, Any]:
230 """
231 Extract text from multiple PDF files.
233 Args:
234 files_data: List of dicts with 'content' (bytes) and 'filename' (str)
236 Returns:
237 Dictionary with:
238 - 'results': List of extraction results
239 - 'total_files': Total number of files processed
240 - 'successful': Number of successfully processed files
241 - 'failed': Number of failed files
242 - 'errors': List of error messages
243 """
244 results: list[Dict[str, Any]] = []
245 successful = 0
246 failed = 0
247 errors: list[str] = []
249 for file_data in files_data:
250 result = PDFExtractionService.extract_text_and_metadata(
251 file_data["content"], file_data["filename"]
252 )
254 results.append(result)
256 if result["success"]:
257 successful += 1
258 else:
259 failed += 1
260 errors.append(f"{file_data['filename']}: {result['error']}")
262 return {
263 "results": results,
264 "total_files": len(files_data),
265 "successful": successful,
266 "failed": failed,
267 "errors": errors,
268 }
271# Singleton pattern for service
272_pdf_extraction_service = None
275def get_pdf_extraction_service() -> PDFExtractionService:
276 """Get the singleton PDF extraction service instance."""
277 global _pdf_extraction_service
278 if _pdf_extraction_service is None:
279 _pdf_extraction_service = PDFExtractionService()
280 return _pdf_extraction_service