Bookkeeping Automation with a Receipt and Invoice API

import os
import requests

def extract_document(file_path: str) -> dict:
    """
    Send an invoice or receipt to DocuParseAPI.
    Returns structured JSON with all financial fields.
    """
    with open(file_path, "rb") as f:
        response = requests.post(
            "https://docuparseapi.com/api/v1/extract",
            headers={"Authorization": f"Bearer {os.environ['DOCUPARSE_API_KEY']}"},
            files={"file": (os.path.basename(file_path), f)},
            timeout=30,
        )
    
    response.raise_for_status()
    data = response.json()
    
    if not data.get("success"):
        code = data.get("error", {}).get("code", "UNKNOWN")
        raise RuntimeError(f"Extraction failed [{code}]")
    
    return data

{
  "document_type": "invoice",
  "merchant": "TechCloud Solutions",
  "invoice_id": "TC-2026-0183",
  "date": "2026-05-01",
  "due_date": "2026-05-31",
  "currency": "USD",
  "subtotal": "4800.00",
  "tax": "480.00",
  "total": "5280.00",
  "line_items": [
    {
      "description": "Cloud Infrastructure — May",
      "quantity": 1,
      "unit_price": "4800.00",
      "total": "4800.00"
    }
  ]
}

import os
import requests

DOCUPARSE_KEY = os.environ["DOCUPARSE_API_KEY"]
XERO_TOKEN = os.environ["XERO_ACCESS_TOKEN"]
XERO_TENANT_ID = os.environ["XERO_TENANT_ID"]


def extract_invoice(file_path: str) -> dict:
    with open(file_path, "rb") as f:
        response = requests.post(
            "https://docuparseapi.com/api/v1/extract",
            headers={"Authorization": f"Bearer {DOCUPARSE_KEY}"},
            files={"file": f},
            timeout=30,
        )
    data = response.json()
    if not data["success"]:
        raise RuntimeError(f"Extraction failed: {data['error']['code']}")
    return data


def find_or_create_xero_contact(merchant_name: str) -> str:
    """Look up a Xero contact by name, or create one if it doesn't exist."""
    # Search for existing contact
    search_response = requests.get(
        f"https://api.xero.com/api.xro/2.0/Contacts",
        headers={
            "Authorization": f"Bearer {XERO_TOKEN}",
            "Xero-tenant-id": XERO_TENANT_ID,
        },
        params={"searchTerm": merchant_name},
    )
    search_data = search_response.json()
    
    contacts = search_data.get("Contacts", [])
    if contacts:
        return contacts[0]["ContactID"]
    
    # Create new contact if not found
    create_response = requests.post(
        "https://api.xero.com/api.xro/2.0/Contacts",
        headers={
            "Authorization": f"Bearer {XERO_TOKEN}",
            "Xero-tenant-id": XERO_TENANT_ID,
            "Content-Type": "application/json",
        },
        json={"Contacts": [{"Name": merchant_name}]},
    )
    created = create_response.json()
    return created["Contacts"][0]["ContactID"]


def create_xero_bill(invoice_data: dict, contact_id: str) -> dict:
    """Create a supplier bill in Xero from extracted invoice data."""
    
    # Build line items — fall back to single line if none extracted
    if invoice_data.get("line_items"):
        xero_line_items = [
            {
                "Description": item.get("description", ""),
                "Quantity": item.get("quantity", 1),
                "UnitAmount": float(
                    item.get("unit_price") or item.get("amount") or 0
                ),
                "AccountCode": "429",  # adjust to your chart of accounts
            }
            for item in invoice_data["line_items"]
        ]
    else:
        xero_line_items = [{
            "Description": f"Invoice {invoice_data.get('invoice_id', '')}",
            "Quantity": 1,
            "UnitAmount": float(
                invoice_data.get("subtotal") or invoice_data.get("total") or 0
            ),
            "AccountCode": "429",
        }]
    
    bill = {
        "Type": "ACCPAY",
        "Contact": {"ContactID": contact_id},
        "Date": invoice_data.get("date"),
        "DueDate": invoice_data.get("due_date"),
        "InvoiceNumber": invoice_data.get("invoice_id"),
        "CurrencyCode": invoice_data.get("currency", "USD"),
        "Status": "DRAFT",  # Use AUTHORISED to skip the approval step
        "LineItems": xero_line_items,
    }
    
    response = requests.post(
        "https://api.xero.com/api.xro/2.0/Invoices",
        headers={
            "Authorization": f"Bearer {XERO_TOKEN}",
            "Xero-tenant-id": XERO_TENANT_ID,
            "Content-Type": "application/json",
        },
        json={"Invoices": [bill]},
    )
    response.raise_for_status()
    return response.json()["Invoices"][0]


