""" 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 ─────────────────────────────────────────────────────── @dataclass(frozen=False, slots=True) 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 = "" @property def is_valid(self) -> bool: """Whether the field has a value and confidence > 0.5.""" return bool(self.value) and self.confidence > 0.5 @dataclass(frozen=False, slots=True) 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) @staticmethod 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 @dataclass(frozen=False, slots=True) 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 = "" @property def is_valid(self) -> bool: """Whether the party has a name.""" return bool(self.name) @dataclass(frozen=False, slots=True) 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) @property 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 @property 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 @property 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") @staticmethod 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)