如何为Discord音乐机器人实现歌曲队列系统与切歌反应交互功能

你的思路完全可行，以下是可直接落地的实现步骤：

1. 全局变量初始化 #

先定义分服务器的存储变量，避免多个服务器使用时数据串扰：

# 存放各服务器的歌曲队列，key为服务器ID，value为歌曲信息列表
guild_queues = {}
# 存放各服务器当前播放的提示消息对象
current_play_msgs = {}
# 存放各服务器当前的反应监听任务，切歌时方便取消
reaction_tasks = {}
# 把公共参数抽出来全局复用
FFMPEG_OPTIONS = {'before_options': '-reconnect 1 -reconnect_streamed 1 -reconnect_delay_max 5', 'options': '-vn'}
YDL_OPTIONS = {'format': 'bestaudio', 'noplaylist':'True'}

2. 封装下一首自动播放函数 #

单独封装play_next异步函数，负责从队列取歌、播放、发送提示消息、绑定播放结束回调：

async def play_next(ctx):
    guild_id = ctx.guild.id
    # 先清空上一首的残留反应监听任务
    if guild_id in reaction_tasks:
        reaction_tasks[guild_id].cancel()
        del reaction_tasks[guild_id]
    # 队列为空直接退出
    if guild_id not in guild_queues or len(guild_queues[guild_id]) == 0:
        await current_play_msgs[guild_id].clear_reactions() if guild_id in current_play_msgs else None
        return
    # 取出队列第一首歌
    song = guild_queues[guild_id].pop(0)
    vc = ctx.voice_client
    # 生成音频源
    source = await discord.FFmpegOpusAudio.from_probe(song['stream_url'], **FFMPEG_OPTIONS)
    # 播放，绑定结束回调，播放完成后自动触发下一首
    vc.play(source, after=lambda e: asyncio.run_coroutine_threadsafe(play_next(ctx), client.loop))
    # 发送播放提示embed
    embed = discord.Embed(title = "Currently Playing", colour = discord.Colour.blue())
    embed.add_field(name = "Song", value = song['title'], inline = False)
    embed.add_field(name = "Length", value = str(datetime.timedelta(seconds = song['length'])), inline = False)
    embed.add_field(name = "Link", value = song['url'], inline = False)
    msg = await ctx.send(embed=embed)
    current_play_msgs[guild_id] = msg
    # 添加控制反应，新增切歌emoji ⏭️
    await msg.add_reaction("\u23F8") # 暂停
    await msg.add_reaction("\u25B6") # 播放
    await msg.add_reaction("\u23F9") # 停止
    await msg.add_reaction("\u23ED") # 切歌
    # 启动反应监听任务
    async def reaction_watcher():
        while True:
            try:
                reaction, user = await client.wait_for("reaction_add", check=lambda reaction, user: user.id == ctx.author.id and reaction.message.id == msg.id and reaction.emoji in ["\u23F8", "\u25B6", "\u23F9", "\u23ED"], timeout = song['length'])
            except asyncio.TimeoutError:
                return await msg.clear_reactions()
            if reaction.emoji == "\u23F8":
                await msg.remove_reaction(reaction.emoji, ctx.author)
                vc.pause()
            elif reaction.emoji == "\u25B6":
                await msg.remove_reaction(reaction.emoji, ctx.author)
                vc.resume()
            elif reaction.emoji == "\u23F9":
                await msg.remove_reaction(reaction.emoji, ctx.author)
                # 停止播放并清空队列
                guild_queues[guild_id] = []
                vc.stop()
                await msg.clear_reactions()
            elif reaction.emoji == "\u23ED":
                await msg.remove_reaction(reaction.emoji, ctx.author)
                # 切歌：停止当前播放，自动触发after回调播放下一首
                vc.stop()
    task = asyncio.create_task(reaction_watcher())
    reaction_tasks[guild_id] = task

3. 修改原play命令逻辑 #

删除原来的直接ctx.voice_client.stop()逻辑，改成先提取歌曲信息，再判断播放状态决定直接播放还是入队：

@client.command()
async def play(ctx, *, url = None):
    guild_id = ctx.guild.id
    # 初始化当前服务器的队列
    if guild_id not in guild_queues:
        guild_queues[guild_id] = []
    if ctx.author.voice is None:
        msg = await ctx.send("You are not in a voice channel.")
        await msg.delete(delay = 3)
        return
    voice_channel = ctx.author.voice.channel
    if ctx.voice_client is None:
        await voice_channel.connect()
    else:
        await ctx.voice_client.move_to(voice_channel)
    vc = ctx.voice_client

    search_keyword = url
    if not ("youtube.com/watch?" in search_keyword):
        search_keyword = search_keyword.replace(" ", "+")
        html = urllib.request.urlopen(f"https://www.youtube.com/results?search_query={search_keyword}")
        video_ids = re.findall(r"watch\?v=(\S{11})", html.read().decode())
        url = str(f"https://www.youtube.com/watch?v={video_ids[0]}")
    
    # 提取歌曲信息
    with youtube_dl.YoutubeDL(YDL_OPTIONS) as ydl:
        info = ydl.extract_info(url, download=False)
        song_info = {
            'title': info.get('title', None),
            'length': info['duration'],
            'url': url,
            'stream_url': info["formats"][0]["url"]
        }
    
    # 判断播放状态
    if not vc.is_playing() and not vc.is_paused():
        # 没有在播放，加入队列后直接播放
        guild_queues[guild_id].append(song_info)
        await play_next(ctx)
    else:
        # 正在播放，加入队列后发送入队提示
        guild_queues[guild_id].append(song_info)
        await ctx.send(f"已将歌曲 {song_info['title']} 加入队列，当前队列位置：{len(guild_queues[guild_id])}")

4. 新增前缀skip命令（可选） #

如果需要支持前缀命令切歌，可以添加以下代码：

@client.command()
async def skip(ctx):
    guild_id = ctx.guild.id
    if ctx.voice_client is None or not ctx.voice_client.is_playing():
        msg = await ctx.send("当前没有正在播放的歌曲")
        await msg.delete(delay=3)
        return
    # 停止当前播放，自动触发下一首
    ctx.voice_client.stop()
    await ctx.send("已跳过当前歌曲")

注意事项 #

• 如果机器人被主动断开语音连接，建议添加事件监听清空对应服务器的队列和缓存，避免残留数据报错
• 可以根据需求新增队列查看、清空队列等功能，直接操作guild_queues[guild_id]列表即可
• youtube_dl已经停止维护，建议替换成yt-dlp，用法完全一致，兼容性更好

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