def process_invoice(file_path: str) -> dict:
    """Full workflow: extract → find/create contact → create bill."""
    
    # 1. Extract from PDF
    extracted = extract_invoice(file_path)
    merchant = extracted.get("merchant") or "Unknown Vendor"
    
    print(f"Extracted: {merchant} — {extracted.get('currency')} {extracted.get('total')}")
    
    # 2. Find or create Xero contact
    contact_id = find_or_create_xero_contact(merchant)
    
    # 3. Create draft bill
    bill = create_xero_bill(extracted, contact_id)
    
    print(f"Bill created in Xero: {bill['InvoiceID']} (status: {bill['Status']})")
    
    return {
        "merchant": merchant,
        "total": extracted.get("total"),
        "currency": extracted.get("currency"),
        "xero_invoice_id": bill["InvoiceID"],
        "xero_status": bill["Status"],
    }


# Process a single invoice
result = process_invoice("supplier_invoice.pdf")

# Process a folder of invoices
from pathlib import Path

invoice_folder = Path("./client_invoices/")
for pdf_file in invoice_folder.glob("*.pdf"):
    try:
        result = process_invoice(str(pdf_file))
        print(f"✓ {result['merchant']}: {result['currency']} {result['total']}")
    except RuntimeError as e:
        print(f"✗ {pdf_file.name}: {e}")

import os
import requests

QBO_COMPANY_ID = os.environ["QBO_COMPANY_ID"]
QBO_TOKEN = os.environ["QBO_ACCESS_TOKEN"]
QBO_EXPENSE_ACCOUNT = os.environ.get("QBO_EXPENSE_ACCOUNT_ID", "7")


def create_qbo_bill(invoice_data: dict, vendor_id: str) -> dict:
    """Create a Bill in QuickBooks Online."""
    
    lines = [
        {
            "Amount": float(item.get("total") or item.get("amount") or 0),
            "DetailType": "AccountBasedExpenseLineDetail",
            "Description": item.get("description", ""),
            "AccountBasedExpenseLineDetail": {
                "AccountRef": {"value": QBO_EXPENSE_ACCOUNT}
            },
        }
        for item in invoice_data.get("line_items", [])
    ] or [{
        "Amount": float(invoice_data.get("subtotal") or invoice_data.get("total") or 0),
        "DetailType": "AccountBasedExpenseLineDetail",
        "Description": f"Invoice {invoice_data.get('invoice_id', '')}",
        "AccountBasedExpenseLineDetail": {
            "AccountRef": {"value": QBO_EXPENSE_ACCOUNT}
        },
    }]

    bill = {
        "VendorRef": {"value": vendor_id},
        "TxnDate": invoice_data.get("date"),
        "DueDate": invoice_data.get("due_date"),
        "DocNumber": invoice_data.get("invoice_id"),
        "TotalAmt": float(invoice_data.get("total") or 0),
        "Line": lines,
    }

    response = requests.post(
        f"https://quickbooks.api.intuit.com/v3/company/{QBO_COMPANY_ID}/bill",
        headers={
            "Authorization": f"Bearer {QBO_TOKEN}",
            "Content-Type": "application/json",
            "Accept": "application/json",
        },
        json={"Bill": bill},
    )
    response.raise_for_status()
    return response.json()["Bill"]

def process_with_review(file_path: str, approval_threshold: float = 1000.0) -> dict:
    """Process invoice but flag high-value ones for approval."""
    extracted = extract_invoice(file_path)
    
    total = float(extracted.get("total") or 0)
    
    if total >= approval_threshold:
        # Don't auto-create — put in review queue
        save_to_review_queue(extracted, reason="ABOVE_THRESHOLD")
        return {"status": "pending_review", "total": total}
    
    if not extracted.get("merchant"):
        save_to_review_queue(extracted, reason="UNKNOWN_VENDOR")
        return {"status": "pending_review", "total": total}
    
    # Safe to auto-process
    contact_id = find_or_create_xero_contact(extracted["merchant"])
    bill = create_xero_bill(extracted, contact_id)
    return {"status": "processed", "xero_id": bill["InvoiceID"]}