discord.py等级系统：如何为用户消息添加XP获取冷却机制？

开发Discord等级机器人时发现XP获取存在滥用漏洞：用户可通过频道刷屏刷取XP，轻松登顶排行榜。目前XP获取、排行榜、升级提示功能均正常运行，需要为用户发送消息添加0.5秒冷却时间，以此防止刷分行为。

import json
import discord
from discord.ext import commands

# 假设bot和log_channel_id已提前定义
bot = commands.Bot(command_prefix='!')
log_channel_id = 123456789012345678

@bot.event
async def on_message(message):
    if not message.author.bot:
        # 指定utf-8编码打开文件
        with open('level.json','r', encoding='utf-8') as f:
            users = json.load(f)
        await update_data(users, message.author, message.guild)
        await add_experience(users, message.author, 4, message.guild)
        await level_up(users, message.author, message.channel, message.guild)

        # 指定utf-8编码写入文件
        with open('level.json','w', encoding='utf-8') as f:
            json.dump(users, f)
    await bot.process_commands(message)

async def update_data(users, user, server):
    if not str(server.id) in users:
        users[str(server.id)] = {}
        if not str(user.id) in users[str(server.id)]:
            users[str(server.id)][str(user.id)] = {
                'experience': 0,
                'level': 1
            }
    elif not str(user.id) in users[str(server.id)]:
        users[str(server.id)][str(user.id)] = {
            'experience': 0,
            'level': 1
        }

async def add_experience(users, user, exp, server):
    users[str(user.guild.id)][str(user.id)]['experience'] += exp

async def level_up(users, user, channel, server):
    experience = users[str(user.guild.id)][str(user.id)]['experience']
    lvl_start = users[str(user.guild.id)][str(user.id)]['level']
    lvl_end = int(experience ** (1/4))
    if str(user.guild.id) != '757383943116030074':
        if lvl_start < lvl_end:
            await channel.send(f':fire: {user.mention} 升级到**Level {lvl_end}!**')
            users[str(user.guild.id)][str(user.id)]['level'] = lvl_end
            embed = discord.Embed(title=f':fire: {user.mention} 升级到**Level {lvl_end}!**', color=0x38ff6a)
            await bot.get_channel(log_channel_id).send(embed=embed)

@bot.command(aliases=['rank','lvl'])
async def level(ctx, member: discord.Member = None):
    target = member or ctx.author
    # 指定utf-8编码打开文件
    with open('level.json','r', encoding='utf-8') as f:
        users = json.load(f)
    lvl = users[str(ctx.guild.id)][str(target.id)]['level']
    exp = users[str(ctx.guild.id)][str(target.id)]['experience']

    embed = discord.Embed(title=f'Level {lvl}', description=f"{exp} XP", color=discord.Color.green())
    embed.set_author(name=target.display_name, icon_url=target.avatar.url)
    await ctx.send(embed=embed)

{
  "757383943116030074": {
    "123456789012345678": {
      "experience": 0,
      "level": 1
    }
  }
}

通过全局字典记录用户最后获取XP的时间戳，在on_message事件中判断时间差是否超过0.5秒，仅满足条件时才发放XP。修改后的完整代码如下：

import json
import discord
from discord.ext import commands
import time

# 假设bot和log_channel_id已提前定义
bot = commands.Bot(command_prefix='!')
log_channel_id = 123456789012345678

# 存储用户最后获得XP的时间：{用户ID: 毫秒级时间戳}
xp_cooldowns = {}
# 冷却时间（毫秒）
COOLDOWN_MS = 500

@bot.event
async def on_message(message):
    if not message.author.bot:
        user_id = message.author.id
        current_time = time.time() * 1000  # 转换为毫秒级时间戳
        
        # 检查冷却状态：用户未记录过时间，或已过冷却期
        if user_id not in xp_cooldowns or current_time - xp_cooldowns[user_id] > COOLDOWN_MS:
            # 更新冷却时间
            xp_cooldowns[user_id] = current_time
            
            # 执行原有XP逻辑
            with open('level.json','r', encoding='utf-8') as f:
                users = json.load(f)
            await update_data(users, message.author, message.guild)
            await add_experience(users, message.author, 4, message.guild)
            await level_up(users, message.author, message.channel, message.guild)

            with open('level.json','w', encoding='utf-8') as f:
                json.dump(users, f)
    await bot.process_commands(message)

async def update_data(users, user, server):
    if not str(server.id) in users:
        users[str(server.id)] = {}
        if not str(user.id) in users[str(server.id)]:
            users[str(server.id)][str(user.id)] = {
                'experience': 0,
                'level': 1
            }
    elif not str(user.id) in users[str(server.id)]:
        users[str(server.id)][str(user.id)] = {
            'experience': 0,
            'level': 1
        }

async def add_experience(users, user, exp, server):
    users[str(user.guild.id)][str(user.id)]['experience'] += exp

async def level_up(users, user, channel, server):
    experience = users[str(user.guild.id)][str(user.id)]['experience']
    lvl_start = users[str(user.guild.id)][str(user.id)]['level']
    lvl_end = int(experience ** (1/4))
    if str(user.guild.id) != '757383943116030074':
        if lvl_start < lvl_end:
            await channel.send(f':fire: {user.mention} 升级到**Level {lvl_end}!**')
            users[str(user.guild.id)][str(user.id)]['level'] = lvl_end
            embed = discord.Embed(title=f':fire: {user.mention} 升级到**Level {lvl_end}!**', color=0x38ff6a)
            await bot.get_channel(log_channel_id).send(embed=embed)

@bot.command(aliases=['rank','lvl'])
async def level(ctx, member: discord.Member = None):
    target = member or ctx.author
    with open('level.json','r', encoding='utf-8') as f:
        users = json.load(f)
    lvl = users[str(ctx.guild.id)][str(target.id)]['level']
    exp = users[str(ctx.guild.id)][str(target.id)]['experience']

    embed = discord.Embed(title=f'Level {lvl}', description=f"{exp} XP", color=discord.Color.green())
    embed.set_author(name=target.display_name, icon_url=target.avatar.url)
    await ctx.send(embed=embed)

核心修改点 #

• 引入time模块处理毫秒级时间戳，确保冷却精度
• 新增全局字典xp_cooldowns存储用户最后获取XP的时间
• 在on_message中添加冷却判断逻辑，只有满足条件才执行XP发放流程

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