基于Node.js的Discord机器人投票系统：7天后自动统计并执行操作

以下是基于discord.py 2.0+和SQLite的落地实现，覆盖投票创建、数据存储、定时结果处理全流程：

1. 核心依赖与初始化 #

先安装必要依赖：

pip install discord.py apscheduler

初始化数据库，存储投票数据、投票记录和注册用户信息：

import sqlite3
from datetime import datetime, timedelta
import discord
from discord import ui
import asyncio
from apscheduler.schedulers.asyncio import AsyncIOScheduler

def init_db():
    conn = sqlite3.connect('bot.db')
    c = conn.cursor()
    # 投票主表：存储投票核心参数
    c.execute('''CREATE TABLE IF NOT EXISTS votes
                 (id INTEGER PRIMARY KEY AUTOINCREMENT,
                  target_user_id INTEGER NOT NULL,
                  initiator_id INTEGER NOT NULL,
                  agree_count INTEGER DEFAULT 0,
                  disagree_count INTEGER DEFAULT 0,
                  expire_at INTEGER NOT NULL,
                  message_id INTEGER NOT NULL,
                  channel_id INTEGER NOT NULL,
                  guild_id INTEGER NOT NULL,
                  is_processed BOOLEAN DEFAULT 0)''')
    # 投票记录表：防止用户重复投票（跨机器人重启生效）
    c.execute('''CREATE TABLE IF NOT EXISTS vote_records
                 (vote_id INTEGER NOT NULL,
                  user_id INTEGER NOT NULL,
                  PRIMARY KEY (vote_id, user_id),
                  FOREIGN KEY (vote_id) REFERENCES votes(id))''')
    # 注册用户表：存储最终通过投票的用户
    c.execute('''CREATE TABLE IF NOT EXISTS registered_users
                 (user_id INTEGER PRIMARY KEY,
                  registered_at INTEGER NOT NULL)''')
    conn.commit()
    conn.close()

2. 投票按钮组件实现 #

创建带超时逻辑的按钮视图，处理投票操作并实时更新票数：

class VoteView(ui.View):
    def __init__(self, vote_id):
        super().__init__(timeout=7*24*3600)  # 7天后自动禁用按钮
        self.vote_id = vote_id

    @ui.button(label="Agree", style=discord.ButtonStyle.green)
    async def agree_button(self, interaction: discord.Interaction, button: ui.Button):
        await self._process_vote(interaction, "agree")

    @ui.button(label="Disagree", style=discord.ButtonStyle.red)
    async def disagree_button(self, interaction: discord.Interaction, button: ui.Button):
        await self._process_vote(interaction, "disagree")

    async def _process_vote(self, interaction: discord.Interaction, vote_type: str):
        # 检查是否已投票
        conn = sqlite3.connect('bot.db')
        c = conn.cursor()
        c.execute('SELECT 1 FROM vote_records WHERE vote_id = ? AND user_id = ?', (self.vote_id, interaction.user.id))
        if c.fetchone():
            await interaction.response.send_message("你已经投过票了！", ephemeral=True)
            conn.close()
            return
        
        # 更新投票计数
        if vote_type == "agree":
            c.execute('UPDATE votes SET agree_count = agree_count + 1 WHERE id = ?', (self.vote_id,))
        else:
            c.execute('UPDATE votes SET disagree_count = disagree_count + 1 WHERE id = ?', (self.vote_id,))
        # 记录投票用户
        c.execute('INSERT INTO vote_records (vote_id, user_id) VALUES (?, ?)', (self.vote_id, interaction.user.id))
        conn.commit()

        # 更新投票消息的实时票数
        c.execute('SELECT agree_count, disagree_count, message_id, channel_id FROM votes WHERE id = ?', (self.vote_id,))
        agree_num, disagree_num, msg_id, chan_id = c.fetchone()
        conn.close()

        channel = await interaction.client.fetch_channel(chan_id)
        msg = await channel.fetch_message(msg_id)
        embed = msg.embeds[0]
        embed.set_field_at(0, name="同意", value=str(agree_num), inline=True)
        embed.set_field_at(1, name="反对", value=str(disagree_num), inline=True)
        await msg.edit(embed=embed)

        await interaction.response.send_message(f"你的{vote_type}票已记录！", ephemeral=True)

    # 超时后禁用所有按钮
    async def on_timeout(self):
        for child in self.children:
            if isinstance(child, ui.Button):
                child.disabled = True
        # 更新消息按钮状态
        conn = sqlite3.connect('bot.db')
        c = conn.cursor()
        c.execute('SELECT message_id, channel_id FROM votes WHERE id = ?', (self.vote_id,))
        msg_id, chan_id = c.fetchone()
        conn.close()
        try:
            channel = await self.bot.fetch_channel(chan_id)
            msg = await channel.fetch_message(msg_id)
            await msg.edit(view=self)
        except discord.NotFound:
            pass

