vgtc-api / src /hermes /tools /pdf_parser.py
vora-sonnet's picture
Upload folder using huggingface_hub (part 2)
b9f94e1 verified
Raw
History Blame Contribute Delete
28.1 kB
"""
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)