如何使用 Smartsheet API 自动采集付款信息？

2026-10-24支付小周

结论

可以自动采集付款结果，但 Smartsheet API 不能直接连接支付网关“收款”。实际部署时，需要增加一个中间服务：

Smartsheet 申请表
        ↓
生成唯一申请编号
        ↓
NIC Inc. / EGOV 完成付款
        ↓
交易通知或付款报告
        ↓
中间服务校验、匹配并去重
        ↓
调用 Smartsheet API 更新申请记录

系统应优先处理每笔交易的实时通知，再用每日付款报告对账，补充遗漏记录。

能否实时接收交易，以及具体使用 webhook、SFTP、API 还是报告下载，取决于所在城市购买的 EGOV 产品和实际配置。需要向城市财务部门或 NIC Inc. 确认，不能根据报告名称猜测接口格式。

为什么需要中间服务

Smartsheet API 可以读取、添加和更新 Smartsheet 中的数据，但它不是支付回调地址，也不会主动解析 EGOV 生成的报告。

Smartsheet webhook 的方向也不同。它会在 Smartsheet 数据发生变化时向外发送通知，不能充当接收支付网关交易通知的通用入口。

因此，需要一个能够运行代码的中间层，例如：

Azure Functions
AWS Lambda
Google Cloud Functions
内部服务器上的定时任务
现有的企业集成平台

中间服务负责接收或下载付款数据，再调用 Smartsheet API 更新记录。

第一步：建立可靠的关联编号

自动更新付款状态时，首要问题不是如何调用 API，而是怎样确认每笔付款对应哪份申请。

建议为每份申请生成唯一的 Application ID，例如：

PM-2026-000123

发起付款时，应将该编号写入 EGOV 支付记录的 merchant reference、order ID 或类似字段。交易完成后，EGOV 返回同一个编号，中间服务便能找到 Smartsheet 中对应的申请。

不要只依赖以下信息进行匹配：

姓名
邮箱地址
付款金额
付款日期

这些字段可能重复，也可能填写错误。只有在 EGOV 无法保存申请编号时，才应考虑组合字段匹配。无法唯一匹配的交易需要转交人工处理。

Smartsheet 中可以设置以下列：

列名	用途
Application ID	申请与付款的唯一关联编号
Applicant Name	申请人
Deposit Expected	应付押金
Payment Status	`Pending`、`Paid`、`Failed`、`Refunded` 等
Amount Paid	实付金额
Transaction ID	支付网关交易编号
Payment Time	交易时间
Reconciliation Status	对账状态
Last Payment Sync	最后同步时间

第二步：确认 EGOV 可以提供什么

需要向 EGOV 或负责支付系统的城市部门确认以下事项：

交易完成后是否提供 HTTP webhook 或 callback。
实时报告通过什么方式交付，例如 SFTP、API、邮件附件或后台下载。
每日汇总报告的文件格式和生成时间。
哪个字段可以保存 Application ID。
是否提供交易状态、退款、撤销和拒付通知。
如何验证通知的真实性，例如数字签名、证书或共享密钥。
同一交易是否可能重复发送。
测试环境与正式环境是否分开。

如果所谓的实时“报告”是每笔交易生成一个文件，就需要监听相应的 SFTP 目录或定时下载文件，而不是建立 webhook。

第三步：取得 Smartsheet 访问凭证

集成程序需要 Smartsheet access token，并通过 HTTP 请求头发送：

Authorization: Bearer YOUR_ACCESS_TOKEN
Content-Type: application/json

正式部署时，应将 token 存放在环境变量或密钥管理服务中，不能写入源代码、日志或 Smartsheet 单元格。

集成程序还需要记录：

目标 sheetId
各列对应的 columnId
申请编号对应的 Smartsheet rowId

columnId 是数字 ID，不是列标题。生产代码也不应假定列顺序永远不变。

第四步：接收交易并更新申请

下面的 Python 示例展示了核心流程。EGOV 的实际字段名、签名算法和通知格式必须按照其接口文档替换。

import os
from datetime import datetime, timezone

