Automate Accounts Payable Invoice Processing with AI: A Complete Pipeline Guide

pip install requests python-dotenv watchdog boto3

APIVULT_API_KEY=YOUR_API_KEY
DATABASE_URL=postgresql://user:pass@host:5432/ap_db

import os
import base64
import requests
from pathlib import Path
from dotenv import load_dotenv
 
load_dotenv()
 
API_KEY = os.getenv("APIVULT_API_KEY")
BASE_URL = "https://apivult.com/api/finaudit"
 
def extract_invoice_data(file_path: str) -> dict:
    """
    Extract structured data from an invoice PDF or image.
    Handles handwritten, printed, and scanned invoices.
    """
    headers = {
        "X-RapidAPI-Key": API_KEY,
        "Content-Type": "application/json"
    }
 
    # Read and encode the file
    with open(file_path, "rb") as f:
        file_content = base64.b64encode(f.read()).decode("utf-8")
 
    file_ext = Path(file_path).suffix.lower().replace(".", "")
 
    payload = {
        "document": file_content,
        "document_format": file_ext,  # "pdf", "jpg", "png", "docx"
        "document_type": "invoice",
        "extraction_fields": [
            "invoice_number",
            "invoice_date",
            "due_date",
            "vendor_name",
            "vendor_address",
            "vendor_tax_id",
            "bill_to",
            "line_items",
            "subtotal",
            "tax_amount",
            "total_amount",
            "payment_terms",
            "bank_details",
            "purchase_order_reference"
        ],
        "currency_normalize": True,  # standardize all amounts to USD
        "date_normalize": "ISO8601"
    }
 
    response = requests.post(
        f"{BASE_URL}/extract",
        json=payload,
        headers=headers,
        timeout=30
    )
    response.raise_for_status()
    return response.json()
 
# Extract from an invoice
result = extract_invoice_data("invoices/vendor_invoice_2026_04.pdf")
invoice = result["extracted_data"]
 
print(f"Invoice #: {invoice['invoice_number']}")
print(f"Vendor: {invoice['vendor_name']}")
print(f"Amount: ${invoice['total_amount']:.2f}")
print(f"Due Date: {invoice['due_date']}")
print(f"PO Reference: {invoice.get('purchase_order_reference', 'NONE')}")
print(f"\nLine Items:")
for item in invoice.get("line_items", []):
    print(f"  - {item['description']}: {item['quantity']} x ${item['unit_price']:.2f}")

{
  "invoice_number": "INV-2026-0847",
  "invoice_date": "2026-04-01",
  "due_date": "2026-05-01",
  "vendor_name": "Acme Office Supplies Ltd",
  "vendor_tax_id": "12-3456789",
  "total_amount": 4872.50,
  "purchase_order_reference": "PO-2026-0312",
  "line_items": [
    {"description": "Laser Paper A4 (case)", "quantity": 50, "unit_price": 42.50},
    {"description": "Toner Cartridge HP 26A", "quantity": 10, "unit_price": 189.75}
  ],
  "confidence": 0.97
}

def match_invoice_to_po(
    invoice: dict,
    po_database: dict,
    receiving_report: dict = None
) -> dict:
    """
    Perform two-way (PO+invoice) or three-way (PO+invoice+receiving) match.
    Returns match result with variance details.
    """
    po_ref = invoice.get("purchase_order_reference")
 
    if not po_ref:
        return {
            "match_status": "NO_PO_REFERENCE",
            "requires_approval": True,
            "reason": "Invoice has no PO number — requires manual PO lookup"
        }
 
    po = po_database.get(po_ref)
 
    if not po:
        return {
            "match_status": "PO_NOT_FOUND",
            "requires_approval": True,
            "block": True,
            "reason": f"PO {po_ref} not found in system — possible unauthorized purchase"
        }
 
    # Check amount tolerance (typically ±5% is acceptable)
    amount_variance = (invoice["total_amount"] - po["approved_amount"]) / po["approved_amount"]
    amount_ok = abs(amount_variance) <= 0.05
 
    # Check vendor match
    vendor_ok = invoice["vendor_name"].lower() in po["approved_vendor"].lower()
 
    match_result = {
        "po_reference": po_ref,
        "po_approved_amount": po["approved_amount"],
        "invoice_amount": invoice["total_amount"],
        "amount_variance_pct": round(amount_variance * 100, 2),
        "amount_match": amount_ok,
        "vendor_match": vendor_ok,
        "match_type": "two_way"
    }
 
    # Three-way match if receiving report available
    if receiving_report:
        match_result["match_type"] = "three_way"
        quantities_ok = all(
            abs(item["received_qty"] - item["invoiced_qty"]) <= 1
            for item in receiving_report.get("items", [])
        )
        match_result["quantity_match"] = quantities_ok
 
    # Overall match decision
    if amount_ok and vendor_ok:
        match_result["match_status"] = "MATCHED"
    elif not vendor_ok:
        match_result["match_status"] = "VENDOR_MISMATCH"
        match_result["block"] = True
    elif amount_variance > 0.05:
        match_result["match_status"] = "OVERBILLED"
        match_result["requires_approval"] = True
    else:
        match_result["match_status"] = "UNDERBILLED"
        match_result["requires_approval"] = True
 
    return match_result

