quickbooks-online-api

QuickBooks Online API 专家指南

概述

QuickBooks Online API 提供对 QuickBooks Online 公司会计数据和操作的全面访问能力。此技能可帮助你构建集成，以处理发票、付款、客户管理、库存跟踪和财务报告。该 API 使用 OAuth 2.0 进行身份验证，并支持客户、发票、付款、商品、账户等所有主要会计实体的操作。

QuickBooks Online API 基于 REST，返回 JSON 或 XML 响应，并提供 Java、Python、PHP、Node.js 和 C# SDK。它同时支持沙盒（开发）和生产环境。

何时使用此技能

在以下场景中使用此技能：

• 构建 QuickBooks 集成，实现会计自动化
• 实现发票工作流或付款处理
• 创建客户或供应商管理功能
• 使用 QuickBooks Online API 身份验证（OAuth2）
• 排查 API 错误或验证失败问题
• 实现批量操作，以批量更新数据
• 设置变更数据捕获（CDC）或 Webhook，以同步数据
• 设计多币种或国际会计集成
• 基于 QuickBooks 数据构建报告或分析功能
• 向 QuickBooks Online 迁移数据或从中迁出数据

会计集成设计

对于会计同步或规范审查，应先梳理现有的单据和付款生命周期，再选择 API 调用。以下要求适用于下方的简化 API 示例；这些示例并非完整的对账实现。

• **明确归属关系：**识别本地发票/账单、其业务所有者，以及有权发布该单据或更改结算状态的系统。其他提供方可能已经将同一笔负债发布到 QuickBooks。创建新记录前应匹配现有记录；不要假设每张外部账单都与一张本地账单一一对应。
• **区分应收和应付：**客户收款使用 Payment；供应商结算使用 BillPayment。保留付款与单据之间的分配关系、部分结算情况，以及尚未分配的客户款项。根据本地付款账本计算余额；反复设置已付款标记或累加已付金额并不等于对账。
• **定义过账边界：**区分可编辑的本地草稿与已开具发票或已批准的应付款项。明确选择导出事件。会计导出不得悄悄再次向客户发送邮件或发起银行付款。
• **保留金额含义：**导出已保存的单据明细和准确金额。定义产品、收入/费用映射、税费、货币、舍入、手续费、贷项，以及按总额或净额结算。即使不涵盖完整的会计科目表同步，也仍需定义过账交易所需的映射。
• **保留交易标识：**按连接、环境、realm、实体类型以及本地/远程 ID 对映射进行限定。映射应涵盖发票、账单、付款及分配关系标识，而不只是客户/供应商。即使本地单据作废，也应保留映射。重新开具的单据属于新单据；重新连接到另一家公司时，不能复用旧映射。
• **确保交付可恢复：**将导出任务与业务事务一同持久化，并通过现有队列/outbox 处理。成功后保存远程 ID，并不能应对创建请求响应丢失的情况。使用受支持且稳定的请求 ID，并对结果不明确的情况进行对账。更新时使用 SyncToken，并在重试存在冲突的会计编辑前，应用明确的字段归属策略。
• **处理更正：**定义作废、贷项、退款、已删除/重新分配的付款，以及重放行为。不支持的情况应作为可见的对账异常处理；不要悄悄将单据标记为已付款。只有在数据已可靠处理后才推进轮询检查点；收到 Webhook 不得导致跳过轮询时间窗口。
• **保护连接：**复用加密凭据存储、租户约束和 RLS、绑定用户/组织且会过期的 OAuth 状态，以及串行执行的刷新令牌轮换。复用现有的代码库模式，而不是另建一套集成框架。

对于 Laneweaver TMS 相关工作，在提出架构、同步触发器、付款导入或 QuickBooks issue #146 的变更之前，请先阅读 Laneweaver 应收与应付设计。其中针对该代码库的规则不会自动适用于其他产品。

身份验证与 OAuth2 设置

OAuth 2.0 流程

QuickBooks Online API 需要使用 OAuth 2.0 进行身份验证。流程如下：

1. 在 developer.intuit.com 注册应用，获取 Client ID 和 Client Secret
2. 将用户引导至授权 URL，让他们授予访问其 QuickBooks 公司账户的权限
3. 使用授权码换取令牌（访问令牌 + 刷新令牌）
4. 在 API 请求中使用访问令牌（Authorization: Bearer 请求头）
5. 在令牌过期前刷新令牌，以保持访问权限

令牌生命周期

访问令牌：

• 有效期为 3600 秒（1 小时）
• 在 Authorization 标头中包含：Authorization: Bearer {access_token}
• 过期后返回 401 Unauthorized

刷新令牌：

• 自签发之日起有效期为 100 天
• 用于获取新的访问令牌 + 刷新令牌对
• 签发新令牌 24 小时后，之前的刷新令牌将过期
• 始终使用最新的刷新令牌

令牌刷新模式

Node.js 示例：

const oauthClient = require('intuit-oauth');

// Refresh access token
oauthClient.refresh()
  .then(function(authResponse) {
    const newAccessToken = authResponse.token.access_token;
    const newRefreshToken = authResponse.token.refresh_token;
    const expiresIn = authResponse.token.expires_in; // 3600 seconds

    // Store new tokens securely (database, encrypted storage)
    console.log('Tokens refreshed successfully');
  })
  .catch(function(e) {
    console.error('Token refresh failed:', e.originalMessage);
    // Handle re-authentication if refresh token is invalid
  });

Python 示例：

from intuitlib.client import AuthClient

auth_client = AuthClient(
    client_id='YOUR_CLIENT_ID',
    client_secret='YOUR_CLIENT_SECRET',
    redirect_uri='YOUR_REDIRECT_URI',
    environment='sandbox'  # or 'production'
)

# Refresh tokens
auth_client.refresh(refresh_token='STORED_REFRESH_TOKEN')

# Get new tokens
new_access_token = auth_client.access_token
new_refresh_token = auth_client.refresh_token

最佳实践

• 主动刷新：在令牌过期前刷新（例如，50 分钟后）
• 安全存储：在数据库中加密令牌，切勿提交到版本控制系统
• 处理 401 响应：遇到身份验证错误时自动尝试刷新令牌
• Realm ID（公司 ID）：存储 OAuth 返回的 realmId，所有 API 调用都需要它
• 作用域：只请求必要的作用域（会计、付款等）

核心实体参考

客户

表示 QuickBooks 中的客户和子客户（工作）。

关键字段：

• Id（字符串，只读）：唯一标识符
• DisplayName（字符串，必填）：客户显示名称（必须唯一）
• GivenName、FamilyName（字符串）：名字和姓氏
• CompanyName（字符串）：企业客户的公司名称
• PrimaryEmailAddr（对象）：电子邮件地址 { "Address": "[email protected]" }
• PrimaryPhone（对象）：电话号码 { "FreeFormNumber": "(555) 123-4567" }
• BillAddr、ShipAddr（对象）：账单地址和送货地址
• Balance（小数，只读）：当前未结余额
• Active（布尔值）：客户是否处于启用状态
• SyncToken（字符串，更新时必填）：用于乐观锁定的版本号

引用类型：在交易中使用 CustomerRef：{ "value": "123", "name": "Customer Name" }

发票

表示发送给客户的销售发票。

关键字段：

• Id（字符串，只读）：唯一标识符
• DocNumber（字符串）：发票编号（未提供时自动生成）
• TxnDate（日期）：交易日期（YYYY-MM-DD 格式）
• DueDate（日期）：付款到期日
• CustomerRef（对象，必填）：客户引用 { "value": "customerId" }
• Line（数组，必填）：发票明细项（参见“明细项”部分）
• TotalAmt（小数，只读）：计算得出的总金额
• Balance（小数，只读）：剩余未付余额
• EmailStatus（枚举）：NotSet、NeedToSend、EmailSent
• BillEmail（对象）：用于发送发票的客户电子邮件地址
• TxnTaxDetail（对象）：税费计算详情
• LinkedTxn（数组）：关联的交易（付款、贷项通知单）
• SyncToken（字符串，更新时必填）：版本号

明细项：

{
  "Line": [
    {
      "Amount": 100.00,
      "DetailType": "SalesItemLineDetail",
      "SalesItemLineDetail": {
        "ItemRef": { "value": "1", "name": "Services" },
        "Qty": 1,
        "UnitPrice": 100.00,
        "TaxCodeRef": { "value": "TAX" }
      }
    },
    {
      "Amount": 100.00,
      "DetailType": "SubTotalLineDetail",
      "SubTotalLineDetail": {}
    }
  ]
}

付款

表示客户针对发票支付的款项。

关键字段：

• Id（字符串，只读）：唯一标识符
• TotalAmt（小数，必填）：付款总额
• CustomerRef（对象，必填）：客户引用
• PaymentMethodRef（对象）：付款方式（现金、支票、信用卡等）
• PaymentRefNum（字符串）：参考编号（支票号码、交易 ID）
• TxnDate（日期）：付款日期
• DepositToAccountRef（对象）：用于存款的银行账户
• Line（数组）：应用于发票/贷项通知单的付款
• UnappliedAmt（小数，只读）：尚未应用于发票的金额
• SyncToken（字符串，更新时必填）：版本号

付款明细项（将付款应用于发票）：

{
  "Line": [
    {
      "Amount": 100.00,
      "LinkedTxn": [
        {
          "TxnId": "123",
          "TxnType": "Invoice"
        }
      ]
    }
  ]
}

项目

表示销售的产品或服务。

类型：

• Service：服务（咨询、劳务等）
• Inventory：纳入库存跟踪的实体产品
• NonInventory：未纳入库存跟踪的实体产品
• Category：其他项目的分组

关键字段：

• Id（string，只读）：唯一标识符
• Name（string，必填）：项目名称（必须唯一）
• Type（enum，必填）：Service、Inventory、NonInventory、Category
• Description（string）：项目描述
• UnitPrice（decimal）：销售价格
• PurchaseCost（decimal）：采购／成本价格
• IncomeAccountRef（object，必填）：收入账户引用
• ExpenseAccountRef（object）：采购支出账户
• TrackQtyOnHand（boolean）：是否跟踪库存数量
• QtyOnHand（decimal）：当前库存数量
• Active（boolean）：项目是否处于启用状态

账户

表示会计科目表中的账户。

关键字段：

• Id（string，只读）：唯一标识符
• Name（string，必填）：账户名称
• AccountType（enum，必填）：Bank、Accounts Receivable、Accounts Payable、Income、Expense 等
• AccountSubType（enum）：更具体的类型（CashOnHand、Checking、Savings 等）
• CurrentBalance（decimal，只读）：当前账户余额
• Active（boolean）：账户是否处于启用状态
• Classification（enum）：资产、负债、权益、收入、支出

常见账户类型：

• Bank：银行和现金账户
• Accounts Receivable：客户欠款
• Accounts Payable：供应商欠款
• Income：收入账户
• Expense：支出账户
• Other Current Asset：短期资产
• Fixed Asset：长期资产

CRUD 操作模式

创建操作

最低必需字段：每种实体都有特定的必填字段（通常是名称／引用和金额）。

端点模式：POST /v3/company/{realmId}/{entityName}

请求标头：

