如何在我的 discord 音乐机器人中添加队列功能？
How can i add a queue function in my discord music bot?
我不知道如何在此处放置队列功能，当我播放另一首正在播放的歌曲时，它会提示我“正在播放音频”错误。
顺便说一句，这是在一个齿轮里面
这是我的播放命令代码：
@commands.command()
async def play(self, ctx, *, url):
if ctx.voice_client is None:
voice_channel = ctx.author.voice.channel
if ctx.author.voice is None:
await ctx.send("`You are not in a voice channel!`")
if (ctx.author.voice):
await voice_channel.connect()
else:
pass
FFMPEG_OPTIONS = {'before_options':'-reconnect 1 -reconnect_streamed 1 -reconnect_delay_max 5', 'options' : '-vn'}
YDL_OPTIONS = {'format':'bestaudio', 'default_search':'auto'}
vc = ctx.voice_client
with youtube_dl.YoutubeDL(YDL_OPTIONS) as ydl:
info = ydl.extract_info(url, download=False)
if 'entries' in info:
url2 = info['entries'][0]['formats'][0]['url']
title = info['entries'][0]['title']
elif 'formats' in info:
url2 = info['formats'][0]['url']
title = info['title']
source = await discord.FFmpegOpusAudio.from_probe(url2, **FFMPEG_OPTIONS)
if ctx.author.voice is None:
await ctx.send("`You are not in a voice channel`")
else:
await ctx.send(f"`Now Playing: {title}`")
vc.play(source)
在任何事情之前，您似乎已经回答了自己的问题？ “排队”。
你所要做的 google 就是“如何在 Python 中制作队列”，这会向你展示如何对 python.[=42= 的标准库进行排队]
现在你需要做的是确定你想要什么类型的队列。
例如，现在假设有三种类型的队列。其中之一是一种队列类型，您可以将所有内容堆叠在一起，然后首先使用顶部的队列，然后一直到底部。如果您想将其形象化，一堆堆叠的盘子可能会有所帮助。当你洗盘子时，你把盘子叠在一起，如果你想要其中一个盘子，你不会从底部开始，因为很难在不打扰顶部的情况下把底部的盘子洗干净。这就是我们所说的 LIFO (last in first out) Queue.
另一种类型的队列是 FIFO (first in first out) Queue，要将其形象化，您需要想象一排人在商店前排队等待 limited game/figurine/item，如果您先去，您将第一个拿东西出去，导致第二个人进来。最后一个人不会是第一个拿东西的人。
还有一种队列叫做Priority Queue，顾名思义就是这个。想象一下我们之前谈论的同一条线路，但除了，你什么时候到达并不重要，有一些贵宾即使你比他们早到达也能抢先获得机会。这种类型的队列在你放东西的时候需要一个迭代器，这个迭代器的第一个元素是一个显示优先级的数字。比方说，我们放 (1, "hello") 然后我们放 (30, "nope")，然后我们放 (15, "wahahah") 然后我们从这个队列中放 get()，它会给出 (1, "hello")，第二个调用 get() 的时间会给出 (15, "wahahah")，最后是 (30, "nope")。它按从低到高的顺序排列值。
毫无疑问，您正在寻找的是 FIFO 队列 first in first out，即您首先播放的歌曲首先播放，然后是其他歌曲。
python 的 queue 标准库有一个 Queue class 提供了这个。类似地，对于 LIFO Queue，有一个 queue.LifoQueue，您可以通过传递 maxsize kwarg 来确定这些队列的大小，默认为 0，这意味着无限项。如果你的 RAM 不足，你会想使用它。
既然您知道什么是队列并且知道队列的基本类型，您想知道如何实现它。
为此，别再看代码了，写点自己想做的事情。或者，可视化用户将做什么以及您希望机器人做什么。
++ User executes play command
++ The bot looks if a `Queue` for the guild.id exists
++ If it does, you add the song to this queue, and notify user "Hey, I
have added your song to the current queue"
-- If it doesn't, you create a new Queue for that guild's ID and then
insert the current song that will be played. Don't play it yet,
otherwise it would be hard to write code in a uniform manner.
++ Now that we have a queue, or at least we know the song has been added, We
check if our voice client is currently playing something or not.
Thankfully, `discord.VoiceClient` offers us a method of `is_playing()`,
this is wonderful as we need this to determine if the voice is currently
playing or not. If it isn't, then we can play the voice, then we will
start a `task` in the background. This is not `ext.tasks` but an
`asyncio.Task`. Think of it as launching a ball in space and never
expecting it return anything, and you move on with your life. That is the
case, until you await it. Awaiting an `asyncio.Task` object will make it
execute on spot, and it will block pretty horribly, we don't want that, so
we will just make it a background task, so it does its thing in the
background.
++ After song is finished, check if our queue of guild.id has any song, if it doesn't, we leave vc notifying user, if it does, we play that, and create a background task again.
从现在开始，无论我说什么都是“我的方法”，我不认为这是最好的方法，但它有效。
所以这是out循环函数的伪逻辑
async def check_play(self, ctx: commands.Context):
get our current voice client
while our voice client exists and it is playing:
sleep for 1 second asynchronously
dispatch an event that tells our bot that a track has ended
现在最后一行涉及内部函数 bot.dispatch。简而言之，这是负责机器人中 'dispatching' “事件”的行。 on_message、on_raw_message、on_reaction_add、on_raw_reaction_add、on_ready 等。这意味着您可以创建 自定义事件 和派遣他们。
我认为上面的段落不够直观，但这里有一个小例子可以提供帮助
bot = commands.Bot(...)
@bot.listen()
async def on_rude(eek: str):
print("Wow the rude person said", eek)
@bot.command()
async def rude(ctx, *, arg):
bot.dispatch('rude', arg)
# on using ?rude what the hell??
# it will print: Wow the rude person said what the hell??
所以现在我们可以利用它来发挥我们的优势。如何调度一个 on_track_end 事件来检查我们的队列是否有更多歌曲，如果有，那么我们播放它，如果没有，我们说我们 运行 没有歌曲并离开VC.
现在结合所有这些，这是代码。
# utils/models.py
from queue import Queue
class Playlist:
def __init__(self, id: int):
self.id = id
self.queue: Queue = Queue(maxsize=0) # maxsize <= 0 means infinite size
def add_song(self, song: str):
self.queue.put(song)
def get_song(self):
return self.queue.get()
def empty_playlist(self):
self.queue.clear()
@property
def is_empty(self):
return self.queue.empty()
@property
def track_count(self):
return self.queue.qsize()
# main.py
import functools
from typing import Dict
import asyncio
import discord
from discord.ext import commands
import youtube_dl
from utils.models import Playlist
class Music(commands.Cog):
def __init__(self, bot):
self.bot = bot
self.playlists: Dict[int, Playlist] = {}
async def check_play(self, ctx: commands.Context):
client = ctx.voice_client
while client and client.is_playing():
await asyncio.sleep(1)
self.bot.dispatch("track_end", ctx)
@commands.command()
async def play(self, ctx: commands.Context, *, url: str):
if ctx.voice_client is None:
voice_channel = ctx.author.voice.channel
if ctx.author.voice is None:
await ctx.send("`You are not in a voice channel!`")
if (ctx.author.voice):
await voice_channel.connect()
else:
pass
FFMPEG_OPTIONS = {'before_options':'-reconnect 1 -reconnect_streamed 1 -reconnect_delay_max 5', 'options' : '-vn'}
YDL_OPTIONS = {'format':'bestaudio', 'default_search':'auto'}
with youtube_dl.YoutubeDL(YDL_OPTIONS) as ydl:
info = ydl.extract_info(url, download=False)
if 'entries' in info:
url2 = info['entries'][0]['formats'][0]['url']
title = info['entries'][0]['title']
elif 'formats' in info:
url2 = info['formats'][0]['url']
title = info['title']
source = await discord.FFmpegOpusAudio.from_probe(url2, **FFMPEG_OPTIONS)
self.bot.dispatch("play_command", ctx, source, title)
@commands.Cog.listener()
async def on_play_command(self, ctx: commands.Context, song, title: str):
playlist = self.playlists.get(ctx.guild.id, Playlist(ctx.guild.id))
self.playlists[ctx.guild.id] = playlist
to_add = (song, title)
playlist.add_song(to_add)
await ctx.send(f"`Added {title} to the playlist.`")
if not ctx.voice_client.is_playing():
self.bot.dispatch("track_end", ctx)
@commands.Cog.listener()
async def on_track_end(self, ctx: commands.Context):
playlist = self.playlists.get(ctx.guild.id)
if playlist and not playlist.is_empty:
song, title = playlist.get_song()
else:
await ctx.send("No more songs in the playlist")
return await ctx.guild.voice_client.disconnect()
await ctx.send(f"Now playing: {title}")
ctx.guild.voice_client.play(song, after=functools.partial(lambda x: self.bot.loop.create_task(self.check_play(ctx))))
# for the above code, instead of functools.partial, you could also create_task on the next line, I just find using the `after` kwargs much better
def setup(bot):
bot.add_cog(Music(bot))
我不知道如何在此处放置队列功能，当我播放另一首正在播放的歌曲时，它会提示我“正在播放音频”错误。
顺便说一句，这是在一个齿轮里面
这是我的播放命令代码：
@commands.command()
async def play(self, ctx, *, url):
if ctx.voice_client is None:
voice_channel = ctx.author.voice.channel
if ctx.author.voice is None:
await ctx.send("`You are not in a voice channel!`")
if (ctx.author.voice):
await voice_channel.connect()
else:
pass
FFMPEG_OPTIONS = {'before_options':'-reconnect 1 -reconnect_streamed 1 -reconnect_delay_max 5', 'options' : '-vn'}
YDL_OPTIONS = {'format':'bestaudio', 'default_search':'auto'}
vc = ctx.voice_client
with youtube_dl.YoutubeDL(YDL_OPTIONS) as ydl:
info = ydl.extract_info(url, download=False)
if 'entries' in info:
url2 = info['entries'][0]['formats'][0]['url']
title = info['entries'][0]['title']
elif 'formats' in info:
url2 = info['formats'][0]['url']
title = info['title']
source = await discord.FFmpegOpusAudio.from_probe(url2, **FFMPEG_OPTIONS)
if ctx.author.voice is None:
await ctx.send("`You are not in a voice channel`")
else:
await ctx.send(f"`Now Playing: {title}`")
vc.play(source)
在任何事情之前，您似乎已经回答了自己的问题？ “排队”。
你所要做的 google 就是“如何在 Python 中制作队列”，这会向你展示如何对 python.[=42= 的标准库进行排队]
现在你需要做的是确定你想要什么类型的队列。
例如，现在假设有三种类型的队列。其中之一是一种队列类型，您可以将所有内容堆叠在一起，然后首先使用顶部的队列，然后一直到底部。如果您想将其形象化，一堆堆叠的盘子可能会有所帮助。当你洗盘子时，你把盘子叠在一起，如果你想要其中一个盘子，你不会从底部开始，因为很难在不打扰顶部的情况下把底部的盘子洗干净。这就是我们所说的 LIFO (last in first out) Queue.
另一种类型的队列是 FIFO (first in first out) Queue，要将其形象化，您需要想象一排人在商店前排队等待 limited game/figurine/item，如果您先去，您将第一个拿东西出去，导致第二个人进来。最后一个人不会是第一个拿东西的人。
还有一种队列叫做Priority Queue，顾名思义就是这个。想象一下我们之前谈论的同一条线路，但除了，你什么时候到达并不重要，有一些贵宾即使你比他们早到达也能抢先获得机会。这种类型的队列在你放东西的时候需要一个迭代器，这个迭代器的第一个元素是一个显示优先级的数字。比方说，我们放 (1, "hello") 然后我们放 (30, "nope")，然后我们放 (15, "wahahah") 然后我们从这个队列中放 get()，它会给出 (1, "hello")，第二个调用 get() 的时间会给出 (15, "wahahah")，最后是 (30, "nope")。它按从低到高的顺序排列值。
毫无疑问，您正在寻找的是 FIFO 队列 first in first out，即您首先播放的歌曲首先播放，然后是其他歌曲。
python 的 queue 标准库有一个 Queue class 提供了这个。类似地，对于 LIFO Queue，有一个 queue.LifoQueue，您可以通过传递 maxsize kwarg 来确定这些队列的大小，默认为 0，这意味着无限项。如果你的 RAM 不足，你会想使用它。
既然您知道什么是队列并且知道队列的基本类型，您想知道如何实现它。
为此，别再看代码了，写点自己想做的事情。或者，可视化用户将做什么以及您希望机器人做什么。
++ User executes play command
++ The bot looks if a `Queue` for the guild.id exists
++ If it does, you add the song to this queue, and notify user "Hey, I
have added your song to the current queue"
-- If it doesn't, you create a new Queue for that guild's ID and then
insert the current song that will be played. Don't play it yet,
otherwise it would be hard to write code in a uniform manner.
++ Now that we have a queue, or at least we know the song has been added, We
check if our voice client is currently playing something or not.
Thankfully, `discord.VoiceClient` offers us a method of `is_playing()`,
this is wonderful as we need this to determine if the voice is currently
playing or not. If it isn't, then we can play the voice, then we will
start a `task` in the background. This is not `ext.tasks` but an
`asyncio.Task`. Think of it as launching a ball in space and never
expecting it return anything, and you move on with your life. That is the
case, until you await it. Awaiting an `asyncio.Task` object will make it
execute on spot, and it will block pretty horribly, we don't want that, so
we will just make it a background task, so it does its thing in the
background.
++ After song is finished, check if our queue of guild.id has any song, if it doesn't, we leave vc notifying user, if it does, we play that, and create a background task again.
从现在开始，无论我说什么都是“我的方法”，我不认为这是最好的方法，但它有效。
所以这是out循环函数的伪逻辑
async def check_play(self, ctx: commands.Context):
get our current voice client
while our voice client exists and it is playing:
sleep for 1 second asynchronously
dispatch an event that tells our bot that a track has ended
现在最后一行涉及内部函数 bot.dispatch。简而言之，这是负责机器人中 'dispatching' “事件”的行。 on_message、on_raw_message、on_reaction_add、on_raw_reaction_add、on_ready 等。这意味着您可以创建 自定义事件 和派遣他们。
我认为上面的段落不够直观，但这里有一个小例子可以提供帮助
bot = commands.Bot(...)
@bot.listen()
async def on_rude(eek: str):
print("Wow the rude person said", eek)
@bot.command()
async def rude(ctx, *, arg):
bot.dispatch('rude', arg)
# on using ?rude what the hell??
# it will print: Wow the rude person said what the hell??
所以现在我们可以利用它来发挥我们的优势。如何调度一个 on_track_end 事件来检查我们的队列是否有更多歌曲，如果有，那么我们播放它，如果没有，我们说我们 运行 没有歌曲并离开VC.
现在结合所有这些，这是代码。
# utils/models.py
from queue import Queue
class Playlist:
def __init__(self, id: int):
self.id = id
self.queue: Queue = Queue(maxsize=0) # maxsize <= 0 means infinite size
def add_song(self, song: str):
self.queue.put(song)
def get_song(self):
return self.queue.get()
def empty_playlist(self):
self.queue.clear()
@property
def is_empty(self):
return self.queue.empty()
@property
def track_count(self):
return self.queue.qsize()
# main.py
import functools
from typing import Dict
import asyncio
import discord
from discord.ext import commands
import youtube_dl
from utils.models import Playlist
class Music(commands.Cog):
def __init__(self, bot):
self.bot = bot
self.playlists: Dict[int, Playlist] = {}
async def check_play(self, ctx: commands.Context):
client = ctx.voice_client
while client and client.is_playing():
await asyncio.sleep(1)
self.bot.dispatch("track_end", ctx)
@commands.command()
async def play(self, ctx: commands.Context, *, url: str):
if ctx.voice_client is None:
voice_channel = ctx.author.voice.channel
if ctx.author.voice is None:
await ctx.send("`You are not in a voice channel!`")
if (ctx.author.voice):
await voice_channel.connect()
else:
pass
FFMPEG_OPTIONS = {'before_options':'-reconnect 1 -reconnect_streamed 1 -reconnect_delay_max 5', 'options' : '-vn'}
YDL_OPTIONS = {'format':'bestaudio', 'default_search':'auto'}
with youtube_dl.YoutubeDL(YDL_OPTIONS) as ydl:
info = ydl.extract_info(url, download=False)
if 'entries' in info:
url2 = info['entries'][0]['formats'][0]['url']
title = info['entries'][0]['title']
elif 'formats' in info:
url2 = info['formats'][0]['url']
title = info['title']
source = await discord.FFmpegOpusAudio.from_probe(url2, **FFMPEG_OPTIONS)
self.bot.dispatch("play_command", ctx, source, title)
@commands.Cog.listener()
async def on_play_command(self, ctx: commands.Context, song, title: str):
playlist = self.playlists.get(ctx.guild.id, Playlist(ctx.guild.id))
self.playlists[ctx.guild.id] = playlist
to_add = (song, title)
playlist.add_song(to_add)
await ctx.send(f"`Added {title} to the playlist.`")
if not ctx.voice_client.is_playing():
self.bot.dispatch("track_end", ctx)
@commands.Cog.listener()
async def on_track_end(self, ctx: commands.Context):
playlist = self.playlists.get(ctx.guild.id)
if playlist and not playlist.is_empty:
song, title = playlist.get_song()
else:
await ctx.send("No more songs in the playlist")
return await ctx.guild.voice_client.disconnect()
await ctx.send(f"Now playing: {title}")
ctx.guild.voice_client.play(song, after=functools.partial(lambda x: self.bot.loop.create_task(self.check_play(ctx))))
# for the above code, instead of functools.partial, you could also create_task on the next line, I just find using the `after` kwargs much better
def setup(bot):
bot.add_cog(Music(bot))