3. .report命令注册 #

实现触发投票的slash命令，初始化投票数据并发送带按钮的消息：

bot = discord.Bot(intents=discord.Intents.all())

@bot.tree.command(name="report", description="发起用户注册投票，7天后统计结果")
async def report(interaction: discord.Interaction, user: discord.User):
    # 计算7天后的过期时间戳
    expire_timestamp = int((datetime.utcnow() + timedelta(days=7)).timestamp())
    
    # 插入初始投票数据到数据库
    conn = sqlite3.connect('bot.db')
    c = conn.cursor()
    c.execute('''INSERT INTO votes 
                 (target_user_id, initiator_id, expire_at, message_id, channel_id, guild_id)
                 VALUES (?, ?, ?, ?, ?, ?)''',
              (user.id, interaction.user.id, expire_timestamp, 0, interaction.channel_id, interaction.guild_id))
    vote_id = c.lastrowid
    conn.commit()
    conn.close()

    # 创建投票视图并发送消息
    view = VoteView(vote_id)
    view.bot = bot

    embed = discord.Embed(title="用户注册投票",
                          description=f"是否同意将 {user.mention} 注册到数据库？",
                          color=discord.Color.blue())
    embed.add_field(name="同意", value="0", inline=True)
    embed.add_field(name="反对", value="0", inline=True)
    embed.set_footer(text=f"投票将在7天后结束 | 投票ID: {vote_id}")
    
    msg = await interaction.response.send_message(embed=embed, view=view)
    
    # 更新数据库中的消息ID
    conn = sqlite3.connect('bot.db')
    c = conn.cursor()
    c.execute('UPDATE votes SET message_id = ? WHERE id = ?', (msg.id, vote_id))
    conn.commit()
    conn.close()

4. 定时处理投票结果 #

使用APScheduler定期扫描过期投票，执行结果判断与操作：

async def process_expired_vote(vote_id):
    conn = sqlite3.connect('bot.db')
    c = conn.cursor()
    # 查询投票数据（确保未处理）
    c.execute('''SELECT target_user_id, agree_count, disagree_count, channel_id, is_processed
                 FROM votes WHERE id = ?''', (vote_id,))
    result = c.fetchone()
    if not result or result[4]:
        conn.close()
        return
    
    target_id, agree_num, disagree_num, chan_id, _ = result
    # 标记投票为已处理，避免重复执行
    c.execute('UPDATE votes SET is_processed = 1 WHERE id = ?', (vote_id,))
    
    channel = await bot.fetch_channel(chan_id)
    if agree_num > disagree_num:
        # 注册用户到数据库
        c.execute('INSERT OR IGNORE INTO registered_users (user_id, registered_at) VALUES (?, ?)',
                  (target_id, int(datetime.utcnow().timestamp())))
        conn.commit()
        await channel.send(f"投票结束！同意票({agree_num})＞反对票({disagree_num})，已将 <@{target_id}> 注册到数据库。")
    else:
        await channel.send(f"Report Voting Failed：同意票({agree_num})未超过反对票({disagree_num})，注册请求被驳回。")
    conn.close()

def check_expired_votes():
    """定时扫描并处理过期投票"""
    conn = sqlite3.connect('bot.db')
    c = conn.cursor()
    current_time = int(datetime.utcnow().timestamp())
    c.execute('SELECT id FROM votes WHERE expire_at <= ? AND is_processed = 0', (current_time,))
    expired_votes = [row[0] for row in c.fetchall()]
    conn.close()
    
    for vote_id in expired_votes:
        asyncio.create_task(process_expired_vote(vote_id))

@bot.event
async def on_ready():
    print(f"Logged in as {bot.user}")
    init_db()
    # 启动定时任务（每30分钟检查一次过期投票）
    scheduler = AsyncIOScheduler(timezone="UTC")
    scheduler.add_job(check_expired_votes, 'interval', minutes=30)
    scheduler.start()
    # 同步slash命令到服务器
    await bot.tree.sync()

# 替换为你的机器人Token后运行
bot.run("YOUR_BOT_TOKEN")

关键细节说明 #

• 重复投票防护：通过vote_records表持久化记录已投票用户，机器人重启后仍能生效。
• 可靠性保障：使用APScheduler替代asyncio.sleep，避免重启后丢失未处理任务。
• 用户体验优化：投票后实时更新消息票数，超时自动禁用按钮，避免无效操作。
• 幂等性设计：处理结果前先标记投票为已处理，防止重复执行注册或提示逻辑。

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