Authorization: Bearer {access_token}
Accept: application/json
Content-Type: application/json

Python 示例——创建发票：

import requests

realm_id = "YOUR_REALM_ID"
access_token = "YOUR_ACCESS_TOKEN"

url = f"https://sandbox-quickbooks.api.intuit.com/v3/company/{realm_id}/invoice"

headers = {
    "Authorization": f"Bearer {access_token}",
    "Accept": "application/json",
    "Content-Type": "application/json"
}

invoice_data = {
    "Line": [
        {
            "Amount": 100.00,
            "DetailType": "SalesItemLineDetail",
            "SalesItemLineDetail": {
                "ItemRef": {"value": "1"}
            }
        }
    ],
    "CustomerRef": {"value": "1"}
}

response = requests.post(url, json=invoice_data, headers=headers)

if response.status_code == 200:
    invoice = response.json()['Invoice']
    print(f"Invoice created: {invoice['Id']}")
else:
    print(f"Error: {response.status_code} - {response.text}")

读取操作

单个实体：GET /v3/company/{realmId}/{entityName}/{entityId}

Node.js 示例——读取客户：

const axios = require('axios');

async function readCustomer(realmId, customerId, accessToken) {
  const url = `https://sandbox-quickbooks.api.intuit.com/v3/company/${realmId}/customer/${customerId}`;

  try {
    const response = await axios.get(url, {
      headers: {
        'Authorization': `Bearer ${accessToken}`,
        'Accept': 'application/json'
      }
    });

    return response.data.Customer;
  } catch (error) {
    if (error.response && error.response.status === 401) {
      // Token expired, refresh and retry
      console.error('Authentication failed - refresh token needed');
    } else {
      console.error('Read failed:', error.response?.data || error.message);
    }
    throw error;
  }
}

更新操作

更新分为两种：

1. 完整更新：必须包含所有可写字段。未提供的字段将设为 NULL。

2. 稀疏更新：仅更新指定字段。在请求正文中设置 "sparse": true。

重要：始终包含最新读取响应中的 SyncToken。这样可以避免并发修改冲突。

Python 示例——稀疏更新客户邮箱：

import requests

def sparse_update_customer(realm_id, customer_id, sync_token, new_email, access_token):
    url = f"https://sandbox-quickbooks.api.intuit.com/v3/company/{realm_id}/customer"

    headers = {
        "Authorization": f"Bearer {access_token}",
        "Accept": "application/json",
        "Content-Type": "application/json"
    }

    # Sparse update - only updating email
    customer_data = {
        "Id": customer_id,
        "SyncToken": sync_token,
        "sparse": True,
        "PrimaryEmailAddr": {
            "Address": new_email
        }
    }

    response = requests.post(url, json=customer_data, headers=headers)

    if response.status_code == 200:
        updated_customer = response.json()['Customer']
        print(f"Customer updated, new SyncToken: {updated_customer['SyncToken']}")
        return updated_customer
    else:
        print(f"Update failed: {response.text}")
        return None

SyncToken 处理：

# 1. Read entity to get latest SyncToken
customer = read_customer(realm_id, customer_id, access_token)

# 2. Update with current SyncToken
updated = sparse_update_customer(
    realm_id,
    customer_id,
    customer['SyncToken'],  # Use current sync token
    "[email protected]",
    access_token
)

# 3. Store new SyncToken for next update
new_sync_token = updated['SyncToken']

删除操作

大多数实体使用软删除（将 Active 设为 false）或作废操作。

软删除模式：

// Mark customer as inactive
const deleteCustomer = {
  Id: customerId,
  SyncToken: currentSyncToken,
  sparse: true,
  Active: false
};

// POST to update endpoint
axios.post(`${baseUrl}/customer`, deleteCustomer, { headers });

硬删除（适用于有限的实体）： POST /v3/company/{realmId}/{entityName}?operation=delete

{
  "Id": "123",
  "SyncToken": "2"
}

查询语言与筛选

QuickBooks 使用类似 SQL 的查询语法，但有所限制。

查询语法

基本模式：

SELECT * FROM {EntityName} WHERE {field} {operator} '{value}'

端点：GET /v3/company/{realmId}/query?query={sqlQuery}

运算符

• =：等于
• <、>、<=、>=：比较
• IN：匹配列表中的任意值
• LIKE：模式匹配（仅支持 % 通配符，不支持 _）

示例

按名称查询客户：

SELECT * FROM Customer WHERE DisplayName LIKE 'Acme%'

按日期范围查询发票：

SELECT * FROM Invoice WHERE TxnDate >= '2024-01-01' AND TxnDate <= '2024-12-31'

按顺序查询：

SELECT * FROM Customer WHERE Active = true ORDERBY DisplayName

分页：

SELECT * FROM Invoice STARTPOSITION 1 MAXRESULTS 100

Python 示例——使用筛选条件查询：

import requests
from urllib.parse import quote

def query_invoices_by_customer(realm_id, customer_id, access_token):
    query = f"SELECT * FROM Invoice WHERE CustomerRef = '{customer_id}' ORDERBY TxnDate DESC"
    encoded_query = quote(query)

    url = f"https://sandbox-quickbooks.api.intuit.com/v3/company/{realm_id}/query?query={encoded_query}"

    headers = {
        "Authorization": f"Bearer {access_token}",
        "Accept": "application/json"
    }

    response = requests.get(url, headers=headers)

    if response.status_code == 200:
        result = response.json()['QueryResponse']
        invoices = result.get('Invoice', [])
        print(f"Found {len(invoices)} invoices")
        return invoices
    else:
        print(f"Query failed: {response.text}")
        return []

查询限制

• 除 % 外不支持通配符：LIKE 仅支持 %（不支持 _）
• 不支持 JOIN 操作：一次只能查询一个实体
• 函数有限：不支持聚合函数（SUM、COUNT 等）
• 最多返回 1000 条结果：结果集更大时请使用分页
• 返回所有字段：无法选择特定字段（始终返回所有字段）

分页模式

def query_all_customers(realm_id, access_token):
    all_customers = []
    start_position = 1
    max_results = 1000

    while True:
        query = f"SELECT * FROM Customer STARTPOSITION {start_position} MAXRESULTS {max_results}"
        encoded_query = quote(query)
        url = f"{base_url}/company/{realm_id}/query?query={encoded_query}"

        response = requests.get(url, headers={"Authorization": f"Bearer {access_token}"})
        result = response.json()['QueryResponse']

        customers = result.get('Customer', [])
        if not customers:
            break

        all_customers.extend(customers)

        # Check if more results exist
        if len(customers) < max_results:
            break

        start_position += max_results

    return all_customers

批量操作

批量操作允许在单个 HTTP 请求中进行多次 API 调用（最多 30 次操作）。

批量请求结构

端点：POST /v3/company/{realmId}/batch

请求正文：

{
  "BatchItemRequest": [
    {
      "bId": "bid1",
      "operation": "create",
      "Customer": {
        "DisplayName": "New Customer 1"
      }
    },
    {
      "bId": "bid2",
      "operation": "update",
      "Invoice": {
        "Id": "123",
        "SyncToken": "1",
        "sparse": true,
        "EmailStatus": "NeedToSend"
      }
    },
    {
      "bId": "bid3",
      "operation": "query",
      "Query": "SELECT * FROM Customer WHERE Active = true MAXRESULTS 10"
    }
  ]
}

批次 ID 跟踪

每个操作都有唯一的 bId（批次 ID），用于跟踪结果：

响应结构：

{
  "BatchItemResponse": [
    {
      "bId": "bid1",
      "Customer": {
        "Id": "456",
        "DisplayName": "New Customer 1"
      }
    },
    {
      "bId": "bid2",
      "Invoice": {
        "Id": "123",
        "SyncToken": "2"
      }
    },
    {
      "bId": "bid3",
      "QueryResponse": {
        "Customer": [...]
      }
    }
  ]
}

Node.js 示例 - 批量更新客户

async function batchUpdateCustomers(realmId, customers, accessToken) {
  const batchItems = customers.map((customer, index) => ({
    bId: `customer_${index}`,
    operation: 'update',
    Customer: {
      Id: customer.Id,
      SyncToken: customer.SyncToken,
      sparse: true,
      Active: true  // Reactivate all customers
    }
  }));

  const url = `https://sandbox-quickbooks.api.intuit.com/v3/company/${realmId}/batch`;

  try {
    const response = await axios.post(url, {
      BatchItemRequest: batchItems
    }, {
      headers: {
        'Authorization': `Bearer ${accessToken}`,
        'Content-Type': 'application/json'
      }
    });

    const results = response.data.BatchItemResponse;

    // Process results by batch ID
    results.forEach(result => {
      if (result.Fault) {
        console.error(`Error for ${result.bId}:`, result.Fault);
      } else {
        console.log(`Success for ${result.bId}: Customer ${result.Customer.Id}`);
      }
    });

    return results;
  } catch (error) {
    console.error('Batch operation failed:', error.response?.data || error.message);
    throw error;
  }
}

批量操作的优势

• 减少 API 调用：一个请求中执行 30 个操作，而不是分别发送 30 个请求
• 降低延迟：单次往返，而非多次往返
• 适合速率限制：速率限制按单次 API 调用计数
• 每个操作独立执行：每个操作都可以独立成功或失败

批量操作类型

• create：创建新实体
• update：更新现有实体
• delete：删除实体
• query：执行查询

错误处理与故障排除

HTTP 状态码

• 200 OK：请求成功（但响应正文中可能包含 <Fault> 元素）
• 400 Bad Request：语法无效或请求格式错误
• 401 Unauthorized：访问令牌无效或已过期
• 403 Forbidden：权限不足或资源受限
• 404 Not Found：资源不存在
• 429 Too Many Requests：超出速率限制
• 500 Internal Server Error：服务端问题（重试一次）
• 503 Service Unavailable：服务暂时不可用（采用退避策略重试）

故障类型

即使状态为 200 OK，响应中也可能包含故障元素：

{
  "Fault": {
    "Error": [
      {
        "Message": "Duplicate Name Exists Error",
        "Detail": "The name supplied already exists.",
        "code": "6240",
        "element": "Customer.DisplayName"
      }
    ],
    "type": "ValidationFault"
  },
  "time": "2024-12-09T10:30:00.000-08:00"
}

故障类型：

1. ValidationFault：请求数据无效或违反业务规则

   • 修复方法：更正请求负载，检查必填字段

2. SystemFault：服务端错误

   • 修复方法：重试请求；如果问题持续存在，请联系支持人员

3. AuthenticationFault：凭据无效

   • 修复方法：刷新访问令牌，重新进行身份验证

4. AuthorizationFault：权限不足

   • 修复方法：检查 OAuth 范围，确保用户拥有管理员权限

常见错误代码

代码	错误	解决方案
6000	业务验证错误	检查 TotalAmt 和必填字段
3200	对象过期（SyncToken 不匹配）	重新读取实体以获取最新 SyncToken
3100	引用无效	确认被引用的实体存在（CustomerRef、ItemRef）
6240	名称重复	为 Customer/Item 使用唯一的 DisplayName
610	找不到对象	检查实体 ID 是否存在
4001	令牌无效	刷新访问令牌

