UiO-Gaming/ShiteMusicBot
1
0 Fork 0
mirror of https://github.com/UiO-Gaming/ShiteMusicBot.git synced 2026-11-05 13:56:51 +01:00
Code Issues Packages Projects Releases Wiki Activity

Use decorators in place of some checks and repeated code.

Browse Source 
Merge decorator changes to master
This commit is contained in:
Ev-1 authored and GitHub committed 2020-09-17 14:30:02 +02:00
2 parents 9ea88815b1 1a9e16d66f
commit eb76ecf3f4
27 files changed
+1211 -1028
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

.vscode/settings.json
Vendored
+12 -4
View File
@@ -1,15 +1,23 @@
{
"python.languageSever": "Pylance",
"python.linting.enabled": true,
"python.linting.pylintEnabled": true,
"python.linting.pylintEnabled": false,
"python.linting.flake8Enabled": true,
"python.linting.pycodestyleEnabled": true,
"python.linting.pycodestyleEnabled": false,
"[python]": {
"editor.rulers": [120],
},
"[Dockerfile]": {
"editor.formatOnSave": false
},
"editor.tabSize": 4
"editor.tabSize": 4,
"files.exclude": {
"**/.git": true,
"**/.svn": true,
"**/.hg": true,
"**/CVS": true,
"**/.DS_Store": true,
"**/*.pyc": {"when": "$(basename).py"},
"**/__pycache__": true,
},
}
bot.py
+1 -1
View File
@@ -20,7 +20,7 @@ from cogs.utils.settingsmanager import Settings
initial_extensions = [
'cogs.errors',
'cogs.cogs',
'cogs.cogmanager',
'cogs.settings',
'cogs.misc',
'cogs.helpformatter'
cogs/cogs.py → cogs/cogmanager.py
+19 -13
View File
@@ -3,20 +3,26 @@ from discord.ext import commands
import traceback
# Bot Utilities
from cogs.helpformatter import commandhelper
from cogs.utils.paginator import Scroller
class Cogs(commands.Cog):
class CogManager(commands.Cog):
def __init__(self, bot):
self.bot = bot
self.settings = self.bot.settings
@commands.group(name='cogs', hidden=True)
@commands.group(name='cogmanager', hidden=True)
@commands.is_owner()
async def _cogs(self, ctx):
async def _cogmanager(self, ctx):
if ctx.invoked_subcommand is None:
await ctx.invoke(self.bot.get_command('help'),
ctx.command.qualified_name)
ctx.localizer.prefix = 'help' # Ensure the bot looks for locales in the context of help, not cogmanager.
paginator = commandhelper(ctx, ctx.command, ctx.invoker, include_subcmd=True)
scroller = Scroller(ctx, paginator)
await scroller.start_scrolling()
@_cogs.command()
@_cogmanager.command()
@commands.is_owner()
async def load(self, ctx, *, module):
"""Loads a module."""
@@ -26,11 +32,11 @@ class Cogs(commands.Cog):
except Exception:
await ctx.send(f'```py\n{traceback.format_exc()}\n```')
@_cogs.command()
@_cogmanager.command()
@commands.is_owner()
async def unload(self, ctx, *, module):
"""Unloads a module."""
if module == "cogs":
if module == "cogmanager":
return await ctx.send('Unloading this cog is not allowed')
try:
self.bot.unload_extension(f'cogs.{module}')
@@ -38,7 +44,7 @@ class Cogs(commands.Cog):
except Exception:
await ctx.send(f'```py\n{traceback.format_exc()}\n```')
@_cogs.command(name='reload')
@_cogmanager.command(name='reload')
@commands.is_owner()
async def _reload(self, ctx, *, module):
"""Reloads a module."""
@@ -49,13 +55,13 @@ class Cogs(commands.Cog):
except Exception:
await ctx.send(f'```py\n{traceback.format_exc()}\n```')
@_cogs.command(name='reloadall')
@_cogmanager.command(name='reloadall')
@commands.is_owner()
async def _relaod_all(self, ctx):
"""Reloads all extensions"""
try:
for extension in self.bot.extensions:
if extension == 'cogs.cogs':
if extension == 'cogs.cogmanager':
continue
self.bot.unload_extension(f'{extension}')
self.bot.load_extension(f'{extension}')
@@ -63,7 +69,7 @@ class Cogs(commands.Cog):
except Exception:
await ctx.send(f'```py\n{traceback.format_exc()}\n```')
@_cogs.command(name='shutdown')
@_cogmanager.command(name='shutdown')
@commands.is_owner()
async def _shutdown(self, ctx):
"""Logs out and stops."""
@@ -72,4 +78,4 @@ class Cogs(commands.Cog):
def setup(bot):
bot.add_cog(Cogs(bot))
bot.add_cog(CogManager(bot))
cogs/errors.py
+41 -4
View File
@@ -4,6 +4,8 @@ from discord.ext import commands
import traceback
from cogs.music.music_errors import WrongTextChannelError, WrongVoiceChannelError
# Bot Utilities
from cogs.helpformatter import commandhelper
from cogs.utils.paginator import Scroller
@@ -34,9 +36,11 @@ class Errors(commands.Cog):
if isinstance(err, (commands.CommandNotFound)):
return
async def send_error_embed(description):
async def send_error_embed(description, **kwargs):
embed = await self.base_msg(ctx, state=0xFC0303)
embed.description = description
if kwargs.get("title"):
embed.title = kwargs["title"]
embed = ctx.localizer.format_embed(embed)
return await ctx.send(embed=embed)
@@ -51,12 +55,45 @@ class Errors(commands.Cog):
elif (err.original == 'I need the `CONNECT` and `SPEAK` permissions.'):
return await send_error_embed('{errors.need_permission}')
elif (err.original == 'You need to be in the right voice channel'):
return await send_error_embed('{errors.right_channel}')
elif (err.original == 'You need to be in my voicechannel.'):
return await send_error_embed('{errors.my_channel}')
elif (err.original == 'Not playing'):
return await send_error_embed('{not_playing}')
elif (err.original == 'Not listening'):
return await send_error_embed('{have_to_listen}')
elif (err.original == 'The channel is currently full'):
return await send_error_embed('{errors.full_channel}')
if isinstance(err, WrongVoiceChannelError):
changed = False
response = ctx.localizer.format_str('{settings_check.voicechannel}')
for channel_id in err.channels:
channel = ctx.guild.get_channel(channel_id)
if channel is not None:
response += f'{channel.name}, '
changed = True
if changed:
return await send_error_embed(response[:-2], title='{errors.right_channel}')
else:
return await send_error_embed('{errors.right_channel}')
if isinstance(err, WrongTextChannelError):
try:
await ctx.message.delete()
except Exception as e:
self.logger.debug("Error deleting message: %s\nTraceback: %s" % (e, err))
response = ctx.localizer.format_str('{settings_check.textchannel}')
for channel_id in err.channels:
response += f'<#{channel_id}>, '
msg = await send_error_embed(response[:-2], title='{errors.right_channel}')
return await msg.delete(delay=5)
if isinstance(err, commands.CheckFailure):
pass
if isinstance(err, commands.CommandOnCooldown):
await ctx.send(f"{ctx.message.author.mention} Command is on cooldown. "
f"Try again in `{err.retry_after:.1f}` seconds.")
cogs/helpformatter.py
+17 -14
View File
@@ -42,21 +42,24 @@ async def helper(ctx):
async def coghelper(ctx, cog, ignore_subcommands=True):
paginator = HelpPaginator(max_size=5000, max_fields=5, color=ctx.me.color, title=cog.__cog_name__)
for cmd in cog.walk_commands():
if ignore_subcommands:
if len(cmd.qualified_name.split()) > 1:
continue
if cmd.hidden:
continue
try:
can_run = await cmd.can_run(ctx)
except commands.CommandError: # Was 'CommandError' guessed it was supposed to be what it is now
can_run = False
if not can_run:
continue
cmd_dict = get_cmd_dict(ctx, cmd.qualified_name)
if cmd_dict:
paginator.add_command_field(cmd_dict)
if ignore_subcommands:
if len(cmd.qualified_name.split()) > 1:
continue
if cmd.hidden:
continue
try:
can_run = await cmd.can_run(ctx)
except commands.CommandError: # Was 'CommandError' guessed it was supposed to be what it is now
can_run = False
if not can_run:
continue
cmd_dict = get_cmd_dict(ctx, cmd.qualified_name)
if cmd_dict:
paginator.add_command_field(cmd_dict)
except Exception as e:
print(e)
print("Help failed for command: ", cmd)
paginator.add_page_indicator(ctx.localizer, "{pageindicator}", _prefix=ctx.prefix)
return paginator
cogs/music.py
-832
View File
@@ -1,832 +0,0 @@
"""
Music commands
"""
# Discord Packages
import discord
import lavalink
from discord.ext import commands
import asyncio
import json
import math
import re
import urllib
from typing import Optional
from bs4 import BeautifulSoup
from .utils import checks, thumbnailer, timeformatter
from .utils.paginator import QueuePaginator, Scroller, TextPaginator
from .utils.selector import Selector
time_rx = re.compile('[0-9]+')
url_rx = re.compile('https?:\\/\\/(?:www\\.)?.+')
class Music(commands.Cog):
def __init__(self, bot):
self.bot = bot
self.logger = self.bot.main_logger.bot_logger.getChild("Music")
async def cog_check(self, ctx):
if not ctx.guild:
raise commands.NoPrivateMessage
textchannels = self.bot.settings.get(ctx.guild, 'channels.text', [])
if textchannels:
if ctx.channel.id not in textchannels:
return False
return True
async def cog_before_invoke(self, ctx):
await self.ensure_voice(ctx)
async def connect_to(self, guild_id: int, channel_id: str):
""" Connects to the given voicechannel ID. A channel_id of `None` means disconnect. """
ws = self.bot._connection._get_websocket(guild_id)
await ws.voice_state(str(guild_id), channel_id)
@commands.command(name='play')
async def _play(self, ctx, *, query: str):
""" Searches and plays a song from a given query. """
player = self.bot.lavalink.player_manager.get(ctx.guild.id)
query = query.strip('<>')
if not url_rx.match(query):
query = f'ytsearch:{query}'
results = await player.node.get_tracks(query)
if not results or not results['tracks']:
return await ctx.send(ctx.localizer.format_str("{nothing_found}"))
embed = discord.Embed(color=ctx.me.color)
if results['loadType'] == 'PLAYLIST_LOADED':
tracks = results['tracks']
maxlength = self.max_track_length(ctx.guild, player)
if maxlength:
numtracks = 0
for track in tracks:
if track['info']['length'] <= maxlength:
thumbnail_url = await thumbnailer.ThumbNailer.identify(
self, track['info']['identifier'], track['info']['uri'])
track = lavalink.models.AudioTrack(track, ctx.author.id, thumbnail_url=thumbnail_url)
player.add(requester=ctx.author.id, track=track)
numtracks += 1
else:
numtracks = len(tracks)
for track in tracks:
thumbnail_url = await thumbnailer.ThumbNailer.identify(
self, track['info']['identifier'], track['info']['uri'])
track = lavalink.models.AudioTrack(track, ctx.author.id, thumbnail_url=thumbnail_url)
player.add(requester=ctx.author.id, track=track)
embed.title = '{playlist_enqued}'
embed.description = f'{results["playlistInfo"]["name"]} - {numtracks} {{tracks}}'
embed = ctx.localizer.format_embed(embed)
await ctx.send(embed=embed)
else:
track = results['tracks'][0]
await self.enqueue(ctx, track, embed)
embed = ctx.localizer.format_embed(embed)
await ctx.send(embed=embed)
if not player.is_playing:
await player.play()
@commands.command(name='seek')
@checks.dj_or(alone=True)
async def _seek(self, ctx, *, time: str):
""" Seeks to a given position in a track. """
player = self.bot.lavalink.player_manager.get(ctx.guild.id)
if not player.is_playing:
return await ctx.send(ctx.localizer.format_str("{not_playing}"))
if ctx.author not in player.listeners:
return await ctx.send(ctx.localizer.format_str("{have_to_listen}"))
seconds = time_rx.search(time)
if not seconds:
return await ctx.send(ctx.localizer.format_str("{seek.missing_amount}"))
seconds = int(seconds.group()) * 1000
if time.startswith('-'):
seconds *= -1
track_time = player.position + seconds
await player.seek(track_time)
msg = ctx.localizer.format_str("{seek.track_moved}", _position=timeformatter.format_ms(track_time))
await ctx.send(msg)
@commands.command(name='skip')
async def _skip(self, ctx):
""" Skips the current track. """
player = self.bot.lavalink.player_manager.get(ctx.guild.id)
if not player.is_playing:
return await ctx.send(ctx.localizer.format_str("{not_playing}"))
player.add_skipper(ctx.author)
total = len(player.listeners)
skips = len(player.skip_voters)
threshold = self.bot.settings.get(ctx.guild, 'vote_threshold', 'default_threshold')
if skips/total >= threshold/100 or player.current.requester == ctx.author.id:
await player.skip()
if player.current:
song = f'**[{player.current.title}]({player.current.uri})**'
embed = discord.Embed(color=ctx.me.color, description=song, title='{now}')
thumbnail_url = player.current.extra["thumbnail_url"]
member = ctx.guild.get_member(player.current.requester)
if thumbnail_url:
embed.set_thumbnail(url=thumbnail_url)
if member.nick:
member_identifier = member.nick
else:
member_identifier = member.name
embed.set_footer(text=f'{{requested_by}} {member_identifier}', icon_url=member.avatar_url)
embed = ctx.localizer.format_embed(embed)
await ctx.send(ctx.localizer.format_str("{skip.skipped}"), embed=embed)
else:
await ctx.send(ctx.localizer.format_str("{skip.skipped}"))
else:
if skips != 0:
needed = math.ceil(total*threshold/100)
msg = ctx.localizer.format_str("{skip.require_vote}", _skips=skips, _total=needed)
await ctx.send(msg)
@commands.command(name='skipto')
@checks.dj_or(alone=True)
async def _skip_to(self, ctx, pos: int = 1):
""" Plays the queue from a specific point. Disregards tracks before the pos. """
player = self.bot.lavalink.player_manager.get(ctx.guild.id)
if ctx.author not in player.listeners:
return await ctx.send(ctx.localizer.format_str("{have_to_listen}"))
if pos < 1:
return await ctx.send(ctx.localizer.format_str("{skip_to.invalid_pos}"))
if len(player.queue) < pos:
return await ctx.send(ctx.localizer.format_str("{skip_to.exceeds_queue}"))
await player.skip(pos - 1)
msg = ctx.localizer.format_str("{skip_to.skipped_to}", _title=player.current.title, _pos=pos)
await ctx.send(msg)
@commands.command(name='stop')
@checks.dj_or(alone=True)
async def _stop(self, ctx):
""" Stops the player and clears its queue. """
player = self.bot.lavalink.player_manager.get(ctx.guild.id)
if ctx.author not in player.listeners:
return await ctx.send(ctx.localizer.format_str("{have_to_listen}"))
if not player.is_playing:
return await ctx.send(ctx.localizer.format_str("{not_playing}"))
player.queue.clear()
await player.stop()
embed = discord.Embed(color=ctx.me.color, title='{stop}')
await ctx.send(embed=ctx.localizer.format_embed(embed))
@commands.command(name='now')
async def _now(self, ctx):
""" Shows some stats about the currently playing song. """
player = self.bot.lavalink.player_manager.get(ctx.guild.id)
if not player.current:
return await ctx.send(ctx.localizer.format_str("{not_playing}"))
position = timeformatter.format_ms(player.position)
if player.current.stream:
duration = '{live}'
else:
duration = timeformatter.format_ms(player.current.duration)
song = f'**[{player.current.title}]({player.current.uri})**\n({position}/{duration})'
member = ctx.guild.get_member(player.current.requester)
embed = discord.Embed(color=ctx.me.color, description=song, title='{now}')
thumbnail_url = player.current.extra["thumbnail_url"]
if thumbnail_url:
embed.set_thumbnail(url=thumbnail_url)
if member.nick:
member_identifier = member.nick
else:
member_identifier = member.name
embed.set_footer(text=f'{{requested_by}} {member_identifier}', icon_url=member.avatar_url)
embed = ctx.localizer.format_embed(embed)
await ctx.send(embed=embed)
@commands.command(name='queue')
async def _queue(self, ctx, user: discord.Member = None):
""" Shows the player's queue. """
player = self.bot.lavalink.player_manager.get(ctx.guild.id)
if player.queue.empty:
embed = discord.Embed(description='{queue.empty}', color=ctx.me.color)
embed = ctx.localizer.format_embed(embed)
return await ctx.send(embed=embed)
if user is None:
queue = player.global_queue()
pagified_queue = QueuePaginator(ctx.localizer, queue, color=ctx.me.color)
else:
user_queue = player.user_queue(user.id, dual=True)
if not user_queue:
return await ctx.send(ctx.localizer.format_str("{queue.empty}", _user=user.name))
pagified_queue = QueuePaginator(ctx.localizer, user_queue, color=ctx.me.color, user_name=user.name)
scroller = Scroller(ctx, pagified_queue)
await scroller.start_scrolling()
@commands.command(name='myqueue')
async def _myqueue(self, ctx):
""" Shows your queue. """
player = self.bot.lavalink.player_manager.get(ctx.guild.id)
user_queue = player.user_queue(ctx.author.id, dual=True)
if not user_queue:
return await ctx.send(ctx.localizer.format_str("{my_queue}"))
pagified_queue = QueuePaginator(ctx.localizer, user_queue, color=ctx.me.color, user_name=ctx.author.name)
scroller = Scroller(ctx, pagified_queue)
await scroller.start_scrolling()
@commands.command(name='pause')
@checks.dj_or(alone=True)
async def _pause(self, ctx):
""" Pauses/Resumes the current track. """
player = self.bot.lavalink.player_manager.get(ctx.guild.id)
if not player.is_playing:
return await ctx.send(ctx.localizer.format_str("{not_playing}"))
if player.paused:
await player.set_pause(False)
await ctx.send(ctx.localizer.format_str("{resume.resumed}"))
else:
await player.set_pause(True)
await ctx.send(ctx.localizer.format_str("{resume.paused}"))
@commands.command(name='shuffle')
async def _shuffle(self, ctx):
""" Shuffles your queue. """
player = self.bot.lavalink.player_manager.get(ctx.guild.id)
user_queue = player.user_queue(ctx.author.id)
if not user_queue:
return await ctx.send(ctx.localizer.format_str("{my_queue}"))
player.shuffle_user_queue(ctx.author.id)
await ctx.send(ctx.localizer.format_str("{shuffle}"))
@commands.command(name='move')
async def _move(self, ctx):
""" Moves a song in your queue. """
player = self.bot.lavalink.player_manager.get(ctx.guild.id)
if player.queue.empty:
return
user_queue = player.user_queue(ctx.author.id, dual=True)
if not user_queue:
return await ctx.send(ctx.localizer.format_str("{my_queue}"))
# Setup for selector
identifiers = []
for index, temp in enumerate(user_queue):
track, globpos = temp
identifiers.append(ctx.localizer.format_str("{queue.usertrack}", _index=index+1,
_globalindex=globpos+1, _title=track.title,
_uri=track.uri))
def ret_first_arg(*args):
return args[0]
functions = [ret_first_arg]*len(user_queue)
arguments = [(i,) for i in range(len(user_queue))]
# Set the different titles for the different selections
round_titles = ["{moved.choose_song}", "{moved.choose_pos}"]
round_titles = [ctx.localizer.format_str(i) for i in round_titles]
selector = Selector(ctx, identifiers, functions, arguments, num_selections=5, round_titles=round_titles,
color=ctx.me.color, title=ctx.localizer.format_str("{moved.choose}"))
# Get the initial and final position
message, page, selections = await selector.start_scrolling()
pos_initial, pos_final = selections[0], selections[1]
# Move the track
moved = player.move_user_track(ctx.author.id, pos_initial, pos_final)
# Create a nice embed explaining what happened
song = f'**[{moved.title}]({moved.uri})**'
embed = discord.Embed(color=ctx.me.color, description=song, title='{moved.moved}')
thumbnail_url = moved.extra["thumbnail_url"]
member = ctx.guild.get_member(moved.requester)
if thumbnail_url:
embed.set_thumbnail(url=thumbnail_url)
if member.nick:
member_identifier = member.nick
else:
member_identifier = member.name
embed.add_field(name="{moved.local}", value=f"`{pos_initial + 1} → {pos_final + 1}`", inline=True)
embed.add_field(name="{moved.global}", value=f"`{player.queue._loc_to_glob(ctx.author.id, pos_initial) + 1}\
→ {player.queue._loc_to_glob(ctx.author.id, pos_final) + 1}`", inline=True)
embed.set_footer(text=f'{{requested_by}} {member_identifier}', icon_url=member.avatar_url)
embed = ctx.localizer.format_embed(embed)
await message.edit(embed=embed)
await message.clear_reactions()
@commands.command(name='remove')
async def _remove(self, ctx):
player = self.bot.lavalink.player_manager.get(ctx.guild.id)
if player.queue.empty:
return
user_queue = player.user_queue(ctx.author.id, dual=True)
if not user_queue:
return await ctx.send(ctx.localizer.format_str("{my_queue}"))
identifiers = []
for index, temp in enumerate(user_queue):
track, globpos = temp
identifiers.append(ctx.localizer.format_str("{queue.usertrack}", _index=index+1,
_globalindex=globpos+1, _title=track.title,
_uri=track.uri))
functions = [player.remove_user_track]*len(user_queue)
arguments = [(ctx.author.id, i) for i in range(len(user_queue))]
selector = Selector(ctx, identifiers, functions, arguments, num_selections=5, color=0xFF0000,
title='Remove song plz')
message, page, removed = await selector.start_scrolling()
await message.edit(content=ctx.localizer.format_str("{remove}", _title=removed.title), embed=None)
await message.clear_reactions()
@commands.command(name="DJremove")
@checks.dj_or()
async def _djremove(self, ctx, pos: int, user: discord.Member = None):
""" Remove a song from either the global queue or a users queue"""
player = self.bot.lavalink.player_manager.get(ctx.guild.id)
if user is None:
if player.queue.empty:
return
if pos > len(player.queue) or pos < 1:
return await ctx.send(ctx.localizer.format_str("{out_of_range}", _len=len(player.queue)))
removed = player.remove_global_track(pos - 1)
requester = self.bot.get_user(removed.requester)
await ctx.send(ctx.localizer.format_str("{dj_removed}", _title=removed.title, _user=requester.name))
else:
if player.queue.empty:
return
user_queue = player.user_queue(user.id)
if not user_queue:
return
if pos > len(user_queue) or pos < 1:
return await ctx.send(ctx.localizer.format_str("{out_of_range}", _len=len(user_queue)))
removed = player.remove_user_track(user.id, pos - 1)
await ctx.send(ctx.localizer.format_str("{dj_removed}", _title=removed.title, _user=removed.requester.name))
@commands.command(name='removeuser')
@checks.dj_or(alone=True)
async def _user_queue_remove(self, ctx, user: discord.Member):
""" Remove a song from either the global queue or a users queue"""
player = self.bot.lavalink.player_manager.get(ctx.guild.id)
if player.queue.empty:
return
user_queue = player.user_queue(user.id)
if not user_queue:
return
player.remove_user_queue(user.id)
embed = discord.Embed(color=ctx.me.color)
embed.description = "{dj_remove_user}"
embed = ctx.localizer.format_embed(embed, _id=user.id)
await ctx.send(embed=embed)
@commands.command(name='search')
async def _search(self, ctx, *, query):
player = self.bot.lavalink.player_manager.get(ctx.guild.id)
if not query.startswith('ytsearch:') and not query.startswith('scsearch:'):
query = 'ytsearch:' + query
results = await player.node.get_tracks(query)
embed = discord.Embed(description='{nothing_found}', color=0x36393F)
if not results or not results['tracks']:
embed = ctx.localizer.format_embed(embed)
return await ctx.send(embed=embed)
tracks = results['tracks']
result_count = min(len(tracks), 15)
# Setup for the selector
identifiers = []
functions = [self.enqueue]*result_count
arguments = [(ctx, tracks[i], discord.Embed(color=ctx.me.color)) for i in range(result_count)]
for index, track in enumerate(tracks, start=1):
track_title = track['info']['title']
track_uri = track['info']['uri']
duration = timeformatter.format_ms(int(track['info']['length']))
identifiers.append(f'`{index}.` [{track_title}]({track_uri}) `{duration}`')
if index == result_count:
break
search_selector = Selector(ctx, identifiers, functions, arguments, num_selections=5,
color=ctx.me.color, title=ctx.localizer.format_str('{results}'))
# Let the user scroll through results
message, current_page, result = await search_selector.start_scrolling()
result = ctx.localizer.format_embed(result)
await message.edit(embed=result)
await message.clear_reactions()
if not player.is_playing:
await player.play()
@commands.command(name='disconnect')
@checks.dj_or(alone=True)
async def _disconnect(self, ctx):
""" Disconnects the player from the voice channel and clears its queue. """
player = self.bot.lavalink.player_manager.get(ctx.guild.id)
if not player.is_connected:
return await ctx.send(ctx.localizer.format_str("{not_connected}"))
if not ctx.author.voice or (player.is_connected and ctx.author.voice.channel.id != int(player.channel_id)):
return await ctx.send(ctx.localizer.format_str("{disconnect.not_in_voice}"))
player.queue.clear()
await player.stop()
await self.connect_to(ctx.guild.id, None)
embed = discord.Embed(description='{disconnect.disconnected}', color=ctx.me.color)
embed = ctx.localizer.format_embed(embed)
await ctx.send(embed=embed)
@commands.command(name='volume')
@checks.dj_or(alone=True, track_requester=True)
async def _volume(self, ctx, volume: int = None):
""" Changes the player's volume. Must be between 0 and 1000. Error Handling for that is done by Lavalink. """
player = self.bot.lavalink.player_manager.get(ctx.guild.id)
if not volume:
return await ctx.send(f'🔈 | {player.volume}%')
is_dj = checks.is_dj(ctx)
is_admin = getattr(ctx.author.guild_permissions, 'administrator', None) is True
if int(player.current.requester) == ctx.author.id and not is_dj and not is_admin:
if not 50 <= volume <= 125:
return await ctx.send(ctx.localizer.format_str("{volume.out_of_range}"))
await player.set_volume(volume)
embed = discord.Embed(color=ctx.me.color)
embed.description = "{volume.set_to}"
embed = ctx.localizer.format_embed(embed, _volume=player.volume)
await ctx.send(embed=embed)
@commands.command(name='normalize')
@checks.dj_or(alone=True)
async def _normalize(self, ctx):
""" Reset the equalizer and """
player = self.bot.lavalink.player_manager.get(ctx.guild.id)
await player.set_volume(100)
await player.bassboost(False)
embed = discord.Embed(color=ctx.me.color)
embed.description = '{volume.reset}'
embed = ctx.localizer.format_embed(embed)
await ctx.send(embed=embed)
@commands.command(name='boost')
@checks.dj_or(alone=True)
async def _boost(self, ctx, boost: bool = None):
""" Set the equalizer to bass boost the music """
player = self.bot.lavalink.player_manager.get(ctx.guild.id)
if boost is not None:
await player.bassboost(boost)
embed = discord.Embed(color=ctx.me.color)
if player.boosted:
embed.description = '{boost.on}'
else:
embed.description = '{boost.off}'
embed = ctx.localizer.format_embed(embed)
await ctx.send(embed=embed)
@commands.command(name='history')
async def _history(self, ctx):
""" Show the last 10 songs played """
player = self.bot.lavalink.player_manager.get(ctx.guild.id)
history = player.get_history()
if not history:
embed = discord.Embed(description='{history.empty}', color=ctx.me.color)
embed = ctx.localizer.format_embed(embed)
return await ctx.send(embed=embed)
track = history[0]
description = ctx.localizer.format_str("{history.current}", _title=track.title, _uri=track.uri,
_id=track.requester) + '\n\n'
description += ctx.localizer.format_str("{history.previous}", _len=len(history)-1) + '\n'
thumbnail_url = track.extra["thumbnail_url"]
for index, track in enumerate(history[1:], start=1):
description += ctx.localizer.format_str("{history.track}", _index=-index, _title=track.title,
_uri=track.uri, _id=track.requester) + '\n'
embed = discord.Embed(title=ctx.localizer.format_str('{history.title}'), color=ctx.me.color,
description=description)
if thumbnail_url:
embed.set_thumbnail(url=thumbnail_url)
await ctx.send(embed=embed)
@commands.cooldown(1, 5, commands.BucketType.guild)
@commands.command(name='lyrics')
async def _lyrics(self, ctx, *query: str):
player = self.bot.lavalink.player_manager.get(ctx.guild.id)
genius_access_token = self.bot.APIkeys.get('genius', None)
if genius_access_token is None:
return await ctx.send('Missing API key')
excluded_words = ['music', 'video', 'version', 'original', 'lyrics', 'lyric',
'official', 'live', 'instrumental', 'audio', 'hd']
query = ' '.join(query)
if not query and player.is_playing:
query = player.current.title
query = re.sub(r'[-()_\[\]]', '', query)
filtered_words = [word for word in query.split() if word.lower() not in excluded_words]
query = ' '.join(filtered_words)
async def get_site_content(url):
async with self.bot.session.get(url) as resp:
response = await resp.read()
return response.decode('utf-8')
embed = discord.Embed(description=':mag_right:')
status_msg = await ctx.send(embed=embed)
try:
url = 'https://api.genius.com/search?' + urllib.parse.urlencode(
{'access_token': genius_access_token, 'q': query})
result = await get_site_content(url)
response = json.loads(result)
song_id = response['response']['hits'][0]['result']['id']
except Exception:
embed = discord.Embed(description=':x:', color=0xFF0000)
return await status_msg.edit(embed=embed)
result = await get_site_content(f'https://api.genius.com/songs/{song_id}?access_token={genius_access_token}')
song = json.loads(result)['response']['song']
response = await get_site_content(song['url'])
scraped_data = BeautifulSoup(response, 'html.parser')
lyrics = scraped_data.find(class_='lyrics').get_text()
paginator = TextPaginator(max_size=2000, max_lines=50, color=0xFFFF64)
for line in lyrics.split('\n'):
paginator.add_line(line)
await status_msg.delete()
paginator.pages[0].url = song['url']
paginator.pages[0].title = song['full_title']
paginator.pages[0].set_thumbnail(url=song['header_image_thumbnail_url'])
paginator.pages[0].set_author(name='Genius', icon_url='https://i.imgur.com/NmCTsoF.png')
if len(paginator.pages) < 4:
for page in paginator.pages:
await ctx.send(embed=page)
else:
paginator.add_page_indicator(ctx.localizer)
await Scroller(ctx, paginator).start_scrolling()
@commands.command(name='scrub')
@checks.dj_or(alone=True)
async def _scrub(self, ctx):
""" Lists the first 10 search results from a given query. """
player = self.bot.lavalink.player_manager.get(ctx.guild.id)
controls = '{scrub.controls}'
embed = discord.Embed(description='{nothing_playing}', color=ctx.me.color)
embed = ctx.localizer.format_embed(embed)
if player.current is None:
return await ctx.send(embed=embed)
scrubber = [
('\N{BLACK LEFT-POINTING DOUBLE TRIANGLE WITH VERTICAL BAR}', player.seek, -1000),
('\N{BLACK LEFT-POINTING DOUBLE TRIANGLE}', player.seek, -15),
('\N{DOUBLE VERTICAL BAR}', player.set_pause, True),
('\N{BLACK RIGHT-POINTING TRIANGLE}', player.set_pause, False),
('\N{BLACK RIGHT-POINTING DOUBLE TRIANGLE}', player.seek, 15),
('\N{BLACK RIGHT-POINTING DOUBLE TRIANGLE WITH VERTICAL BAR}', player.seek, 1000),
]
selection = None
arg = None
def check(reaction, user):
if user is None or user.id != ctx.author.id:
return False
if reaction.message.id != msg.id:
return False
for (emoji, func, _arg) in scrubber:
if emoji == reaction.emoji:
nonlocal selection
nonlocal arg
selection = func
arg = _arg
return True
return False
embed.description = '{scrub.add}'
embed = ctx.localizer.format_embed(embed)
msg = await ctx.send(embed=embed)
for (emoji, _, _) in scrubber:
await msg.add_reaction(emoji)
embed.description = controls
embed = ctx.localizer.format_embed(embed)
await msg.edit(embed=embed)
scrubbing = True
while scrubbing:
try:
reaction, user = await self.bot.wait_for('reaction_add', timeout=15.0, check=check)
except asyncio.TimeoutError:
# PEP8EDIT scrolling = False
try:
await msg.delete()
await ctx.message.delete()
except discord.Forbidden:
pass
except discord.HTTPException:
pass
finally:
break
try:
await msg.remove_reaction(reaction, user)
except discord.Forbidden:
pass
if selection is not None:
if not isinstance(arg, bool):
arg = player.position + arg * 1000
await selection(arg)
async def ensure_voice(self, ctx, do_connect: Optional[bool] = None):
""" This check ensures that the bot and command author are in the same voicechannel. """
# If region doesn't return a known server location
try:
region = ctx.guild.region.value
except AttributeError:
region = ctx.guild.region
player = self.bot.lavalink.player_manager.create(ctx.guild.id, endpoint=region)
# Create returns a player if one exists, otherwise creates.
# Add commands that require joining voice to work.
should_connect = ctx.command.callback.__name__ in ('_play', '_find', '_search', '_sk')
# Add commands that don't require the you being in voice.
without_connect = ctx.command.callback.__name__ in ('_queue', '_history', '_now', '_lyrics')
if without_connect:
return
if not ctx.author.voice or not ctx.author.voice.channel:
raise commands.CommandInvokeError('Join a voicechannel first.')
if not player.is_connected:
if not should_connect:
raise commands.CommandInvokeError('Not connected.')
voice_channel = ctx.author.voice.channel
permissions = voice_channel.permissions_for(ctx.me)
if not permissions.connect or not permissions.speak: # Check user limit too?
raise commands.CommandInvokeError('I need the `CONNECT` and `SPEAK` permissions.')
voice_channels = self.bot.settings.get(ctx.guild, 'channels.music', [])
if voice_channels:
if voice_channel.id not in voice_channels:
response = ctx.localizer.format_str('{settings_check.voicechannel}')
for channel_id in voice_channels:
channel = ctx.guild.get_channel(channel_id)
if channel is not None:
response += f'{channel.name}, '
await ctx.send(response[:-2])
raise commands.CommandInvokeError('You need to be in the right voice channel')
player.store('channel', ctx.channel.id)
await self.connect_to(ctx.guild.id, str(ctx.author.voice.channel.id))
else:
if int(player.channel_id) != ctx.author.voice.channel.id:
raise commands.CommandInvokeError('You need to be in my voicechannel.')
async def enqueue(self, ctx, track, embed):
player = self.bot.lavalink.player_manager.get(ctx.guild.id)
maxlength = self.max_track_length(ctx.guild, player)
if maxlength and track['info']['length'] > maxlength:
embed.description = ctx.localizer.format_str("{enqueue.toolong}",
_length=timeformatter.format_ms(track['info']['length']),
_max=timeformatter.format_ms(maxlength))
return
if isinstance(track, dict):
thumbnail_url = await thumbnailer.ThumbNailer.identify(
self, track['info']['identifier'], track['info']['uri'])
track = lavalink.models.AudioTrack(track, ctx.author.id, thumbnail_url=thumbnail_url)
track, pos_global, pos_local = player.add(requester=ctx.author.id, track=track)
if player.current is not None:
queue_duration = 0
for i, track in enumerate(player.queue):
if i == pos_global:
break
queue_duration += int(track.duration)
until_play = queue_duration + player.current.duration - player.position
until_play = timeformatter.format_ms(until_play)
embed.add_field(name="{enqueue.position}", value=f"`{pos_local + 1}({pos_global + 1})`", inline=True)
embed.add_field(name="{enqueue.playing_in}", value=f"`{until_play} ({{enqueue.estimated}})`", inline=True)
embed.title = '{enqueue.enqueued}'
thumbnail_url = track.extra["thumbnail_url"]
if thumbnail_url:
embed.set_thumbnail(url=thumbnail_url)
duration = timeformatter.format_ms(int(track.duration))
embed.description = f'[{track.title}]({track.uri})\n**{duration}**'
return embed
def max_track_length(self, guild, player):
is_dynamic = self.bot.settings.get(guild, 'duration.is_dynamic', 'default_duration_type')
maxlength = self.bot.settings.get(guild, 'duration.max', None)
if maxlength is None:
return None
if len(player.listeners): # Avoid division by 0.
listeners = len(player.listeners)
else:
listeners = 1
if maxlength > 10 and is_dynamic:
return max(maxlength*60*1000/listeners, 60*10*1000)
else:
return maxlength*60*1000
@commands.Cog.listener()
async def on_command_error(self, ctx, err):
if isinstance(err, commands.CheckFailure):
textchannels = self.bot.settings.get(ctx.guild, 'channels.text', [])
if textchannels:
if ctx.channel.id not in textchannels:
response = ctx.localizer.format_str('{settings_check.textchannel}')
for channel_id in textchannels:
response += f'<#{channel_id}>, '
await ctx.send(response[:-2])
def setup(bot):
bot.add_cog(Music(bot))
cogs/music/__init__.py
+122
View File
@@ -0,0 +1,122 @@
# Discord Packages
import discord
import lavalink
from discord.ext import commands
from ..utils import thumbnailer, timeformatter
from .music_errors import WrongTextChannelError
class Music(commands.Cog):
def __init__(self, bot):
self.bot = bot
self.logger = self.bot.main_logger.bot_logger.getChild("Music")
self.bot = bot
self.leave_timer.start()
self.logger = self.bot.main_logger.bot_logger.getChild("Errors")
bot.lavalink.add_event_hook(self.track_hook)
async def cog_check(self, ctx):
if not ctx.guild:
raise commands.NoPrivateMessage
if textchannels := self.bot.settings.get(ctx.guild, 'channels.text', []):
if ctx.channel.id not in textchannels:
raise WrongTextChannelError('You need to be in the right text channel', channels=textchannels)
return True
async def cog_before_invoke(self, ctx):
""" Ensures a valic player exists whenever a command is run """
region = ctx.guild.region if isinstance(ctx.guild.region, str) else ctx.guild.region.value
# Creates a new only if one doesn't exist, ensures a valid player for all checks.
self.bot.lavalink.player_manager.create(ctx.guild.id, endpoint=region)
async def connect_to(self, guild_id: int, channel_id: str):
""" Connects to the given voicechannel ID. A channel_id of `None` means disconnect. """
ws = self.bot._connection._get_websocket(guild_id)
await ws.voice_state(str(guild_id), channel_id)
async def enqueue(self, ctx, track, embed, silent=False):
player = self.bot.lavalink.player_manager.get(ctx.guild.id)
# Only add tracks that don't exceed the max track length
if maxlength := self.max_track_length(ctx.guild, player):
if track['info']['length'] > maxlength:
embed.description = ctx.localizer.format_str("{enqueue.toolong}",
_length=timeformatter.format_ms(track['info']['length']),
_max=timeformatter.format_ms(maxlength))
return embed, False
# Add thumbnail, turn track into track class
thumbnail_url = await thumbnailer.ThumbNailer.identify(self, track['info']['identifier'], track['info']['uri'])
track = lavalink.models.AudioTrack(track, ctx.author.id, thumbnail_url=thumbnail_url)
# Add to player
track, pos_global, pos_local = player.add(requester=ctx.author.id, track=track)
if player.current is not None and not silent:
until_play = player.queue_duration(include_current=True, end_pos=pos_global)
embed.add_field(name="{enqueue.position}", value=f"`{pos_local + 1}({pos_global + 1})`", inline=True)
embed.add_field(name="{enqueue.playing_in}", value=f"`{until_play} ({{enqueue.estimated}})`",
inline=True)
embed.title = '{enqueue.enqueued}'
thumbnail_url = track.extra["thumbnail_url"]
if thumbnail_url:
embed.set_thumbnail(url=thumbnail_url)
duration = timeformatter.format_ms(int(track.duration))
embed.description = f'[{track.title}]({track.uri})\n**{duration}**'
return embed, True
def get_current_song_embed(self, ctx, include_time=False):
player = self.bot.lavalink.player_manager.get(ctx.guild.id)
if include_time:
position = timeformatter.format_ms(player.position)
if player.current.stream:
duration = '{live}'
else:
duration = timeformatter.format_ms(player.current.duration)
song = f'**[{player.current.title}]({player.current.uri})**\n({position}/{duration})'
else:
song = f'**[{player.current.title}]({player.current.uri})**'
embed = discord.Embed(color=ctx.me.color, description=song, title='{now}')
thumbnail_url = player.current.extra["thumbnail_url"]
member = ctx.guild.get_member(player.current.requester)
if thumbnail_url:
embed.set_thumbnail(url=thumbnail_url)
embed.set_footer(text=f'{{requested_by}} {member.display_name}', icon_url=member.avatar_url)
embed = ctx.localizer.format_embed(embed)
return embed
def max_track_length(self, guild, player):
if maxlength := self.bot.settings.get(guild, 'duration.max', None):
is_dynamic = self.bot.settings.get(guild, 'duration.is_dynamic', 'default_duration_type')
listeners = max(1, len(player.listeners)) # Avoid division by 0
if maxlength > 10 and is_dynamic:
return max(maxlength*60*1000/listeners, 60*10*1000)
else:
return maxlength*60*1000
else:
return None
# commands
from .basic_commands import (
_disconnect, _djremove, _move, _myqueue, _now, _pause, _play, _queue, _reconnect, _remove, _search, _seek,
_shuffle, _skip, _skip_to, _stop, _user_queue_remove, _volume)
# events
from .events import check_leave_voice, cog_unload, leave_check, leave_timer, on_voice_state_update, track_hook
# commands
from .extra_commands import _boost, _history, _lyrics, _normalize, _scrub
def setup(bot):
bot.add_cog(Music(bot))
cogs/music/basic_commands.py
+392
View File
@@ -0,0 +1,392 @@
"""
Commands that interact with just discord and lavalink and only present the basic commands needed for a music bot
"""
# Discord Packages
import discord
from discord.ext import commands
import asyncio
import re
from ..utils import checks, timeformatter
from ..utils.paginator import QueuePaginator, Scroller
from ..utils.selector import Selector
from .decorators import require_playing, require_queue, require_voice_connection, voteable
time_rx = re.compile('[0-9]+')
url_rx = re.compile('https?:\\/\\/(?:www\\.)?.+')
@commands.command(name='play')
@require_voice_connection(should_connect=True)
async def _play(self, ctx, *, query: str):
""" Searches and plays a song from a given query. """
player = self.bot.lavalink.player_manager.get(ctx.guild.id)
query = query.strip('<>')
if not url_rx.match(query):
query = f'ytsearch:{query}'
results = await player.node.get_tracks(query)
if not results or not results['tracks']:
return await ctx.send(ctx.localizer.format_str("{nothing_found}"))
embed = discord.Embed(color=ctx.me.color)
if results['loadType'] == 'PLAYLIST_LOADED':
numtracks = 0
for track in results['tracks']:
_, track_added = await self.enqueue(ctx, track, embed, silent=True)
if track_added:
numtracks += 1
embed.title = '{playlist_enqued}'
embed.description = f'{results["playlistInfo"]["name"]} - {numtracks} {{tracks}}'
else:
track = results['tracks'][0]
await self.enqueue(ctx, track, embed)
embed = ctx.localizer.format_embed(embed)
await ctx.send(embed=embed)
if not player.is_playing:
await player.play()
@commands.command(name='seek')
@checks.dj_or(alone=True, track_requester=True)
@require_voice_connection()
@require_playing(require_user_listening=True)
async def _seek(self, ctx, *, time: str):
""" Seeks to a given position in a track. """
player = self.bot.lavalink.player_manager.get(ctx.guild.id)
if seconds := time_rx.search(time):
# Convert to milliseconds, include sign
milliseconds = int(seconds.group())*1000 * (-1 if time.startswith('-1') else 1)
track_time = player.position + milliseconds
await player.seek(track_time)
msg = ctx.localizer.format_str("{seek.track_moved}", _position=timeformatter.format_ms(track_time))
await ctx.send(msg)
else:
await ctx.send(ctx.localizer.format_str("{seek.missing_amount}"))
@commands.command(name='skip')
@require_voice_connection()
@require_playing()
@voteable(requester_override=True)
async def _skip(self, ctx):
""" Skips the current track. """
player = self.bot.lavalink.player_manager.get(ctx.guild.id)
await player.skip()
if player.current:
embed = self.get_current_song_embed(ctx)
await ctx.send(ctx.localizer.format_str("{skip.skipped}"), embed=embed)
else:
await ctx.send(ctx.localizer.format_str("{skip.skipped}"))
@commands.command(name='skipto')
@checks.dj_or(alone=True)
@require_playing(require_user_listening=True)
async def _skip_to(self, ctx, pos: int = 1):
""" Plays the queue from a specific point. Disregards tracks before the pos. """
player = self.bot.lavalink.player_manager.get(ctx.guild.id)
# TODO: Do all queue out of range messages the same way
if pos < 1:
return await ctx.send(ctx.localizer.format_str("{skip_to.invalid_pos}"))
if len(player.queue) < pos:
return await ctx.send(ctx.localizer.format_str("{skip_to.exceeds_queue}"))
await player.skip(pos - 1)
msg = ctx.localizer.format_str("{skip_to.skipped_to}", _title=player.current.title, _pos=pos)
await ctx.send(msg)
@commands.command(name='stop')
@require_voice_connection()
@require_playing(require_user_listening=True)
@voteable(DJ_override=True)
async def _stop(self, ctx):
""" Stops the player and clears its queue. """
player = self.bot.lavalink.player_manager.get(ctx.guild.id)
player.queue.clear()
await player.stop()
embed = discord.Embed(color=ctx.me.color, title='{stop}')
await ctx.send(embed=ctx.localizer.format_embed(embed))
@commands.command(name='now')
@require_playing()
async def _now(self, ctx):
embed = self.get_current_song_embed(ctx, include_time=True)
embed = ctx.localizer.format_embed(embed)
await ctx.send(embed=embed)
@commands.command(name='queue')
@require_queue(require_member_queue=True)
async def _queue(self, ctx, *, member: discord.Member = None):
""" Shows the player's queue. """
player = self.bot.lavalink.player_manager.get(ctx.guild.id)
pagified_queue = QueuePaginator(ctx.localizer, player, color=ctx.me.color, member=member)
scroller = Scroller(ctx, pagified_queue)
await scroller.start_scrolling()
@commands.command(name='myqueue')
@require_queue(require_author_queue=True)
async def _myqueue(self, ctx):
""" Shows your queue. """
await self._queue(ctx, member=ctx.author)
@commands.command(name='pause')
@checks.dj_or(alone=True)
@require_voice_connection()
@require_playing()
async def _pause(self, ctx):
""" Pauses/Resumes the current track. """
player = self.bot.lavalink.player_manager.get(ctx.guild.id)
await player.set_pause(not player.paused)
if player.paused:
await ctx.send(ctx.localizer.format_str("{resume.paused}"))
else:
await ctx.send(ctx.localizer.format_str("{resume.resumed}"))
@commands.command(name='shuffle')
@require_voice_connection()
@require_queue(require_author_queue=True)
async def _shuffle(self, ctx):
""" Shuffles your queue. """
player = self.bot.lavalink.player_manager.get(ctx.guild.id)
player.shuffle_user_queue(ctx.author.id)
await ctx.send(ctx.localizer.format_str("{shuffle}"))
@commands.command(name='move')
@require_voice_connection()
@require_queue(require_author_queue=True)
async def _move(self, ctx):
""" Moves a song in your queue. """
player = self.bot.lavalink.player_manager.get(ctx.guild.id)
user_queue = player.user_queue(ctx.author.id, dual=True)
# Setup for selector
identifiers = []
for index, temp in enumerate(user_queue):
track, globpos = temp
identifiers.append(ctx.localizer.format_str("{queue.usertrack}", _index=index+1,
_globalindex=globpos+1, _title=track.title,
_uri=track.uri))
def ret_first_arg(*args):
return args[0]
functions = [ret_first_arg]*len(user_queue)
arguments = [(i,) for i in range(len(user_queue))]
# Set the different titles for the different selections
round_titles = ["{moved.choose_song}", "{moved.choose_pos}"]
round_titles = [ctx.localizer.format_str(i) for i in round_titles]
selector = Selector(ctx, identifiers, functions, arguments, num_selections=5, round_titles=round_titles,
color=ctx.me.color, title=ctx.localizer.format_str("{moved.choose}"))
# Get the initial and final position
message, page, selections = await selector.start_scrolling()
pos_initial, pos_final = selections[0], selections[1]
# Move the track
moved = player.move_user_track(ctx.author.id, pos_initial, pos_final)
# Create a nice embed explaining what happened
song = f'**[{moved.title}]({moved.uri})**'
embed = discord.Embed(color=ctx.me.color, description=song, title='{moved.moved}')
thumbnail_url = moved.extra["thumbnail_url"]
member = ctx.guild.get_member(moved.requester)
if thumbnail_url:
embed.set_thumbnail(url=thumbnail_url)
embed.add_field(name="{moved.local}", value=f"`{pos_initial + 1} → {pos_final + 1}`", inline=True)
embed.add_field(name="{moved.global}", value=f"`{player.queue._loc_to_glob(ctx.author.id, pos_initial) + 1}\
→ {player.queue._loc_to_glob(ctx.author.id, pos_final) + 1}`", inline=True)
embed.set_footer(text=f'{{requested_by}} {member.display_name}', icon_url=member.avatar_url)
embed = ctx.localizer.format_embed(embed)
await message.edit(embed=embed)
@commands.command(name='remove')
@require_voice_connection()
@require_queue(require_author_queue=True)
async def _remove(self, ctx):
player = self.bot.lavalink.player_manager.get(ctx.guild.id)
user_queue = player.user_queue(ctx.author.id, dual=True)
functions = [player.remove_user_track]*len(user_queue)
arguments = [(ctx.author.id, i) for i in range(len(user_queue))]
identifiers = []
for index, temp in enumerate(user_queue):
track, globpos = temp
identifiers.append(ctx.localizer.format_str("{queue.usertrack}", _index=index+1,
_globalindex=globpos+1, _title=track.title,
_uri=track.uri))
selector = Selector(ctx, identifiers, functions, arguments, num_selections=5, color=0xFF0000,
title='Remove song plz')
message, _, removed = await selector.start_scrolling()
await message.edit(content=ctx.localizer.format_str("{remove}", _title=removed.title), embed=None)
@commands.command(name="DJremove")
@checks.dj_or()
@require_voice_connection()
@require_queue()
async def _djremove(self, ctx, pos: int, member: discord.Member = None):
""" Remove a song from either the global queue or a users queue"""
player = self.bot.lavalink.player_manager.get(ctx.guild.id)
# TODO: add way to allow "member" as regular arg in require_queue() instead of only kwargs
queue = player.user_queue(member.id) if member else player.queue
# TODO: Do all queue out of range messages the same way
if pos <= len(queue) and pos >= 1:
if member is None:
removed = player.remove_global_track(pos - 1)
else:
removed = player.remove_user_track(member.id, pos - 1)
requester = self.bot.get_user(removed.requester)
await ctx.send(ctx.localizer.format_str("{dj_removed}", _title=removed.title, _user=requester.display_name))
else:
await ctx.send(ctx.localizer.format_str("{out_of_range}", _len=len(queue)))
@commands.command(name='removeuser')
@checks.dj_or(alone=True)
@require_voice_connection()
@require_queue(require_member_queue=True)
async def _user_queue_remove(self, ctx, *, member: discord.Member):
""" Remove a song from either the global queue or a users queue"""
player = self.bot.lavalink.player_manager.get(ctx.guild.id)
player.remove_user_queue(member.id)
embed = discord.Embed(description="{dj_remove_user}", color=ctx.me.color)
embed = ctx.localizer.format_embed(embed, _id=member.id)
await ctx.send(embed=embed)
@commands.command(name='search')
@require_voice_connection(should_connect=True)
async def _search(self, ctx, *, query):
player = self.bot.lavalink.player_manager.get(ctx.guild.id)
if not query.startswith('ytsearch:') and not query.startswith('scsearch:'):
query = 'ytsearch:' + query
results = await player.node.get_tracks(query)
embed = discord.Embed(description='{nothing_found}', color=0x36393F)
if not results or not results['tracks']:
embed = ctx.localizer.format_embed(embed)
return await ctx.send(embed=embed)
tracks = results['tracks']
result_count = min(len(tracks), 15)
# Setup for the selector
identifiers = []
functions = [self.enqueue]*result_count
arguments = [(ctx, tracks[i], discord.Embed(color=ctx.me.color)) for i in range(result_count)]
for index, track in enumerate(tracks, start=1):
track = track['info']
duration = timeformatter.format_ms(int(track['length']))
identifiers.append(f'`{index}.` [{track["title"]}]({track["uri"]}) `{duration}`')
if index == result_count:
break
search_selector = Selector(ctx, identifiers, functions, arguments, num_selections=5,
color=ctx.me.color, title=ctx.localizer.format_str('{results}'))
# Let the user scroll through results
message, current_page, (embed, added) = await search_selector.start_scrolling()
embed = ctx.localizer.format_embed(embed)
await message.edit(embed=embed)
if not player.is_playing:
await player.play()
@commands.command(name='disconnect')
@require_voice_connection()
@voteable(DJ_override=True)
async def _disconnect(self, ctx):
""" Disconnects the player from the voice channel and clears its queue. """
player = self.bot.lavalink.player_manager.get(ctx.guild.id)
player.queue.clear()
await player.stop()
await self.connect_to(ctx.guild.id, None)
embed = discord.Embed(description='{disconnect.disconnected}', color=ctx.me.color)
embed = ctx.localizer.format_embed(embed)
await ctx.send(embed=embed)
@commands.command(name='reconnect')
@require_voice_connection()
@voteable(DJ_override=True)
async def _reconnect(self, ctx):
""" Tries to disconnect then reconnect the player in case the bot gets stuck on a song """
player = self.bot.lavalink.player_manager.get(ctx.guild.id)
current_channel = player.channel_id
async def recon():
await player.stop()
await self.connect_to(ctx.guild.id, None)
await asyncio.sleep(1) # Pretend stuff is happening/give everything some time to reset.
await self.connect_to(ctx.guild.id, current_channel)
if player.current:
track = player.current
start_time = player.position
await recon()
if track:
await player.play(track, start_time=start_time)
else:
await recon()
@commands.command(name='volume')
@checks.dj_or(alone=True, track_requester=True)
@require_playing(require_user_listening=True)
async def _volume(self, ctx, volume: int = None):
""" Changes the player's volume. Must be between 0 and 1000. Error Handling for that is done by Lavalink. """
player = self.bot.lavalink.player_manager.get(ctx.guild.id)
if not volume:
return await ctx.send(f'🔈 | {player.volume}%')
is_dj = checks.is_dj(ctx)
is_admin = getattr(ctx.author.guild_permissions, 'administrator', None) is True
if player.current and (int(player.current.requester) == ctx.author.id and not is_dj and not is_admin):
if not 50 <= volume <= 125:
return await ctx.send(ctx.localizer.format_str("{volume.out_of_range}"))
await player.set_volume(volume)
embed = discord.Embed(description="{volume.set_to}", color=ctx.me.color)
embed = ctx.localizer.format_embed(embed, _volume=player.volume)
await ctx.send(embed=embed)
cogs/music/decorators.py
+156
View File
@@ -0,0 +1,156 @@
# Discord Packages
import discord
from discord.ext import commands
import functools
import inspect
import math
from cogs.music.music_errors import WrongVoiceChannelError
from ..utils.checks import is_dj
def require_voice_connection(should_connect=False):
"""
Checks if the bot is in a valid voice channel for the command
should_connect indicates whether the bot should try to join a channel
"""
def ensure_voice(func):
@functools.wraps(func)
async def ensure_voice_inner(self, ctx, *command_args, **kwargs):
""" This check ensures that the bot and command author are in the same voicechannel. """
player = self.bot.lavalink.player_manager.get(ctx.guild.id)
if not ctx.author.voice or not ctx.author.voice.channel:
raise commands.CommandInvokeError('Join a voicechannel first.')
if not player.is_connected:
if not should_connect:
raise commands.CommandInvokeError('Not connected.')
user_voice_channel = ctx.author.voice.channel
permissions = user_voice_channel.permissions_for(ctx.me)
if not permissions.connect or not permissions.speak: # Check user limit too?
raise commands.CommandInvokeError('I need the `CONNECT` and `SPEAK` permissions.')
if (len(user_voice_channel.members) == user_voice_channel.user_limit) and not permissions.administrator:
raise commands.CommandInvokeError('The channel is currently full')
# Check against music channel restrictions in bot settings
if voice_channels := self.bot.settings.get(ctx.guild, 'channels.music', []):
if user_voice_channel.id not in voice_channels:
raise WrongVoiceChannelError(
'You need to be in the right voice channel', channels=voice_channels)
player.store('channel', ctx.channel.id)
await self.connect_to(ctx.guild.id, str(ctx.author.voice.channel.id))
elif int(player.channel_id) != ctx.author.voice.channel.id:
raise commands.CommandInvokeError('You need to be in my voicechannel.')
await func(self, ctx, *command_args, **kwargs)
return ensure_voice_inner
return ensure_voice
def require_playing(require_user_listening=False):
"""
Checks if the bot is currently playing a track
ensure_user_listening: also checks if the user is listening to the bot
"""
def ensure_play(func):
@functools.wraps(func)
async def ensure_play_inner(self, ctx, *command_args, **kwargs):
if player := self.bot.lavalink.player_manager.get(ctx.guild.id):
if not player.is_playing:
raise commands.CommandInvokeError('Not playing')
if require_user_listening and (ctx.author not in player.listeners):
raise commands.CommandInvokeError('Not listening')
else:
raise Exception('Player object does not yet exist')
await func(self, ctx, *command_args, **kwargs)
return ensure_play_inner
return ensure_play
def require_queue(require_member_queue=False, require_author_queue=False):
"""
Checks if there is something queued
require_member_queue also checks if the queue of a member is empty, only works when member is an argument
"""
def ensure_queue(func):
if require_member_queue:
inspection = inspect.getfullargspec(func)
assert 'member' in inspection.kwonlyargs, """require_member_error can only be used
on commands with a member keyword argument"""
# assert str(inspection.annotations['member']) == "<class 'discord.member.Member'>", """
# Member must be a discord member type"""
@functools.wraps(func)
async def ensure_queue_inner(self, ctx, *command_args, **kwargs):
player = self.bot.lavalink.player_manager.get(ctx.guild.id)
if require_member_queue:
try:
if member := kwargs['member']: # Ignore if member is None
user_queue = player.user_queue(member.id)
if not user_queue:
embed = discord.Embed(description='{queue.user_empty}', color=ctx.me.color)
embed = ctx.localizer.format_embed(embed, _user=member.display_name)
return await ctx.send(embed=embed)
except KeyError:
raise Exception("require_member_error can only be used on commands with a member keyword argument")
if require_author_queue:
user_queue = player.user_queue(ctx.author.id)
if not user_queue:
embed = discord.Embed(description='{my_queue}', color=ctx.me.color)
embed = ctx.localizer.format_embed(embed)
return await ctx.send(embed=embed)
if player.queue.empty:
embed = discord.Embed(description='{queue.empty}', color=ctx.me.color)
embed = ctx.localizer.format_embed(embed)
return await ctx.send(embed=embed)
await func(self, ctx, *command_args, **kwargs)
return ensure_queue_inner
return ensure_queue
def voteable(requester_override=False, DJ_override=False):
def make_voteable(func):
@functools.wraps(func)
async def voteable_inner(self, ctx, *command_args, **kwargs):
player = self.bot.lavalink.player_manager.get(ctx.guild.id)
player.add_vote(func.__name__, ctx.author)
total = len(player.listeners)
votes = len(player.get_voters(func.__name__))
threshold = self.bot.settings.get(ctx.guild, 'vote_threshold', 'default_threshold')
enough_votes = votes/total >= threshold/100
DJ = DJ_override and is_dj(ctx)
requester = player.current.requester == ctx.author.id and requester_override
if enough_votes or requester or DJ:
await func(self, ctx, *command_args, **kwargs)
player.clear_votes()
else:
if votes != 0:
needed = math.ceil(total*threshold/100)
# TODO: redo this message to allow for other vote types
msg = ctx.localizer.format_str("{skip.require_vote}", _skips=votes, _total=needed)
await ctx.send(msg)
return voteable_inner
return make_voteable
cogs/music/events.py
+73
View File
@@ -0,0 +1,73 @@
# Discord Packages
import lavalink
import lavalink.events
from discord.ext import commands, tasks
def cog_unload(self):
self.bot.lavalink._event_hooks.clear()
async def track_hook(self, event):
if isinstance(event, lavalink.events.TrackEndEvent):
channel = self.bot.get_channel(event.player.fetch('channel'))
player = self.bot.lavalink.player_manager.get(channel.guild.id)
player.skip_voters.clear()
if isinstance(event, lavalink.events.TrackStartEvent):
pass
if isinstance(event, lavalink.events.QueueEndEvent):
channel = self.bot.get_channel(event.player.fetch('channel'))
await self.check_leave_voice(channel.guild)
if isinstance(event, lavalink.events.PlayerUpdateEvent):
pass
if isinstance(event, lavalink.events.NodeDisconnectedEvent):
pass
if isinstance(event, lavalink.events.NodeConnectedEvent):
pass
if isinstance(event, lavalink.events.NodeChangedEvent):
pass
if isinstance(event, lavalink.events.TrackStuckEvent):
print(event)
@commands.Cog.listener()
async def on_voice_state_update(self, member, before, after):
""" Updates listeners when the bot or a user changes voice state """
if member.id == self.bot.user.id and after.channel is not None:
voice_channel = after.channel
player = self.bot.lavalink.player_manager.get(member.guild.id)
player.clear_listeners()
for member in voice_channel.members:
if not member.bot:
player.update_listeners(member, member.voice)
if not member.bot:
player = self.bot.lavalink.player_manager.get(member.guild.id)
if player is not None:
player.update_listeners(member, after)
await self.check_leave_voice(member.guild)
async def check_leave_voice(self, guild):
""" Checks if the bot should leave the voice channel """
# TODO, disconnect timer?
player = self.bot.lavalink.player_manager.get(guild.id)
if len(player.listeners) == 0 and player.is_connected:
if player.queue.empty and player.current is None:
await player.stop()
await self.connect_to(guild.id, None)
async def leave_check(self):
for player_id in self.bot.lavalink.player_manager.players:
await self.check_leave_voice(self.bot.get_guild(player_id))
@tasks.loop(seconds=10.0)
async def leave_timer(self):
try:
await self.leave_check()
except Exception as err:
self.logger.debug("Error in leave_timer loop.\nTraceback: %s" % (err))
cogs/music/extra_commands.py
+215
View File
@@ -0,0 +1,215 @@
"""
Commands that add to the music bot experience, like equalizers, commands that access the web, like lyrics, and
commands that are complex and mostly for show, like scrubber.
"""
# Discord Packages
import discord
from discord.ext import commands
import asyncio
import json
import re
import urllib
from bs4 import BeautifulSoup
from ..utils import checks
from ..utils.paginator import Scroller, TextPaginator
from .decorators import require_playing, require_voice_connection
@commands.command(name='normalize')
@checks.dj_or(alone=True)
@require_voice_connection()
async def _normalize(self, ctx):
""" Reset the equalizer and volume """
player = self.bot.lavalink.player_manager.get(ctx.guild.id)
await player.set_volume(100)
await player.bassboost(False)
embed = discord.Embed(description='{volume.reset}', color=ctx.me.color)
embed = ctx.localizer.format_embed(embed)
await ctx.send(embed=embed)
@commands.command(name='boost')
@checks.dj_or(alone=True)
async def _boost(self, ctx, boost: bool = None):
""" Set the equalizer to bass boost the music """
player = self.bot.lavalink.player_manager.get(ctx.guild.id)
if boost is not None:
await player.bassboost(boost)
embed = discord.Embed(color=ctx.me.color)
embed.description = '{boost.on}' if player.boosted else '{boost.off}'
embed = ctx.localizer.format_embed(embed)
await ctx.send(embed=embed)
@commands.command(name='history')
async def _history(self, ctx):
""" Show the last 10 songs played """
player = self.bot.lavalink.player_manager.get(ctx.guild.id)
history = player.get_history()
if not history:
embed = discord.Embed(description='{history.empty}', color=ctx.me.color)
embed = ctx.localizer.format_embed(embed)
return await ctx.send(embed=embed)
track = history[0]
description = ctx.localizer.format_str("{history.current}", _title=track.title, _uri=track.uri,
_id=track.requester) + '\n\n'
description += ctx.localizer.format_str("{history.previous}", _len=len(history)-1) + '\n'
thumbnail_url = track.extra["thumbnail_url"]
for index, track in enumerate(history[1:], start=1):
description += ctx.localizer.format_str("{history.track}", _index=-index, _title=track.title,
_uri=track.uri, _id=track.requester) + '\n'
embed = discord.Embed(title=ctx.localizer.format_str('{history.title}'), color=ctx.me.color,
description=description)
if thumbnail_url:
embed.set_thumbnail(url=thumbnail_url)
await ctx.send(embed=embed)
@commands.cooldown(1, 5, commands.BucketType.guild)
@commands.command(name='lyrics')
async def _lyrics(self, ctx, *query: str):
# TODO:
player = self.bot.lavalink.player_manager.get(ctx.guild.id)
genius_access_token = self.bot.APIkeys.get('genius', None)
if genius_access_token is None:
return await ctx.send('Missing API key')
excluded_words = ['music', 'video', 'version', 'original', 'lyrics', 'lyric',
'official', 'live', 'instrumental', 'audio', 'hd']
query = ' '.join(query)
if not query and player.is_playing:
query = player.current.title
query = re.sub(r'[-()_\[\]]', '', query)
filtered_words = [word for word in query.split() if word.lower() not in excluded_words]
query = ' '.join(filtered_words)
async def get_site_content(url):
async with self.bot.session.get(url) as resp:
response = await resp.read()
return response.decode('utf-8')
embed = discord.Embed(description=':mag_right:')
status_msg = await ctx.send(embed=embed)
try:
url = 'https://api.genius.com/search?' + urllib.parse.urlencode(
{'access_token': genius_access_token, 'q': query})
result = await get_site_content(url)
response = json.loads(result)
song_id = response['response']['hits'][0]['result']['id']
except Exception:
embed = discord.Embed(description=':x:', color=0xFF0000)
return await status_msg.edit(embed=embed)
result = await get_site_content(f'https://api.genius.com/songs/{song_id}?access_token={genius_access_token}')
song = json.loads(result)['response']['song']
response = await get_site_content(song['url'])
scraped_data = BeautifulSoup(response, 'html.parser')
lyrics = scraped_data.find(class_='lyrics').get_text()
paginator = TextPaginator(max_size=2000, max_lines=50, color=0xFFFF64)
for line in lyrics.split('\n'):
paginator.add_line(line)
await status_msg.delete()
paginator.pages[0].url = song['url']
paginator.pages[0].title = song['full_title']
paginator.pages[0].set_thumbnail(url=song['header_image_thumbnail_url'])
paginator.pages[0].set_author(name='Genius', icon_url='https://i.imgur.com/NmCTsoF.png')
if len(paginator.pages) < 4:
for page in paginator.pages:
await ctx.send(embed=page)
else:
paginator.add_page_indicator(ctx.localizer)
await Scroller(ctx, paginator).start_scrolling()
@commands.command(name='scrub')
@checks.dj_or(alone=True)
@require_voice_connection()
@require_playing(require_user_listening=True)
async def _scrub(self, ctx):
""" Lists the first 10 search results from a given query. """
player = self.bot.lavalink.player_manager.get(ctx.guild.id)
controls = '{scrub.controls}'
scrubber = [
('\N{BLACK LEFT-POINTING DOUBLE TRIANGLE WITH VERTICAL BAR}', player.seek, -1000),
('\N{BLACK LEFT-POINTING DOUBLE TRIANGLE}', player.seek, -15),
('\N{DOUBLE VERTICAL BAR}', player.set_pause, True),
('\N{BLACK RIGHT-POINTING TRIANGLE}', player.set_pause, False),
('\N{BLACK RIGHT-POINTING DOUBLE TRIANGLE}', player.seek, 15),
('\N{BLACK RIGHT-POINTING DOUBLE TRIANGLE WITH VERTICAL BAR}', player.seek, 1000),
]
selection = None
arg = None
def check(reaction, user):
if user is None or user.id != ctx.author.id:
return False
if reaction.message.id != msg.id:
return False
for (emoji, func, _arg) in scrubber:
if emoji == reaction.emoji:
nonlocal selection
nonlocal arg
selection = func
arg = _arg
return True
return False
embed = discord.Embed(description='{scrub.add}', color=ctx.me.color)
embed = ctx.localizer.format_embed(embed)
msg = await ctx.send(embed=embed)
for (emoji, _, _) in scrubber:
await msg.add_reaction(emoji)
embed.description = controls
embed = ctx.localizer.format_embed(embed)
await msg.edit(embed=embed)
scrubbing = True
while scrubbing:
try:
reaction, user = await self.bot.wait_for('reaction_add', timeout=15.0, check=check)
except asyncio.TimeoutError:
# PEP8EDIT scrolling = False
try:
await msg.delete()
await ctx.message.delete()
except (discord.Forbidden, discord.HTTPException):
pass
finally:
break
try:
await msg.remove_reaction(reaction, user)
except discord.Forbidden:
pass
if selection is not None:
if not isinstance(arg, bool):
arg = player.position + arg * 1000
await selection(arg)
cogs/music/music_errors.py
+20
View File
@@ -0,0 +1,20 @@
# define Python user-defined exceptions
# Discord Packages
from discord.ext.commands import CommandError
class WrongChannelError(CommandError):
def __init__(self, e: str, channels=None):
self.original = e
self.channels = channels
super().__init__()
class WrongVoiceChannelError(WrongChannelError):
def __init__(self, e: str, channels=None):
super().__init__(e, channels=channels)
class WrongTextChannelError(WrongChannelError):
def __init__(self, e: str, channels=None):
super().__init__(e, channels=channels)
cogs/musicevents.py
-82
View File
@@ -1,82 +0,0 @@
"""
A cog to separate events from regular music commands
"""
# Discord Packages
import lavalink
import lavalink.events
from discord.ext import commands, tasks
class MusicEvents(commands.Cog):
def __init__(self, bot):
self.bot = bot
self.leave_timer.start()
self.logger = self.bot.main_logger.bot_logger.getChild("Errors")
bot.lavalink.add_event_hook(self.track_hook)
def cog_unload(self):
self.bot.lavalink._event_hooks.clear()
async def track_hook(self, event):
if isinstance(event, lavalink.events.TrackEndEvent):
pass # Send track ended message to channel.
if isinstance(event, lavalink.events.TrackStartEvent):
pass
if isinstance(event, lavalink.events.QueueEndEvent):
channel = self.bot.get_channel(event.player.fetch('channel'))
await self.check_leave_voice(channel.guild)
if isinstance(event, lavalink.events.PlayerUpdateEvent):
pass
if isinstance(event, lavalink.events.NodeDisconnectedEvent):
pass
if isinstance(event, lavalink.events.NodeConnectedEvent):
pass
if isinstance(event, lavalink.events.NodeChangedEvent):
pass
async def connect_to(self, guild_id: int, channel_id: str):
""" Connects to the given voicechannel ID. A channel_id of `None` means disconnect. """
ws = self.bot._connection._get_websocket(guild_id)
await ws.voice_state(str(guild_id), channel_id)
@commands.Cog.listener()
async def on_voice_state_update(self, member, before, after):
""" Updates listeners when the bot or a user changes voice state """
if member.id == self.bot.user.id and after.channel is not None:
voice_channel = after.channel
player = self.bot.lavalink.player_manager.get(member.guild.id)
player.clear_listeners()
for member in voice_channel.members:
if not member.bot:
player.update_listeners(member, member.voice)
if not member.bot:
player = self.bot.lavalink.player_manager.get(member.guild.id)
if player is not None:
player.update_listeners(member, after)
await self.check_leave_voice(member.guild)
async def check_leave_voice(self, guild):
""" Checks if the bot should leave the voice channel """
# TODO, disconnect timer?
player = self.bot.lavalink.player_manager.get(guild.id)
if len(player.listeners) == 0 and player.is_connected:
if player.queue.empty and player.current is None:
await player.stop()
await self.connect_to(guild.id, None)
async def leave_check(self):
for player_id in self.bot.lavalink.player_manager.players:
await self.check_leave_voice(self.bot.get_guild(player_id))
@tasks.loop(seconds=10.0)
async def leave_timer(self):
try:
await self.leave_check()
except Exception as err:
self.logger.debug("Error in leave_timer loop.\nTraceback: %s" % (err))
def setup(bot):
bot.add_cog(MusicEvents(bot))
cogs/nodemanager.py
+45 -3
View File
@@ -22,7 +22,6 @@ class NodeManager(commands.Cog):
music_extensions = [
'cogs.music',
'cogs.musicevents'
]
if not hasattr(bot, 'lavalink'):
@@ -49,9 +48,10 @@ class NodeManager(commands.Cog):
for node in self.bot.lavalink.node_manager.nodes:
name_cache.append(node.name)
for node in conf['lavalink nodes']:
for node in conf['lavalink nodes'] + self.settings.get('lavalink', 'nodes', []):
if node['name'] in name_cache:
continue
name_cache.append(node['name'])
self.bot.lavalink.add_node(**node)
self.logger.debug("Adding Lavalink node: %s on %s with the port %s in %s" % (
@@ -100,10 +100,20 @@ class NodeManager(commands.Cog):
return embed
def get_node_properties(self, node):
return {
'name': node.name,
'host': node.host,
'port': node.port,
'password': node.password,
'region': node.region
}
@commands.group(name='node', hidden=True)
@commands.is_owner()
async def _node(self, ctx):
if ctx.invoked_subcommand is None:
ctx.localizer.prefix = 'help' # Ensure the bot looks for locales in the context of help, not cogmanager.
paginator = commandhelper(ctx, ctx.command, ctx.invoker, include_subcmd=True)
scroller = Scroller(ctx, paginator)
await scroller.start_scrolling()
@@ -119,11 +129,17 @@ class NodeManager(commands.Cog):
@_node.command(name='add')
@commands.is_owner()
async def _add(self, ctx, host, port, password, region, name=None):
self.bot.lavalink.add_node(host, port, password, region, name)
if name in [n.name for n in self.bot.lavalink.node_manager.nodes]:
return await ctx.send("A node with that name already exists")
self.bot.lavalink.add_node(host, port, password, region, name=name)
self.logger.debug("Adding Lavalink node: %s on %s with the port %s in %s" % (host, port, region, name,))
embed = await self._node_presenter(ctx, {'host': host, 'port': port, 'password': password,
'region': region, 'name': name})
embed.title = 'Added new node!'
self.settings.set('lavalink', 'nodes', [self.get_node_properties(n) for
n in self.bot.lavalink.node_manager.nodes])
await ctx.send(embed=embed)
@_node.command(name='list')
@@ -149,9 +165,35 @@ class NodeManager(commands.Cog):
await ctx.send(embed=embed)
self.bot.lavalink.node_manager.remove_node(_node)
self.logger.info("Removed Lavalink node: %s" % _node.host)
self.settings.set('lavalink', 'nodes', [self.get_node_properties(n) for
n in self.bot.lavalink.node_manager.nodes])
if not sent_feedback:
await ctx.send('No node found')
@_node.command(name='change')
@commands.is_owner()
async def _nodechange(self, ctx, node=None):
player = self.bot.lavalink.player_manager.get(ctx.guild.id)
if not player:
return
# "Switches" to the current node
if node is None:
await player.change_node(player.node)
else:
newnode = None
for _node in self.bot.lavalink.node_manager.nodes:
if (_node.name or _node.host) == node:
newnode = _node
if not newnode:
return
await player.change_node(newnode)
def setup(bot):
bot.add_cog(NodeManager(bot))
cogs/settings.py
+11 -29
View File
@@ -34,7 +34,6 @@ class Settings(commands.Cog):
self.bot.aliaser.load_localizations()
if locale in self.bot.aliaser.localization_table.keys():
self.settings.set(ctx.guild, 'locale', locale)
locale = self.settings.get(ctx.guild, 'locale', 'default_lang')
await ctx.send(f'Locale set to `{locale}`')
@@ -45,8 +44,7 @@ class Settings(commands.Cog):
@commands.guild_only()
@_set.command(name='serverprefix')
async def _set_guild_prefix(self, ctx, *prefixes):
prefixes = list(prefixes)
if prefixes:
if prefixes := list(prefixes):
self.settings.set(ctx.guild, 'prefixes', prefixes)
prefixes = self.settings.get(ctx.guild, 'prefixes')
await ctx.send(f'Server prefixes: {self.format_prefixes(prefixes)}')
@@ -86,9 +84,7 @@ class Settings(commands.Cog):
else:
self.bot.settings.set(ctx.guild, 'channels.text', None)
textchannels = self.bot.settings.get(ctx.guild, 'channels.text')
if textchannels:
if textchannels := self.bot.settings.get(ctx.guild, 'channels.text'):
embed = discord.Embed(title='Music command channels set', color=ctx.me.color)
channels = [ctx.guild.get_channel(channel) for channel in textchannels]
mentioned = [channel.mention for channel in channels if channel is not None]
@@ -108,9 +104,7 @@ class Settings(commands.Cog):
else:
self.bot.settings.set(ctx.guild, 'channels.music', None)
musicchannels = self.bot.settings.get(ctx.guild, 'channels.music')
if musicchannels:
if musicchannels := self.bot.settings.get(ctx.guild, 'channels.music'):
embed = discord.Embed(title='Music channels set', color=ctx.me.color)
channels = [ctx.guild.get_channel(channel) for channel in musicchannels]
mentioned = [channel.name for channel in channels if channel is not None]
@@ -130,9 +124,7 @@ class Settings(commands.Cog):
else:
self.bot.settings.set(ctx.guild, 'channels.listen_only', None)
listenchannels = self.bot.settings.get(ctx.guild, 'channels.listen_only')
if listenchannels:
if listenchannels := self.bot.settings.get(ctx.guild, 'channels.listen_only'):
embed = discord.Embed(title='Listen only channels set', color=ctx.me.color)
channels = [ctx.guild.get_channel(channel) for channel in listenchannels]
mentioned = [channel.name for channel in channels if channel is not None]
@@ -152,9 +144,7 @@ class Settings(commands.Cog):
else:
self.bot.settings.set(ctx.guild, 'roles.dj', None)
djroles = self.bot.settings.get(ctx.guild, 'roles.dj')
if djroles:
if djroles := self.bot.settings.get(ctx.guild, 'roles.dj'):
embed = discord.Embed(title='DJ roles set to', color=ctx.me.color)
roles = [ctx.guild.get_role(role) for role in djroles]
mentioned = [role.name for role in roles if role is not None]
@@ -170,9 +160,7 @@ class Settings(commands.Cog):
async def set_max_track_duration(self, ctx, duration: int = None):
self.bot.settings.set(ctx.guild, 'duration.max', duration)
duration = self.bot.settings.get(ctx.guild, 'duration.max')
if duration:
if duration := self.bot.settings.get(ctx.guild, 'duration.max'):
embed = discord.Embed(title='Max track duration set to', color=ctx.me.color)
embed.description = f'{duration} minutes'
await ctx.send(embed=embed)
@@ -186,10 +174,8 @@ class Settings(commands.Cog):
async def set_track_duration_type(self, ctx, dynamic: bool = False):
self.bot.settings.set(ctx.guild, 'duration.is_dynamic', dynamic)
is_dynamic = self.bot.settings.get(ctx.guild, 'duration.is_dynamic')
embed = discord.Embed(title='Max duration type set to', color=ctx.me.color)
if is_dynamic:
if self.bot.settings.get(ctx.guild, 'duration.is_dynamic'):
embed.description = 'Dynamic'
else:
embed.description = 'Static'
@@ -214,24 +200,20 @@ class Settings(commands.Cog):
is_dynamic = self.bot.settings.get(ctx.guild, 'duration.is_dynamic', 'default_is_dynamic')
embed.add_field(name='{current.dynamicmax}', value=is_dynamic)
duration = self.bot.settings.get(ctx.guild, 'maxduration')
if duration:
if duration := self.bot.settings.get(ctx.guild, 'maxduration'):
embed.add_field(name='{current.maxduration}', value=f'{duration} minutes')
textchannels = self.bot.settings.get(ctx.guild, 'channels.text')
if textchannels:
if textchannels := self.bot.settings.get(ctx.guild, 'channels.text'):
channels = [ctx.guild.get_channel(channel) for channel in textchannels]
mentioned = [channel.mention for channel in channels if channel is not None]
embed.add_field(name='{current.textchannels}', value='\n'.join(mentioned))
voicechannels = self.bot.settings.get(ctx.guild, 'channels.music')
if voicechannels:
if voicechannels := self.bot.settings.get(ctx.guild, 'channels.music'):
channels = [ctx.guild.get_channel(channel) for channel in voicechannels]
mentioned = [channel.name for channel in channels if channel is not None]
embed.add_field(name='{current.musicchannels}', value='\n'.join(mentioned))
listenchannels = self.bot.settings.get(ctx.guild, 'channels.listen_only')
if listenchannels:
if listenchannels := self.bot.settings.get(ctx.guild, 'channels.listen_only'):
channels = [ctx.guild.get_channel(channel) for channel in listenchannels]
mentioned = [channel.name for channel in channels if channel is not None]
embed.add_field(name='{current.listenchannels}', value='\n'.join(mentioned))
cogs/utils/checks.py
-1
View File
@@ -70,6 +70,5 @@ def dj_or(alone: bool = False, track_requester: bool = False):
is_alone = False
dj = is_dj(ctx)
return dj or is_alone or requester
return commands.check(predicate)
cogs/utils/mixplayer.py
+36 -10
View File
@@ -1,4 +1,5 @@
# Discord Packages
import discord
import lavalink
from lavalink import AudioTrack, DefaultPlayer, Node
from lavalink.events import QueueEndEvent, TrackEndEvent, TrackExceptionEvent, TrackStartEvent, TrackStuckEvent
@@ -16,6 +17,7 @@ class MixPlayer(DefaultPlayer):
self.queue = MixQueue()
self.listeners = set()
self.voteables = {}
self.skip_voters = set()
self.boosted = False
@@ -55,14 +57,21 @@ class MixPlayer(DefaultPlayer):
def get_history(self):
return self.queue.history
def queue_duration(self, include_current: bool = True):
def queue_duration(self, include_current: bool = False, member: discord.Member = None, end_pos: int = None):
duration = 0
for track in self.queue:
duration += track.duration
queue = self.user_queue(member.id) if member else self.queue
for i, track in enumerate(queue):
if i == end_pos:
break
duration += int(track.duration)
remaining = self.current.duration - self.position
if include_current:
return lavalink.Utils.format_time(duration + remaining)
return lavalink.Utils.format_time(duration)
if self.current:
remaining = self.current.duration - self.position
return lavalink.utils.format_time(duration + remaining)
return lavalink.utils.format_time(duration)
async def play(self, track: AudioTrack = None, start_time: int = 0):
@@ -90,7 +99,7 @@ class MixPlayer(DefaultPlayer):
""" Plays the next track in the queue, if any. """
for i in range(pos):
_ = self.queue.pop_first()
self.skip_voters.clear()
self.clear_votes()
await self.play()
def update_listeners(self, member, voice_state):
@@ -98,26 +107,43 @@ class MixPlayer(DefaultPlayer):
vc = int(self.channel_id)
if voice_state.channel is None or voice_state.channel.id != vc:
self.listeners.discard(member)
self.skip_voters.discard(member)
self.remove_member_votes(member)
else:
if voice_state.deaf or voice_state.self_deaf:
self.listeners.discard(member)
self.skip_voters.discard(member)
self.remove_member_votes(member)
else:
self.listeners.add(member)
def clear_listeners(self):
self.listeners.clear()
def add_skipper(self, member):
def add_vote(self, category, member):
if member in self.listeners:
self.skip_voters.add(member)
if category not in self.voteables:
self.voteables[category] = set()
self.voteables[category].add(member)
def remove_member_votes(self, member):
for category, votes in self.voteables.items():
votes.discard(member)
def get_voters(self, category):
if category not in self.voteables:
self.voteables[category] = set()
return self.voteables[category]
def clear_votes(self):
for category, votes in self.voteables.items():
votes.clear()
async def handle_event(self, event):
""" Handles the given event as necessary. """
if isinstance(event, (TrackStuckEvent, TrackExceptionEvent)) or \
isinstance(event, TrackEndEvent) and event.reason == 'FINISHED':
self.skip_voters.clear()
for category, votes in self.voteables.items():
votes.clear()
await self.play()
async def bassboost(self, boost: bool = False):
cogs/utils/paginator.py
+9 -6
View File
@@ -110,19 +110,22 @@ class FieldPaginator(BasePaginator):
class QueuePaginator(TextPaginator):
def __init__(self, localizer, queue, color, user_name: str = None):
def __init__(self, localizer, player, color, member: discord.Member = None, include_current: bool = False):
self.localizer = localizer
self.queue = queue
if user_name is None:
title = localizer.format_str("{queue.length}", _length=len(queue))
queue = player.global_queue() if not member else player.user_queue(member.id, dual=True)
duration = player.queue_duration(member=member, include_current=include_current)
if member:
title = localizer.format_str("{queue.userqueue}", _user=member.display_name, _length=len(queue),
_duration=duration)
else:
title = localizer.format_str("{queue.userqueue}", _user=user_name, _length=len(queue))
title = localizer.format_str("{queue.length}", _length=len(queue), _duration=duration)
super().__init__(max_lines=10, **{"color": color, "title": title})
for index, temp in enumerate(queue):
if user_name is None:
if member is None:
track = temp
queued_track = localizer.format_str("{queue.globaltrack}", _index=index+1, _title=track.title,
_uri=track.uri, _user_id=track.requester)
cogs/utils/selector.py
+4
View File
@@ -191,6 +191,10 @@ class Selector(TextPaginator):
self.args = None
self.control_input = False
if len(results) == 1:
self.stopped = True
await self.message.clear_reactions()
return self.message, self.pages[self.current_page], results[0]
else:
self.stopped = True
await self.message.clear_reactions()
return self.message, self.pages[self.current_page], results
cogs/utils/settingsmanager.py
+21 -13
View File
@@ -1,3 +1,6 @@
# Discord Packages
import discord
import codecs
import locale as localee
import os
@@ -60,34 +63,39 @@ class Settings:
except KeyError:
return None
def set(self, guild, setting, value):
def set(self, identifier, setting, value):
""" Set value in settings, will overwrite any existing values. """
guild_id = str(guild.id)
guild_name = None
if isinstance(identifier, discord.Guild):
guild_name = identifier.name
identifier = str(identifier.id)
if guild_id not in self.settings.keys():
self.settings[guild_id] = {}
if identifier not in self.settings.keys():
self.settings[identifier] = {}
self.settings[guild_id]["_servername"] = guild.name
self._set(self.settings[guild_id], setting.split('.'), value)
if guild_name:
self.settings[identifier]["_servername"] = guild_name
self._set(self.settings[identifier], setting.split('.'), value)
with codecs.open(self._SETTINGS_PATH, 'w', encoding='utf8') as f:
yaml.dump(self.settings, f, indent=2)
def get(self, guild, setting, default=''):
def get(self, identifier, setting, default=''):
""" Gets a value from the settings if a default return value is specified
it will return the default if no setting is found. If that default is a
class attribute the value of the attribute will get returned."""
guild_id = str(guild.id)
it will return the default if no setting is found. If that default is an
attribute of the settings class, the value of the attribute will get returned."""
if isinstance(identifier, discord.Guild):
identifier = str(identifier.id)
if default and isinstance(default, str) and hasattr(self, default):
default = getattr(self, default)
elif not default:
elif default == '':
default = None
if guild_id not in self.settings.keys():
if identifier not in self.settings.keys():
return default
value = self._get(self.settings[guild_id], setting.split('.'))
value = self._get(self.settings[identifier], setting.split('.'))
if value is not None:
return value
else:
localization/en_en/commands.yaml
+6 -3
View File
@@ -72,7 +72,7 @@ volume:
args: '<vol>'
description: 'Sets the volume of the music.'
normalize:
aliases: [nomralize, normal]
aliases: [normalize, normal]
description: 'Resets the volume and the equalizer.'
boost:
aliases: [boost, bassboost]
@@ -88,6 +88,9 @@ lyrics:
scrub:
aliases: [scrub]
description: 'Shows a temporary controller/scrubber.'
reconnect:
aliases: [rc]
description: 'disconnects then reconnects the bot in case of bugged songs'
# Misc
info:
@@ -183,8 +186,8 @@ settings:
description: 'Set the vote threshold required to skip a song.'
# Cogs
cogs:
aliases: [cogs]
cogmanager:
aliases: [cogmanager]
description: 'cog management'
sub_commands:
load:
localization/en_en/music.yaml
+3 -2
View File
@@ -21,8 +21,8 @@ now: Now Playing
queue:
empty: The queue is empty
user_empty: "{_user}'s queue is empty"
length: "**Queue** `{_length} songs`"
userqueue: "**{_user}\\'s queue** `{_length}`"
length: "**Queue** `{_length}` songs `{_duration}`"
userqueue: "**{_user}\\'s queue** `{_length}` songs `{_duration}`"
globaltrack: "`{_index}.` **[{_title}]({_uri})** _by <@{_user_id}>_"
usertrack: "`{_index}({_globalindex}).` **[{_title}]({_uri})**"
pageindicator: Viewing page {_current}/{_total}
@@ -81,3 +81,4 @@ errors:
need_permission: I need the `CONNECT` and `SPEAK` permissions
right_channel: You need to be in the right voice channel
my_channel: You need to be in my voicechannel
full_channel: The channel is currently full
localization/en_en/nodemanager.yaml
-2
View File
@@ -1,2 +0,0 @@
pageindicator: Viewing page {_current}/{_total} | Type "{_prefix}help <command>" for more information
notcommand: Looks like "{_command}" is not a command or category.
localization/nb_no/commands.yaml
+2 -2
View File
@@ -165,8 +165,8 @@ settings:
description: 'Setter grensen for hvor mange prosent som må stemme for å hoppe over en sang.'
# Cogs
cogs:
aliases: [cogs]
cogmanager:
aliases: [cogmanager]
description: 'behandle cogs'
sub_commands:
load:
localization/nb_no/music.yaml
+3 -2
View File
@@ -21,8 +21,8 @@ now: Spiller nå
queue:
empty: Køen er tom
user_empty: "{_user} sin kø er tom"
length: "**Kø** `{_length} sanger`"
userqueue: "**{_user} sin kø** `{_length}` sanger"
length: "**Kø** `{_length}` sanger `{_duration}`"
userqueue: "**{_user} sin kø** `{_length}` sanger `{_duration}`"
globaltrack: "`{_index}.` **[{_title}]({_uri})** _av <@{_user_id}>_"
usertrack: "`{_index}({_globalindex}).` **[{_title}]({_uri})**"
pageindicator: Viser side {_current}/{_total}
@@ -81,3 +81,4 @@ errors:
need_permission: Jeg trenger tillgang til `CONNECT` og `SPEAK`
right_channel: Du må være i riktig kanal
my_channel: Du må være i samme stemmekanal som meg
full_channel: Den kanalen er full
localization/nb_no/nodemanager.yaml
-2
View File
@@ -1,2 +0,0 @@
pageindicator: Viser side {_current}/{_total} | Skriv "{_prefix}hjelp <kommando>" for mer informasjon
notcommand: Ser ut som "{_command}" ikke er en kategori eller en kommando.
localization/we_eb/commands.yaml
+3 -3
View File
@@ -72,7 +72,7 @@ volume:
args: '<vol>'
description: 'Sets the vowume of the music.'
normalize:
aliases: [nomralize, normal]
aliases: [normalize, normal]
description: 'Resets the volume and the equalizer.'
boost:
aliases: [boost, bassboost]
@@ -165,8 +165,8 @@ settings:
description: 'Set the vote thweshowd wequiwed to skip a sonyg.'
# Cogs
cogs:
aliases: [cogs]
cogmanager:
aliases: [cogmanager]
description: 'cog management'
sub_commands:
load: