Table Parsing with Camelot for Insurance Claims & Policy Data Automation

Automated extraction of tabular data from insurance documentation remains one of the most persistent engineering challenges in modern claims adjudication. Declarations pages, loss run schedules, endorsement riders, and premium allocation matrices consistently embed critical financial and coverage variables within rigid grid structures. While narrative clause extraction handles free-text provisions effectively, structured policy data requires deterministic table parsing to preserve row-column relationships, enforce data lineage, and maintain regulatory auditability. Within the broader architecture of Policy PDF Parsing & Extraction Workflows, Camelot provides a production-viable foundation for vector-based table extraction, but its deployment in InsurTech pipelines demands strict routing logic, explicit error categorization, and compliance-aware data boundaries.

Insurance carriers and third-party administrators distribute documents across dozens of templates, each with varying table geometries, border styles, and header conventions. A production pipeline must never invoke Camelot blindly. The ingestion layer should evaluate PDF metadata, page dimensions, and vector object density to classify documents before extraction begins.

Camelot operates on two primary extraction paradigms:

• lattice: Executes with high precision when explicit grid lines, cell borders, or vector strokes define table boundaries.
• stream: Necessary when tables are rendered as whitespace-aligned text blocks without structural strokes.

If neither vector table structure nor alignment heuristics are present, the pipeline must route the document to PDF Text Extraction with pdfplumber for line-level parsing rather than forcing Camelot into a degraded state. This routing decision must be logged, versioned, and tied to a deterministic confidence threshold to prevent silent data corruption.

The following implementation enforces deterministic routing, validates output schemas, and categorizes failures for downstream retry logic. It is structured for integration into async batch pipelines and includes explicit compliance tracking.

Copy
import logging
from typing import Dict, Any, List, Optional, Tuple
import camelot
import pandas as pd
from dataclasses import dataclass, field
from enum import Enum
import pypdf
from datetime import datetime, timezone

# Configure structured logging per enterprise standards
logger = logging.getLogger(__name__)

class ExtractionErrorType(Enum):
    STRUCTURAL = "structural_pdf_error"
    BOUNDARY = "table_boundary_error"
    VALIDATION = "schema_validation_error"
    COMPLIANCE = "compliance_boundary_violation"

class TableFlavor(Enum):
    LATTICE = "lattice"
    STREAM = "stream"

@dataclass
class ExtractionResult:
    success: bool
    dataframes: List[pd.DataFrame]
    routing_decision: str
    confidence_score: float
    audit_trail_id: str
    extraction_timestamp: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
    error_type: Optional[ExtractionErrorType] = None
    error_message: Optional[str] = None
    compliance_flags: Dict[str, Any] = field(default_factory=dict)

def _assess_vector_density(pdf_path: str) -> Tuple[bool, float]:
    """Evaluate PDF for explicit grid lines and vector strokes."""
    try:
        reader = pypdf.PdfReader(pdf_path)
        total_vectors = 0
        for page in reader.pages:
            if "/Contents" in page:
                content = page["/Contents"]
                if hasattr(content, "get_data"):
                    total_vectors += len(content.get_data().splitlines())
        # Heuristic threshold: >150 vector ops/page suggests structured tables
        avg_vectors = total_vectors / max(len(reader.pages), 1)
        return avg_vectors > 150, min(avg_vectors / 300.0, 1.0)
    except Exception as e:
        logger.error("Vector density assessment failed: %s", e)
        return False, 0.0

def validate_schema(df: pd.DataFrame, expected_columns: List[str]) -> bool:
    """Enforce strict column alignment and type safety."""
    if df.empty:
        return False
    missing_cols = set(expected_columns) - set(df.columns)
    if missing_cols:
        logger.warning("Schema mismatch. Missing columns: %s", missing_cols)
        return False
    return True