def audit_invoice_for_anomalies(
    invoice: dict,
    vendor_history: list
) -> dict:
    """
    Run AI anomaly detection on an invoice.
    Checks for: duplicate detection, round-number bias, unusual timing,
    vendor pattern deviations, and synthetic invoice indicators.
    """
    headers = {
        "X-RapidAPI-Key": API_KEY,
        "Content-Type": "application/json"
    }
 
    payload = {
        "invoice": invoice,
        "vendor_history": vendor_history[-50:],  # last 50 invoices from vendor
        "anomaly_checks": [
            "duplicate_detection",
            "round_number_bias",
            "amount_splitting",  # invoice splitting to avoid approval thresholds
            "unusual_frequency",
            "weekend_submission",
            "bank_detail_change",
            "synthetic_invoice_detection"
        ]
    }
 
    response = requests.post(
        f"{BASE_URL}/audit",
        json=payload,
        headers=headers,
        timeout=20
    )
    response.raise_for_status()
    return response.json()
 
# Example: check for amount splitting
# Fraudsters sometimes split one large invoice into several below approval threshold
audit_result = audit_invoice_for_anomalies(invoice, vendor_history=[])
 
if audit_result["anomaly_score"] > 0.7:
    print(f"⚠️ ANOMALY DETECTED (score: {audit_result['anomaly_score']:.2f})")
    for flag in audit_result["flags"]:
        print(f"   - {flag['type']}: {flag['description']}")

from dataclasses import dataclass
from enum import Enum
 
class ApprovalAction(Enum):
    AUTO_APPROVE = "auto_approve"
    ROUTE_TO_MANAGER = "route_to_manager"
    ROUTE_TO_DIRECTOR = "route_to_director"
    ROUTE_TO_CFO = "route_to_cfo"
    BLOCK = "block"
 
@dataclass
class ApprovalDecision:
    action: ApprovalAction
    approver: str
    reason: str
    invoice_id: str
 
APPROVAL_THRESHOLDS = {
    ApprovalAction.AUTO_APPROVE: 2500,
    ApprovalAction.ROUTE_TO_MANAGER: 10000,
    ApprovalAction.ROUTE_TO_DIRECTOR: 50000,
    ApprovalAction.ROUTE_TO_CFO: float("inf")
}
 
def route_for_approval(
    invoice: dict,
    match_result: dict,
    audit_result: dict
) -> ApprovalDecision:
    """
    Determine approval routing based on amount, match status, and risk score.
    """
    amount = invoice["total_amount"]
    anomaly_score = audit_result.get("anomaly_score", 0)
 
    # Block conditions — require investigation before processing
    if match_result.get("block"):
        return ApprovalDecision(
            action=ApprovalAction.BLOCK,
            approver="compliance-team",
            reason=f"Blocked: {match_result.get('reason', 'PO mismatch')}",
            invoice_id=invoice["invoice_number"]
        )
 
    if anomaly_score > 0.85:
        return ApprovalDecision(
            action=ApprovalAction.BLOCK,
            approver="ap-fraud-team",
            reason=f"High fraud risk score: {anomaly_score:.2f}",
            invoice_id=invoice["invoice_number"]
        )
 
    # Escalate based on anomaly score
    if anomaly_score > 0.5:
        # Add one approval tier for suspicious invoices
        for action, threshold in APPROVAL_THRESHOLDS.items():
            if amount <= threshold / 5:
                return ApprovalDecision(
                    action=action,
                    approver=action.value.replace("route_to_", ""),
                    reason=f"Elevated risk — anomaly score {anomaly_score:.2f}",
                    invoice_id=invoice["invoice_number"]
                )
 
    # Standard routing by amount
    for action, threshold in sorted(APPROVAL_THRESHOLDS.items(), key=lambda x: x[1]):
        if amount <= threshold:
            return ApprovalDecision(
                action=action,
                approver="auto" if action == ApprovalAction.AUTO_APPROVE else action.value.replace("route_to_", ""),
                reason="Standard approval routing",
                invoice_id=invoice["invoice_number"]
            )

def process_invoice_file(file_path: str, po_db: dict) -> dict:
    """
    Run the complete AP processing pipeline for a single invoice.
    """
    print(f"\nProcessing: {file_path}")
 
    # 1. Extract data
    extraction = extract_invoice_data(file_path)
    invoice = extraction["extracted_data"]
 
    # 2. PO match
    match_result = match_invoice_to_po(invoice, po_db)
 
    # 3. Fraud audit
    audit_result = audit_invoice_for_anomalies(invoice, vendor_history=[])
 
    # 4. Route
    decision = route_for_approval(invoice, match_result, audit_result)
 
    result = {
        "invoice_number": invoice["invoice_number"],
        "vendor": invoice["vendor_name"],
        "amount": invoice["total_amount"],
        "match_status": match_result["match_status"],
        "anomaly_score": audit_result.get("anomaly_score", 0),
        "action": decision.action.value,
        "approver": decision.approver,
        "processed_at": datetime.utcnow().isoformat()
    }
 
    print(f"  Invoice: {result['invoice_number']} | ${result['amount']:.2f}")
    print(f"  Match: {result['match_status']} | Risk: {result['anomaly_score']:.2f}")
    print(f"  Decision: {result['action'].upper()} → {result['approver']}")
 
    return result