Python调用Stripe API仅获取最新交易数据的最优方案咨询

这绝对是更高效、更可靠的方案——直接让Stripe API只返回你上次同步之后产生的新交易，从根源上避免重复数据，不用事后去重。

为什么这个方案更好？ #

• 减少API请求量：不用拉取全量交易，节省带宽和Stripe API调用次数（虽然Stripe免费额度足够，但好习惯要保持）
• 避免本地去重的潜在问题：比如CSV文件被手动修改、格式错乱导致的对比失败
• 逻辑更简洁：不用处理本地数据的读取和对比，只需要跟踪上次同步的时间点

具体实现步骤： #

1. 记录上次同步的时间戳：每次程序运行后，把本次拉取到的最新交易的created时间戳存到一个本地文件（比如last_sync.txt）。如果是第一次运行，就设置一个初始时间（比如程序开发完成的日期，或者你开始需要同步的起始日期）。
2. 用Stripe API的created参数过滤交易：调用stripe.BalanceTransaction.list()时，传入created={'gte': last_sync_timestamp}，这样只会返回指定时间之后创建的交易。
3. 更新同步时间戳：每次拉取完成后，找到本次交易中最大的created时间戳，更新到last_sync.txt里，作为下次同步的起始点。

修正后的代码示例（包含分页处理+时间过滤） #

注意：你的原代码没有处理Stripe列表的分页问题（Stripe API默认只返回10条数据，超过的话需要翻页），我也一并修正了这个问题：

import stripe
import datetime
import csv

# 初始化Stripe（记得替换成你的API密钥）
stripe.api_key = "your_stripe_secret_key"

def get_last_sync_timestamp():
    """读取上次同步的时间戳，没有则返回初始时间（比如2024-01-01）"""
    try:
        with open('last_sync.txt', 'r') as f:
            return int(f.read().strip())
    except FileNotFoundError:
        # 初始时间：2024年1月1日的时间戳
        return datetime.datetime(2024, 1, 1).timestamp()

def update_last_sync_timestamp(new_timestamp):
    """更新上次同步的时间戳到文件"""
    with open('last_sync.txt', 'w') as f:
        f.write(str(int(new_timestamp)))

def get_customer_order_number(charge_id):
    """根据charge ID获取对应的订单号"""
    try:
        charge = stripe.Charge.retrieve(charge_id)
        return charge['metadata'].get('orderNumber', 'No Order Number')
    except stripe.error.StripeError:
        return 'Charge Not Found'

def get_transaction_data():
    last_sync = get_last_sync_timestamp()
    max_created = last_sync  # 记录本次同步的最新时间戳

    # 拉取上次同步之后的所有余额交易（自动处理分页）
    balance_transactions = stripe.BalanceTransaction.list(
        created={'gte': last_sync},
        limit=100  # 每次拉取100条，Stripe允许的最大单次数量
    ).auto_paging_iter()

    with open('stripe_data.csv', 'a', newline='') as csv_file:  # 用追加模式而不是覆盖
        stripe_data = csv.DictWriter(csv_file, fieldnames=['date', 'Description', 'Credit', 'Debit'])
        
        # 如果文件是新创建的，写入表头
        if csv_file.tell() == 0:
            stripe_data.writeheader()

        for transaction in balance_transactions:
            if transaction['description'] == "STRIPE PAYOUT":
                continue

            # 更新最新时间戳
            if transaction['created'] > max_created:
                max_created = transaction['created']

            # 处理交易金额
            amount = transaction['amount']
            charge_id = transaction.get('source')  # 获取关联的charge ID
            order_number = get_customer_order_number(charge_id)
            created_date = datetime.datetime.fromtimestamp(transaction['created']).strftime('%d/%m/%y')
            
            # 写入交易记录（贷方）
            stripe_data.writerow({
                'date': created_date,
                'Description': order_number,
                'Credit': amount / 100,
                'Debit': ''
            })

            # 处理手续费记录（借方）
            for fee in transaction['fee_details']:
                stripe_data.writerow({
                    'date': created_date,
                    'Description': fee['description'],
                    'Credit': '',
                    'Debit': fee['amount'] / 100
                })

    # 更新同步时间戳
    update_last_sync_timestamp(max_created)

# 执行同步
get_transaction_data()

关键改进点说明： #

• 用a模式追加写入CSV，而不是w+覆盖，保留历史数据
• 处理Stripe的分页问题，用auto_paging_iter()自动遍历所有符合条件的交易
• 通过charge_id关联获取对应的订单号，替代原代码中全局变量循环的错误逻辑
• 记录并更新同步时间戳，确保每次只拉新数据

如果因为某些原因无法使用API过滤（比如需要回溯历史数据补全），可以选择这种方案，但效率较低：

实现思路： #

1. 先读取已有的CSV文件，把所有已存在的交易记录（比如通过交易ID、日期+金额组合）存入一个集合
2. 拉取全量交易数据，每条交易生成一个唯一标识（比如transaction['id']，Stripe的交易ID是唯一的）
3. 只有当这个唯一标识不在集合里时，才写入CSV

缺点： #

• 需要拉取全量交易，数据量大时API请求耗时久
• 依赖本地CSV文件的完整性，如果文件被删除或修改，会导致重复数据
• 逻辑比API过滤复杂

内容的提问来源于stack exchange，提问作者SockWhisperer