使用steampy开发Discord CS:GO贴纸价格追踪机器人遇TooManyRequests错误

问题描述 #

我开发了一款追踪CS:GO贴纸市场价格的Discord机器人，使用steampy调用接口获取价格时频繁触发TooManyRequests错误，甚至首次请求也会出现。更换Steam API Key、改用steammarket包都无法解决问题，今日测试时成功获取前两款贴纸价格后，第三款请求直接失败。

相关代码片段：

@client.command()
async def cena(ctx):   
    global steam_client
    f = open('items.json', "r")
    data = json.load(f)
    item_dict = data['items_info']
    for x in item_dict:
        item = x['item']
        quantity = x['quantity']
        cost = x['cost']
        market = steam_client.market.fetch_price(item, game=GameOptions.CS, currency=Currency.EURO)
        print(market)
        if  market == None or market['success'] == False:
            await ctx.send("error") 
            break
        else:      
            lowest = market['lowest_price'].replace("€", "").replace(",", ".")
            ans = f"{item} : **{lowest}€** ||{cost}€ ({quantity})||"
            await ctx.send(ans)
    f.close()
    time.sleep(6)

对应的items.json配置：

{
        "items_info" : [
        { 
            "item" : "Sticker | Spirit (Holo) | 2020 RMR",
            "quantity" : 1,
            "cost" : 0.50,
            "sellat" : 0.65
        },
        {
            "item" : "Sticker | Virtus.pro (Holo) | 2020 RMR",
            "quantity" : 1,
            "cost" : 0.61,
            "sellat" : 0.90
        },
        { 
            "item" : "Sticker | FURIA (Holo) | 2020 RMR",
            "quantity" : 2,
            "cost" : 0.48,
            "sellat" : 0.60
        },
        { 
            "item" : "Sticker | Renegades (Holo) | 2020 RMR",
            "quantity" : 5,
            "cost" : 0.20,
            "sellat" : 0.30
        }
       ]
}

解决方案 #

1. 调整请求间隔适配异步环境 #

Steam市场API对请求频率限制严格，你当前的time.sleep(6)放在循环结束后，对单个请求间隔无作用。要把延迟加到每个请求之间，且用asyncio.sleep适配Discord的异步框架：

import asyncio

@client.command()
async def cena(ctx):   
    global steam_client
    with open('items.json', "r") as f:  # 用with语句自动关闭文件，更安全
        data = json.load(f)
    item_dict = data['items_info']
    for x in item_dict:
        item = x['item']
        quantity = x['quantity']
        cost = x['cost']
        try:
            market = steam_client.market.fetch_price(item, game=GameOptions.CS, currency=Currency.EURO)
            print(market)
            if not market or not market['success']:
                await ctx.send(f"获取{item}价格失败") 
                continue  # 不要直接中断，继续处理其他物品
            lowest = market['lowest_price'].replace("€", "").replace(",", ".")
            ans = f"{item} : **{lowest}€** ||{cost}€ ({quantity})||"
            await ctx.send(ans)
        except Exception as e:
            await ctx.send(f"请求{item}时出错: {str(e)}")
        finally:
            await asyncio.sleep(10)  # 每个请求后延迟10秒，避免触发频率限制

2. 确认Steam客户端登录有效性 #

steampy的fetch_price依赖已登录的会话状态，如果会话过期或初始化错误，会触发异常请求行为导致被限制。确保客户端正确登录：

from steampy.client import SteamClient
from steampy.models import GameOptions, Currency

# 初始化时完成有效登录
steam_client = SteamClient('你的API_KEY')
steam_client.login('你的账号', '你的密码', 'SteamGuard文件路径')

3. 改用公开市场API（无需登录） #

如果登录态请求容易触发限制，可以直接调用Steam公开市场接口，无需登录态：

import requests

def fetch_public_price(item_name, currency=3):  # 3对应欧元
    encoded_name = requests.utils.quote(item_name)
    url = f"https://steamcommunity.com/market/priceoverview/?appid=730&currency={currency}&market_hash_name={encoded_name}"
    response = requests.get(url)
    return response.json() if response.status_code == 200 else None

# 在命令中替换调用
market = fetch_public_price(item)

4. 增加失败重试机制 #

用重试库处理临时的限制错误，遇到TooManyRequests时自动延迟重试：

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
def fetch_price_with_retry(steam_client, item):
    market = steam_client.market.fetch_price(item, game=GameOptions.CS, currency=Currency.EURO)
    if not market or not market['success']:
        raise Exception("价格获取失败")
    return market

# 循环中调用重试函数
market = fetch_price_with_retry(steam_client, item)

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