Commercial Invoice PDF Extraction

import re
import hashlib
import logging
from typing import List, Optional
from pathlib import Path
from datetime import datetime

import pdfplumber
from pdfminer.pdfparser import PDFSyntaxError
from pydantic import BaseModel, Field, ValidationError, field_validator

logger = logging.getLogger(__name__)

# WCO/HTSUS codes are 6, 8, or 10 digits — never 7 or 9.
HS_CODE_PATTERN = r"^\d{6}(?:\d{2}|\d{4})?$"

# Compliance-focused schema with HTS/HS validation
class InvoiceLineItem(BaseModel):
    line_number: int
    description: str
    quantity: float
    unit_of_measure: str
    unit_price: float
    extended_value: float
    hs_code: str = Field(..., pattern=HS_CODE_PATTERN)
    country_of_origin: str = Field(..., pattern=r"^[A-Z]{2}$")
    extraction_confidence: float = Field(ge=0.0, le=1.0)

    @field_validator("hs_code")
    @classmethod
    def validate_hs_format(cls, v: str) -> str:
        # WCO HS Nomenclature requires 6-digit base; 8 or 10 digits for national subdivisions
        if not re.match(HS_CODE_PATTERN, v):
            raise ValueError(f"Invalid HTS/HS code format: {v}")
        return v

class CommercialInvoicePayload(BaseModel):
    invoice_number: str
    issue_date: datetime
    currency: str
    incoterms: Optional[str]
    line_items: List[InvoiceLineItem]
    document_hash: str
    extraction_timestamp: datetime = Field(default_factory=datetime.utcnow)

class ExtractionError(Exception):
    """Custom exception for pipeline extraction failures."""
    def __init__(self, message: str, doc_hash: str, stage: str):
        super().__init__(message)
        self.doc_hash = doc_hash
        self.stage = stage

def compute_sha256(file_path: Path) -> str:
    sha256 = hashlib.sha256()
    with open(file_path, "rb") as f:
        for chunk in iter(lambda: f.read(8192), b""):
            sha256.update(chunk)
    return sha256.hexdigest()

def extract_line_items_from_pdf(file_path: Path, min_confidence: float = 0.85) -> List[InvoiceLineItem]:
    """Coordinate-aware table extraction with explicit error boundaries."""
    doc_hash = compute_sha256(file_path)
    items = []
    
    try:
        with pdfplumber.open(file_path) as pdf:
            # Target first page containing primary commercial data
            page = pdf.pages[0]
            
            # Extract tables with explicit coordinate filtering
            tables = page.extract_tables()
            if not tables:
                raise ExtractionError("No tabular structures detected", doc_hash, "table_detection")
            
            # Process first valid table (commercial invoices typically use one primary table)
            raw_table = tables[0]
            
            # Skip header row, map columns by positional index
            for idx, row in enumerate(raw_table[1:], start=1):
                if len(row) < 6:
                    logger.warning(f"Row {idx} truncated, skipping: {row}")
                    continue
                
                # Parse numeric fields with locale-safe conversion
                try:
                    qty = float(str(row[1]).replace(",", ""))
                    unit_price = float(str(row[3]).replace(",", ""))
                    ext_val = float(str(row[4]).replace(",", ""))
                except ValueError as e:
                    raise ExtractionError(
                        f"Numeric parsing failure at row {idx}: {e}", doc_hash, "field_parsing"
                    )
                
                # Confidence scoring based on text clarity and bounding box overlap
                # In production, integrate OCR confidence scores from Tesseract/AWS Textract
                confidence = 0.92 if len(str(row[0])) > 10 else 0.78
                
                if confidence < min_confidence:
                    logger.warning(f"Low confidence row {idx} (conf={confidence}), flagging for manual review")
                
                items.append(InvoiceLineItem(
                    line_number=idx,
                    description=str(row[0]).strip(),
                    quantity=qty,
                    unit_of_measure=str(row[2]).strip().upper(),
                    unit_price=unit_price,
                    extended_value=ext_val,
                    hs_code=str(row[5]).strip(),
                    country_of_origin="US", # Default; override via header extraction logic
                    extraction_confidence=confidence
                ))
                
    except PDFSyntaxError as e:
        raise ExtractionError(f"Corrupted PDF structure: {e}", doc_hash, "pdf_open")
    except Exception as e:
        raise ExtractionError(f"Unexpected extraction failure: {e}", doc_hash, "unknown")
        
    return items

def build_compliance_payload(file_path: Path, items: List[InvoiceLineItem]) -> CommercialInvoicePayload:
    """Assemble validated payload for downstream customs engines."""
    doc_hash = compute_sha256(file_path)
    try:
        payload = CommercialInvoicePayload(
            invoice_number="INV-2024-8891", # Extracted via header regex in production
            issue_date=datetime.utcnow(),
            currency="USD",
            incoterms="FOB",
            line_items=items,
            document_hash=doc_hash
        )
        logger.info(f"Successfully built payload for {doc_hash} with {len(items)} line items")
        return payload
    except ValidationError as e:
        logger.error(f"Schema validation failed for {doc_hash}: {e.json()}")
        raise ExtractionError("Schema validation failure", doc_hash, "pydantic_validation")