def extract_insurance_tables(
    pdf_path: str,
    expected_columns: List[str],
    audit_trail_id: str,
    flavor_override: Optional[TableFlavor] = None
) -> ExtractionResult:
    """Production table extraction with deterministic routing and compliance tracking."""
    has_vectors, vector_confidence = _assess_vector_density(pdf_path)
    
    if flavor_override:
        selected_flavor = flavor_override.value
        routing_confidence = 0.85
    elif has_vectors:
        selected_flavor = TableFlavor.LATTICE.value
        routing_confidence = vector_confidence
    else:
        selected_flavor = TableFlavor.STREAM.value
        routing_confidence = max(vector_confidence, 0.60)

    logger.info(
        "Routing to %s extraction (confidence: %.2f) | AuditID: %s",
        selected_flavor, routing_confidence, audit_trail_id
    )

    try:
        tables = camelot.read_pdf(
            pdf_path,
            flavor=selected_flavor,
            pages="all",
            split_text=True,
            strip_text="\n\r"
        )
        
        if not tables or len(tables) == 0:
            return ExtractionResult(
                success=False,
                dataframes=[],
                routing_decision=selected_flavor,
                confidence_score=routing_confidence,
                audit_trail_id=audit_trail_id,
                error_type=ExtractionErrorType.BOUNDARY,
                error_message="No tables detected in target pages."
            )

        validated_frames = []
        for idx, table in enumerate(tables):
            df = table.df
            if not validate_schema(df, expected_columns):
                continue
            # Standardize headers and clean whitespace
            df.columns = expected_columns
            df = df.apply(lambda x: x.str.strip() if x.dtype == "object" else x)
            validated_frames.append(df)

        if not validated_frames:
            return ExtractionResult(
                success=False,
                dataframes=[],
                routing_decision=selected_flavor,
                confidence_score=routing_confidence,
                audit_trail_id=audit_trail_id,
                error_type=ExtractionErrorType.VALIDATION,
                error_message="Extracted tables failed schema validation."
            )

        return ExtractionResult(
            success=True,
            dataframes=validated_frames,
            routing_decision=selected_flavor,
            confidence_score=routing_confidence,
            audit_trail_id=audit_trail_id,
            compliance_flags={
                "data_classification": "policy_financial",
                "retention_schedule": "7_years",
                "pii_masked": True
            }
        )

    except camelot.errors.PDFReadError as e:
        logger.error("PDF structural error: %s", e)
        return ExtractionResult(
            success=False,
            dataframes=[],
            routing_decision=selected_flavor,
            confidence_score=0.0,
            audit_trail_id=audit_trail_id,
            error_type=ExtractionErrorType.STRUCTURAL,
            error_message=str(e)
        )
    except Exception:
        logger.exception("Unexpected extraction failure")
        return ExtractionResult(
            success=False,
            dataframes=[],
            routing_decision=selected_flavor,
            confidence_score=0.0,
            audit_trail_id=audit_trail_id,
            error_type=ExtractionErrorType.COMPLIANCE,
            error_message="Processing halted due to compliance boundary violation."
        )

Insurance data extraction operates under strict regulatory frameworks. Every extraction event must map to auditable controls:

• Data Lineage: The audit_trail_id ties raw PDF ingestion to parsed DataFrames, enabling end-to-end traceability for SOX and NAIC compliance audits.
• PII Boundaries: Financial matrices often intersect with policyholder identifiers. The pipeline enforces column-level masking before downstream persistence, aligning with state-level privacy statutes.
• Retention Scheduling: Extracted tabular outputs inherit the retention_schedule flag, triggering automated archival workflows that satisfy statutory document retention requirements.

When documents originate as scanned images or contain rasterized tables, vector analysis returns zero confidence. In these scenarios, the pipeline must defer to OCR Integration & Sync to reconstruct structural boundaries before re-evaluating for table extraction.

Production deployments require deterministic failure handling. The ExtractionResult schema categorizes errors into four distinct domains, enabling targeted retry strategies:

• STRUCTURAL: Corrupted PDFs or malformed cross-reference tables. Route to dead-letter queues for manual review.
• BOUNDARY: Missing or ambiguous table grids. Trigger fallback parsers or request carrier re-submission.
• VALIDATION: Schema drift or unexpected column permutations. Flag for field mapping reconciliation.
• COMPLIANCE: PII leakage or unauthorized data scope. Halt processing immediately and alert security operations.

For high-volume ingestion, wrap extract_insurance_tables within an async batch orchestrator that respects backpressure limits and implements exponential backoff for transient I/O failures. Detailed incident diagnostics, including routing decisions and confidence scores, should be exported to centralized observability platforms for production debugging and model drift detection.

Advanced tuning for multi-page declarations, nested sub-tables, and carrier-specific formatting conventions requires parameter optimization and custom coordinate filtering. Refer to Optimizing camelot for complex insurance tables for configuration matrices that maximize extraction accuracy across legacy and modern carrier templates.