使用Python开发Discord Bot如何实现离线仍保存临时Ban到期解封时间？

可选存储方案（按需求选择） #

• 轻量场景选SQLite：Python原生支持，无需额外部署服务，数据存在本地.db文件，适合单服务器小型Bot
• 多服务器/高并发场景选MySQL/PostgreSQL：支持分布式访问，性能更高
• 测试场景可直接用JSON文件存储，但读写并发高时容易损坏，不建议生产环境使用

核心实现步骤 #

1. 改造现有ban命令，新增时长参数，计算封禁到期的UTC时间戳，将封禁记录写入持久化存储
2. Bot启动时自动读取所有未到期的封禁记录，加入定时任务队列
3. 后台运行定时任务，按固定间隔扫描到期的封禁记录，自动执行解封操作并删除对应记录

代码实现示例 #

首先安装依赖：pip install discord.py apscheduler aiosqlite

1. 数据库与定时任务初始化（Bot启动时执行） #

import aiosqlite
import datetime
from apscheduler.schedulers.asyncio import AsyncIOScheduler

scheduler = AsyncIOScheduler(timezone="UTC")

# 启动时创建封禁记录表
async def init_db():
    async with aiosqlite.connect("ban_records.db") as db:
        await db.execute("""
            CREATE TABLE IF NOT EXISTS temp_bans (
                user_id INTEGER NOT NULL,
                guild_id INTEGER NOT NULL,
                end_time INTEGER NOT NULL,
                reason TEXT,
                PRIMARY KEY (user_id, guild_id)
            )
        """)
        await db.commit()
        
        # 加载未到期的封禁任务
        now = int(datetime.datetime.utcnow().timestamp())
        async with db.execute("SELECT user_id, guild_id, end_time FROM temp_bans WHERE end_time > ?", (now,)) as cursor:
            async for row in cursor:
                user_id, guild_id, end_time = row
                run_time = datetime.datetime.fromtimestamp(end_time, tz=datetime.timezone.utc)
                scheduler.add_job(unban_user, "date", run_date=run_time, args=(guild_id, user_id))

# 自动解封函数
async def unban_user(guild_id, user_id):
    guild = client.get_guild(guild_id)
    if not guild:
        return
    user = await client.fetch_user(user_id)
    try:
        await guild.unban(user, reason="临时封禁到期自动解封")
        # 删除已完成的封禁记录
        async with aiosqlite.connect("ban_records.db") as db:
            await db.execute("DELETE FROM temp_bans WHERE user_id = ? AND guild_id = ?", (user_id, guild_id))
            await db.commit()
    except:
        # 处理用户已经被提前解封的异常情况
        pass

2. 改造后的ban命令 #

@client.command()
@commands.has_any_role("Admin", "Server Owner")
# 新增duration参数，格式示例：1d=1天、1h=1小时、1m=1分钟
async def ban(ctx, member : discord.Member, duration: str = None, *, reason=None):
    # 原代码bug修复：原有逻辑是给操作人发消息，实际应该给被封禁用户发
    await member.ban(reason=reason)
    await ctx.channel.send(f'已封禁用户 {member.mention}')
    
    # 永久封禁直接返回，不存记录
    if not duration:
        await member.send(f'你已被服务器 {ctx.guild.name} 永久封禁\n操作人：{ctx.author.mention}\n原因：{reason if reason else "无"}')
        return
    
    # 解析时长，计算到期时间戳
    unit = duration[-1].lower()
    try:
        num = int(duration[:-1])
    except:
        await ctx.send("时长格式错误，示例：1d（1天）、1h（1小时）、1m（1分钟）")
        return
    
    now = datetime.datetime.utcnow()
    if unit == 'm':
        end_time = now + datetime.timedelta(minutes=num)
    elif unit == 'h':
        end_time = now + datetime.timedelta(hours=num)
    elif unit == 'd':
        end_time = now + datetime.timedelta(days=num)
    else:
        await ctx.send("仅支持m（分钟）、h（小时）、d（天）单位")
        return
    
    end_timestamp = int(end_time.timestamp())
    # 写入数据库
    async with aiosqlite.connect("ban_records.db") as db:
        await db.execute("""
            INSERT OR REPLACE INTO temp_bans (user_id, guild_id, end_time, reason)
            VALUES (?, ?, ?, ?)
        """, (member.id, ctx.guild.id, end_timestamp, reason))
        await db.commit()
    
    # 添加定时解封任务
    scheduler.add_job(unban_user, "date", run_date=end_time, args=(ctx.guild.id, member.id))
    await member.send(f'你已被服务器 {ctx.guild.name} 临时封禁\n时长：{duration}\n操作人：{ctx.author.mention}\n原因：{reason if reason else "无"}')

3. Bot启动时触发初始化 #

在启动client的逻辑中添加：

@client.event
async def on_ready():
    await init_db()
    scheduler.start()
    print("Bot已上线")

注意事项 #

• 所有时间统一使用UTC时间，避免服务器时区变化导致解封时间错误
• 可以在定时任务之外增加每小时全表扫描的兜底逻辑，避免个别定时任务遗漏
• 如果Bot服务跨机器部署，建议改用Redis做定时任务+MySQL存数据，保证任务不重复执行

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