"""PDF content extraction for NSCT crawler. Extracts text from PDF documents using pdfminer.six or pdfplumber as fallback. Returns structured error markers when text extraction is not possible. """ from __future__ import annotations import io import logging from typing import BinaryIO logger = logging.getLogger(__name__) try: from pdfminer.high_level import extract_text as pdfminer_extract_text from pdfminer.layout import LAParams _HAS_PDFMINER = True except ImportError: _HAS_PDFMINER = False pdfminer_extract_text = None # type: ignore[assignment,misc] try: import pdfplumber _HAS_PDFPLUMBER = True except ImportError: _HAS_PDFPLUMBER = False pdfplumber = None # type: ignore[assignment,misc] try: from pdfminer.pdfinterp import PDFResourceManager, PDFPageInterpreter from pdfminer.pdfpage import PDFPage from pdfminer.converter import XMLConverter, TextConverter from pdfminer.layout import LAParams _HAS_PDFMINER_FULL = True except ImportError: _HAS_PDFMINER_FULL = False PDFResourceManager = None # type: ignore[assignment,misc] PDFPage = None # type: ignore[assignment,misc] PDFPageInterpreter = None # type: ignore[assignment,misc] XMLConverter = None # type: ignore[assignment,misc] TextConverter = None # type: ignore[assignment,misc] def extract_pdf_from_bytes(pdf_bytes: bytes) -> str: """Extract text content from raw PDF bytes. Priority: pdfminer.six -> pdfplumber -> minimal fallback. Args: pdf_bytes: Raw PDF content. Returns: Extracted text content, or an error marker string if extraction fails. """ if not pdf_bytes: return "pdf_extract_failed" content_types = "pdf_no_text" # --- Try pdfminer.six --- if _HAS_PDFMINER: try: stream = io.BytesIO(pdf_bytes) text = pdfminer_extract_text( stream, laparams=LAParams( line_margin=0.5, word_margin=0.1, char_margin=2.0, boxes_flow=0.5, ), ) if text and text.strip(): return text.strip() except Exception as exc: logger.debug("pdfminer extraction failed: %s", exc) # --- Try pdfplumber --- if _HAS_PDFPLUMBER: try: stream = io.BytesIO(pdf_bytes) with pdfplumber.open(stream) as pdf: parts: list[str] = [] for page in pdf.pages: page_text = page.extract_text() if page_text: parts.append(page_text) if parts: text = "\n\n".join(parts) return text.strip() except Exception as exc: logger.debug("pdfplumber extraction failed: %s", exc) return f"pdf_extract_failed" def detect_pdf_content_type(pdf_bytes: bytes) -> str: """Detect the content type of PDF bytes. Args: pdf_bytes: Raw PDF content. Returns: Content type string. """ if not pdf_bytes: return "unknown" # Check PDF magic bytes if pdf_bytes[:4] == b"%PDF": return "application/pdf" return "unknown"