-
Notifications
You must be signed in to change notification settings - Fork 33
/
advanced.py
391 lines (311 loc) · 14 KB
/
advanced.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
# type: ignore
"""
This example aims to show the full capabilities of the library.
This is in the form of a drop-in cog you can use and modify to your liking.
This example aims to include everything you would need to make a fully functioning music bot,
from a queue system, advanced queue control and more.
"""
import math
from contextlib import suppress
import discord
from discord.ext import commands
import pomice
class Player(pomice.Player):
"""Custom pomice Player class."""
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.queue = pomice.Queue()
self.controller: discord.Message = None
# Set context here so we can send a now playing embed
self.context: commands.Context = None
self.dj: discord.Member = None
self.pause_votes = set()
self.resume_votes = set()
self.skip_votes = set()
self.shuffle_votes = set()
self.stop_votes = set()
async def do_next(self) -> None:
# Clear the votes for a new song
self.pause_votes.clear()
self.resume_votes.clear()
self.skip_votes.clear()
self.shuffle_votes.clear()
self.stop_votes.clear()
# Check if theres a controller still active and deletes it
if self.controller:
with suppress(discord.HTTPException):
await self.controller.delete()
# Queue up the next track, else teardown the player
try:
track: pomice.Track = self.queue.get()
except pomice.QueueEmpty:
return await self.teardown()
await self.play(track)
# Call the controller (a.k.a: The "Now Playing" embed) and check if one exists
if track.is_stream:
embed = discord.Embed(
title="Now playing",
description=f":red_circle: **LIVE** [{track.title}]({track.uri}) [{track.requester.mention}]",
)
self.controller = await self.context.send(embed=embed)
else:
embed = discord.Embed(
title=f"Now playing",
description=f"[{track.title}]({track.uri}) [{track.requester.mention}]",
)
self.controller = await self.context.send(embed=embed)
async def teardown(self):
"""Clear internal states, remove player controller and disconnect."""
with suppress((discord.HTTPException), (KeyError)):
await self.destroy()
if self.controller:
await self.controller.delete()
async def set_context(self, ctx: commands.Context):
"""Set context for the player"""
self.context = ctx
self.dj = ctx.author
class Music(commands.Cog):
def __init__(self, bot: commands.Bot) -> None:
self.bot = bot
# In order to initialize a node, or really do anything in this library,
# you need to make a node pool
self.pomice = pomice.NodePool()
# Start the node
bot.loop.create_task(self.start_nodes())
async def start_nodes(self):
# Waiting for the bot to get ready before connecting to nodes.
await self.bot.wait_until_ready()
# You can pass in Spotify credentials to enable Spotify querying.
# If you do not pass in valid Spotify credentials, Spotify querying will not work
await self.pomice.create_node(
bot=self.bot,
host="127.0.0.1",
port=3030,
password="youshallnotpass",
identifier="MAIN",
)
print(f"Node is ready!")
def required(self, ctx: commands.Context):
"""Method which returns required votes based on amount of members in a channel."""
player: Player = ctx.voice_client
channel = self.bot.get_channel(int(player.channel.id))
required = math.ceil((len(channel.members) - 1) / 2.5)
if ctx.command.name == "stop":
if len(channel.members) == 3:
required = 2
return required
def is_privileged(self, ctx: commands.Context):
"""Check whether the user is an Admin or DJ."""
player: Player = ctx.voice_client
return player.dj == ctx.author or ctx.author.guild_permissions.kick_members
# The following are events from pomice.events
# We are using these so that if the track either stops or errors,
# we can just skip to the next track
# Of course, you can modify this to do whatever you like
@commands.Cog.listener()
async def on_pomice_track_end(self, player: Player, track, _):
await player.do_next()
@commands.Cog.listener()
async def on_pomice_track_stuck(self, player: Player, track, _):
await player.do_next()
@commands.Cog.listener()
async def on_pomice_track_exception(self, player: Player, track, _):
await player.do_next()
@commands.command(aliases=["joi", "j", "summon", "su", "con", "connect"])
async def join(self, ctx: commands.Context, *, channel: discord.VoiceChannel = None) -> None:
if not channel:
channel = getattr(ctx.author.voice, "channel", None)
if not channel:
return await ctx.send(
"You must be in a voice channel in order to use this command!",
)
# With the release of discord.py 1.7, you can now add a compatible
# VoiceProtocol class as an argument in VoiceChannel.connect().
# This library takes advantage of that and is how you initialize a player.
await ctx.author.voice.channel.connect(cls=Player)
player: Player = ctx.voice_client
# Set the player context so we can use it so send messages
await player.set_context(ctx=ctx)
await ctx.send(f"Joined the voice channel `{channel.name}`")
@commands.command(aliases=["disconnect", "dc", "disc", "lv", "fuckoff"])
async def leave(self, ctx: commands.Context):
if not (player := ctx.voice_client):
return await ctx.send(
"You must have the bot in a channel in order to use this command",
delete_after=7,
)
await player.destroy()
await ctx.send("Player has left the channel.")
@commands.command(aliases=["pla", "p"])
async def play(self, ctx: commands.Context, *, search: str) -> None:
# Checks if the player is in the channel before we play anything
if not (player := ctx.voice_client):
await ctx.author.voice.channel.connect(cls=Player)
player: Player = ctx.voice_client
await player.set_context(ctx=ctx)
# If you search a keyword, Pomice will automagically search the result using YouTube
# You can pass in "search_type=" as an argument to change the search type
# i.e: player.get_tracks("query", search_type=SearchType.ytmsearch)
# will search up any keyword results on YouTube Music
# We will also set the context here to get special features, like a track.requester object
results = await player.get_tracks(search, ctx=ctx)
if not results:
return await ctx.send("No results were found for that search term", delete_after=7)
if isinstance(results, pomice.Playlist):
for track in results.tracks:
player.queue.put(track)
else:
track = results[0]
player.queue.put(track)
if not player.is_playing:
await player.do_next()
@commands.command(aliases=["pau", "pa"])
async def pause(self, ctx: commands.Context):
"""Pause the currently playing song."""
if not (player := ctx.voice_client):
return await ctx.send(
"You must have the bot in a channel in order to use this command",
delete_after=7,
)
if player.is_paused or not player.is_connected:
return
if self.is_privileged(ctx):
await ctx.send("An admin or DJ has paused the player.", delete_after=10)
player.pause_votes.clear()
return await player.set_pause(True)
required = self.required(ctx)
player.pause_votes.add(ctx.author)
if len(player.pause_votes) >= required:
await ctx.send("Vote to pause passed. Pausing player.", delete_after=10)
player.pause_votes.clear()
await player.set_pause(True)
else:
await ctx.send(
f"{ctx.author.mention} has voted to pause the player. Votes: {len(player.pause_votes)}/{required}",
delete_after=15,
)
@commands.command(aliases=["res", "r"])
async def resume(self, ctx: commands.Context):
"""Resume a currently paused player."""
if not (player := ctx.voice_client):
return await ctx.send(
"You must have the bot in a channel in order to use this command",
delete_after=7,
)
if not player.is_paused or not player.is_connected:
return
if self.is_privileged(ctx):
await ctx.send("An admin or DJ has resumed the player.", delete_after=10)
player.resume_votes.clear()
return await player.set_pause(False)
required = self.required(ctx)
player.resume_votes.add(ctx.author)
if len(player.resume_votes) >= required:
await ctx.send("Vote to resume passed. Resuming player.", delete_after=10)
player.resume_votes.clear()
await player.set_pause(False)
else:
await ctx.send(
f"{ctx.author.mention} has voted to resume the player. Votes: {len(player.resume_votes)}/{required}",
delete_after=15,
)
@commands.command(aliases=["n", "nex", "next", "sk"])
async def skip(self, ctx: commands.Context):
"""Skip the currently playing song."""
if not (player := ctx.voice_client):
return await ctx.send(
"You must have the bot in a channel in order to use this command",
delete_after=7,
)
if not player.is_connected:
return
if self.is_privileged(ctx):
await ctx.send("An admin or DJ has skipped the song.", delete_after=10)
player.skip_votes.clear()
return await player.stop()
if ctx.author == player.current.requester:
await ctx.send("The song requester has skipped the song.", delete_after=10)
player.skip_votes.clear()
return await player.stop()
required = self.required(ctx)
player.skip_votes.add(ctx.author)
if len(player.skip_votes) >= required:
await ctx.send("Vote to skip passed. Skipping song.", delete_after=10)
player.skip_votes.clear()
await player.stop()
else:
await ctx.send(
f"{ctx.author.mention} has voted to skip the song. Votes: {len(player.skip_votes)}/{required} ",
delete_after=15,
)
@commands.command()
async def stop(self, ctx: commands.Context):
"""Stop the player and clear all internal states."""
if not (player := ctx.voice_client):
return await ctx.send(
"You must have the bot in a channel in order to use this command",
delete_after=7,
)
if not player.is_connected:
return
if self.is_privileged(ctx):
await ctx.send("An admin or DJ has stopped the player.", delete_after=10)
return await player.teardown()
required = self.required(ctx)
player.stop_votes.add(ctx.author)
if len(player.stop_votes) >= required:
await ctx.send("Vote to stop passed. Stopping the player.", delete_after=10)
await player.teardown()
else:
await ctx.send(
f"{ctx.author.mention} has voted to stop the player. Votes: {len(player.stop_votes)}/{required}",
delete_after=15,
)
@commands.command(aliases=["mix", "shuf"])
async def shuffle(self, ctx: commands.Context):
"""Shuffle the players queue."""
if not (player := ctx.voice_client):
return await ctx.send(
"You must have the bot in a channel in order to use this command",
delete_after=7,
)
if not player.is_connected:
return
if player.queue.qsize() < 3:
return await ctx.send(
"The queue is empty. Add some songs to shuffle the queue.",
delete_after=15,
)
if self.is_privileged(ctx):
await ctx.send("An admin or DJ has shuffled the queue.", delete_after=10)
player.shuffle_votes.clear()
return player.queue.shuffle()
required = self.required(ctx)
player.shuffle_votes.add(ctx.author)
if len(player.shuffle_votes) >= required:
await ctx.send("Vote to shuffle passed. Shuffling the queue.", delete_after=10)
player.shuffle_votes.clear()
player.queue.shuffle()
else:
await ctx.send(
f"{ctx.author.mention} has voted to shuffle the queue. Votes: {len(player.shuffle_votes)}/{required}",
delete_after=15,
)
@commands.command(aliases=["v", "vol"])
async def volume(self, ctx: commands.Context, *, vol: int):
"""Change the players volume, between 1 and 100."""
if not (player := ctx.voice_client):
return await ctx.send(
"You must have the bot in a channel in order to use this command",
delete_after=7,
)
if not player.is_connected:
return
if not self.is_privileged(ctx):
return await ctx.send("Only the DJ or admins may change the volume.")
if not 0 < vol < 101:
return await ctx.send("Please enter a value between 1 and 100.")
await player.set_volume(vol)
await ctx.send(f"Set the volume to **{vol}**%", delete_after=7)
async def setup(bot: commands.Bot):
await bot.add_cog(Music(bot))