SDK 异常处理

Java SDK 异常：

• ValidationException：验证故障
• ServiceException：服务故障
• AuthenticationException：身份验证故障
• BadRequestException：400 状态码
• InvalidTokenException：401 状态码
• InternalServiceException：500 状态码

Python 异常处理：

from intuitlib.exceptions import AuthClientError

try:
    response = requests.post(url, json=data, headers=headers)
    response.raise_for_status()

    # Check for fault in response body
    result = response.json()
    if 'Fault' in result:
        fault = result['Fault']
        print(f"Fault Type: {fault['type']}")
        for error in fault['Error']:
            print(f"  Code {error['code']}: {error['Message']}")
            print(f"  Element: {error.get('element', 'N/A')}")
        return None

    return result

except requests.exceptions.HTTPError as e:
    if e.response.status_code == 401:
        # Token expired, refresh
        print("Token expired, refreshing...")
        # Implement token refresh logic
    elif e.response.status_code == 429:
        # Rate limited, implement backoff
        print("Rate limited, backing off...")
    else:
        print(f"HTTP Error: {e.response.status_code}")
        print(f"Response: {e.response.text}")

except AuthClientError as e:
    print(f"Auth error: {str(e)}")

调试策略

1. 即使状态为 200，也要检查响应正文：成功响应中也可能出现故障元素
2. 记录 intuit_tid：在支持请求中提供此信息，以便更快解决问题
3. 验证 SyncToken：始终使用读取操作返回的最新版本
4. 先在沙盒中测试：使用沙盒公司进行开发
5. 实现重试逻辑：针对 500/503 错误使用指数退避
6. 解析错误详情：检查 error.code、element、message 字段

指数退避重试模式

async function apiCallWithRetry(apiFunction, maxRetries = 3) {
  for (let attempt = 0; attempt < maxRetries; attempt++) {
    try {
      return await apiFunction();
    } catch (error) {
      const status = error.response?.status;

      // Retry on server errors
      if (status >= 500 && status < 600 && attempt < maxRetries - 1) {
        const delay = Math.pow(2, attempt) * 1000; // 1s, 2s, 4s
        console.log(`Attempt ${attempt + 1} failed, retrying in ${delay}ms...`);
        await new Promise(resolve => setTimeout(resolve, delay));
        continue;
      }

      // Don't retry on client errors
      throw error;
    }
  }
}

变更检测与 Webhook

变更数据捕获（CDC）

CDC 会返回指定时间范围内（最长 30 天）发生变更的实体。

端点：GET /v3/company/{realmId}/cdc?entities={entityList}&changedSince={dateTime}

参数：

• entities：以逗号分隔的列表（例如，"Invoice,Customer,Payment"）
• changedSince：ISO 8601 时间戳（例如，"2024-12-01T09:00:00-07:00"）

Python 示例：

from datetime import datetime, timedelta
from urllib.parse import urlencode

def get_changed_entities(realm_id, entity_types, since_datetime, access_token):
    # Format: 2024-12-01T09:00:00-07:00
    changed_since = since_datetime.strftime('%Y-%m-%dT%H:%M:%S-07:00')

    params = {
        'entities': ','.join(entity_types),
        'changedSince': changed_since
    }

    url = f"https://sandbox-quickbooks.api.intuit.com/v3/company/{realm_id}/cdc"

    response = requests.get(
        url,
        params=params,
        headers={"Authorization": f"Bearer {access_token}"}
    )

    if response.status_code == 200:
        cdc_response = response.json()['CDCResponse']

        # Process changed entities
        for query_response in cdc_response:
            entity_type = query_response.get('QueryResponse', [{}])[0]

            for entity_name, entities in entity_type.items():
                if entities:
                    for entity in entities:
                        status = entity.get('status', 'Updated')
                        if status == 'Deleted':
                            print(f"Deleted {entity_name}: {entity['Id']}")
                        else:
                            print(f"Changed {entity_name}: {entity['Id']}")

        return cdc_response
    else:
        print(f"CDC request failed: {response.text}")
        return None

# Usage: Get all invoices and customers changed in last 24 hours
since = datetime.now() - timedelta(hours=24)
changes = get_changed_entities(realm_id, ['Invoice', 'Customer'], since, access_token)

响应结构：

{
  "CDCResponse": [
    {
      "QueryResponse": [
        {
          "Invoice": [
            {
              "Id": "123",
              "MetaData": {
                "LastUpdatedTime": "2024-12-09T10:30:00-08:00"
              },
              "TotalAmt": 100.00,
              "Balance": 50.00
              // ... full invoice object
            }
          ]
        }
      ]
    },
    {
      "QueryResponse": [
        {
          "Customer": [
            {
              "Id": "456",
              "status": "Deleted"
            }
          ]
        }
      ]
    }
  ],
  "time": "2024-12-09T11:00:00.000-08:00"
}

CDC 最佳实践

• 查询较短时间段：每次响应最多返回 1000 个实体，应按小时或按天检查
• 存储上次同步时间：跟踪 LastUpdatedTime，以设置 changedSince 参数
• 处理删除操作：status: "Deleted" 的实体仅包含 ID
• 获取完整实体：CDC 返回完整负载，而不只是变更内容
• 与 webhook 结合使用：使用 webhook 实现实时更新，并以 CDC 作为备份

Webhook（实时通知）

数据发生变化时，Webhook 会发送 HTTP POST 通知。

设置：

1. 在开发者控制面板中配置 Webhook URL
2. 实现用于接收通知的 POST 端点
3. 在 1 秒内返回 200 OK
4. 异步处理通知

通知负载：

{
  "eventNotifications": [
    {
      "realmId": "123456789",
      "dataChangeEvent": {
        "entities": [
          {
            "name": "Invoice",
            "id": "145",
            "operation": "Create",
            "lastUpdated": "2024-12-09T10:30:00.000Z"
          },
          {
            "name": "Payment",
            "id": "456",
            "operation": "Update",
            "lastUpdated": "2024-12-09T10:31:00.000Z"
          },
          {
            "name": "Customer",
            "id": "789",
            "operation": "Merge",
            "lastUpdated": "2024-12-09T10:32:00.000Z",
            "deletedId": "788"
          }
        ]
      }
    }
  ]
}

Node.js Webhook 处理程序：

const express = require('express');
const crypto = require('crypto');

const app = express();
app.use(express.json());

// Webhook endpoint
app.post('/webhooks/quickbooks', async (req, res) => {
  // Verify webhook signature (recommended)
  const signature = req.headers['intuit-signature'];
  const payload = JSON.stringify(req.body);

  // Return 200 immediately (process async)
  res.status(200).send('OK');

  // Process notifications asynchronously
  processWebhook(req.body).catch(console.error);
});

async function processWebhook(notification) {
  for (const event of notification.eventNotifications) {
    const realmId = event.realmId;

    for (const entity of event.dataChangeEvent.entities) {
      console.log(`${entity.operation} on ${entity.name} ID ${entity.id}`);

      // Fetch full entity data
      if (entity.operation !== 'Delete') {
        await fetchAndProcessEntity(realmId, entity.name, entity.id);
      } else {
        await handleEntityDeletion(realmId, entity.name, entity.id);
      }
    }
  }
}

async function fetchAndProcessEntity(realmId, entityType, entityId) {
  // Fetch full entity using read endpoint
  const url = `https://quickbooks.api.intuit.com/v3/company/${realmId}/${entityType.toLowerCase()}/${entityId}`;
  // ... implement fetch and processing logic
}

Webhook 与 CDC 决策矩阵

使用场景	建议
实时同步	Webhook
定期同步（每小时／每天）	CDC
初始数据加载	CDC
停机后重新连接	CDC
高容量变更	CDC（可减少通知开销）
低延迟要求	Webhook
备份／冗余	两者兼用（Webhook 为主，CDC 为备份）

组合方案模式

class QuickBooksSync:
    def __init__(self):
        self.last_cdc_sync = self.load_last_sync_time()

    def handle_webhook(self, notification):
        """Process real-time webhook"""
        for entity in notification['dataChangeEvent']['entities']:
            self.process_entity_change(entity)

        # Update last known change time
        self.last_cdc_sync = datetime.now()
        self.save_last_sync_time()

    def periodic_cdc_sync(self):
        """Catch any missed changes"""
        changes = get_changed_entities(
            self.realm_id,
            ['Invoice', 'Customer', 'Payment'],
            self.last_cdc_sync,
            self.access_token
        )

        for entity in self.extract_entities(changes):
            if not self.entity_exists_locally(entity):
                # Missed by webhook, process now
                self.process_entity_change(entity)

        self.last_cdc_sync = datetime.now()
        self.save_last_sync_time()

最佳实践

性能优化

1. 批量变更时使用批处理操作

   • 在单个请求中合并最多 30 项操作
   • 减少 API 调用并提高吞吐量
   • 示例：批量更新 30 位客户，而不是分别更新 30 次

2. 使用 CDC 或 Webhook 进行同步

   • 避免反复轮询所有实体
   • CDC 仅返回已更改的实体
   • Webhook 无需轮询即可提供实时通知

3. 稀疏更新可缩减负载

   • 仅发送要更改的字段
   • 减少数据传输和处理时间
   • 防止意外覆盖字段

4. 在本地缓存参考数据

   • 付款方式、税码和账户很少变化
   • 查询一次并设置 TTL 进行缓存
   • 减少冗余 API 调用

5. 对大型结果集进行分页

   • 使用 MAXRESULTS 限制查询结果
   • 分批处理以避免内存问题
   • 示例：每次查询 100 位客户

数据完整性

1. 更新时始终使用 SyncToken

   • 防止并发修改冲突
   • 更新前读取实体以获取最新令牌
   • 遇到 3200 错误时，重新读取并重试

2. 妥善处理并发修改

   def safe_update(realm_id, customer_id, changes, access_token):
       max_attempts = 3
       for attempt in range(max_attempts):
           # Read latest version
           customer = read_customer(realm_id, customer_id, access_token)
   
           # Apply changes
           customer.update(changes)
           customer['sparse'] = True
   
           # Attempt update
           try:
               return update_customer(realm_id, customer, access_token)
           except SyncTokenError:
               if attempt == max_attempts - 1:
                   raise
               continue  # Retry with fresh SyncToken
   

3. 在调用 API 前验证必填字段

   • 先在本地检查业务规则
   • 减少 API 返回的验证错误
   • 示例：创建发票前确认客户存在

4. 结合使用 Webhook 和 CDC 以可靠地跟踪变更

   • 使用 Webhook 获取实时更新
   • 定期使用 CDC 作为备份，以查找遗漏的变更
   • 保存上次同步时间戳

令牌管理

1. 访问令牌会在 3600 秒后过期

   • 设置自动刷新，在过期前完成刷新
   • 为稳妥起见，在第 50 分钟时刷新

