基于Discord.py的定时密钥角色自动增减系统开发需求

核心思路 #

要实现定时移除角色，不能用阻塞式的time.sleep()（会卡住Bot的事件循环），必须用asyncio.sleep()启动异步定时任务。同时要处理持久化——Bot重启后不能丢失未到期的订阅任务，所以需要把用户的订阅信息（到期时间、角色ID）存在JSON文件里，Bot启动时重新加载并恢复定时任务。

步骤1：替换密钥存储格式（从TXT到JSON） #

原来的keys.txt只能存纯密钥，现在需要给每个密钥绑定时长，改用JSON文件存储键值对：

修改密钥创建命令 #

import asyncio
import json
import datetime
import discord
from discord.ext import commands

@bot.slash_command(name="createkey", description="创建带时长的订阅密钥")
@commands.has_role("Litteral Owner")
async def createkey(ctx, key: str, duration_hours: int):
    # 读取现有密钥数据，文件不存在则初始化空字典
    try:
        with open("keys.json", "r", encoding="utf-8") as f:
            keys_data = json.load(f)
    except FileNotFoundError:
        keys_data = {}
    
    # 存储密钥和对应的时长（转成秒）
    keys_data[key] = duration_hours * 3600
    with open("keys.json", "w", encoding="utf-8") as f:
        json.dump(keys_data, f, indent=4, ensure_ascii=False)
    
    await ctx.respond(f"密钥创建成功！订阅时长：{duration_hours}小时", ephemeral=True)

@createkey.error
async def createkeyerror(ctx, error):
    if isinstance(error, commands.MissingRole):
        await ctx.respond("你没有使用该命令的权限（需要Litteral Owner角色）！", ephemeral=True)
    else:
        raise error

步骤2：修改领取密钥逻辑，添加定时任务 #

领取密钥时验证有效性，添加角色后启动异步定时任务，并记录订阅信息到JSON文件：

修改领取密钥命令 #

@bot.slash_command(name="claim", description="领取密钥获取订阅角色")
async def claim(ctx, key: str, member: discord.Member = None):
    # 斜杠命令用ctx.user，而非ctx.message.author
    target_member = member or ctx.user
    guild = ctx.guild
    # 获取目标角色
    role = discord.utils.get(guild.roles, name="Buyer")
    if not role:
        await ctx.respond("服务器内找不到名为Buyer的角色！", ephemeral=True)
        return
    
    # 读取密钥数据
    try:
        with open("keys.json", "r", encoding="utf-8") as f:
            keys_data = json.load(f)
    except FileNotFoundError:
        await ctx.respond("当前没有可用密钥！", ephemeral=True)
        return
    
    # 验证密钥有效性
    if key not in keys_data:
        await ctx.respond("无效密钥！", ephemeral=True)
        return
    
    # 获取订阅时长并移除已使用的密钥
    duration_seconds = keys_data.pop(key)
    with open("keys.json", "w", encoding="utf-8") as f:
        json.dump(keys_data, f, indent=4, ensure_ascii=False)
    
    # 为用户添加角色
    await target_member.add_roles(role)
    await ctx.respond(
        f"密钥验证成功！已为你添加Buyer角色，将在{duration_seconds//3600}小时后自动移除",
        ephemeral=True
    )
    
    # 记录订阅信息到JSON，用于Bot重启后恢复任务
    try:
        with open("subscriptions.json", "r", encoding="utf-8") as f:
            subs_data = json.load(f)
    except FileNotFoundError:
        subs_data = {}
    
    # 计算到期时间（时间戳格式）
    expire_timestamp = (datetime.datetime.now() + datetime.timedelta(seconds=duration_seconds)).timestamp()
    subs_data[str(target_member.id)] = {
        "role_id": str(role.id),
        "expire_time": expire_timestamp
    }
    
    with open("subscriptions.json", "w", encoding="utf-8") as f:
        json.dump(subs_data, f, indent=4, ensure_ascii=False)
    
    # 启动定时移除角色的异步任务
    asyncio.create_task(remove_subscription_role(target_member, role, duration_seconds))

步骤3：定时移除角色的核心函数 #

这个函数会等待指定时长后，检查用户是否还拥有该角色，若有则移除，并更新订阅记录：

async def remove_subscription_role(member: discord.Member, role: discord.Role, duration: int):
    # 等待指定时长（非阻塞，不影响Bot其他功能）
    await asyncio.sleep(duration)
    
    # 检查用户是否仍在服务器且拥有该角色
    if member.guild and role in member.roles:
        await member.remove_roles(role)
        
        # 移除订阅记录
        try:
            with open("subscriptions.json", "r", encoding="utf-8") as f:
                subs_data = json.load(f)
            if str(member.id) in subs_data:
                del subs_data[str(member.id)]
                with open("subscriptions.json", "w", encoding="utf-8") as f:
                    json.dump(subs_data, f, indent=4, ensure_ascii=False)
        except Exception as e:
            print(f"移除订阅记录时出错：{e}")

步骤4：Bot启动时恢复未到期的订阅任务 #

避免Bot重启后丢失正在进行的订阅，启动时读取订阅记录，计算剩余时间并重新启动定时任务：

@bot.event
async def on_ready():
    print(f"已登录为：{bot.user.name}")
    # 替换成你的服务器ID（开启Discord开发者模式后右键服务器可复制）
    target_guild_id = 123456789012345678
    
    try:
        with open("subscriptions.json", "r", encoding="utf-8") as f:
            subs_data = json.load(f)
    except FileNotFoundError:
        subs_data = {}
    
    now_timestamp = datetime.datetime.now().timestamp()
    guild = bot.get_guild(target_guild_id)
    if not guild:
        print("无法连接到目标服务器，跳过订阅恢复")
        return
    
    # 遍历所有订阅记录
    to_remove = []
    for user_id_str, sub_info in subs_data.items():
        user_id = int(user_id_str)
        role_id = int(sub_info["role_id"])
        expire_time = sub_info["expire_time"]
        
        # 获取用户和角色
        try:
            member = await guild.fetch_member(user_id)
            role = guild.get_role(role_id)
        except discord.NotFound:
            to_remove.append(user_id_str)
            continue
        
        if not role:
            to_remove.append(user_id_str)
            continue
        
        # 计算剩余时间
        remaining_time = expire_time - now_timestamp
        if remaining_time > 0:
            # 启动剩余时长的定时任务
            asyncio.create_task(remove_subscription_role(member, role, remaining_time))
            print(f"恢复用户{member.name}的订阅任务，剩余时长：{remaining_time//3600}小时")
        else:
            # 已到期，移除角色并标记删除记录
            if role in member.roles:
                await member.remove_roles(role)
            to_remove.append(user_id_str)
    
    # 清理已过期或无效的订阅记录
    for user_id_str in to_remove:
        del subs_data[user_id_str]
    
    with open("subscriptions.json", "w", encoding="utf-8") as f:
        json.dump(subs_data, f, indent=4, ensure_ascii=False)

注意事项 #

• 替换代码中的target_guild_id为你的服务器实际ID（开启Discord开发者模式后右键服务器可复制）
• 确保Bot拥有管理角色的权限，否则无法添加/移除角色
• JSON文件会自动创建，无需手动初始化
• 若需要支持不同时长单位（比如分钟），只需调整createkey命令中的时长转换逻辑

内容的提问来源于stack exchange，提问作者Meme.rip