Discord.py基于SQLite3自定义时间定时推送防重复方案问询

你最初用@tasks.loop()轮询的思路完全可行，不需要引入复杂的调度框架，针对小时级匹配、每日仅推一次的需求，用两层过滤+数据库持久化推送标记的方案就能100%解决重复推送问题，逻辑非常稳。

核心实现逻辑 #

• 轮询间隔设为每10分钟1次即可，不用设太短增加无意义的数据库查询，也不要设为1小时整避免卡整点漏推，同一个小时内多次检查天然带重试能力。
• 所有时间计算统一使用UTC时间，绝对不要用本地时区，避免服务器时区变动、夏令时导致的推送时间错乱。
• 第一层时间匹配：每次轮询触发时，只取当前UTC时间的小时字段，和数据库中各公会配置的推送小时做比对，先筛出配置小时和当前小时一致的公会，缩小待处理范围。
• 第二层防重过滤：在现有SQLite公会配置表新增一个last_push_date字段（TEXT类型，存YYYY-MM-DD格式的日期字符串即可），从第一层筛出的公会里，再过滤掉last_push_date等于当前UTC日期的公会，剩下的就是本次需要推送的目标。
• 推送完成后立刻将对应公会的last_push_date更新为当前UTC日期。哪怕同一小时内轮询再跑多次，因为日期标记已经更新，会直接跳过该公会，完全不会重复推送。等到第二天日期变更后，到了配置的推送小时才会再次触发，天然保证每24小时仅推送一次。

注意：如果推送过程中API调用失败，不要更新last_push_date字段，同一小时内的后续轮询会自动重试，既不会漏推也不会重复发。

参考代码（基于discord.py） #

import sqlite3
from datetime import datetime, timezone
import discord
from discord.ext import tasks, commands

def get_utc_now():
    """统一获取UTC当前时间，所有时间计算都走这个方法避免时区混乱"""
    return datetime.now(timezone.utc)

class DailyQuestion(commands.Cog):
    def __init__(self, bot):
        self.bot = bot
        self.daily_push_loop.start()

    def cog_unload(self):
        self.daily_push_loop.cancel()

    @tasks.loop(minutes=10)
    async def daily_push_loop(self):
        now = get_utc_now()
        current_hour = now.hour
        current_date = now.strftime("%Y-%m-%d")

        # 替换为你自己的数据库路径
        conn = sqlite3.connect("bot_config.db")
        cursor = conn.cursor()
        # 同时匹配小时、校验最后推送日期，一次查询就筛出所有待推送公会
        cursor.execute("""
            SELECT guild_id, push_channel_id FROM guild_config
            WHERE push_hour = ? AND last_push_date != ?
        """, (current_hour, current_date))
        pending_list = cursor.fetchall()

        for guild_id, channel_id in pending_list:
            channel = self.bot.get_channel(channel_id)
            if not channel:
                continue
            try:
                # 替换为你自己的问题生成、发送逻辑
                await channel.send("今日每日一问：xxx")
                # 发送成功才更新推送标记
                cursor.execute("""
                    UPDATE guild_config SET last_push_date = ?
                    WHERE guild_id = ?
                """, (current_date, guild_id))
                conn.commit()
            except Exception as e:
                print(f"公会{guild_id}推送失败，将在本小时内自动重试: {str(e)}")
        conn.close()

    @daily_push_loop.before_loop
    async def loop_prepare(self):
        # 等机器人完全连接成功后再启动轮询，避免空跑报错
        await self.bot.wait_until_ready()

避坑提示 #

• 不要用内存变量存储已推送标记，机器人重启、热重载后内存数据会直接丢失，必然出现重复推送。把推送标记存在SQLite里和公会配置放一起，没有额外依赖，重启也不影响状态。
• 不需要校验分、秒字段，只要日期标记逻辑正常，哪怕轮询在同一个小时内跑6次，也只会在第一次命中时发送一次，后续全部被过滤。
• 给用户展示推送时间的时候，再把UTC时间转换成用户所在时区的时间即可，存储和校验环节全程用UTC能避开90%的时间相关bug。

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