2. 主动刷新令牌

   class TokenManager {
     constructor() {
       this.refreshTimer = null;
     }
   
     scheduleRefresh(expiresIn) {
       // Refresh 5 minutes before expiration
       const refreshTime = (expiresIn - 300) * 1000;
   
       this.refreshTimer = setTimeout(() => {
         this.refreshAccessToken();
       }, refreshTime);
     }
   
     async refreshAccessToken() {
       try {
         const newTokens = await oauthClient.refresh();
         this.storeTokens(newTokens);
         this.scheduleRefresh(newTokens.expires_in);
       } catch (error) {
         // Refresh failed, need re-authentication
         this.handleReauthentication();
       }
     }
   }
   

3. 始终使用最新的刷新令牌

   • 新刷新令牌签发 24 小时后，之前的刷新令牌会过期
   • 刷新后立即保存刷新令牌
   • 切勿使用旧刷新令牌

4. 安全存储令牌

   • 在数据库中加密
   • 切勿提交到版本控制系统
   • 在开发过程中使用环境变量

5. 自动处理 401 响应

   def api_call_with_auto_refresh(api_function):
       try:
           return api_function()
       except Unauthorized401Error:
           # Attempt token refresh
           refresh_tokens()
           # Retry with new token
           return api_function()
   

API 速率限制

1. 针对 429 实施指数退避

   def call_with_rate_limit_handling(api_function):
       max_retries = 5
       base_delay = 1
   
       for attempt in range(max_retries):
           try:
               return api_function()
           except RateLimitError as e:
               if attempt == max_retries - 1:
                   raise
   
               delay = base_delay * (2 ** attempt)  # 1s, 2s, 4s, 8s, 16s
               time.sleep(delay)
               continue
   

2. 使用批处理操作以减少调用次数

   • 1 次批处理请求相比 30 次单独请求，可减少 30 倍调用
   • 在速率限制中，批处理计为 1 次 API 调用

3. 监控速率限制标头（如果提供）

   • 某些端点会在标头中返回速率限制信息
   • 跟踪使用量以确保不超出限制

多币种注意事项

1. 启用多币种时必须提供 CurrencyRef

   {
     "Invoice": {
       "CurrencyRef": {
         "value": "USD",
         "name": "United States Dollar"
       }
     }
   }
   

2. 汇率处理

   • API 会自动应用汇率
   • ExchangeRate 字段显示所使用的换算汇率
   • 自动计算本位币金额

3. 特定地区的必填字段

   • 法国：启用自定义交易编号时，DocNumber 为必填项
   • 英国：税务处理方式不同（VAT）
   • 请查阅特定地区的文档

测试与开发

1. 使用沙盒公司（开发者账户免费提供）

   • 在 developer.intuit.com 创建
   • 与生产数据分开
   • API 功能完全一致

2. 端到端测试 OAuth 流程

   • 授权 URL → 代码交换 → 令牌刷新
   • 测试令牌过期处理
   • 验证刷新令牌轮换

3. 验证 webhook 端点

   • 使用示例载荷进行测试
   • 确保响应时间小于 1 秒
   • 处理 webhook 签名验证

4. 在生产环境中处理所有故障类型

   • ValidationFault、SystemFault、AuthenticationFault、AuthorizationFault
   • 记录错误详情（代码、消息、元素）
   • 实现适当的重试逻辑

5. 监控 API 调用和错误

   • 跟踪成功率和失败率
   • 错误率升高时发出警报
   • 记录支持请求的 intuit_tid

常见工作流

工作流 1：创建并发送发票

场景：为客户创建发票并通过电子邮件发送。

步骤：

1. 查询或创建客户

# Check if customer exists
customers = query_customers_by_email(realm_id, "[email protected]", access_token)

if not customers:
    # Create new customer
    customer = create_customer(realm_id, {
        "DisplayName": "Acme Corp",
        "PrimaryEmailAddr": {"Address": "[email protected]"},
        "BillAddr": {
            "Line1": "123 Main St",
            "City": "San Francisco",
            "CountrySubDivisionCode": "CA",
            "PostalCode": "94105"
        }
    }, access_token)
else:
    customer = customers[0]

customer_id = customer['Id']

2. 查询用于行项目的商品

# Get service item
query = "SELECT * FROM Item WHERE Type = 'Service' AND Name = 'Consulting'"
items = query_entity(realm_id, query, access_token)
service_item = items[0]

3. 创建包含行项目的发票

invoice_data = {
    "TxnDate": "2024-12-09",
    "DueDate": "2024-12-23",
    "CustomerRef": {"value": customer_id},
    "BillEmail": {"Address": "[email protected]"},
    "EmailStatus": "NeedToSend",  # Mark for email sending
    "Line": [
        {
            "Amount": 1500.00,
            "DetailType": "SalesItemLineDetail",
            "SalesItemLineDetail": {
                "ItemRef": {"value": service_item['Id']},
                "Qty": 10,
                "UnitPrice": 150.00,
                "TaxCodeRef": {"value": "NON"}  # Non-taxable
            },
            "Description": "Consulting services - December 2024"
        },
        {
            "Amount": 1500.00,
            "DetailType": "SubTotalLineDetail",
            "SubTotalLineDetail": {}
        }
    ]
}

invoice = create_invoice(realm_id, invoice_data, access_token)
print(f"Invoice {invoice['DocNumber']} created: ${invoice['TotalAmt']}")

4. 发送发票电子邮件（如果 EmailStatus = "NeedToSend"，则自动发送）

# QuickBooks automatically sends email when EmailStatus is NeedToSend
# Alternatively, use send endpoint:
send_url = f"{base_url}/company/{realm_id}/invoice/{invoice['Id']}/send"
params = {"sendTo": "[email protected]"}

response = requests.post(send_url, params=params, headers=headers)
if response.status_code == 200:
    print(f"Invoice sent to {customer['PrimaryEmailAddr']['Address']}")

5. 处理响应和关联交易

# Check invoice status
print(f"Invoice ID: {invoice['Id']}")
print(f"Balance: ${invoice['Balance']}")
print(f"Email Status: {invoice['EmailStatus']}")

# Track linked transactions
if 'LinkedTxn' in invoice:
    for linked in invoice['LinkedTxn']:
        print(f"Linked {linked['TxnType']}: {linked['TxnId']}")

工作流 2：记录发票付款

场景：客户通过支票支付发票。

步骤：

1. 按 DocNumber 查询发票

def find_invoice_by_number(realm_id, doc_number, access_token):
    query = f"SELECT * FROM Invoice WHERE DocNumber = '{doc_number}'"
    invoices = query_entity(realm_id, query, access_token)

    if not invoices:
        raise ValueError(f"Invoice {doc_number} not found")

    return invoices[0]

invoice = find_invoice_by_number(realm_id, "1045", access_token)
customer_id = invoice['CustomerRef']['value']
balance = invoice['Balance']

2. 创建付款实体

# Get payment method (Check)
payment_methods = query_entity(realm_id, "SELECT * FROM PaymentMethod WHERE Name = 'Check'", access_token)
payment_method_id = payment_methods[0]['Id']

payment_data = {
    "TotalAmt": balance,  # Pay full amount
    "CustomerRef": {"value": customer_id},
    "PaymentMethodRef": {"value": payment_method_id},
    "PaymentRefNum": "1234",  # Check number
    "TxnDate": "2024-12-09",
    "Line": [
        {
            "Amount": balance,
            "LinkedTxn": [
                {
                    "TxnId": invoice['Id'],
                    "TxnType": "Invoice"
                }
            ]
        }
    ]
}

payment = create_payment(realm_id, payment_data, access_token)

3. 验证余额更新

# Re-read invoice to see updated balance
updated_invoice = read_invoice(realm_id, invoice['Id'], access_token)

print(f"Original balance: ${balance}")
print(f"Payment amount: ${payment['TotalAmt']}")
print(f"New balance: ${updated_invoice['Balance']}")
print(f"Unapplied payment amount: ${payment.get('UnappliedAmt', 0)}")

4. 处理部分付款

def apply_partial_payment(realm_id, invoice_id, payment_amount, customer_id, access_token):
    payment_data = {
        "TotalAmt": payment_amount,  # Less than invoice balance
        "CustomerRef": {"value": customer_id},
        "Line": [
            {
                "Amount": payment_amount,
                "LinkedTxn": [
                    {
                        "TxnId": invoice_id,
                        "TxnType": "Invoice"
                    }
                ]
            }
        ]
    }

    payment = create_payment(realm_id, payment_data, access_token)

    # Check unapplied amount
    if payment['UnappliedAmt'] > 0:
        print(f"Warning: ${payment['UnappliedAmt']} unapplied (overpayment or error)")

    return payment

5. 将付款应用于多张发票

def pay_multiple_invoices(realm_id, invoice_ids, amounts, customer_id, total_paid, access_token):
    lines = []

    for invoice_id, amount in zip(invoice_ids, amounts):
        lines.append({
            "Amount": amount,
            "LinkedTxn": [{
                "TxnId": invoice_id,
                "TxnType": "Invoice"
            }]
        })

    payment_data = {
        "TotalAmt": total_paid,
        "CustomerRef": {"value": customer_id},
        "Line": lines
    }

    return create_payment(realm_id, payment_data, access_token)

# Example: Pay two invoices with single check
payment = pay_multiple_invoices(
    realm_id,
    ["145", "146"],  # Invoice IDs
    [100.00, 50.00],  # Amounts applied to each
    customer_id,
    150.00,  # Total check amount
    access_token
)

工作流 3：客户管理

场景：完整管理客户生命周期。

1. 创建带地址的客户

def create_customer_complete(realm_id, customer_info, access_token):
    customer_data = {
        "DisplayName": customer_info['display_name'],
        "GivenName": customer_info.get('first_name'),
        "FamilyName": customer_info.get('last_name'),
        "CompanyName": customer_info.get('company_name'),
        "PrimaryEmailAddr": {
            "Address": customer_info['email']
        },
        "PrimaryPhone": {
            "FreeFormNumber": customer_info.get('phone')
        },
        "BillAddr": {
            "Line1": customer_info['address_line1'],
            "City": customer_info['city'],
            "CountrySubDivisionCode": customer_info['state'],
            "PostalCode": customer_info['zip']
        },
        "ShipAddr": {
            "Line1": customer_info.get('ship_line1', customer_info['address_line1']),
            "City": customer_info.get('ship_city', customer_info['city']),
            "CountrySubDivisionCode": customer_info.get('ship_state', customer_info['state']),
            "PostalCode": customer_info.get('ship_zip', customer_info['zip'])
        }
    }

    return create_customer(realm_id, customer_data, access_token)

2. 稀疏更新以修改电子邮件

def update_customer_email(realm_id, customer_id, new_email, access_token):
    # Read current customer
    customer = read_customer(realm_id, customer_id, access_token)

    # Sparse update - only email
    update_data = {
        "Id": customer_id,
        "SyncToken": customer['SyncToken'],
        "sparse": True,
        "PrimaryEmailAddr": {
            "Address": new_email
        }
    }

    return update_customer(realm_id, update_data, access_token)

3. 查询客户交易

