Donationalerts捐赠数据获取求助：API鉴权失败+WebSocket断连丢数据

一、修复C# API的Unauthorized错误 #

• 添加权限范围参数：Donationalerts的捐赠数据接口需要donations_read权限，在获取token的请求中补充scope字段：

var values = new Dictionary<string, string>
{
    { "client_id", clientId },
    { "client_secret", clientSecret },
    { "grant_type", "client_credentials" },
    { "scope", "donations_read" } // 新增权限范围
};

• 验证Token有效性：解析token响应时，检查token_type是否为Bearer，同时确认expires_in字段的有效期，避免使用过期或类型错误的token。
• 核对后台配置：确认Donationalerts开发者后台的客户端ID、密钥完全正确，且账号已开通访问捐赠数据的权限（部分权限需平台审核）。

二、解决Python WebSocket断连丢数据问题 #

1. 新增心跳机制防止主动断连 #

Donationalerts服务器会因长时间无通信断开连接，定期发送心跳包维持连接：

import time
from threading import Thread

# 心跳发送线程
def send_heartbeat():
    while True:
        if sio.connected:
            sio.emit('ping')
        time.sleep(20) # 每20秒发送一次心跳

@sio.on('connect')
def on_connect():
    sio.emit('add-user', {'token': da_alert_widget_token, "type": "alert_widget"})
    print('Start')
    # 启动心跳线程
    Thread(target=send_heartbeat, daemon=True).start()

2. 断连后补取缺失的捐赠记录 #

记录最后一次收到的捐赠时间戳，重连后调用API获取该时间之后的捐赠：

import requests

last_donation_time = None  # 存储最后一条捐赠的时间戳

@sio.on('donation')
def on_donation(data):
    global last_donation_time
    event = json.loads(data)
    print(f'Donat: {event["username"]} - {event["amount"]} {event["currency"]}\nMessage: {event["message"]}')
    # 更新最后捐赠时间（格式需匹配API返回的时间格式，如ISO 8601）
    last_donation_time = event['created_at']

@sio.on('reconnect')
def on_reconnect():
    global last_donation_time
    print('Reconnect')
    if last_donation_time:
        # 调用API补取断连期间的捐赠
        api_url = f"https://www.donationalerts.com/api/v1/alerts/donations?from={last_donation_time}"
        headers = {'Authorization': f'Bearer {你的API访问令牌}'}
        response = requests.get(api_url, headers=headers)
        if response.status_code == 200:
            donations = response.json()['data']
            for donation in donations:
                print(f'补录Donat: {donation["username"]} - {donation["amount"]} {donation["currency"]}\nMessage: {donation["message"]}')
                if donation['created_at'] > last_donation_time:
                    last_donation_time = donation['created_at']

3. 持久化最后捐赠时间 #

将最后捐赠时间保存到本地文件，程序重启后读取，避免丢失记录节点：

import os

# 读取本地存储的最后时间
def load_last_time():
    if os.path.exists('last_donation.txt'):
        with open('last_donation.txt', 'r') as f:
            return f.read().strip()
    return None

# 保存最后时间到本地
def save_last_time(time_str):
    with open('last_donation.txt', 'w') as f:
        f.write(time_str)

# 初始化时加载本地记录
last_donation_time = load_last_time()

@sio.on('donation')
def on_donation(data):
    global last_donation_time
    event = json.loads(data)
    print(f'Donat: {event["username"]} - {event["amount"]} {event["currency"]}\nMessage: {event["message"]}')
    last_donation_time = event['created_at']
    save_last_time(last_donation_time)

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