如何实现Discord Python机器人跨重启与关机执行延迟任务

嘿，这个问题我之前开发Discord机器人时也遇到过！要让这类延迟任务在机器人重启/关机后还能继续执行，核心就是把任务的关键信息持久化存储下来，等机器人重新启动后，再读取这些信息重新调度未完成的任务。下面给你两个实用的方案，从简单到进阶都有：

这种方案适合小型机器人，不需要额外依赖，实现成本极低。思路是：每次创建延迟任务时，把任务的执行时间、目标频道、要发送的内容等信息写入JSON文件；机器人启动时，读取文件里的任务，检查哪些任务还没到执行时间，重新调度。

示例代码：

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

client = commands.Bot(command_prefix='!', intents=...)

# 存储任务的JSON文件路径
TASK_FILE = "pending_tasks.json"

def save_task(task_data):
    """保存任务到JSON文件"""
    try:
        with open(TASK_FILE, 'r') as f:
            tasks = json.load(f)
    except FileNotFoundError:
        tasks = []
    
    tasks.append(task_data)
    with open(TASK_FILE, 'w') as f:
        json.dump(tasks, f, default=str)  # default=str处理datetime类型

def load_pending_tasks():
    """加载未完成的任务"""
    try:
        with open(TASK_FILE, 'r') as f:
            return json.load(f)
    except FileNotFoundError:
        return []

async def execute_delayed_task(channel_id, message):
    """执行延迟任务，并从文件中移除已完成的任务"""
    channel = client.get_channel(channel_id)
    if channel:
        await channel.send(message)
    
    # 更新任务列表，移除已执行的任务
    with open(TASK_FILE, 'r') as f:
        tasks = json.load(f)
    updated_tasks = [
        t for t in tasks 
        if not (t['channel_id'] == channel_id and t['message'] == message and datetime.datetime.fromisoformat(t['execute_at']) <= datetime.datetime.now())
    ]
    with open(TASK_FILE, 'w') as f:
        json.dump(updated_tasks, f, default=str)

@client.command(pass_context=True)
async def test(ctx):
    delay_seconds = 500
    execute_at = datetime.datetime.now() + datetime.timedelta(seconds=delay_seconds)
    
    # 保存任务信息
    task_data = {
        "type": "test",
        "channel_id": ctx.channel.id,
        "message": "Hello!",
        "execute_at": execute_at.isoformat()
    }
    save_task(task_data)
    
    # 执行原延迟逻辑
    await asyncio.sleep(delay_seconds)
    await ctx.send("Hello!")
    # 任务正常完成后，从文件中移除
    execute_delayed_task(ctx.channel.id, "Hello!")

@client.event
async def on_ready():
    print(f'Logged in as {client.user}')
    # 启动时加载并调度未完成的任务
    pending_tasks = load_pending_tasks()
    now = datetime.datetime.now()
    for task in pending_tasks:
        execute_at = datetime.datetime.fromisoformat(task['execute_at'])
        if execute_at > now:
            # 计算剩余延迟时间
            remaining_seconds = (execute_at - now).total_seconds()
            # 异步调度任务
            asyncio.create_task(
                asyncio.sleep(remaining_seconds) then execute_delayed_task(task['channel_id'], task['message'])
            )
        else:
            # 已经过了执行时间，直接执行
            await execute_delayed_task(task['channel_id'], task['message'])

client.run('YOUR_BOT_TOKEN')

优缺点：

• 优点：零额外依赖，代码简单易上手
• 缺点：任务量多的时候读写效率低，并发场景下可能出现文件冲突

如果你的机器人任务量较大，或者需要更稳定的存储方案，SQLite是个绝佳选择——它是轻量的文件型数据库，无需额外安装服务，支持事务和查询操作。

示例代码：

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

client = commands.Bot(command_prefix='!', intents=...)

def init_db():
    """初始化数据库表"""
    conn = sqlite3.connect('tasks.db')
    c = conn.cursor()
    c.execute('''CREATE TABLE IF NOT EXISTS pending_tasks
                 (id INTEGER PRIMARY KEY AUTOINCREMENT,
                  type TEXT,
                  channel_id INTEGER,
                  message TEXT,
                  execute_at TEXT)''')
    conn.commit()
    conn.close()

def save_task_to_db(task_data):
    """保存任务到数据库"""
    conn = sqlite3.connect('tasks.db')
    c = conn.cursor()
    c.execute('INSERT INTO pending_tasks (type, channel_id, message, execute_at) VALUES (?, ?, ?, ?)',
              (task_data['type'], task_data['channel_id'], task_data['message'], task_data['execute_at']))
    conn.commit()
    conn.close()

def load_pending_tasks_from_db():
    """加载未完成的任务"""
    conn = sqlite3.connect('tasks.db')
    c = conn.cursor()
    c.execute('SELECT * FROM pending_tasks')
    tasks = c.fetchall()
    conn.close()
    
    # 转换为字典格式方便处理
    return [
        {
            "id": t[0],
            "type": t[1],
            "channel_id": t[2],
            "message": t[3],
            "execute_at": t[4]
        }
        for t in tasks
    ]

def remove_task_from_db(task_id):
    """从数据库移除已完成的任务"""
    conn = sqlite3.connect('tasks.db')
    c = conn.cursor()
    c.execute('DELETE FROM pending_tasks WHERE id = ?', (task_id,))
    conn.commit()
    conn.close()

async def execute_db_task(task_id, channel_id, message):
    """执行数据库中的延迟任务"""
    channel = client.get_channel(channel_id)
    if channel:
        await channel.send(message)
    remove_task_from_db(task_id)

@client.command(pass_context=True)
async def test(ctx):
    delay_seconds = 500
    execute_at = datetime.datetime.now() + datetime.timedelta(seconds=delay_seconds)
    
    task_data = {
        "type": "test",
        "channel_id": ctx.channel.id,
        "message": "Hello!",
        "execute_at": execute_at.isoformat()
    }
    save_task_to_db(task_data)
    
    await asyncio.sleep(delay_seconds)
    await ctx.send("Hello!")
    # 任务完成后从数据库删除
    conn = sqlite3.connect('tasks.db')
    c = conn.cursor()
    c.execute('DELETE FROM pending_tasks WHERE channel_id = ? AND message = ? AND execute_at = ?',
              (ctx.channel.id, "Hello!", execute_at.isoformat()))
    conn.commit()
    conn.close()

@client.event
async def on_ready():
    print(f'Logged in as {client.user}')
    init_db()
    pending_tasks = load_pending_tasks_from_db()
    now = datetime.datetime.now()
    for task in pending_tasks:
        execute_at = datetime.datetime.fromisoformat(task['execute_at'])
        if execute_at > now:
            remaining_seconds = (execute_at - now).total_seconds()
            asyncio.create_task(
                asyncio.sleep(remaining_seconds) then execute_db_task(task['id'], task['channel_id'], task['message'])
            )
        else:
            await execute_db_task(task['id'], task['channel_id'], task['message'])

client.run('YOUR_BOT_TOKEN')

优缺点：

• 优点：存储更稳定，支持查询和事务，适合任务量较大的场景
• 缺点：比JSON方案稍微复杂一点，但学习成本很低

• 尽量使用UTC时间存储执行时间，避免本地时区变化导致任务执行时间错误
• 如果任务需要传递复杂参数（比如自定义指令、用户ID等），可以把参数序列化成JSON字符串存储
• 高并发场景下，注意处理数据库/文件的读写锁问题（SQLite本身支持事务，能避免大部分冲突）

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