import requests
from fastapi import FastAPI, Header, HTTPException, Request

app = FastAPI()

SMARTSHEET_TOKEN = os.environ["SMARTSHEET_TOKEN"]
SMARTSHEET_SHEET_ID = os.environ["SMARTSHEET_SHEET_ID"]

APPLICATION_ID_COLUMN_ID = int(os.environ["APPLICATION_ID_COLUMN_ID"])
PAYMENT_STATUS_COLUMN_ID = int(os.environ["PAYMENT_STATUS_COLUMN_ID"])
AMOUNT_PAID_COLUMN_ID = int(os.environ["AMOUNT_PAID_COLUMN_ID"])
TRANSACTION_ID_COLUMN_ID = int(os.environ["TRANSACTION_ID_COLUMN_ID"])
PAYMENT_TIME_COLUMN_ID = int(os.environ["PAYMENT_TIME_COLUMN_ID"])
LAST_SYNC_COLUMN_ID = int(os.environ["LAST_SYNC_COLUMN_ID"])

SMARTSHEET_BASE_URL = "https://api.smartsheet.com/2.0"

HEADERS = {
    "Authorization": f"Bearer {SMARTSHEET_TOKEN}",
    "Content-Type": "application/json",
}


def verify_egov_signature(body: bytes, signature: str | None) -> bool:
    """
    必须按 EGOV 提供的签名规范实现。
    在签名算法尚未确认前，不能把此函数直接用于生产环境。
    """
    raise NotImplementedError("Implement the official EGOV signature verification")


def find_row_by_application_id(application_id: str) -> int | None:
    response = requests.get(
        f"{SMARTSHEET_BASE_URL}/sheets/{SMARTSHEET_SHEET_ID}",
        headers=HEADERS,
        timeout=30,
    )
    response.raise_for_status()

    sheet = response.json()

    for row in sheet.get("rows", []):
        for cell in row.get("cells", []):
            if (
                cell.get("columnId") == APPLICATION_ID_COLUMN_ID
                and str(cell.get("value", "")).strip() == application_id
            ):
                return row["id"]

    return None


def update_payment_row(
    row_id: int,
    status: str,
    amount: float,
    transaction_id: str,
    payment_time: str,
) -> None:
    payload = [
        {
            "id": row_id,
            "cells": [
                {
                    "columnId": PAYMENT_STATUS_COLUMN_ID,
                    "value": status,
                },
                {
                    "columnId": AMOUNT_PAID_COLUMN_ID,
                    "value": amount,
                },
                {
                    "columnId": TRANSACTION_ID_COLUMN_ID,
                    "value": transaction_id,
                },
                {
                    "columnId": PAYMENT_TIME_COLUMN_ID,
                    "value": payment_time,
                },
                {
                    "columnId": LAST_SYNC_COLUMN_ID,
                    "value": datetime.now(timezone.utc).isoformat(),
                },
            ],
        }
    ]

    response = requests.put(
        f"{SMARTSHEET_BASE_URL}/sheets/{SMARTSHEET_SHEET_ID}/rows",
        headers=HEADERS,
        json=payload,
        timeout=30,
    )
    response.raise_for_status()


@app.post("/payments/egov")
async def receive_payment(
    request: Request,
    x_egov_signature: str | None = Header(default=None),
):
    raw_body = await request.body()

    if not verify_egov_signature(raw_body, x_egov_signature):
        raise HTTPException(status_code=401, detail="Invalid signature")

    event = await request.json()

    # 以下字段仅表示集成所需的数据结构，
    # 必须替换成 EGOV 实际提供的字段名。
    application_id = str(event["application_id"]).strip()
    transaction_id = str(event["transaction_id"]).strip()
    status = str(event["status"]).strip()
    amount = float(event["amount"])
    payment_time = str(event["payment_time"]).strip()

    row_id = find_row_by_application_id(application_id)

    if row_id is None:
        # 生产环境应保存为待人工匹配记录，不能直接丢弃。
        raise HTTPException(status_code=404, detail="Application not found")

    # 此处还应先检查 transaction_id 是否已经处理，防止重复更新。
    update_payment_row(
        row_id=row_id,
        status=status,
        amount=amount,
        transaction_id=transaction_id,
        payment_time=payment_time,
    )

    return {"accepted": True}