def get_customer_transactions(realm_id, customer_id, access_token):
    transactions = {}

    # Query invoices
    invoice_query = f"SELECT * FROM Invoice WHERE CustomerRef = '{customer_id}'"
    transactions['invoices'] = query_entity(realm_id, invoice_query, access_token)

    # Query payments
    payment_query = f"SELECT * FROM Payment WHERE CustomerRef = '{customer_id}'"
    transactions['payments'] = query_entity(realm_id, payment_query, access_token)

    # Query estimates
    estimate_query = f"SELECT * FROM Estimate WHERE CustomerRef = '{customer_id}'"
    transactions['estimates'] = query_entity(realm_id, estimate_query, access_token)

    # Calculate totals
    total_invoiced = sum(inv['TotalAmt'] for inv in transactions['invoices'])
    total_paid = sum(pmt['TotalAmt'] for pmt in transactions['payments'])

    transactions['summary'] = {
        'total_invoiced': total_invoiced,
        'total_paid': total_paid,
        'balance': total_invoiced - total_paid
    }

    return transactions

4. 更新应收账款账户引用

# Change default AR account for customer
def update_customer_ar_account(realm_id, customer_id, new_ar_account_id, access_token):
    customer = read_customer(realm_id, customer_id, access_token)

    update_data = {
        "Id": customer_id,
        "SyncToken": customer['SyncToken'],
        "sparse": True,
        "ARAccountRef": {
            "value": new_ar_account_id
        }
    }

    return update_customer(realm_id, update_data, access_token)

工作流 4：批量同步操作

场景：使用 CDC 和批量更新同步已更改的实体。

1. 使用 CDC 获取已更改的实体

from datetime import datetime, timedelta

def sync_changed_entities(realm_id, last_sync_time, access_token):
    # Get changes since last sync
    entity_types = ['Invoice', 'Customer', 'Payment', 'Item']
    changes = get_changed_entities(realm_id, entity_types, last_sync_time, access_token)

    return changes

2. 构建包含更新的批量操作

def build_batch_updates(changes):
    batch_items = []
    bid_counter = 0

    # Process each entity type
    for entity_type in ['Customer', 'Invoice', 'Payment']:
        entities = extract_entities_by_type(changes, entity_type)

        for entity in entities:
            if entity.get('status') == 'Deleted':
                # Skip deleted entities or handle separately
                continue

            # Example: Mark all invoices as reviewed
            if entity_type == 'Invoice':
                batch_items.append({
                    "bId": f"invoice_{bid_counter}",
                    "operation": "update",
                    "Invoice": {
                        "Id": entity['Id'],
                        "SyncToken": entity['SyncToken'],
                        "sparse": True,
                        "PrivateNote": f"Synced at {datetime.now().isoformat()}"
                    }
                })
                bid_counter += 1

    return batch_items

3. 异步执行批量操作

async def execute_batch_sync(realm_id, batch_items, access_token):
    # Split into batches of 30 (API limit)
    batch_size = 30
    results = []

    for i in range(0, len(batch_items), batch_size):
        batch_chunk = batch_items[i:i+batch_size]

        batch_request = {
            "BatchItemRequest": batch_chunk
        }

        url = f"https://sandbox-quickbooks.api.intuit.com/v3/company/{realm_id}/batch"

        response = await async_post(url, batch_request, {
            "Authorization": f"Bearer {access_token}",
            "Content-Type": "application/json"
        })

        if response.status == 200:
            batch_response = response.json()
            results.extend(batch_response['BatchItemResponse'])
        else:
            print(f"Batch {i//batch_size + 1} failed: {response.text}")

    return results

4. 按批次 ID 处理批量响应

def process_batch_results(batch_results):
    success_count = 0
    error_count = 0
    errors = []

    for result in batch_results:
        bid = result['bId']

        if 'Fault' in result:
            error_count += 1
            fault = result['Fault']
            errors.append({
                'batch_id': bid,
                'error_code': fault['Error'][0]['code'],
                'message': fault['Error'][0]['Message']
            })
            print(f"Error in {bid}: {fault['Error'][0]['Message']}")
        else:
            success_count += 1
            # Extract updated entity
            entity_type = list(result.keys())[0]
            if entity_type != 'bId':
                entity = result[entity_type]
                print(f"Success {bid}: {entity_type} {entity['Id']} updated")

    summary = {
        'total': len(batch_results),
        'success': success_count,
        'errors': error_count,
        'error_details': errors
    }

    return summary

# Complete workflow
async def sync_workflow(realm_id, last_sync_time, access_token):
    # 1. Get changes via CDC
    changes = sync_changed_entities(realm_id, last_sync_time, access_token)

    # 2. Build batch updates
    batch_items = build_batch_updates(changes)

    if not batch_items:
        print("No changes to sync")
        return

    # 3. Execute batch
    results = await execute_batch_sync(realm_id, batch_items, access_token)

    # 4. Process results
    summary = process_batch_results(results)

    print(f"Sync complete: {summary['success']}/{summary['total']} successful")
    if summary['errors'] > 0:
        print(f"Errors encountered: {summary['errors']}")
        for error in summary['error_details']:
            print(f"  {error['batch_id']}: {error['message']}")

    return summary

代码示例

示例 1：OAuth2 令牌刷新（Node.js）

包含自动刷新的完整令牌管理：

const OAuthClient = require('intuit-oauth');

class QuickBooksAuth {
  constructor(clientId, clientSecret, redirectUri, environment = 'sandbox') {
    this.oauthClient = new OAuthClient({
      clientId: clientId,
      clientSecret: clientSecret,
      environment: environment,
      redirectUri: redirectUri
    });

    this.accessToken = null;
    this.refreshToken = null;
    this.tokenExpiry = null;
    this.refreshTimer = null;
  }

  // Store tokens after authorization
  async storeTokens(authResponse) {
    this.accessToken = authResponse.token.access_token;
    this.refreshToken = authResponse.token.refresh_token;

    // Calculate expiry time
    const expiresIn = authResponse.token.expires_in; // 3600 seconds
    this.tokenExpiry = Date.now() + (expiresIn * 1000);

    // Schedule automatic refresh (5 minutes before expiry)
    this.scheduleRefresh(expiresIn - 300);

    // Persist tokens to secure storage
    await this.saveToDatabase({
      access_token: this.accessToken,
      refresh_token: this.refreshToken,
      expiry: this.tokenExpiry
    });
  }

  // Schedule automatic token refresh
  scheduleRefresh(delaySeconds) {
    if (this.refreshTimer) {
      clearTimeout(this.refreshTimer);
    }

    this.refreshTimer = setTimeout(async () => {
      try {
        await this.refreshAccessToken();
      } catch (error) {
        console.error('Scheduled token refresh failed:', error);
        // Notify admin that re-authentication needed
        this.notifyReauthenticationNeeded();
      }
    }, delaySeconds * 1000);
  }

  // Refresh access token
  async refreshAccessToken() {
    try {
      // Set refresh token in client
      this.oauthClient.setToken({
        refresh_token: this.refreshToken
      });

      // Refresh
      const authResponse = await this.oauthClient.refresh();

      console.log('Token refreshed successfully');

      // Store new tokens
      await this.storeTokens(authResponse);

      return authResponse;

    } catch (error) {
      console.error('Token refresh failed:', error.originalMessage);

      // Check if refresh token is invalid
      if (error.error === 'invalid_grant') {
        console.error('Refresh token invalid - re-authentication required');
        this.accessToken = null;
        this.refreshToken = null;
        throw new Error('Re-authentication required');
      }

      throw error;
    }
  }

  // Get valid access token (refresh if needed)
  async getAccessToken() {
    // Check if token is about to expire (within 5 minutes)
    const bufferTime = 5 * 60 * 1000; // 5 minutes

    if (!this.accessToken || Date.now() >= (this.tokenExpiry - bufferTime)) {
      console.log('Token expired or expiring soon, refreshing...');
      await this.refreshAccessToken();
    }

    return this.accessToken;
  }

  // Make API call with automatic token refresh
  async apiCall(url, options = {}) {
    try {
      const token = await this.getAccessToken();

      const response = await fetch(url, {
        ...options,
        headers: {
          ...options.headers,
          'Authorization': `Bearer ${token}`,
          'Accept': 'application/json'
        }
      });

      // Handle 401 - token might have expired
      if (response.status === 401) {
        console.log('401 Unauthorized - refreshing token and retrying...');
        await this.refreshAccessToken();

        // Retry with new token
        const newToken = await this.getAccessToken();
        return fetch(url, {
          ...options,
          headers: {
            ...options.headers,
            'Authorization': `Bearer ${newToken}`,
            'Accept': 'application/json'
          }
        });
      }

      return response;

    } catch (error) {
      console.error('API call failed:', error);
      throw error;
    }
  }

  // Save tokens to database (implement based on your storage)
  async saveToDatabase(tokens) {
    // Example: Save to database
    // await db.tokens.update({ realmId }, tokens);
  }

  // Notify admin about re-auth requirement
  notifyReauthenticationNeeded() {
    // Example: Send email or notification
    console.error('Re-authentication required for QuickBooks integration');
  }
}

// Usage
const auth = new QuickBooksAuth(
  'YOUR_CLIENT_ID',
  'YOUR_CLIENT_SECRET',
  'https://yourapp.com/callback',
  'sandbox'
);

// After OAuth authorization
auth.storeTokens(authResponse);

// Make API calls - automatic token refresh
const response = await auth.apiCall(
  'https://sandbox-quickbooks.api.intuit.com/v3/company/123/customer/456',
  { method: 'GET' }
);

示例 2：创建发票（Python）

包含行项目和税费的完整发票创建流程：

import requests
from datetime import datetime, timedelta

