Automated Invoice Processing: How to Eliminate Manual Data Entry for Good

Human uploads file
       ↓
API extracts data
       ↓
Form pre-filled for review
       ↓
Human approves → data saved

Email arrives with PDF attachment
       ↓
Automation detects attachment
       ↓
API extracts data
       ↓
Validation passes? → record created automatically
Validation fails?  → routed to human review queue

Invoice received (email, portal, EDI)
       ↓
Document queued for extraction
       ↓
Webhook fires when extraction completes
       ↓
Automated validation + vendor matching
       ↓
Threshold exceeded or unknown vendor? → Review queue
Otherwise? → Accounting record created

import os
import requests

def extract_invoice(file_content: bytes, filename: str) -> dict:
    """
    Extract structured data from an invoice PDF.
    
    Returns dict with: merchant, invoice_id, date, due_date,
    currency, subtotal, tax, total, line_items
    """
    response = requests.post(
        "https://docuparseapi.com/api/v1/extract",
        headers={"Authorization": f"Bearer {os.environ['DOCUPARSE_API_KEY']}"},
        files={"file": (filename, file_content)},
        timeout=30,
    )
    
    data = response.json()
    
    if not data.get("success"):
        raise RuntimeError(
            f"Extraction failed [{data.get('error', {}).get('code', 'UNKNOWN')}]"
        )
    
    return data

def validate_invoice(extracted: dict, config: dict) -> tuple[bool, list[str]]:
    """
    Check extracted invoice data against business rules.
    
    Returns (is_valid, list_of_issues)
    Issues don't necessarily block processing — some route to review, some reject.
    """
    issues = []
    
    # Required fields
    if not extracted.get("total"):
        issues.append("MISSING_TOTAL")
    
    if not extracted.get("merchant"):
        issues.append("MISSING_VENDOR")
    
    # Approval threshold
    try:
        total = float(extracted.get("total") or 0)
        if total > config.get("approval_threshold", 2000):
            issues.append("ABOVE_APPROVAL_THRESHOLD")
    except (ValueError, TypeError):
        issues.append("INVALID_TOTAL_FORMAT")
    
    # Currency check
    allowed_currencies = config.get("allowed_currencies", ["USD", "EUR", "GBP"])
    if extracted.get("currency") and extracted["currency"] not in allowed_currencies:
        issues.append("UNEXPECTED_CURRENCY")
    
    # Duplicate check (you need to implement the DB lookup)
    if extracted.get("invoice_id") and extracted.get("merchant"):
        if is_duplicate(extracted["invoice_id"], extracted["merchant"]):
            issues.append("DUPLICATE_INVOICE")
    
    # Auto-approve if no blocking issues
    blocking = {"MISSING_TOTAL", "DUPLICATE_INVOICE", "INVALID_TOTAL_FORMAT"}
    is_valid = not any(i in blocking for i in issues)
    
    return is_valid, issues


def is_duplicate(invoice_id: str, merchant: str) -> bool:
    """Check database for existing invoice with same number + vendor."""
    # Implement with your DB of choice
    # e.g.: return db.invoices.exists(invoice_id=invoice_id, merchant=merchant)
    return False

REVIEW_TRIGGERS = {
    "MISSING_VENDOR",
    "ABOVE_APPROVAL_THRESHOLD", 
    "UNEXPECTED_CURRENCY",
}

REJECT_TRIGGERS = {
    "DUPLICATE_INVOICE",
    "MISSING_TOTAL",
    "INVALID_TOTAL_FORMAT",
}

def route_invoice(extracted: dict, issues: list[str], is_valid: bool) -> str:
    """Determine where the invoice goes based on validation issues."""
    
    if any(i in REJECT_TRIGGERS for i in issues):
        return "REJECT"
    
    if any(i in REVIEW_TRIGGERS for i in issues):
        return "REVIEW"
    
    if is_valid and not issues:
        return "AUTO_APPROVE"
    
    return "REVIEW"  # Default to review for anything unclear


def process_invoice_from_email(file_content: bytes, filename: str) -> dict:
    """
    Full processing pipeline for an email attachment.
    Called by your email monitoring workflow.
    """
    config = {
        "approval_threshold": 2000,
        "allowed_currencies": ["USD", "EUR", "GBP"],
    }
    
    # Extract
    try:
        extracted = extract_invoice(file_content, filename)
    except RuntimeError as e:
        # Extraction itself failed — send to manual review
        save_to_review_queue(
            file_content=file_content,
            filename=filename,
            reason="EXTRACTION_FAILED",
            error=str(e),
        )
        return {"status": "review", "reason": "EXTRACTION_FAILED"}
    
    # Validate
    is_valid, issues = validate_invoice(extracted, config)
    
    # Route
    destination = route_invoice(extracted, issues, is_valid)
    
    if destination == "AUTO_APPROVE":
        record_id = create_accounting_record(extracted)
        return {
            "status": "processed",
            "record_id": record_id,
            "merchant": extracted.get("merchant"),
            "total": extracted.get("total"),
        }
    
    elif destination == "REVIEW":
        queue_id = save_to_review_queue(
            extracted_data=extracted,
            filename=filename,
            issues=issues,
        )
        return {
            "status": "review",
            "queue_id": queue_id,
            "issues": issues,
        }
    
    else:  # REJECT
        log_rejected_invoice(extracted, issues)
        return {
            "status": "rejected",
            "issues": issues,
        }