如何在Discord.py中实现带自动过期与案例编号的警告系统？

存储选型对比 #

直接给你明确几种常见存储方案的差异，按需选择：

• SQLite：轻量文件型数据库，无需额外搭建服务，适合单服务器或小型机器人。支持完整SQL语法，能轻松实现案例编号、过期过滤等需求，开发成本低。
• PostgreSQL/MySQL：客户端-服务器型数据库，适合大型机器人、多服务器集群场景。稳定性和并发处理能力更强，支持数据库层面的定时事件（比如自动删除过期警告），不用在代码里写定时任务。
• JSON/CSV纯文件：不推荐使用。查询指定警告、处理过期数据效率极低，并发操作还容易出现文件损坏，完全满足不了你的需求。

核心功能实现（以SQLite为例） #

1. 数据库初始化 #

先创建存储警告的表，包含自增案例编号、用户ID、服务器ID、警告原因、警告时间、过期时间：

import sqlite3
import time
import discord
from discord.ext import commands, tasks

bot = commands.Bot(command_prefix="!", intents=discord.Intents.all())

def init_db():
    conn = sqlite3.connect('warnings.db')
    cursor = conn.cursor()
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS warnings (
            case_id INTEGER PRIMARY KEY AUTOINCREMENT,
            user_id INTEGER NOT NULL,
            guild_id INTEGER NOT NULL,
            reason TEXT NOT NULL,
            warn_time INTEGER NOT NULL,
            expire_time INTEGER NOT NULL
        )
    ''')
    conn.commit()
    conn.close()

2. 添加警告命令 #

解析目标用户和警告原因，计算30天后的过期时间，插入数据库并返回案例编号：

@bot.command()
async def warn(ctx, member: discord.Member, *, reason: str):
    if not ctx.author.guild_permissions.administrator:
        await ctx.send("你没有权限执行警告操作")
        return
    
    current_ts = int(time.time())
    expire_ts = current_ts + 30 * 24 * 60 * 60  # 换算30天的秒数
    
    conn = sqlite3.connect('warnings.db')
    cursor = conn.cursor()
    cursor.execute('''
        INSERT INTO warnings (user_id, guild_id, reason, warn_time, expire_time)
        VALUES (?, ?, ?, ?, ?)
    ''', (member.id, ctx.guild.id, reason, current_ts, expire_ts))
    case_id = cursor.lastrowid
    conn.commit()
    conn.close()
    
    await ctx.send(
        f"已警告用户 {member.mention}\n"
        f"案例编号：{case_id}\n"
        f"警告原因：{reason}\n"
        "该警告将在30天后自动失效"
    )

3. 自动清理过期警告 #

用discord.py的定时任务，每天检查并删除过期的警告记录：

@tasks.loop(hours=24)
async def clean_expired_warnings():
    current_ts = int(time.time())
    conn = sqlite3.connect('warnings.db')
    cursor = conn.cursor()
    cursor.execute('DELETE FROM warnings WHERE expire_time < ?', (current_ts,))
    deleted_count = cursor.rowcount
    conn.commit()
    conn.close()
    print(f"自动清理了 {deleted_count} 条过期警告")

@bot.event
async def on_ready():
    init_db()
    clean_expired_warnings.start()
    print(f"机器人已登录：{bot.user}")

4. 删除指定警告命令 #

根据案例编号精准删除警告，同时验证权限和案例归属：

@bot.command()
async def deletewarn(ctx, case_id: int):
    if not ctx.author.guild_permissions.administrator:
        await ctx.send("你没有权限删除警告")
        return
    
    conn = sqlite3.connect('warnings.db')
    cursor = conn.cursor()
    
    # 检查案例是否存在且属于当前服务器
    cursor.execute('SELECT guild_id FROM warnings WHERE case_id = ?', (case_id,))
    guild_check = cursor.fetchone()
    if not guild_check:
        await ctx.send(f"未找到案例编号为 {case_id} 的警告")
        conn.close()
        return
    if guild_check[0] != ctx.guild.id:
        await ctx.send("该警告不属于当前服务器")
        conn.close()
        return
    
    cursor.execute('DELETE FROM warnings WHERE case_id = ?', (case_id,))
    conn.commit()
    conn.close()
    await ctx.send(f"已成功删除案例编号为 {case_id} 的警告")

额外注意事项 #

• 所有数据库操作都用参数化查询，避免SQL注入风险；
• 如果用PostgreSQL/MySQL，可以直接在数据库层面设置定时事件自动清理过期数据，不用写Python定时任务；
• 可以扩展查询命令（比如!warns [user]），让管理员查看用户的有效警告记录。

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