class QuickBooksInvoice:
    def __init__(self, realm_id, access_token, base_url='https://sandbox-quickbooks.api.intuit.com'):
        self.realm_id = realm_id
        self.access_token = access_token
        self.base_url = base_url

    def create_invoice(self, customer_id, line_items, due_days=30, tax_code='TAX', memo=None):
        """
        Create an invoice with multiple line items

        Args:
            customer_id: QuickBooks customer ID
            line_items: List of dicts with 'item_id', 'quantity', 'unit_price', 'description'
            due_days: Days until due date
            tax_code: Tax code ('TAX' for taxable, 'NON' for non-taxable)
            memo: Customer memo

        Returns:
            Created invoice dict or None if error
        """
        # Calculate dates
        txn_date = datetime.now().strftime('%Y-%m-%d')
        due_date = (datetime.now() + timedelta(days=due_days)).strftime('%Y-%m-%d')

        # Build line items
        lines = []
        subtotal = 0

        for idx, item in enumerate(line_items, start=1):
            amount = item['quantity'] * item['unit_price']
            subtotal += amount

            lines.append({
                "LineNum": idx,
                "Amount": amount,
                "DetailType": "SalesItemLineDetail",
                "Description": item.get('description', ''),
                "SalesItemLineDetail": {
                    "ItemRef": {
                        "value": item['item_id']
                    },
                    "Qty": item['quantity'],
                    "UnitPrice": item['unit_price'],
                    "TaxCodeRef": {
                        "value": tax_code
                    }
                }
            })

        # Add subtotal line
        lines.append({
            "Amount": subtotal,
            "DetailType": "SubTotalLineDetail",
            "SubTotalLineDetail": {}
        })

        # Build invoice payload
        invoice_data = {
            "TxnDate": txn_date,
            "DueDate": due_date,
            "CustomerRef": {
                "value": customer_id
            },
            "Line": lines,
            "BillEmail": {},  # Will be populated from customer
            "EmailStatus": "NotSet"
        }

        # Add memo if provided
        if memo:
            invoice_data["CustomerMemo"] = {
                "value": memo
            }

        # Make API request
        url = f"{self.base_url}/v3/company/{self.realm_id}/invoice"

        headers = {
            "Authorization": f"Bearer {self.access_token}",
            "Accept": "application/json",
            "Content-Type": "application/json"
        }

        try:
            response = requests.post(url, json=invoice_data, headers=headers)
            response.raise_for_status()

            # Check for fault in response
            result = response.json()

            if 'Fault' in result:
                self._handle_fault(result['Fault'])
                return None

            invoice = result['Invoice']

            print(f"✓ Invoice {invoice['DocNumber']} created")
            print(f"  Customer: {invoice['CustomerRef']['value']}")
            print(f"  Total: ${invoice['TotalAmt']:.2f}")
            print(f"  Due: {invoice['DueDate']}")
            print(f"  Balance: ${invoice['Balance']:.2f}")

            return invoice

        except requests.exceptions.HTTPError as e:
            print(f"✗ HTTP Error: {e.response.status_code}")
            print(f"  Response: {e.response.text}")
            return None
        except Exception as e:
            print(f"✗ Error creating invoice: {str(e)}")
            return None

    def send_invoice(self, invoice_id, email_address):
        """Send invoice via email"""
        url = f"{self.base_url}/v3/company/{self.realm_id}/invoice/{invoice_id}/send"

        headers = {
            "Authorization": f"Bearer {self.access_token}",
            "Accept": "application/json"
        }

        params = {"sendTo": email_address}

        try:
            response = requests.post(url, params=params, headers=headers)
            response.raise_for_status()

            result = response.json()

            if 'Fault' in result:
                self._handle_fault(result['Fault'])
                return False

            invoice = result['Invoice']
            print(f"✓ Invoice {invoice['DocNumber']} sent to {email_address}")
            print(f"  Email Status: {invoice['EmailStatus']}")

            return True

        except Exception as e:
            print(f"✗ Error sending invoice: {str(e)}")
            return False

    def _handle_fault(self, fault):
        """Handle fault responses"""
        print(f"✗ Fault Type: {fault['type']}")
        for error in fault['Error']:
            print(f"  Error {error['code']}: {error['Message']}")
            if 'element' in error:
                print(f"  Element: {error['element']}")

# Usage example
invoice_manager = QuickBooksInvoice(realm_id='123456789', access_token='your_token')

# Create invoice with multiple items
invoice = invoice_manager.create_invoice(
    customer_id='42',
    line_items=[
        {
            'item_id': '1',
            'quantity': 10,
            'unit_price': 150.00,
            'description': 'Consulting services - December 2024'
        },
        {
            'item_id': '5',
            'quantity': 1,
            'unit_price': 500.00,
            'description': 'Project management - December 2024'
        }
    ],
    due_days=30,
    tax_code='TAX',  # or 'NON' for non-taxable
    memo='Thank you for your business!'
)

if invoice:
    # Send invoice via email
    invoice_manager.send_invoice(invoice['Id'], '[email protected]')

示例 3：稀疏更新客户（Node.js）

演示带错误处理的稀疏更新模式：

const axios = require('axios');

class QuickBooksCustomer {
  constructor(realmId, accessToken, baseUrl = 'https://sandbox-quickbooks.api.intuit.com') {
    this.realmId = realmId;
    this.accessToken = accessToken;
    this.baseUrl = baseUrl;
  }

  async readCustomer(customerId) {
    const url = `${this.baseUrl}/v3/company/${this.realmId}/customer/${customerId}`;

    try {
      const response = await axios.get(url, {
        headers: {
          'Authorization': `Bearer ${this.accessToken}`,
          'Accept': 'application/json'
        }
      });

      return response.data.Customer;
    } catch (error) {
      console.error('Read customer failed:', error.response?.data || error.message);
      throw error;
    }
  }

  async sparseUpdate(customerId, updates) {
    // First, read customer to get current SyncToken
    const customer = await this.readCustomer(customerId);

    // Build sparse update payload
    const updateData = {
      Id: customerId,
      SyncToken: customer.SyncToken,
      sparse: true,
      ...updates
    };

    const url = `${this.baseUrl}/v3/company/${this.realmId}/customer`;

    try {
      const response = await axios.post(url, updateData, {
        headers: {
          'Authorization': `Bearer ${this.accessToken}`,
          'Accept': 'application/json',
          'Content-Type': 'application/json'
        }
      });

      // Check for fault
      if (response.data.Fault) {
        this.handleFault(response.data.Fault);
        return null;
      }

      const updatedCustomer = response.data.Customer;

      console.log(`✓ Customer ${updatedCustomer.DisplayName} updated`);
      console.log(`  New SyncToken: ${updatedCustomer.SyncToken}`);

      return updatedCustomer;

    } catch (error) {
      if (error.response?.status === 400) {
        const fault = error.response.data.Fault;

        // Handle SyncToken mismatch
        if (fault.Error[0].code === '3200') {
          console.log('SyncToken mismatch - retrying with fresh token...');
          // Recursive retry with new token
          return this.sparseUpdate(customerId, updates);
        }
      }

      console.error('Update failed:', error.response?.data || error.message);
      throw error;
    }
  }

  // Example: Update email
  async updateEmail(customerId, newEmail) {
    return this.sparseUpdate(customerId, {
      PrimaryEmailAddr: {
        Address: newEmail
      }
    });
  }

  // Example: Update phone
  async updatePhone(customerId, newPhone) {
    return this.sparseUpdate(customerId, {
      PrimaryPhone: {
        FreeFormNumber: newPhone
      }
    });
  }

  // Example: Update billing address
  async updateBillingAddress(customerId, address) {
    return this.sparseUpdate(customerId, {
      BillAddr: {
        Line1: address.line1,
        City: address.city,
        CountrySubDivisionCode: address.state,
        PostalCode: address.zip
      }
    });
  }

  // Example: Deactivate customer
  async deactivateCustomer(customerId) {
    return this.sparseUpdate(customerId, {
      Active: false
    });
  }

  // Example: Update multiple fields at once
  async updateMultipleFields(customerId, updates) {
    const sparseUpdates = {};

    if (updates.email) {
      sparseUpdates.PrimaryEmailAddr = { Address: updates.email };
    }

    if (updates.phone) {
      sparseUpdates.PrimaryPhone = { FreeFormNumber: updates.phone };
    }

    if (updates.displayName) {
      sparseUpdates.DisplayName = updates.displayName;
    }

    if (updates.notes) {
      sparseUpdates.Notes = updates.notes;
    }

    return this.sparseUpdate(customerId, sparseUpdates);
  }

  handleFault(fault) {
    console.error(`✗ Fault Type: ${fault.type}`);
    fault.Error.forEach(error => {
      console.error(`  Error ${error.code}: ${error.Message}`);
      if (error.element) {
        console.error(`  Element: ${error.element}`);
      }
    });
  }
}

// Usage
const customerManager = new QuickBooksCustomer('123456789', 'your_access_token');

// Update email
await customerManager.updateEmail('42', '[email protected]');

// Update phone
await customerManager.updatePhone('42', '(555) 987-6543');

// Update address
await customerManager.updateBillingAddress('42', {
  line1: '456 New Street',
  city: 'San Francisco',
  state: 'CA',
  zip: '94105'
});

// Update multiple fields
await customerManager.updateMultipleFields('42', {
  email: '[email protected]',
  phone: '(555) 111-2222',
  displayName: 'Updated Customer Name',
  notes: 'VIP customer - priority support'
});

// Deactivate customer
await customerManager.deactivateCustomer('42');

示例 4：带筛选条件的查询（Python）

包含日期范围和排序的复杂查询：

import requests
from urllib.parse import quote
from datetime import datetime, timedelta

class QuickBooksQuery:
    def __init__(self, realm_id, access_token, base_url='https://sandbox-quickbooks.api.intuit.com'):
        self.realm_id = realm_id
        self.access_token = access_token
        self.base_url = base_url

    def query(self, sql_query):
        """Execute SQL-like query"""
        encoded_query = quote(sql_query)
        url = f"{self.base_url}/v3/company/{self.realm_id}/query?query={encoded_query}"

        headers = {
            "Authorization": f"Bearer {self.access_token}",
            "Accept": "application/json"
        }

        try:
            response = requests.get(url, headers=headers)
            response.raise_for_status()

            result = response.json()

            if 'Fault' in result:
                print(f"Query error: {result['Fault']}")
                return []

            query_response = result.get('QueryResponse', {})

            # Extract entities (keys vary by entity type)
            for key, value in query_response.items():
                if key not in ['startPosition', 'maxResults', 'totalCount']:
                    return value if isinstance(value, list) else []

            return []

        except Exception as e:
            print(f"Query failed: {str(e)}")
            return []

    def query_invoices_by_date_range(self, start_date, end_date, customer_id=None):
        """Query invoices within date range, optionally filtered by customer"""
        query = f"SELECT * FROM Invoice WHERE TxnDate >= '{start_date}' AND TxnDate <= '{end_date}'"

        if customer_id:
            query += f" AND CustomerRef = '{customer_id}'"

        query += " ORDERBY TxnDate DESC"

        invoices = self.query(query)

        print(f"Found {len(invoices)} invoices between {start_date} and {end_date}")

        # Calculate totals
        total_amount = sum(inv['TotalAmt'] for inv in invoices)
        total_balance = sum(inv['Balance'] for inv in invoices)

        print(f"Total invoiced: ${total_amount:.2f}")
        print(f"Outstanding balance: ${total_balance:.2f}")

        return invoices

    def query_overdue_invoices(self, as_of_date=None):
        """Query invoices past due date"""
        if not as_of_date:
            as_of_date = datetime.now().strftime('%Y-%m-%d')

        query = f"SELECT * FROM Invoice WHERE Balance > '0' AND DueDate < '{as_of_date}' ORDERBY DueDate"

        invoices = self.query(query)

        print(f"Found {len(invoices)} overdue invoices as of {as_of_date}")

        # Group by customer
        by_customer = {}
        for inv in invoices:
            customer_id = inv['CustomerRef']['value']
            if customer_id not in by_customer:
                by_customer[customer_id] = {
                    'customer_name': inv['CustomerRef'].get('name', 'Unknown'),
                    'invoices': [],
                    'total_overdue': 0
                }

            by_customer[customer_id]['invoices'].append(inv)
            by_customer[customer_id]['total_overdue'] += inv['Balance']

        # Print summary
        for customer_id, data in by_customer.items():
            print(f"\nCustomer: {data['customer_name']}")
            print(f"  Overdue invoices: {len(data['invoices'])}")
            print(f"  Total overdue: ${data['total_overdue']:.2f}")

        return invoices

    def query_customers_by_balance(self, min_balance=0):
        """Query customers with balance greater than minimum"""
        query = f"SELECT * FROM Customer WHERE Balance > '{min_balance}' ORDERBY Balance DESC"

        customers = self.query(query)

        print(f"Found {len(customers)} customers with balance > ${min_balance}")

        total_ar = sum(cust['Balance'] for cust in customers)
        print(f"Total accounts receivable: ${total_ar:.2f}")

        return customers

    def query_items_by_type(self, item_type='Service'):
        """Query items by type (Service, Inventory, NonInventory, Category)"""
        query = f"SELECT * FROM Item WHERE Type = '{item_type}' AND Active = true ORDERBY Name"

        items = self.query(query)

        print(f"Found {len(items)} active {item_type} items")

        return items

    def search_customers_by_name(self, search_term):
        """Search customers by display name"""
        query = f"SELECT * FROM Customer WHERE DisplayName LIKE '%{search_term}%' ORDERBY DisplayName"

        customers = self.query(query)

        print(f"Found {len(customers)} customers matching '{search_term}'")

        for cust in customers:
            print(f"  {cust['DisplayName']} - Balance: ${cust['Balance']:.2f}")

        return customers

    def query_recent_payments(self, days=30):
        """Query payments from last N days"""
        start_date = (datetime.now() - timedelta(days=days)).strftime('%Y-%m-%d')

        query = f"SELECT * FROM Payment WHERE TxnDate >= '{start_date}' ORDERBY TxnDate DESC"

        payments = self.query(query)

        print(f"Found {len(payments)} payments in last {days} days")

        total_received = sum(pmt['TotalAmt'] for pmt in payments)
        print(f"Total payments received: ${total_received:.2f}")

        return payments

