如何在Discord bot中实现歌曲流媒体播放与播放队列管理功能

前置准备 #

先安装需要的依赖：

• 替换已停止维护的youtube-dl为性能更好的yt-dlp：pip install yt-dlp
• 安装/升级discord.py到2.0及以上版本：pip install -U discord.py
• 确保本地已安装FFmpeg并添加到系统环境变量（可直接在终端执行ffmpeg -version验证）

核心改造说明 #

针对你提的4项需求，我们做如下改造：

• 全局创建字典维护每个服务器的独立播放队列，避免多服务器冲突
• 调用yt-dlp直接提取音频流直链，不需要下载到本地即可播放
• 支持关键词搜索：如果传入内容不是链接，自动调用YouTube搜索取第一个匹配结果
• 新增播放完成回调，当前歌曲播放结束后自动从队列取下一首播放
• 播放启动时自动输出指定格式的提示信息

完整实现代码 #

import discord
from discord.ext import commands
import yt_dlp
import asyncio

# 全局变量：存储每个服务器的播放队列，key为服务器ID，value为歌曲列表
guild_queues = {}

# yt-dlp配置，仅提取音频流不下载
ydl_opts = {
    'format': 'bestaudio/best',
    'quiet': True,
    'extract_flat': 'in_playlist',
    'noplaylist': True
}

# FFmpeg流媒体播放配置
ffmpeg_options = {
    'before_options': '-reconnect 1 -reconnect_streamed 1 -reconnect_delay_max 5',
    'options': '-vn'
}

def get_guild_queue(guild_id):
    """获取对应服务器的播放队列，不存在则初始化"""
    if guild_id not in guild_queues:
        guild_queues[guild_id] = []
    return guild_queues[guild_id]

async def play_next(ctx):
    """播放下一首歌曲的回调函数"""
    queue = get_guild_queue(ctx.guild.id)
    voice = discord.utils.get(ctx.bot.voice_clients, guild=ctx.guild)
    if len(queue) > 0:
        next_song = queue.pop(0)
        voice.play(discord.FFmpegPCMAudio(next_song['url'], **ffmpeg_options), after=lambda e: asyncio.run_coroutine_threadsafe(play_next(ctx), ctx.bot.loop))
        await ctx.send(f"Playing {next_song['webpage_url']}")
    else:
        # 队列空了自动断开语音连接
        await voice.disconnect()
        del guild_queues[ctx.guild.id]

@commands.command(name='play')
async def play(ctx, *, query: str):
    # 检查用户是否在语音频道
    if not ctx.author.voice:
        await ctx.send("请先加入一个语音频道")
        return
    
    channel = ctx.author.voice.channel
    voice = discord.utils.get(ctx.bot.voice_clients, guild=ctx.guild)
    
    # 如果bot没加入语音频道就自动加入
    if not voice:
        voice = await channel.connect()
    
    # 处理搜索：如果不是http链接就用关键词搜索
    if not query.startswith('http'):
        query = f'ytsearch:{query}'
    
    # 提取音频流信息
    with yt_dlp.YoutubeDL(ydl_opts) as ydl:
        info = ydl.extract_info(query, download=False)
        # 如果是搜索结果取第一条
        if 'entries' in info:
            info = info['entries'][0]
        audio_url = info['url']
        webpage_url = info['webpage_url']
    
    song_info = {
        'url': audio_url,
        'webpage_url': webpage_url
    }
    
    queue = get_guild_queue(ctx.guild.id)
    # 如果当前没有播放就直接播放，否则加入队列
    if not voice.is_playing() and not voice.is_paused():
        voice.play(discord.FFmpegPCMAudio(audio_url, **ffmpeg_options), after=lambda e: asyncio.run_coroutine_threadsafe(play_next(ctx), ctx.bot.loop))
        await ctx.send(f"Playing {webpage_url}")
    else:
        queue.append(song_info)
        await ctx.send(f"已将歌曲加入队列，当前队列长度：{len(queue)}")

使用说明 #

• 命令调用方式：/play 关键词 或者 /play YouTube链接
• 不需要额外配置本地存储，所有音频均为流媒体播放
• 多服务器使用时队列互相独立，不会出现串歌问题

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