安装示例依赖：

pip install fastapi uvicorn requests

本地启动方式：

uvicorn app:app --host 0.0.0.0 --port 8000

支付回调地址必须使用 HTTPS，并部署在 EGOV 可以访问的位置。本地测试地址不能直接用于正式回调。

大量申请时不要遍历整张表

为了便于说明基本逻辑，上面的示例会在每次请求中读取整张 sheet，再查找 Application ID。这种方式可用于数据量较小的场景，但不适合处理大量交易。

正式系统最好在数据库中保存以下映射：

Application ID → Smartsheet rowId
Transaction ID → 处理状态

数据库可以使用 PostgreSQL、SQL Server、DynamoDB 或其他现有数据存储。这样可以快速定位对应行，也能可靠判断一笔交易是否已经处理。

如果不希望增加数据库，至少应缓存 Application ID 和 rowId 的对应关系，并定期重新同步。

每日报告应作为对账来源

即使系统已经接收实时交易，也建议每天处理一次汇总报告。网络异常、服务重启或网关重试失败都可能导致实时通知遗漏。

每日任务可以执行以下检查：

下载当天或前一天的 EGOV 报告。
根据 Transaction ID 去重。
根据 Application ID 查找申请。
比较应付金额和实付金额。
补写遗漏的成功交易。
标记金额不符、重复付款、退款或无法匹配的记录。
输出对账结果，交由财务人员复核。

如果报告采用 CSV 格式，可以先将其解析为统一的数据结构：

import csv
from decimal import Decimal


def read_payment_report(file_path: str):
    with open(file_path, newline="", encoding="utf-8-sig") as csv_file:
        reader = csv.DictReader(csv_file)

        for record in reader:
            yield {
                # 列名必须替换为 EGOV 实际报告中的列名。
                "application_id": record["Application Reference"].strip(),
                "transaction_id": record["Transaction ID"].strip(),
                "status": record["Status"].strip(),
                "amount": Decimal(record["Amount"]),
                "payment_time": record["Payment Time"].strip(),
            }

处理金额时建议使用 Decimal，不要使用二进制浮点数进行财务比较：

from decimal import Decimal

expected = Decimal("250.00")
received = Decimal("250.00")

if expected == received:
    reconciliation_status = "Matched"
else:
    reconciliation_status = "Amount mismatch"

必须处理重复通知和状态变化

支付系统可能重复发送同一事件。如果系统每次收到通知都将其视为一笔新付款，就可能产生错误。

应将 Transaction ID 作为幂等键，并记录以下数据：

transaction_id
application_id
payment_status
amount
event_time
received_time
processed_time
source

交易状态也可能发生变化，例如：

Pending → Paid → Refunded

不能规定记录一旦变为 Paid 就不再更新。应根据 EGOV 的状态定义建立明确的状态映射，并保留原始网关状态，方便后续审计。

安全和合规注意事项

Smartsheet 中只应保存业务需要的付款结果，例如交易编号、金额、时间和状态。

不要保存：

完整银行卡号
CVV/CVC
银行账户密码
支付凭证中的敏感认证信息
EGOV 的共享密钥
Smartsheet access token

即使 EGOV 报告中包含卡号，也要确认卡号已经脱敏，并在写入 Smartsheet 前删除不需要的字段。支付数据还可能受到 PCI DSS、当地隐私制度和城市内部信息安全政策的约束，因此字段设计应由合规和信息安全人员共同参与。

系统还需要采取以下措施：

验证每个 EGOV 通知的签名或来源。
全程使用 HTTPS。
对日志中的姓名、邮箱和交易信息进行脱敏。
为 Smartsheet token 配置最小必要权限。
设置失败重试，并限制重试次数。
保存失败队列，防止数据在没有提示的情况下丢失。
监控 Smartsheet API 的错误响应和请求限制。
定期轮换 token 和支付网关密钥。