# Usage
query_service = QuickBooksQuery(realm_id='123456789', access_token='your_token')

# Query invoices for date range
invoices = query_service.query_invoices_by_date_range(
    start_date='2024-01-01',
    end_date='2024-12-31'
)

# Query invoices for specific customer
customer_invoices = query_service.query_invoices_by_date_range(
    start_date='2024-01-01',
    end_date='2024-12-31',
    customer_id='42'
)

# Find overdue invoices
overdue = query_service.query_overdue_invoices()

# Find customers with high balances
high_balance_customers = query_service.query_customers_by_balance(min_balance=1000.00)

# Search for customer
customers = query_service.search_customers_by_name('Acme')

# Get recent payments
recent_payments = query_service.query_recent_payments(days=30)

示例 5：批量操作（Node.js）

批量创建或更新多个实体：

const axios = require('axios');

class QuickBooksBatch {
  constructor(realmId, accessToken, baseUrl = 'https://sandbox-quickbooks.api.intuit.com') {
    this.realmId = realmId;
    this.accessToken = accessToken;
    this.baseUrl = baseUrl;
  }

  async executeBatch(batchItems) {
    const url = `${this.baseUrl}/v3/company/${this.realmId}/batch`;

    try {
      const response = await axios.post(url, {
        BatchItemRequest: batchItems
      }, {
        headers: {
          'Authorization': `Bearer ${this.accessToken}`,
          'Content-Type': 'application/json',
          'Accept': 'application/json'
        }
      });

      const results = response.data.BatchItemResponse;

      // Process results
      const summary = {
        total: results.length,
        success: 0,
        errors: 0,
        results: []
      };

      results.forEach(result => {
        if (result.Fault) {
          summary.errors++;
          console.error(`✗ Error for ${result.bId}:`);
          result.Fault.Error.forEach(err => {
            console.error(`  ${err.code}: ${err.Message}`);
          });
          summary.results.push({
            bId: result.bId,
            status: 'error',
            error: result.Fault
          });
        } else {
          summary.success++;
          // Extract entity from result
          const entityType = Object.keys(result).find(k => k !== 'bId');
          const entity = result[entityType];

          console.log(`✓ Success for ${result.bId}: ${entityType} ${entity.Id}`);
          summary.results.push({
            bId: result.bId,
            status: 'success',
            entityType: entityType,
            entity: entity
          });
        }
      });

      console.log(`\nBatch complete: ${summary.success}/${summary.total} successful`);

      return summary;

    } catch (error) {
      console.error('Batch operation failed:', error.response?.data || error.message);
      throw error;
    }
  }

  // Batch create customers
  async batchCreateCustomers(customers) {
    const batchItems = customers.map((customer, index) => ({
      bId: `customer_create_${index}`,
      operation: 'create',
      Customer: {
        DisplayName: customer.displayName,
        PrimaryEmailAddr: { Address: customer.email },
        PrimaryPhone: { FreeFormNumber: customer.phone },
        BillAddr: {
          Line1: customer.address,
          City: customer.city,
          CountrySubDivisionCode: customer.state,
          PostalCode: customer.zip
        }
      }
    }));

    return this.executeBatch(batchItems);
  }

  // Batch update invoices
  async batchUpdateInvoices(updates) {
    const batchItems = updates.map((update, index) => ({
      bId: `invoice_update_${index}`,
      operation: 'update',
      Invoice: {
        Id: update.id,
        SyncToken: update.syncToken,
        sparse: true,
        ...update.changes
      }
    }));

    return this.executeBatch(batchItems);
  }

  // Batch query multiple entities
  async batchQuery(queries) {
    const batchItems = queries.map((query, index) => ({
      bId: `query_${index}`,
      operation: 'query',
      Query: query.sql
    }));

    const results = await this.executeBatch(batchItems);

    // Extract query results
    const queryResults = {};
    results.results.forEach(result => {
      if (result.status === 'success' && result.entity.QueryResponse) {
        queryResults[result.bId] = result.entity.QueryResponse;
      }
    });

    return queryResults;
  }

  // Mixed batch operations
  async mixedBatch(operations) {
    const batchItems = operations.map((op, index) => {
      const item = {
        bId: `op_${index}_${op.type}`,
        operation: op.operation
      };

      // Add entity or query data
      if (op.operation === 'query') {
        item.Query = op.data;
      } else {
        item[op.entityType] = op.data;
      }

      return item;
    });

    return this.executeBatch(batchItems);
  }
}

// Usage Examples

const batchService = new QuickBooksBatch('123456789', 'your_access_token');

// Example 1: Batch create customers
const newCustomers = [
  {
    displayName: 'Acme Corp',
    email: '[email protected]',
    phone: '(555) 111-1111',
    address: '123 Main St',
    city: 'San Francisco',
    state: 'CA',
    zip: '94105'
  },
  {
    displayName: 'TechStart Inc',
    email: '[email protected]',
    phone: '(555) 222-2222',
    address: '456 Market St',
    city: 'San Francisco',
    state: 'CA',
    zip: '94103'
  }
];

const createResults = await batchService.batchCreateCustomers(newCustomers);

// Example 2: Batch update invoices (mark as sent)
const invoiceUpdates = [
  { id: '145', syncToken: '0', changes: { EmailStatus: 'NeedToSend' } },
  { id: '146', syncToken: '0', changes: { EmailStatus: 'NeedToSend' } },
  { id: '147', syncToken: '0', changes: { EmailStatus: 'NeedToSend' } }
];

const updateResults = await batchService.batchUpdateInvoices(invoiceUpdates);

// Example 3: Batch queries
const queries = [
  { sql: 'SELECT * FROM Customer WHERE Active = true MAXRESULTS 10' },
  { sql: 'SELECT * FROM Invoice WHERE Balance > 0 MAXRESULTS 10' },
  { sql: 'SELECT * FROM Payment MAXRESULTS 10' }
];

const queryResults = await batchService.batchQuery(queries);

// Example 4: Mixed batch operations
const mixedOps = [
  {
    type: 'create_customer',
    operation: 'create',
    entityType: 'Customer',
    data: {
      DisplayName: 'New Customer',
      PrimaryEmailAddr: { Address: '[email protected]' }
    }
  },
  {
    type: 'update_invoice',
    operation: 'update',
    entityType: 'Invoice',
    data: {
      Id: '145',
      SyncToken: '1',
      sparse: true,
      CustomerMemo: { value: 'Thank you!' }
    }
  },
  {
    type: 'query_items',
    operation: 'query',
    data: 'SELECT * FROM Item WHERE Type = \'Service\' MAXRESULTS 5'
  }
];

const mixedResults = await batchService.mixedBatch(mixedOps);

console.log(`Mixed batch: ${mixedResults.success}/${mixedResults.total} successful`);

示例 6：应用付款（Python）

将付款应用于多张发票：

import requests

