How to Automate Purchase Order to Invoice Matching with FinAudit AI API

pip install requests python-dotenv

import os
import json
import requests
from pathlib import Path
from dotenv import load_dotenv
 
load_dotenv()
 
FINAUDIT_KEY = os.getenv("YOUR_API_KEY")
FINAUDIT_BASE = "https://apivult.com/api/finaudit"
 
HEADERS = {
    "X-RapidAPI-Key": FINAUDIT_KEY,
    "Content-Type": "application/json"
}

def extract_document(document_path: str, doc_type: str) -> dict:
    """
    Extract structured data from a financial document.
    doc_type: 'purchase_order' | 'goods_receipt' | 'invoice'
    """
    with open(document_path, "rb") as f:
        file_content = f.read()
 
    import base64
    encoded = base64.b64encode(file_content).decode("utf-8")
 
    payload = {
        "document": {
            "content": encoded,
            "filename": Path(document_path).name,
            "document_type": doc_type
        },
        "extraction": {
            "fields": [
                "document_number",
                "vendor_id",
                "vendor_name",
                "document_date",
                "line_items",
                "subtotal",
                "tax_amount",
                "total_amount",
                "currency"
            ],
            "normalize_amounts": True,
            "normalize_dates": True
        }
    }
 
    resp = requests.post(
        f"{FINAUDIT_BASE}/extract",
        json=payload,
        headers=HEADERS
    )
    resp.raise_for_status()
    result = resp.json()
 
    if not result["success"]:
        raise ValueError(f"Extraction failed for {document_path}: {result['error']}")
 
    return result["data"]

TOLERANCE_PCT = 0.02  # 2% tolerance for minor rounding differences
 
 
def match_line_items(
    po_items: list[dict],
    grn_items: list[dict],
    inv_items: list[dict]
) -> list[dict]:
    """Match line items across PO, GRN, and Invoice."""
    results = []
 
    # Build lookup dicts keyed by item/part number
    po_lookup = {item["item_number"]: item for item in po_items}
    grn_lookup = {item["item_number"]: item for item in grn_items}
 
    for inv_item in inv_items:
        item_num = inv_item["item_number"]
        match_result = {
            "item_number": item_num,
            "description": inv_item.get("description", ""),
            "invoice_qty": inv_item["quantity"],
            "invoice_unit_price": inv_item["unit_price"],
            "invoice_total": inv_item["line_total"],
            "status": "MATCHED",
            "exceptions": []
        }
 
        po_item = po_lookup.get(item_num)
        grn_item = grn_lookup.get(item_num)
 
        # Check 1: Item exists in PO
        if not po_item:
            match_result["status"] = "EXCEPTION"
            match_result["exceptions"].append({
                "code": "NO_PO_LINE",
                "message": f"Item {item_num} not found in purchase order"
            })
        else:
            match_result["po_qty"] = po_item["quantity"]
            match_result["po_unit_price"] = po_item["unit_price"]
 
            # Check 2: Price variance
            price_var = abs(
                inv_item["unit_price"] - po_item["unit_price"]
            ) / po_item["unit_price"]
 
            if price_var > TOLERANCE_PCT:
                match_result["status"] = "EXCEPTION"
                match_result["exceptions"].append({
                    "code": "PRICE_VARIANCE",
                    "message": (
                        f"Unit price variance {price_var:.1%}: "
                        f"PO=${po_item['unit_price']:.4f}, "
                        f"Invoice=${inv_item['unit_price']:.4f}"
                    ),
                    "variance_pct": round(price_var * 100, 2)
                })
 
            # Check 3: Quantity overbilling vs PO
            if inv_item["quantity"] > po_item["quantity"]:
                match_result["status"] = "EXCEPTION"
                match_result["exceptions"].append({
                    "code": "QTY_OVERBILL_VS_PO",
                    "message": (
                        f"Invoice qty {inv_item['quantity']} exceeds "
                        f"PO qty {po_item['quantity']}"
                    )
                })
 
        # Check 4: Quantity vs GRN (received goods)
        if not grn_item:
            match_result["exceptions"].append({
                "code": "NO_GRN_LINE",
                "message": f"Item {item_num} has no goods receipt record"
            })
            if match_result["status"] == "MATCHED":
                match_result["status"] = "REVIEW"
        else:
            match_result["grn_qty"] = grn_item["quantity"]
            if inv_item["quantity"] > grn_item["quantity"]:
                match_result["status"] = "EXCEPTION"
                match_result["exceptions"].append({
                    "code": "QTY_OVERBILL_VS_GRN",
                    "message": (
                        f"Invoice qty {inv_item['quantity']} exceeds "
                        f"received qty {grn_item['quantity']}"
                    )
                })
 
        results.append(match_result)
 
    return results
 
 
