Python调用Roblox API获取用户数据时持续返回429错误的求助

我尝试通过Roblox API格式化JSON数据，有一个存储用户名的names.txt文件，内容如下：

rip_robson0007
Abobausrip
app_58230
kakoytochelik123
Ameliathebest727
Sherri0708
HixPlayk
mekayla_091
ddddorffg
ghfgrgt7nfdbfj
TheWolfylikedog
paquita12345jeje
hfsgfhsgfhgfhds

文件里的用户名以换行符分隔。我的代码本应读取这些用户名，逐个调用https://api.roblox.com/users/get-by-username?username={name}接口获取JSON数据并格式化，但始终返回429错误，也无法保存任何数据。

我的代码如下：

import json
import requests
import time

# Read the names from the text file
with open("./txt/names.txt", "r") as f:
    names = f.read().split("\n")
    
# Initialize an empty dictionary to store the users
users = {}

# Iterate through the names
for name in names:
    time.sleep(5)
    response = requests.get(f"https://api.roblox.com/users/get-by-username?username={name}")

    # Check the status code of the response
    if response.status_code != 200:
        print(f"Failed to get data for {name}: {response.status_code}")
        continue

    # Try to parse the response as JSON
    try:
        user_data = response.json()
    except ValueError:
        print(f"Failed to parse JSON for {name}")
        continue

    # Extract the necessary information from the response
    user_id = user_data["Id"]
    username = user_data["Username"]
    avatar_uri = user_data["AvatarUri"]
    avatar_final = user_data["AvatarFinal"]
    is_online = user_data["IsOnline"]

    # Add the user's information to the dictionary
    users[user_id] = {
        "Id": user_id,
        "Username": username,
        "AvatarUri": avatar_uri,
        "AvatarFinal": avatar_final,
        "IsOnline": is_online
    }

# Save the dictionary to a JSON file
with open("users.json", "w") as f:
    json.dump(users, f)

429错误代表请求频率超过了Roblox API的限制，即便加了5秒延迟也可能触发，以下是几个可行的修复方案：

1. 遵循API返回的限流提示，动态调整等待时间 #

Roblox返回429时，响应头通常会带Retry-After字段，直接用这个值来等待比固定5秒更可靠：

import json
import requests
import time

with open("./txt/names.txt", "r") as f:
    names = f.read().split("\n")
    
users = {}
session = requests.Session()  # 复用连接，减少开销

for name in names:
    if not name.strip():  # 跳过空行
        continue
        
    response = session.get(f"https://api.roblox.com/users/get-by-username?username={name}")

    if response.status_code == 429:
        retry_after = int(response.headers.get("Retry-After", 10))
        print(f"请求超限，等待{retry_after}秒后重试 {name}")
        time.sleep(retry_after)
        # 重试一次
        response = session.get(f"https://api.roblox.com/users/get-by-username?username={name}")
        if response.status_code != 200:
            print(f"重试后仍失败：{name} - {response.status_code}")
            continue
    elif response.status_code != 200:
        print(f"获取数据失败：{name} - {response.status_code}")
        continue
    
    # 解析数据逻辑
    try:
        user_data = response.json()
    except ValueError:
        print(f"解析JSON失败：{name}")
        continue
    
    user_id = user_data["Id"]
    users[user_id] = {
        "Id": user_id,
        "Username": user_data["Username"],
        "AvatarUri": user_data["AvatarUri"],
        "AvatarFinal": user_data["AvatarFinal"],
        "IsOnline": user_data["IsOnline"]
    }
    
    # 实时保存，避免中途丢失数据
    with open("users.json", "w") as f:
        json.dump(users, f, indent=2)

2. 改用批量请求接口，大幅减少请求次数 #

Roblox提供了批量查询用户的接口，一次可以提交多个用户名，能直接降低触发限流的概率：

import json
import requests
import time

def batch_fetch_users(names, batch_size=10):
    users = {}
    session = requests.Session()
    
    # 按批次分割用户名
    for i in range(0, len(names), batch_size):
        batch = [name.strip() for name in names[i:i+batch_size] if name.strip()]
        if not batch:
            continue
            
        payload = {"usernames": batch, "excludeBannedUsers": True}
        response = session.post("https://users.roblox.com/v1/usernames/users", json=payload)
        
        if response.status_code == 429:
            retry_after = int(response.headers.get("Retry-After", 15))
            print(f"批量请求超限，等待{retry_after}秒")
            time.sleep(retry_after)
            response = session.post("https://users.roblox.com/v1/usernames/users", json=payload)
            if response.status_code != 200:
                print(f"批量重试失败：{response.status_code}")
                continue
        
        if response.status_code != 200:
            print(f"批量请求失败：{response.status_code}")
            continue
            
        data = response.json()
        for user in data.get("data", []):
            # 映射字段，部分字段需自行补充或通过其他接口获取
            users[user["id"]] = {
                "Id": user["id"],
                "Username": user["name"],
                "AvatarUri": f"https://www.roblox.com/Thumbs/Avatar.ashx?x=100&y=100&username={user['name']}",
                "AvatarFinal": True,
                "IsOnline": False
            }
        # 每批次后保存数据
        with open("users.json", "w") as f:
            json.dump(users, f, indent=2)
    return users

# 读取用户名文件
with open("./txt/names.txt", "r") as f:
    names = f.read().split("\n")

batch_fetch_users(names)

3. 增加中途数据保存机制 #

原代码只有全部请求完成后才保存数据，一旦中途出错会丢失已获取的信息。上面的示例已经加入了实时保存的逻辑，确保每获取到一个用户（或一批用户）就立刻写入文件。

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