class QuickBooksPayment:
    def __init__(self, realm_id, access_token, base_url='https://sandbox-quickbooks.api.intuit.com'):
        self.realm_id = realm_id
        self.access_token = access_token
        self.base_url = base_url

    def create_payment(self, customer_id, total_amount, payment_method_id,
                       payment_ref_num, txn_date, invoice_applications):
        """
        Create payment and apply to one or more invoices

        Args:
            customer_id: QuickBooks customer ID
            total_amount: Total payment amount
            payment_method_id: Payment method ID
            payment_ref_num: Check number or transaction reference
            txn_date: Payment date (YYYY-MM-DD)
            invoice_applications: List of {'invoice_id': str, 'amount': float}

        Returns:
            Created payment dict or None if error
        """
        # Build line items for invoice applications
        lines = []
        total_applied = 0

        for application in invoice_applications:
            lines.append({
                "Amount": application['amount'],
                "LinkedTxn": [
                    {
                        "TxnId": application['invoice_id'],
                        "TxnType": "Invoice"
                    }
                ]
            })
            total_applied += application['amount']

        # Check for unapplied amount
        unapplied = total_amount - total_applied
        if unapplied < 0:
            print(f"Warning: Applied amount (${total_applied}) exceeds payment (${total_amount})")
            return None

        # Build payment payload
        payment_data = {
            "TotalAmt": total_amount,
            "CustomerRef": {
                "value": customer_id
            },
            "TxnDate": txn_date,
            "PaymentMethodRef": {
                "value": payment_method_id
            },
            "PaymentRefNum": payment_ref_num,
            "Line": lines
        }

        # Make API request
        url = f"{self.base_url}/v3/company/{self.realm_id}/payment"

        headers = {
            "Authorization": f"Bearer {self.access_token}",
            "Accept": "application/json",
            "Content-Type": "application/json"
        }

        try:
            response = requests.post(url, json=payment_data, headers=headers)
            response.raise_for_status()

            result = response.json()

            if 'Fault' in result:
                self._handle_fault(result['Fault'])
                return None

            payment = result['Payment']

            print(f"✓ Payment created: ID {payment['Id']}")
            print(f"  Customer: {payment['CustomerRef']['value']}")
            print(f"  Total Amount: ${payment['TotalAmt']:.2f}")
            print(f"  Applied Amount: ${total_applied:.2f}")
            print(f"  Unapplied Amount: ${payment.get('UnappliedAmt', 0):.2f}")
            print(f"  Reference: {payment.get('PaymentRefNum', 'N/A')}")

            # Show invoice applications
            for line in payment['Line']:
                if 'LinkedTxn' in line:
                    for linked in line['LinkedTxn']:
                        print(f"  Applied ${line['Amount']:.2f} to {linked['TxnType']} {linked['TxnId']}")

            return payment

        except requests.exceptions.HTTPError as e:
            print(f"✗ HTTP Error: {e.response.status_code}")
            print(f"  Response: {e.response.text}")
            return None
        except Exception as e:
            print(f"✗ Error creating payment: {str(e)}")
            return None

    def apply_payment_to_invoices(self, customer_id, check_number, check_amount,
                                   check_date, invoices):
        """
        Apply a single check payment to multiple invoices

        Args:
            customer_id: Customer ID
            check_number: Check number
            check_amount: Total check amount
            check_date: Check date
            invoices: List of {'id': str, 'amount_to_apply': float, 'balance': float}

        Returns:
            Payment dict or None
        """
        # Get check payment method ID
        payment_methods = self.query_payment_methods()
        check_method = next((pm for pm in payment_methods if pm['Name'].lower() == 'check'), None)

        if not check_method:
            print("Check payment method not found")
            return None

        # Build invoice applications
        applications = []
        total_to_apply = 0

        for invoice in invoices:
            amount = min(invoice['amount_to_apply'], invoice['balance'])
            applications.append({
                'invoice_id': invoice['id'],
                'amount': amount
            })
            total_to_apply += amount

            print(f"Will apply ${amount:.2f} to Invoice {invoice['id']}")

        # Check if payment covers all applications
        if total_to_apply > check_amount:
            print(f"Warning: Total applications (${total_to_apply}) exceeds check amount (${check_amount})")
            return None

        # Create payment
        payment = self.create_payment(
            customer_id=customer_id,
            total_amount=check_amount,
            payment_method_id=check_method['Id'],
            payment_ref_num=check_number,
            txn_date=check_date,
            invoice_applications=applications
        )

        return payment

    def apply_partial_payment(self, customer_id, payment_amount, payment_method_id,
                              invoice_id, partial_amount):
        """Apply partial payment to invoice"""
        if partial_amount > payment_amount:
            print("Partial amount cannot exceed total payment")
            return None

        applications = [{
            'invoice_id': invoice_id,
            'amount': partial_amount
        }]

        payment = self.create_payment(
            customer_id=customer_id,
            total_amount=payment_amount,
            payment_method_id=payment_method_id,
            payment_ref_num='',
            txn_date=datetime.now().strftime('%Y-%m-%d'),
            invoice_applications=applications
        )

        if payment and payment.get('UnappliedAmt', 0) > 0:
            print(f"\nNote: ${payment['UnappliedAmt']:.2f} remains unapplied")
            print("This amount can be applied to future invoices or refunded")

        return payment

    def query_payment_methods(self):
        """Get available payment methods"""
        from urllib.parse import quote

        query = "SELECT * FROM PaymentMethod"
        encoded_query = quote(query)

        url = f"{self.base_url}/v3/company/{self.realm_id}/query?query={encoded_query}"

        headers = {
            "Authorization": f"Bearer {self.access_token}",
            "Accept": "application/json"
        }

        response = requests.get(url, headers=headers)
        result = response.json()

        return result.get('QueryResponse', {}).get('PaymentMethod', [])

    def _handle_fault(self, fault):
        """Handle fault responses"""
        print(f"✗ Fault Type: {fault['type']}")
        for error in fault['Error']:
            print(f"  Error {error['code']}: {error['Message']}")
            if 'element' in error:
                print(f"  Element: {error['element']}")

# Usage Examples

payment_service = QuickBooksPayment(realm_id='123456789', access_token='your_token')

# Example 1: Apply single check to multiple invoices
payment = payment_service.apply_payment_to_invoices(
    customer_id='42',
    check_number='1234',
    check_amount=1500.00,
    check_date='2024-12-09',
    invoices=[
        {'id': '145', 'amount_to_apply': 1000.00, 'balance': 1000.00},
        {'id': '146', 'amount_to_apply': 500.00, 'balance': 750.00}
    ]
)

# Example 2: Partial payment on single invoice
partial_payment = payment_service.apply_partial_payment(
    customer_id='42',
    payment_amount=500.00,
    payment_method_id='1',  # Cash
    invoice_id='147',
    partial_amount=500.00  # Invoice balance is $1000, paying $500
)

# Example 3: Payment with unapplied amount (credit for future invoices)
credit_payment = payment_service.create_payment(
    customer_id='42',
    total_amount=2000.00,  # Customer pays $2000
    payment_method_id='1',
    payment_ref_num='',
    txn_date='2024-12-09',
    invoice_applications=[
        {'invoice_id': '145', 'amount': 1000.00}  # Only $1000 applied
    ]
    # $1000 remains unapplied as credit
)

API 参考快速链接

Context7 Library：使用 Context7 MCP 获取最新文档：

• Library ID：/websites/developer_intuit_app_developer_qbo
• 用于获取最新代码示例和 API 变更

官方资源：

• QuickBooks API Explorer：https://developer.intuit.com/app/developer/qbo/docs/api/accounting/all-entities/account

  • 提供支持沙盒测试的交互式 API 参考
  • 提供实体专属文档和示例请求

• Developer Dashboard：https://developer.intuit.com/app/developer/myapps

  • 管理应用、密钥和 Webhook
  • 创建沙盒公司

SDK：

• Node.js：https://github.com/intuit/intuit-oauth（OAuth）+ axios 用于 API 调用
• Python：https://github.com/intuit/intuit-oauth-python（OAuth）+ requests
• Java：https://github.com/intuit/QuickBooks-V3-Java-SDK
• PHP：https://github.com/intuit/QuickBooks-V3-PHP-SDK
• C#/.NET：https://github.com/intuit/QuickBooks-V3-DotNET-SDK

常用端点：

• 基础 URL（沙盒）：https://sandbox-quickbooks.api.intuit.com/v3/company/{realmId}
• 基础 URL（生产环境）：https://quickbooks.api.intuit.com/v3/company/{realmId}
• Token 端点：https://oauth.platform.intuit.com/oauth2/v1/tokens/bearer
• OAuth 授权：https://appcenter.intuit.com/connect/oauth2

常见问题排查

SyncToken 不匹配（错误 3200）

症状：更新实体时出现 “stale object error”

原因：请求中的 SyncToken 与当前版本不匹配（发生并发修改）

解决方案：

def safe_update_with_retry(entity_id, updates, max_attempts=3):
    for attempt in range(max_attempts):
        try:
            # Read latest version
            entity = read_entity(entity_id)

            # Apply changes
            entity.update(updates)
            entity['sparse'] = True

            # Attempt update
            return update_entity(entity)

        except SyncTokenError as e:
            if attempt == max_attempts - 1:
                raise
            print(f"SyncToken mismatch, retrying... (attempt {attempt + 1})")
            continue

缺少必填字段（错误 6000）

症状：出现 “business validation error” 或 “required field missing”

原因：缺少 TotalAmt、CustomerRef 等必填字段，或缺少实体特有的必填字段

解决方案：

• 查看 API 文档，确认实体特有的必填字段
• 对于 Payment：必须提供 TotalAmt 和 CustomerRef
• 对于 Invoice：必须提供 CustomerRef 和 Line 数组
• 在调用 API 前，在本地验证数据

常见必填字段：

• Customer：DisplayName（必须唯一）
• Invoice：CustomerRef、Line（至少一项）
• Payment：TotalAmt、CustomerRef
• Item：Name、Type、IncomeAccountRef（适用于 Service）

OAuth Token 过期（401 Unauthorized）

症状：出现 “invalid_token” 或 “token_expired” 错误

原因：Access token 已过期（3600 秒后）

解决方案：

async function apiCallWithAutoRefresh(apiFunction) {
  try {
    return await apiFunction();
  } catch (error) {
    if (error.response?.status === 401) {
      // Token expired, refresh
      await refreshAccessToken();
      // Retry with new token
      return await apiFunction();
    }
    throw error;
  }
}

预防措施：

• 主动刷新 Token（每 50 分钟一次）
• 存储 Token 到期时间，并在请求前检查
• 自动处理 401 响应

无效引用（错误 3100）

症状：引用 CustomerRef、ItemRef 等时出现 “object not found”

原因：引用的实体不存在或已被删除

解决方案：

def validate_reference(entity_type, entity_id):
    """Verify entity exists before creating reference"""
    try:
        entity = read_entity(entity_type, entity_id)
        return True
    except NotFoundError:
        print(f"{entity_type} {entity_id} not found")
        return False

# Before creating invoice
if validate_reference('Customer', customer_id):
    if validate_reference('Item', item_id):
        create_invoice(customer_id, item_id)

速率限制（429 Too Many Requests）

症状：出现 “throttle_limit_exceeded” 或 429 状态码

原因：超出 API 速率限制

解决方案——带抖动的指数退避：

import time
import random

def api_call_with_backoff(api_function, max_retries=5):
    for attempt in range(max_retries):
        try:
            return api_function()
        except RateLimitError:
            if attempt == max_retries - 1:
                raise

            # Exponential backoff with jitter
            delay = (2 ** attempt) + random.uniform(0, 1)
            print(f"Rate limited, waiting {delay:.1f}s...")
            time.sleep(delay)

预防措施：

• 使用批量操作，减少调用次数
• 实现带速率限制的请求队列
• 缓存频繁访问的引用数据

批量操作失败

症状：批量操作中的部分操作失败，而其他操作成功

原因：每个批量操作彼此独立；一个操作失败不会影响其他操作

解决方案：

function processBatchResults(results) {
  const failed = results.filter(r => r.Fault);
  const succeeded = results.filter(r => !r.Fault);

  console.log(`Batch: ${succeeded.length} success, ${failed.length} failed`);

  // Retry failed operations individually
  for (const failure of failed) {
    console.log(`Retrying ${failure.bId}...`);
    // Implement individual retry logic
  }

  return { succeeded, failed };
}

多币种验证错误

症状：出现 “currency not enabled” 或 “exchange rate required”

原因：未启用多币种功能，或缺少 CurrencyRef

解决方案：

{
  "Invoice": {
    "CurrencyRef": {
      "value": "USD",
      "name": "United States Dollar"
    },
    "ExchangeRate": 1.0
  }
}

检查：

• 确认已在 QuickBooks 公司首选项中启用多币种
• 启用多币种时，务必包含 CurrencyRef
• 对于外币，API 会自动计算汇率

Webhook 未收到通知

症状：已配置 Webhook 端点，但未收到 POST 请求

原因：端点问题、SSL 问题或响应时间过长

解决方案：

1. 确认端点可通过公网访问（不能是 localhost）
2. 使用 HTTPS（Webhook 必须使用 HTTPS）
3. 在 1 秒内响应（立即返回 200 OK，异步处理）
4. 使用示例负载进行测试：

   curl -X POST https://yourapp.com/webhooks/quickbooks \
     -H "Content-Type: application/json" \
     -d '{"eventNotifications":[]}'
   

5. 在开发者控制面板中检查 Webhook 日志

CDC 响应中的已删除实体

症状：状态为 "Deleted" 的实体仅包含 ID

原因：CDC 对已删除实体仅返回最少的数据

解决方案：

def process_cdc_changes(changes):
    for entity in changes:
        if entity.get('status') == 'Deleted':
            # Only ID available
            handle_deletion(entity['Id'])
        else:
            # Full entity data available
            process_entity_update(entity)

---

本技能为 QuickBooks Online API 集成提供了全面的指导。如需获取最新的 API 文档和变更，请使用 Context7，库 ID 为 /websites/developer_intuit_app_developer_qbo。