def run_three_way_match(
    po_data: dict,
    grn_data: dict,
    inv_data: dict
) -> dict:
    """Run full three-way matching and generate exception summary."""
    line_results = match_line_items(
        po_data["line_items"],
        grn_data["line_items"],
        inv_data["line_items"]
    )
 
    matched = [r for r in line_results if r["status"] == "MATCHED"]
    exceptions = [r for r in line_results if r["status"] == "EXCEPTION"]
    reviews = [r for r in line_results if r["status"] == "REVIEW"]
 
    # Calculate total overbilling risk
    overbill_risk = sum(
        r["invoice_total"]
        for r in exceptions
        if any(e["code"].startswith("QTY_OVERBILL") or e["code"] == "PRICE_VARIANCE"
               for e in r["exceptions"])
    )
 
    # Determine approval recommendation
    if exceptions:
        recommendation = "HOLD_FOR_REVIEW"
        reason = f"{len(exceptions)} exception(s) require resolution"
    elif reviews:
        recommendation = "CONDITIONAL_APPROVAL"
        reason = f"{len(reviews)} line(s) lack GRN confirmation"
    else:
        recommendation = "APPROVED_FOR_PAYMENT"
        reason = "All lines matched within tolerance"
 
    return {
        "invoice_number": inv_data["document_number"],
        "vendor": inv_data["vendor_name"],
        "invoice_total": inv_data["total_amount"],
        "currency": inv_data["currency"],
        "line_count": len(line_results),
        "matched_lines": len(matched),
        "exception_lines": len(exceptions),
        "review_lines": len(reviews),
        "match_rate": len(matched) / len(line_results) if line_results else 0,
        "overbilling_risk_amount": round(overbill_risk, 2),
        "recommendation": recommendation,
        "recommendation_reason": reason,
        "line_results": line_results
    }

def log_match_result(match_result: dict) -> str:
    """Submit match result to FinAudit for audit trail logging."""
    payload = {
        "audit_event": {
            "event_type": "PO_INVOICE_MATCH",
            "document_reference": match_result["invoice_number"],
            "vendor": match_result["vendor"],
            "outcome": match_result["recommendation"],
            "exception_count": match_result["exception_lines"],
            "financial_risk_amount": match_result["overbilling_risk_amount"],
            "currency": match_result["currency"],
            "details": match_result
        }
    }
 
    resp = requests.post(
        f"{FINAUDIT_BASE}/audit-log",
        json=payload,
        headers=HEADERS
    )
    resp.raise_for_status()
    return resp.json()["data"]["audit_id"]

def process_invoice_batch(document_sets: list[dict]) -> list[dict]:
    """
    Process a batch of invoice document sets.
    Each set: { "po": path, "grn": path, "invoice": path }
    """
    results = []
 
    for doc_set in document_sets:
        print(f"Processing invoice: {doc_set['invoice']}")
 
        po_data = extract_document(doc_set["po"], "purchase_order")
        grn_data = extract_document(doc_set["grn"], "goods_receipt")
        inv_data = extract_document(doc_set["invoice"], "invoice")
 
        match_result = run_three_way_match(po_data, grn_data, inv_data)
        audit_id = log_match_result(match_result)
 
        match_result["audit_id"] = audit_id
        results.append(match_result)
 
        status_icon = "✅" if match_result["recommendation"] == "APPROVED_FOR_PAYMENT" else "⚠"
        print(
            f"  {status_icon} {match_result['recommendation']} — "
            f"{match_result['invoice_number']} "
            f"(${match_result['invoice_total']:,.2f} {match_result['currency']})"
        )
 
    return results
 
 
if __name__ == "__main__":
    batch = [
        {
            "po": "documents/PO-2026-00142.pdf",
            "grn": "documents/GRN-00142.pdf",
            "invoice": "documents/INV-ACME-20260405.pdf"
        },
        {
            "po": "documents/PO-2026-00143.pdf",
            "grn": "documents/GRN-00143.pdf",
            "invoice": "documents/INV-GLOBEX-20260404.pdf"
        }
    ]
 
    results = process_invoice_batch(batch)
 
    approved = sum(1 for r in results if r["recommendation"] == "APPROVED_FOR_PAYMENT")
    held = len(results) - approved
    total_risk = sum(r["overbilling_risk_amount"] for r in results)
 
    print(f"\nBatch Summary: {approved} approved, {held} held")
    print(f"Total overbilling risk identified: ${total_risk:,.2f}")