Spaces:
Paused
Paused
| """ | |
| PDF Parsing Engine for trade documents. | |
| Uses pypdf (BSD-3) for PDF text extraction. | |
| Provides structured extraction of trade document fields: | |
| - Invoice number, date, total | |
| - Shipper/consignee information | |
| - Item descriptions, quantities, prices | |
| - HS codes (if present in document) | |
| - Country of origin/destination | |
| Supports: | |
| - Text-based PDFs (direct text extraction) | |
| - Scanned PDFs (limited β pypdf extracts embedded text only) | |
| Security: | |
| - File size limit: 50 MB (MAX_PDF_SIZE_BYTES) | |
| - Page limit: 500 pages (MAX_PDF_PAGES) | |
| - PDF magic bytes validation (%PDF-) | |
| - try/finally for all PDF handle operations | |
| Typing conventions: | |
| All public APIs use explicit type hints. Final constants prevent mutation. | |
| """ | |
| from __future__ import annotations | |
| import logging | |
| import re | |
| import threading | |
| import time as _time | |
| from dataclasses import dataclass, field | |
| from pathlib import Path | |
| from typing import Annotated, Any, Final, Literal, Optional | |
| logger: Final = logging.getLogger(__name__) | |
| # Security limits for PDF parsing | |
| MAX_PDF_SIZE_BYTES: Final[int] = 50 * 1024 * 1024 # 50 MB | |
| MAX_PDF_PAGES: Final[int] = 500 | |
| ALLOWED_MIME_TYPES: Final[frozenset[str]] = frozenset({"application/pdf"}) | |
| PDF_MAGIC_BYTES: Final[bytes] = b"%PDF-" | |
| try: | |
| from pypdf import PdfReader | |
| PYPDF_AVAILABLE: Final[bool] = True | |
| except ImportError: | |
| PdfReader = None # type: ignore[assignment,misc] | |
| PYPDF_AVAILABLE = False | |
| logger.warning("pypdf not installed. Install with: pip install pypdf") | |
| # ββ Domain Exceptions βββββββββββββββββββββββββββββββββββββββββββββββββ | |
| class PDFParsingError(Exception): | |
| """Base exception for all PDF parsing errors.""" | |
| class PDFStructureExploitException(PDFParsingError): | |
| """Raised when a PDF file exhibits malicious structure. | |
| Attributes: | |
| file_path: Path to the malicious file. | |
| reason: Description of the exploit detected. | |
| """ | |
| def __init__(self, file_path: str, reason: str) -> None: | |
| self.file_path = file_path | |
| self.reason = reason | |
| super().__init__(f"PDF exploit detected in {file_path}: {reason}") | |
| class PDFSizeLimitExceededError(PDFParsingError): | |
| """Raised when PDF file exceeds MAX_PDF_SIZE_BYTES. | |
| Attributes: | |
| file_size: Actual file size in bytes. | |
| limit: Maximum allowed size in bytes. | |
| """ | |
| def __init__(self, file_size: int, limit: int) -> None: | |
| self.file_size = file_size | |
| self.limit = limit | |
| super().__init__( | |
| f"PDF file too large: {file_size / (1024*1024):.1f} MB " | |
| f"(max {limit / (1024*1024):.0f} MB)" | |
| ) | |
| class PDFPageLimitExceededError(PDFParsingError): | |
| """Raised when PDF exceeds MAX_PDF_PAGES. | |
| Attributes: | |
| page_count: Actual page count. | |
| limit: Maximum allowed pages. | |
| """ | |
| def __init__(self, page_count: int, limit: int) -> None: | |
| self.page_count = page_count | |
| self.limit = limit | |
| super().__init__(f"PDF has {page_count} pages (max {limit})") | |
| class PDFNotValidError(PDFParsingError): | |
| """Raised when file is not a valid PDF.""" | |
| class PDFEmptyError(PDFParsingError): | |
| """Raised when PDF file is empty.""" | |
| # ββ Constants βββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| # Type aliases | |
| DocumentType = Literal[ | |
| "unknown", "invoice", "packing_list", "bill_of_lading", | |
| "certificate_of_origin", "quotation", "purchase_order", | |
| ] | |
| ExtractionSource = Literal["regex", "table", "ocr", "manual"] | |
| # Common invoice field patterns | |
| INVOICE_PATTERNS: Final[dict[str, list[str]]] = { | |
| "invoice_number": [ | |
| r"invoice\s*(?:#|no\.?|number)\s*[:\-]?\s*(\S+)", | |
| r"inv\s*(?:#|no\.?|number)\s*[:\-]?\s*(\S+)", | |
| r"invoice\s*(\d{4,})", | |
| ], | |
| "invoice_date": [ | |
| r"date\s*[:\-]?\s*(\d{1,2}[/\-]\d{1,2}[/\-]\d{2,4})", | |
| r"invoice\s*date\s*[:\-]?\s*(\d{1,2}[/\-]\d{1,2}[/\-]\d{2,4})", | |
| r"(\d{1,2}\s+(?:jan|feb|mar|apr|may|jun|jul|aug|sep|oct|nov|dec)\w*\s+\d{4})", | |
| ], | |
| "total_amount": [ | |
| r"total\s*[:\-]?\s*[\$β¬Β£]?\s*([\d,]+\.?\d*)", | |
| r"amount\s*due\s*[:\-]?\s*[\$β¬Β£]?\s*([\d,]+\.?\d*)", | |
| r"grand\s*total\s*[:\-]?\s*[\$β¬Β£]?\s*([\d,]+\.?\d*)", | |
| r"balance\s*due\s*[:\-]?\s*[\$β¬Β£]?\s*([\d,]+\.?\d*)", | |
| ], | |
| "subtotal": [ | |
| r"subtotal\s*[:\-]?\s*[\$β¬Β£]?\s*([\d,]+\.?\d*)", | |
| r"sub\s*total\s*[:\-]?\s*[\$β¬Β£]?\s*([\d,]+\.?\d*)", | |
| ], | |
| "tax": [ | |
| r"tax\s*[:\-]?\s*[\$β¬Β£]?\s*([\d,]+\.?\d*)", | |
| r"vat\s*[:\-]?\s*[\$β¬Β£]?\s*([\d,]+\.?\d*)", | |
| r"gst\s*[:\-]?\s*[\$β¬Β£]?\s*([\d,]+\.?\d*)", | |
| ], | |
| "hs_code": [ | |
| r"hs\s*(?:code|no\.?|number)\s*[:\-]?\s*(\d{4,12})", | |
| r"harmonized\s*(?:code|system)\s*[:\-]?\s*(\d{4,12})", | |
| r"taric\s*[:\-]?\s*(\d{4,12})", | |
| r"hts\s*[:\-]?\s*(\d{4,12})", | |
| ], | |
| "country_origin": [ | |
| r"country\s*of\s*origin\s*[:\-]?\s*([A-Za-z ]+(?:\n|$))", | |
| r"origin\s*[:\-]?\s*([A-Za-z ]+(?:\n|$))", | |
| r"made\s*in\s*[:\-]?\s*([A-Za-z ]+(?:\n|$))", | |
| ], | |
| "country_destination": [ | |
| r"destination\s*[:\-]?\s*([A-Za-z ]+(?:\n|$))", | |
| r"country\s*of\s*destination\s*[:\-]?\s*([A-Za-z ]+(?:\n|$))", | |
| r"ship\s*to\s*[:\-]?\s*([A-Za-z\s]+?)(?:\n|$)", | |
| ], | |
| } | |
| # Party identification patterns | |
| PARTY_PATTERNS: Final[dict[str, list[str]]] = { | |
| "shipper": [ | |
| r"shipper\s*[:\-]?\s*(.+?)(?:\n|$)", | |
| r"sender\s*[:\-]?\s*(.+?)(?:\n|$)", | |
| r"from\s*[:\-]?\s*(.+?)(?:\n|$)", | |
| r"exporter\s*[:\-]?\s*(.+?)(?:\n|$)", | |
| ], | |
| "consignee": [ | |
| r"consignee\s*[:\-]?\s*(.+?)(?:\n|$)", | |
| r"recipient\s*[:\-]?\s*(.+?)(?:\n|$)", | |
| r"bill\s*to\s*[:\-]?\s*(.+?)(?:\n|$)", | |
| r"ship\s*to\s*[:\-]?\s*(.+?)(?:\n|$)", | |
| r"buyer\s*[:\-]?\s*(.+?)(?:\n|$)", | |
| ], | |
| "notify_party": [ | |
| r"notify\s*(?:party)?\s*[:\-]?\s*(.+?)(?:\n|$)", | |
| ], | |
| } | |
| # Item line patterns | |
| ITEM_PATTERNS: Final[list[str]] = [ | |
| r"(\d+)\s+(.+?)\s+[\$β¬Β£]?\s*([\d,]+\.?\d*)\s+[\$β¬Β£]?\s*([\d,]+\.?\d*)", | |
| r"(\d+)\s*x\s*(.+?)\s*@\s*[\$β¬Β£]?\s*([\d,]+\.?\d*)\s*=\s*[\$β¬Β£]?\s*([\d,]+\.?\d*)", | |
| ] | |
| # Document type detection keywords | |
| _DOC_TYPE_KEYWORDS: Final[dict[str, list[str]]] = { | |
| "invoice": ["invoice", "bill to", "amount due"], | |
| "packing_list": ["packing list", "packing", "carton"], | |
| "bill_of_lading": ["bill of lading", "b/l", "shipper", "consignee"], | |
| "certificate_of_origin": ["certificate", "origin", "certify"], | |
| "quotation": ["quotation", "quote", "proposal"], | |
| "purchase_order": ["purchase order", "po#", "order"], | |
| } | |
| # ββ Data Models βββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| class ExtractedField: | |
| """Represents an extracted field with confidence score. | |
| Attributes: | |
| value: The extracted string value. | |
| confidence: Extraction confidence from 0.0 to 1.0. | |
| source: How the field was extracted (regex, table, ocr, manual). | |
| field_name: Name of the field (e.g., 'invoice_number'). | |
| """ | |
| value: str | |
| confidence: float | |
| source: ExtractionSource | |
| field_name: str = "" | |
| def is_valid(self) -> bool: | |
| """Whether the field has a value and confidence > 0.5.""" | |
| return bool(self.value) and self.confidence > 0.5 | |
| class InvoiceItem: | |
| """Represents a line item from an invoice. | |
| Attributes: | |
| description: Item description. | |
| quantity: Quantity ordered. | |
| unit_price: Price per unit. | |
| total_price: Total price for this line. | |
| hs_code: HS code for this item (if present). | |
| country_origin: Country of origin for this item. | |
| """ | |
| description: str | |
| quantity: float = 0.0 | |
| unit_price: float = 0.0 | |
| total_price: float = 0.0 | |
| hs_code: str = "" | |
| country_origin: str = "" | |
| def __post_init__(self) -> None: | |
| """Parse numeric values from strings if needed.""" | |
| if isinstance(self.quantity, str): | |
| self.quantity = self._parse_number(self.quantity) | |
| if isinstance(self.unit_price, str): | |
| self.unit_price = self._parse_number(self.unit_price) | |
| if isinstance(self.total_price, str): | |
| self.total_price = self._parse_number(self.total_price) | |
| def _parse_number(value: str) -> float: | |
| """Parse a numeric value from string. | |
| Handles US format (1,234,567.89), European format (1.234.567,89), | |
| and simple format (1234.56). | |
| Args: | |
| value: String representation of a number. | |
| Returns: | |
| Parsed float value, or 0.0 on failure. | |
| """ | |
| try: | |
| if not value or not value.strip(): | |
| return 0.0 | |
| value = value.strip() | |
| cleaned: str = re.sub(r"[$β¬Β£Β₯\s]", "", value) | |
| last_comma: int = cleaned.rfind(",") | |
| last_dot: int = cleaned.rfind(".") | |
| if last_comma > last_dot: | |
| cleaned = cleaned.replace(".", "").replace(",", ".") | |
| else: | |
| cleaned = cleaned.replace(",", "") | |
| cleaned = re.sub(r"[^\d.\-]", "", cleaned) | |
| return float(cleaned) if cleaned else 0.0 | |
| except (ValueError, TypeError): | |
| return 0.0 | |
| class Party: | |
| """Represents a trade party (shipper, consignee, etc.). | |
| Attributes: | |
| name: Party name. | |
| address: Street address. | |
| city: City name. | |
| country: Country name. | |
| contact: Contact information. | |
| """ | |
| name: str = "" | |
| address: str = "" | |
| city: str = "" | |
| country: str = "" | |
| contact: str = "" | |
| def is_valid(self) -> bool: | |
| """Whether the party has a name.""" | |
| return bool(self.name) | |
| class ParsedDocument: | |
| """Represents a parsed trade document. | |
| Attributes: | |
| file_path: Path to the source PDF file. | |
| document_type: Detected document type. | |
| raw_text: Full extracted text from the PDF. | |
| invoice_number: Extracted invoice number. | |
| invoice_date: Extracted invoice date. | |
| total_amount: Extracted total amount. | |
| subtotal: Extracted subtotal. | |
| tax: Extracted tax amount. | |
| hs_code: Extracted HS code. | |
| country_origin: Extracted country of origin. | |
| country_destination: Extracted destination country. | |
| shipper: Shipper party information. | |
| consignee: Consignee party information. | |
| notify_party: Notify party information. | |
| items: List of extracted line items. | |
| page_count: Number of pages in the PDF. | |
| parse_time_ms: Parsing time in milliseconds. | |
| warnings: List of warnings encountered during parsing. | |
| """ | |
| file_path: str | |
| document_type: DocumentType = "unknown" | |
| raw_text: str = "" | |
| invoice_number: Optional[ExtractedField] = None | |
| invoice_date: Optional[ExtractedField] = None | |
| total_amount: Optional[ExtractedField] = None | |
| subtotal: Optional[ExtractedField] = None | |
| tax: Optional[ExtractedField] = None | |
| hs_code: Optional[ExtractedField] = None | |
| country_origin: Optional[ExtractedField] = None | |
| country_destination: Optional[ExtractedField] = None | |
| shipper: Optional[Party] = None | |
| consignee: Optional[Party] = None | |
| notify_party: Optional[Party] = None | |
| items: list[InvoiceItem] = field(default_factory=list) | |
| page_count: int = 0 | |
| parse_time_ms: float = 0.0 | |
| warnings: list[str] = field(default_factory=list) | |
| def has_invoice_number(self) -> bool: | |
| """Whether a valid invoice number was extracted.""" | |
| return self.invoice_number is not None and self.invoice_number.is_valid | |
| def has_total(self) -> bool: | |
| """Whether a valid total amount was extracted.""" | |
| return self.total_amount is not None and self.total_amount.is_valid | |
| def has_items(self) -> bool: | |
| """Whether any line items were extracted.""" | |
| return len(self.items) > 0 | |
| def to_dict(self) -> dict[str, Any]: | |
| """Convert to dictionary for serialization. | |
| Returns: | |
| Dict with all document fields suitable for JSON serialization. | |
| """ | |
| return { | |
| "file_path": self.file_path, | |
| "document_type": self.document_type, | |
| "invoice_number": self.invoice_number.value if self.invoice_number else None, | |
| "invoice_date": self.invoice_date.value if self.invoice_date else None, | |
| "total_amount": self.total_amount.value if self.total_amount else None, | |
| "subtotal": self.subtotal.value if self.subtotal else None, | |
| "tax": self.tax.value if self.tax else None, | |
| "hs_code": self.hs_code.value if self.hs_code else None, | |
| "country_origin": self.country_origin.value if self.country_origin else None, | |
| "country_destination": self.country_destination.value if self.country_destination else None, | |
| "shipper": self.shipper.__dict__ if self.shipper else None, | |
| "consignee": self.consignee.__dict__ if self.consignee else None, | |
| "items_count": len(self.items), | |
| "page_count": self.page_count, | |
| "parse_time_ms": self.parse_time_ms, | |
| "warnings": self.warnings, | |
| } | |
| # ββ PDF Parser ββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| class PDFParser: | |
| """Main PDF parsing engine using pypdf (BSD-3). | |
| Provides secure PDF text extraction with: | |
| - File size validation (MAX_PDF_SIZE_BYTES) | |
| - Page count limits (MAX_PDF_PAGES) | |
| - PDF magic bytes validation | |
| - Proper resource cleanup via try/finally | |
| Raises: | |
| ImportError: If pypdf is not installed. | |
| """ | |
| def __init__(self) -> None: | |
| """Initialize the parser. | |
| Raises: | |
| ImportError: If pypdf is not installed. | |
| """ | |
| if not PYPDF_AVAILABLE: | |
| raise ImportError("pypdf not installed. Install with: pip install pypdf") | |
| def _validate_pdf(file_path: Path) -> None: | |
| """Validate PDF file before parsing. | |
| Checks file existence, size limits, and PDF magic bytes. | |
| Args: | |
| file_path: Path to the PDF file. | |
| Raises: | |
| FileNotFoundError: If file doesn't exist. | |
| PDFEmptyError: If file is empty. | |
| PDFSizeLimitExceededError: If file exceeds size limit. | |
| PDFNotValidError: If file doesn't start with %PDF-. | |
| """ | |
| if not file_path.exists(): | |
| raise FileNotFoundError(f"PDF file not found: {file_path}") | |
| file_size: int = file_path.stat().st_size | |
| if file_size == 0: | |
| raise PDFEmptyError(str(file_path)) | |
| if file_size > MAX_PDF_SIZE_BYTES: | |
| raise PDFSizeLimitExceededError(file_size, MAX_PDF_SIZE_BYTES) | |
| with open(file_path, "rb") as f: | |
| header: bytes = f.read(5) | |
| if header != PDF_MAGIC_BYTES: | |
| raise PDFNotValidError( | |
| f"Not a valid PDF file (header: {header!r})" | |
| ) | |
| def parse(self, file_path: str | Path) -> ParsedDocument: | |
| """Parse a PDF file and extract trade document fields. | |
| Args: | |
| file_path: Path to the PDF file. | |
| Returns: | |
| ParsedDocument with extracted fields and any warnings. | |
| Raises: | |
| FileNotFoundError: If file doesn't exist. | |
| PDFSizeLimitExceededError: If file is too large. | |
| PDFPageLimitExceededError: If PDF has too many pages. | |
| PDFNotValidError: If file is not a valid PDF. | |
| """ | |
| start_time: float = _time.time() | |
| path: Path = Path(file_path) | |
| self._validate_pdf(path) | |
| document = ParsedDocument(file_path=str(path)) | |
| try: | |
| reader = PdfReader(str(path)) | |
| document.page_count = len(reader.pages) | |
| if document.page_count > MAX_PDF_PAGES: | |
| raise PDFPageLimitExceededError( | |
| document.page_count, MAX_PDF_PAGES | |
| ) | |
| full_text: list[str] = [] | |
| for page_num in range(document.page_count): | |
| page = reader.pages[page_num] | |
| text: str = page.extract_text() or "" | |
| full_text.append(text) | |
| document.raw_text = "\n".join(full_text) | |
| self._extract_invoice_fields(document) | |
| self._extract_parties(document) | |
| self._extract_items(document) | |
| self._detect_document_type(document) | |
| except (PDFParsingError, FileNotFoundError): | |
| raise | |
| except Exception as exc: | |
| logger.error("Failed to parse PDF: %s", exc) | |
| document.warnings.append(f"Parse error: {exc}") | |
| document.parse_time_ms = (_time.time() - start_time) * 1000 | |
| return document | |
| def _extract_invoice_fields(self, document: ParsedDocument) -> None: | |
| """Extract invoice-specific fields from document text. | |
| Args: | |
| document: ParsedDocument to populate with extracted fields. | |
| """ | |
| text: str = document.raw_text | |
| document.invoice_number = self._extract_field( | |
| text, INVOICE_PATTERNS["invoice_number"], "invoice_number" | |
| ) | |
| document.invoice_date = self._extract_field( | |
| text, INVOICE_PATTERNS["invoice_date"], "invoice_date" | |
| ) | |
| document.total_amount = self._extract_field( | |
| text, INVOICE_PATTERNS["total_amount"], "total_amount" | |
| ) | |
| document.subtotal = self._extract_field( | |
| text, INVOICE_PATTERNS["subtotal"], "subtotal" | |
| ) | |
| document.tax = self._extract_field( | |
| text, INVOICE_PATTERNS["tax"], "tax" | |
| ) | |
| document.hs_code = self._extract_field( | |
| text, INVOICE_PATTERNS["hs_code"], "hs_code" | |
| ) | |
| document.country_origin = self._extract_field( | |
| text, INVOICE_PATTERNS["country_origin"], "country_origin" | |
| ) | |
| document.country_destination = self._extract_field( | |
| text, INVOICE_PATTERNS["country_destination"], "country_destination" | |
| ) | |
| def _extract_parties(self, document: ParsedDocument) -> None: | |
| """Extract party information (shipper, consignee, etc.). | |
| Args: | |
| document: ParsedDocument to populate with party data. | |
| """ | |
| text: str = document.raw_text | |
| shipper_name: Optional[str] = self._extract_first_match( | |
| text, PARTY_PATTERNS["shipper"] | |
| ) | |
| if shipper_name: | |
| document.shipper = Party(name=shipper_name.strip()) | |
| consignee_name: Optional[str] = self._extract_first_match( | |
| text, PARTY_PATTERNS["consignee"] | |
| ) | |
| if consignee_name: | |
| document.consignee = Party(name=consignee_name.strip()) | |
| notify_name: Optional[str] = self._extract_first_match( | |
| text, PARTY_PATTERNS["notify_party"] | |
| ) | |
| if notify_name: | |
| document.notify_party = Party(name=notify_name.strip()) | |
| def _extract_items(self, document: ParsedDocument) -> None: | |
| """Extract line items from the document. | |
| Args: | |
| document: ParsedDocument to populate with items. | |
| """ | |
| text: str = document.raw_text | |
| items: list[InvoiceItem] = [] | |
| for pattern in ITEM_PATTERNS: | |
| matches = re.finditer(pattern, text, re.MULTILINE | re.IGNORECASE) | |
| for match in matches: | |
| try: | |
| groups: tuple[str, ...] = match.groups() | |
| if len(groups) >= 4: | |
| item = InvoiceItem( | |
| description=groups[1].strip(), | |
| quantity=groups[0], | |
| unit_price=groups[2], | |
| total_price=groups[3], | |
| ) | |
| if item.description and item.total_price > 0: | |
| items.append(item) | |
| except (ValueError, IndexError): | |
| continue | |
| # Deduplicate by description | |
| seen: set[str] = set() | |
| unique_items: list[InvoiceItem] = [] | |
| for item in items: | |
| desc_key: str = item.description.lower().strip() | |
| if desc_key not in seen: | |
| seen.add(desc_key) | |
| unique_items.append(item) | |
| document.items = unique_items | |
| def _detect_document_type(self, document: ParsedDocument) -> None: | |
| """Detect the type of trade document. | |
| Args: | |
| document: ParsedDocument to classify. | |
| """ | |
| text: str = document.raw_text.lower() | |
| for doc_type, keywords in _DOC_TYPE_KEYWORDS.items(): | |
| if any(term in text for term in keywords): | |
| document.document_type = doc_type # type: ignore[assignment] | |
| return | |
| document.document_type = "unknown" | |
| def _extract_field( | |
| self, | |
| text: str, | |
| patterns: list[str], | |
| field_name: str, | |
| ) -> Optional[ExtractedField]: | |
| """Extract a field using multiple regex patterns. | |
| Args: | |
| text: Source text to search. | |
| patterns: List of regex patterns to try. | |
| field_name: Name of the field being extracted. | |
| Returns: | |
| ExtractedField if a match is found, None otherwise. | |
| """ | |
| for pattern in patterns: | |
| match: Optional[re.Match[str]] = re.search( | |
| pattern, text, re.IGNORECASE | re.MULTILINE | |
| ) | |
| if match: | |
| value: str = match.group(1).strip() | |
| if value: | |
| confidence: float = 0.8 if len(value) > 2 else 0.6 | |
| return ExtractedField( | |
| value=value, | |
| confidence=confidence, | |
| source="regex", | |
| field_name=field_name, | |
| ) | |
| return None | |
| def _extract_first_match( | |
| self, | |
| text: str, | |
| patterns: list[str], | |
| ) -> Optional[str]: | |
| """Extract first matching text from patterns. | |
| Args: | |
| text: Source text to search. | |
| patterns: List of regex patterns to try. | |
| Returns: | |
| First captured group if a match is found, None otherwise. | |
| """ | |
| for pattern in patterns: | |
| match: Optional[re.Match[str]] = re.search( | |
| pattern, text, re.IGNORECASE | re.MULTILINE | |
| ) | |
| if match: | |
| return match.group(1) | |
| return None | |
| def extract_text_only(self, file_path: str | Path) -> str: | |
| """Extract raw text from PDF without field parsing. | |
| Args: | |
| file_path: Path to the PDF file. | |
| Returns: | |
| Extracted text string. | |
| Raises: | |
| FileNotFoundError: If file doesn't exist. | |
| PDFNotValidError: If file is not a valid PDF. | |
| """ | |
| path: Path = Path(file_path) | |
| self._validate_pdf(path) | |
| try: | |
| reader = PdfReader(str(path)) | |
| full_text: list[str] = [] | |
| for page_num in range(min(len(reader.pages), MAX_PDF_PAGES)): | |
| page = reader.pages[page_num] | |
| text: str = page.extract_text() or "" | |
| full_text.append(text) | |
| return "\n".join(full_text) | |
| except (PDFParsingError, FileNotFoundError): | |
| raise | |
| except Exception as exc: | |
| raise PDFParsingError(f"Text extraction failed: {exc}") from exc | |
| def extract_tables(self, file_path: str | Path) -> list[list[list[str]]]: | |
| """Extract tables from PDF using text-based heuristics. | |
| pypdf does not have native table extraction. This method extracts | |
| text from each page and applies heuristic line-based table detection. | |
| Args: | |
| file_path: Path to the PDF file. | |
| Returns: | |
| List of tables, each table is a list of rows, each row is a list of cells. | |
| Raises: | |
| FileNotFoundError: If file doesn't exist. | |
| PDFNotValidError: If file is not a valid PDF. | |
| """ | |
| path: Path = Path(file_path) | |
| self._validate_pdf(path) | |
| tables: list[list[list[str]]] = [] | |
| try: | |
| reader = PdfReader(str(path)) | |
| for page_num in range(min(len(reader.pages), MAX_PDF_PAGES)): | |
| page = reader.pages[page_num] | |
| text: str = page.extract_text() or "" | |
| lines: list[str] = [ln.strip() for ln in text.split("\n") if ln.strip()] | |
| current_table: list[list[str]] = [] | |
| for line in lines: | |
| cells: list[str] = [c.strip() for c in re.split(r"\s{2,}|\t", line) if c.strip()] | |
| if len(cells) >= 2: | |
| current_table.append(cells) | |
| else: | |
| if len(current_table) >= 2: | |
| tables.append(current_table) | |
| current_table = [] | |
| if len(current_table) >= 2: | |
| tables.append(current_table) | |
| return tables | |
| except (PDFParsingError, FileNotFoundError): | |
| raise | |
| except Exception as exc: | |
| logger.warning("Table extraction failed: %s", exc) | |
| return tables | |
| # ββ Convenience Functions βββββββββββββββββββββββββββββββββββββββββββββ | |
| _parser: Optional[PDFParser] = None | |
| _parser_lock: threading.Lock = threading.Lock() | |
| def get_parser() -> PDFParser: | |
| """Get or create the global PDF parser (thread-safe). | |
| Returns: | |
| The singleton PDFParser instance. | |
| """ | |
| global _parser | |
| if _parser is None: | |
| with _parser_lock: | |
| if _parser is None: | |
| _parser = PDFParser() | |
| return _parser | |
| def parse_pdf(file_path: str | Path) -> ParsedDocument: | |
| """Parse a PDF file and extract trade document fields. | |
| Args: | |
| file_path: Path to the PDF file. | |
| Returns: | |
| ParsedDocument with extracted fields. | |
| """ | |
| parser: PDFParser = get_parser() | |
| return parser.parse(file_path) | |
| def extract_text(file_path: str | Path) -> str: | |
| """Extract raw text from PDF. | |
| Args: | |
| file_path: Path to the PDF file. | |
| Returns: | |
| Extracted text string. | |
| """ | |
| parser: PDFParser = get_parser() | |
| return parser.extract_text_only(file_path) | |
| def extract_tables(file_path: str | Path) -> list[list[list[str]]]: | |
| """Extract tables from PDF. | |
| Args: | |
| file_path: Path to the PDF file. | |
| Returns: | |
| List of tables. | |
| """ | |
| parser: PDFParser = get_parser() | |
| return parser.extract_tables(file_path) | |