How to Automatically Parse Invoices into QuickBooks Online

import os, time, requests

QBO_TOKEN = os.environ["QBO_ACCESS_TOKEN"]
QBO_COMPANY_ID = os.environ["QBO_COMPANY_ID"]
QBO_BASE = f"https://quickbooks.api.intuit.com/v3/company/{QBO_COMPANY_ID}"
QBO_HEADERS = {
    "Authorization": f"Bearer {QBO_TOKEN}",
    "Accept": "application/json",
    "Content-Type": "application/json",
}


API = "https://docuparseapi.com/api/v1"
HEADERS = {"Authorization": f"Bearer {os.environ['DOCUPARSE_API_KEY']}"}

def parse_document(path: str) -> dict:
    # 1. Submit
    with open(path, "rb") as f:
        submit = requests.post(f"{API}/extract", headers=HEADERS,
                               files={"file": (os.path.basename(path), f)}, timeout=30).json()
    if not submit.get("success"):
        raise RuntimeError(f"[{submit['error']['code']}] {submit['error']['message']}")

    # 2. Fetch the result
    for _ in range(40):
        doc = requests.get(f"{API}/documents/{submit['document_id']}",
                           headers=HEADERS, timeout=30).json()
        if doc["status"] == "completed":
            return doc["data"]
        if doc["status"] == "failed":
            raise RuntimeError(f"Extraction failed: {submit['document_id']}")
        time.sleep(1.5)
    raise TimeoutError(submit["document_id"])


def find_or_create_vendor(vendor_name: str) -> str:
    # Search
    safe = vendor_name.replace("'", "\\'")
    query = f"SELECT * FROM Vendor WHERE DisplayName LIKE '%{safe}%'"
    response = requests.get(
        f"{QBO_BASE}/query",
        headers=QBO_HEADERS,
        params={"query": query, "minorversion": "65"},
    )
    vendors = response.json().get("QueryResponse", {}).get("Vendor", [])
    if vendors:
        return vendors[0]["Id"]

    # Create
    response = requests.post(
        f"{QBO_BASE}/vendor",
        headers=QBO_HEADERS,
        params={"minorversion": "65"},
        json={"DisplayName": vendor_name, "PrintOnCheckName": vendor_name},
    )
    return response.json()["Vendor"]["Id"]


def create_qbo_bill(invoice_data: dict, vendor_id: str) -> dict:
    lines = [
        {
            "Amount": float(item["line_total"] or (float(item["price"] or 0) * (item["quantity"] or 1))),
            "DetailType": "AccountBasedExpenseLineDetail",
            "Description": item.get("description", ""),
            "AccountBasedExpenseLineDetail": {
                "AccountRef": {"value": os.environ.get("QBO_EXPENSE_ACCOUNT_ID", "7")}
            },
        }
        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": os.environ.get("QBO_EXPENSE_ACCOUNT_ID", "7")}
        },
    }]

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

    response = requests.post(
        f"{QBO_BASE}/bill",
        headers=QBO_HEADERS,
        params={"minorversion": "65"},
        json=bill,
    )
    response.raise_for_status()
    return response.json()["Bill"]


def process_invoice(file_path: str) -> dict:
    extracted = parse_document(file_path)
    merchant = extracted.get("merchant") or "Unknown Vendor"
    vendor_id = find_or_create_vendor(merchant)
    bill = create_qbo_bill(extracted, vendor_id)

    return {
        "bill_id": bill["Id"],
        "vendor": merchant,
        "total": extracted.get("total"),
        "currency": extracted.get("currency"),
    }

// List all expense accounts
const query = "SELECT * FROM Account WHERE AccountType = 'Expense'";
const response = await fetch(
  `${QBO_BASE}/query?query=${encodeURIComponent(query)}&minorversion=65`,
  { headers: { Authorization: `Bearer ${QBO_ACCESS_TOKEN}`, Accept: "application/json" } }
);
const data = await response.json();
data.QueryResponse.Account.forEach(acc => {
  console.log(`${acc.Id}: ${acc.Name}`);
});

async function processSafely(filePath) {
  try {
    return await processPDFInvoice(filePath);
  } catch (error) {
    if (error.message.includes("LIMIT_EXCEEDED")) {
      return { status: "error", reason: "DocuParse API monthly limit reached" };
    }
    if (error.message.includes("EXTRACTION_FAILED")) {
      return { status: "review", reason: "Could not extract — needs manual review" };
    }
    if (error.message.includes("QBO bill creation failed")) {
      return { status: "error", reason: "QuickBooks API error", detail: error.message };
    }
    return { status: "error", reason: error.message };
  }
}

import json

def attach_pdf_to_bill(bill_id, pdf_path):
    meta = {
        "AttachableRef": [{"EntityRef": {"type": "Bill", "value": bill_id}}],
        "FileName": os.path.basename(pdf_path),
        "ContentType": "application/pdf",
    }
    with open(pdf_path, "rb") as f:
        r = requests.post(
            f"{QBO_BASE}/upload",
            headers={"Authorization": QBO_HEADERS["Authorization"], "Accept": "application/json"},
            files={
                "file_metadata_0": (None, json.dumps(meta), "application/json"),
                "file_content_0": (os.path.basename(pdf_path), f, "application/pdf"),
            },
        )
    r.raise_for_status()
    return r.json()