diff --git a/bot.py b/bot.py index bcf9336..b222e06 100644 --- a/bot.py +++ b/bot.py @@ -29,6 +29,7 @@ from tunebot.redis import GlobalRedisPlaylistSource from tunebot.redis import RedisAutoJoin from tunebot.redis import RedisPlaylistSource from utils.assets import process_colours +from lavalink_player import CustomPlayer from utils.log import logger if TYPE_CHECKING: @@ -129,7 +130,7 @@ class TuneBot(commands.Bot): def create_lavalink(self, user_id: int) -> "lavalink.Client": cfg = self.config["lavalink"] - client: lavalink.Client = lavalink.Client(user_id) + client: lavalink.Client = lavalink.Client(user_id, player=CustomPlayer) client.add_node( cfg["host"], cfg["port"], diff --git a/cogs/music/__init__.py b/cogs/music/__init__.py index 174bf52..f53b720 100644 --- a/cogs/music/__init__.py +++ b/cogs/music/__init__.py @@ -7,13 +7,13 @@ from discord.ext import commands from bot import TuneBot from cogs.music import helper +from cogs.music import track_embed from cogs.music.interactions import MusicCommands from cogs.music.interactions import QUEUE_SIZE from cogs.music.voice_client import LavalinkVoiceClient from utils.classes import BaseCog from utils.log import logger - if typing.TYPE_CHECKING: from bot import TuneBot @@ -41,25 +41,29 @@ class MusicCog(BaseCog, name="Music"): if not player.is_playing: await helper.fill_player_queue(self.bot, player, QUEUE_SIZE) await player.play() - except: + except: logger.error(f"Failed to autojoin guild {guild_id}.") - async def track_hook(self, event: lavalink.Event): if isinstance(event, lavalink.events.QueueEndEvent): - # When this track_hook receives a "QueueEndEvent" from lavalink.py - # it indicates that there are no tracks left in the player's queue. - # To save on resources, we can tell the bot to disconnect from the voicechannel. guild_id = event.player.guild_id guild = self.bot.get_guild(guild_id) await guild.voice_client.disconnect(force=True) - elif isinstance(event, lavalink.events.TrackStartEvent): + elif isinstance(event, lavalink.events.TrackEndEvent): channel_id = int(event.player.fetch("channel")) channel: TextChannel = self.bot.get_channel(channel_id) - embed = helper.create_track_embed(event.player.current) - await channel.send(embed=embed) - elif isinstance(event, lavalink.events.TrackEndEvent): await helper.fill_player_queue(self.bot, event.player, 1) + event.player.append_history(event.track) + + if event.reason == "FINISHED": + queue = event.player.queue[1:] + current = event.player.queue[0] + else: + queue = event.player.queue + current = event.player.current + + embed = track_embed.create_track_embed(current, queue, event.player.history) + await channel.send(embed=embed) async def setup(bot: "TuneBot"): diff --git a/cogs/music/helper.py b/cogs/music/helper.py index 224c72d..28f6e0a 100644 --- a/cogs/music/helper.py +++ b/cogs/music/helper.py @@ -10,10 +10,11 @@ from utils.log import logger if typing.TYPE_CHECKING: from lavalink import DefaultPlayer + from lavalink_player import CustomPlayer from bot import TuneBot -def get_player(bot: "TuneBot", guild_id: int) -> "DefaultPlayer": +def get_player(bot: "TuneBot", guild_id: int) -> "CustomPlayer": if player := bot.lavalink.player_manager.get(guild_id=guild_id): return player @@ -37,24 +38,6 @@ async def fill_player_queue( player.add(requester=bot.user.id, track=track) -def create_track_embed(track: lavalink.models.AudioTrack) -> discord.Embed: - embed = discord.Embed( - title="Now playing...", - colour=colors["embed"], - ) - embed.description = f"[{track.title}]({track.uri})" - embed.set_thumbnail(url=f"https://i3.ytimg.com/vi/{track.identifier}/mqdefault.jpg") - - try: - duration = str(datetime.timedelta(milliseconds=int(track.duration))) - except OverflowError: - duration = "🔴 LIVE" - - embed.add_field(name="Duration", value=duration) - embed.add_field(name="Author", value=track.author) - return embed - - async def ensure_voice( permissions: discord.Permissions, player: "DefaultPlayer", @@ -77,4 +60,4 @@ async def ensure_voice( raise Exception("You need to be in my voicechannel") -__all__ = ("fill_player_queue", "create_track_embed", "get_player", "ensure_voice") +__all__ = ("fill_player_queue", "get_player", "ensure_voice") diff --git a/cogs/music/interactions.py b/cogs/music/interactions.py index d4d5a51..f7a89b5 100644 --- a/cogs/music/interactions.py +++ b/cogs/music/interactions.py @@ -6,11 +6,12 @@ from discord import Interaction from bot import config from cogs.music import helper from utils.embed import create_embed +from cogs.music import track_embed if typing.TYPE_CHECKING: from bot import TuneBot -QUEUE_SIZE = config["queue_buffer_size"] + 1 +QUEUE_SIZE = config["queue_buffer_size"] class MusicCommands(app_commands.Group): @@ -46,7 +47,7 @@ class MusicCommands(app_commands.Group): if not player.is_playing: await player.play() - embed.title = "*⃣ | Connected." + embed = track_embed.create_track_embed(player.current, player.queue, []) await ctx.response.send_message(embed=embed) @app_commands.command( @@ -84,7 +85,9 @@ class MusicCommands(app_commands.Group): await ctx.response.send_message(embed=embed) return - embed = helper.create_track_embed(player.current) + embed = track_embed.create_track_embed( + player.current, player.queue, player.history + ) await ctx.response.send_message(embed=embed) @app_commands.command(name="queue", description="See what's ahead") diff --git a/cogs/music/track_embed.py b/cogs/music/track_embed.py new file mode 100644 index 0000000..2a6096f --- /dev/null +++ b/cogs/music/track_embed.py @@ -0,0 +1,60 @@ +import datetime +import typing +import discord +from bot import colors +from cogs.music import helper + +if typing.TYPE_CHECKING: + from lavalink.models import AudioTrack + + +def format_track(track: "AudioTrack", max_length: int = 0): + try: + duration = str(datetime.timedelta(milliseconds=int(track.duration))) + except OverflowError: + duration = "0:00:00" + + if max_length == 0: + return f"`{duration}` [{track.title}]({track.uri})" + + max_length -= len(duration) + 1 + if len(track.title) > max_length: + track_title = track.title[: max_length - 3] + "..." + else: + track_title = track.title + + return f"`{duration}` [{track_title}]({track.uri})" + + +def create_track_embed( + current: "AudioTrack", + queue: typing.Sequence["AudioTrack"], + history: typing.Sequence["AudioTrack"], +) -> discord.Embed: + embed_width = 52 + embed = discord.Embed( + title="Now playing...", + colour=colors["embed"], + ) + + embed.description = format_track(current, max_length=embed_width) + + if len(queue) > 0: + upcoming_fmt = "\n".join( + format_track(track, max_length=embed_width) for track in queue + ) + else: + upcoming_fmt = "No tracks have been queued yet..." + embed.add_field(name="Coming up", value=upcoming_fmt, inline=False) + + if len(history) > 0: + history_fmt = "\n".join( + format_track(track, max_length=embed_width) for track in history + ) + else: + history_fmt = "No history yet..." + embed.add_field(name="Previously played", value=history_fmt, inline=False) + + embed.set_image(url=f"https://i3.ytimg.com/vi/{current.identifier}/mqdefault.jpg") + embed.set_footer(text=f"Uploaded by: {current.author}") + return embed diff --git a/config.json.sample b/config.json.sample index 2bca8c9..f5f2bbc 100644 --- a/config.json.sample +++ b/config.json.sample @@ -36,5 +36,6 @@ "cogs.music" ], "queue_buffer_size": 5, + "history_size": 2, "slash_descriptions": {} } diff --git a/lavalink_player.py b/lavalink_player.py new file mode 100644 index 0000000..eeb1cc4 --- /dev/null +++ b/lavalink_player.py @@ -0,0 +1,21 @@ +import typing +from lavalink import DefaultPlayer + +if typing.TYPE_CHECKING: + from lavalink import Node + from lavalink.models import AudioTrack + + +class CustomPlayer(DefaultPlayer): + def __init__(self, guild_id: int, node: "Node"): + from bot import config + + self.history: typing.List["AudioTrack"] = [] + self.max_size = config["history_size"] + + super().__init__(guild_id, node) + + def append_history(self, track: "AudioTrack"): + self.history.insert(0, track) + if len(self.history) > self.max_size: + self.history.pop(-1) diff --git a/utils/EmbedGenerator.py b/utils/EmbedGenerator.py deleted file mode 100644 index 1ac67e5..0000000 --- a/utils/EmbedGenerator.py +++ /dev/null @@ -1,47 +0,0 @@ -from typing import Optional -from typing import Union - -import discord -from discord import Embed -from discord.ext.commands import Context - - -class EmbedGenerator: - @staticmethod - async def Error(ctx: Context, message: str, **kwargs) -> Embed: - color = ctx.bot.colors["embed"] - em = Embed(title="Error:", description=message, color=color) - return await EmbedGenerator.SendWithFooter(ctx, em, **kwargs) - - @staticmethod - async def Message( - ctx: Context, title: str, message: Optional[str] = "", **kwargs - ) -> Embed: - color = ctx.bot.colors["embed"] - em = Embed(title=title, description=message, color=color) - return await EmbedGenerator.SendWithFooter(ctx, em, **kwargs) - - @staticmethod - async def Image( - ctx: Context, title: str, url: str, message: Optional[str] = "", **kwargs - ) -> Embed: - color = ctx.bot.colors["embed"] - em = Embed(title=title, description=message, url=url, color=color) - em.set_image(url=url) - return await EmbedGenerator.SendWithFooter(ctx, em, **kwargs) - - @staticmethod - async def Title(ctx: Context, title: str, **kwargs) -> Embed: - color = ctx.bot.colors["embed"] - em = Embed(title=title, color=color) - return await EmbedGenerator.SendWithFooter(ctx, em, **kwargs) - - @staticmethod - async def SendWithFooter( - ctx: Context, em: Embed, **kwargs - ) -> Union[discord.Message, Embed]: - avatar = ctx.author.avatar.with_static_format("jpeg") - em.set_footer(text=f"Requested by: {ctx.author}", icon_url=avatar) - if kwargs.get("no_send", False): - return em - return await ctx.send(embed=em, **kwargs) diff --git a/utils/exceptions.py b/utils/exceptions.py deleted file mode 100644 index b6a08e0..0000000 --- a/utils/exceptions.py +++ /dev/null @@ -1,12 +0,0 @@ -from discord.embeds import Embed -from discord.ext.commands import CommandError - -from context import CustomContext - - -class EmbeddedCommandException(CommandError): - def __init__(self, embed: Embed) -> None: - self.embed = embed - - async def send(self, ctx: CustomContext): - await ctx.send(embed=self.embed) diff --git a/utils/paginator.py b/utils/paginator.py deleted file mode 100644 index e574a93..0000000 --- a/utils/paginator.py +++ /dev/null @@ -1,843 +0,0 @@ -# Original work Copyright (c) 2015 Rapptz (https://github.com/Rapptz/RoboDanny) -# Modified work Copyright (c) 2017 Perry Fraser -# -# Licensed under the MIT License. https://opensource.org/licenses/MIT -# Stolen line for line from paginator.py in R. Danny's code -# Added formatting and lots of blocking of inspections -import asyncio -import copy -import inspect -import itertools -import re - -import discord - - -class CannotPaginate(Exception): - pass - - -class Pages: - """Implements a paginator that queries the user for the - pagination interface. - Pages are 1-index based, not 0-index based. - If the user does not reply within 2 minutes then the pagination - interface exits automatically. - Parameters - ------------ - ctx: Context - The context of the command. - entries: List[str] - A list of entries to paginate. - per_page: int - How many entries show up per page. - show_entry_count: bool - Whether to show an entry count in the footer. - Attributes - ----------- - embed: discord.Embed - The embed object that is being used to send pagination info. - Feel free to modify this externally. Only the description, - footer fields, and colour are internally modified. - permissions: discord.Permissions - Our permissions for the channel. - """ - - def __init__( - self, ctx, *, entries, per_page=12, show_entry_count=True, hide_no_results=False - ): - self.hide_no_results = hide_no_results - self.bot = ctx.bot - self.entries = entries - self.message = ctx.message - self.channel = ctx.channel - self.author = ctx.author - self.per_page = per_page - pages, left_over = divmod(len(self.entries), self.per_page) - if left_over: - pages += 1 - self.maximum_pages = pages - self.embed = discord.Embed(color=0xDEADBF) - self.paginating = len(entries) > per_page - self.show_entry_count = show_entry_count - self.reaction_emojis = [ - ( - "\N{BLACK LEFT-POINTING DOUBLE TRIANGLE WITH VERTICAL BAR}", - self.first_page, - ), - ("\N{BLACK LEFT-POINTING TRIANGLE}", self.previous_page), - ("\N{BLACK RIGHT-POINTING TRIANGLE}", self.next_page), - ( - "\N{BLACK RIGHT-POINTING DOUBLE TRIANGLE WITH VERTICAL BAR}", - self.last_page, - ), - ("\N{INPUT SYMBOL FOR NUMBERS}", self.numbered_page), - ("\N{BLACK SQUARE FOR STOP}", self.stop_pages), - ("\N{INFORMATION SOURCE}", self.show_help), - ] - - if ctx.guild is not None: - self.permissions = self.channel.permissions_for(ctx.guild.me) - else: - self.permissions = self.channel.permissions_for(ctx.bot.user) - - if not self.permissions.embed_links: - raise CannotPaginate("Bot does not have embed links permission.") - - if not self.permissions.send_messages: - raise CannotPaginate("Bot cannot send messages.") - - if self.paginating: - # verify we can actually use the pagination session - if not self.permissions.add_reactions: - raise CannotPaginate("Bot does not have add reactions permission.") - - if not self.permissions.read_message_history: - raise CannotPaginate( - "Bot does not have Read Message History permission." - ) - - def get_page(self, page): - base = (page - 1) * self.per_page - return self.entries[base : base + self.per_page] - - async def show_page(self, page, *, first=False): - # noinspection PyAttributeOutsideInit - self.current_page = page - entries = self.get_page(page) - p = [] - for index, entry in enumerate(entries, 1 + ((page - 1) * self.per_page)): - p.append(f"{index}. {entry}") - - if self.maximum_pages > 1: - if self.show_entry_count: - text = ( - f"Page {page}/{self.maximum_pages}" - f" ({len(self.entries)} entries)" - ) - else: - text = f"Page {page}/{self.maximum_pages}" - - self.embed.set_footer(text=text) - - if not self.paginating: - self.embed.description = "\n".join(p) - return await self.channel.send(embed=self.embed) - - if not first: - self.embed.description = "\n".join(p) - await self.message.edit(embed=self.embed) - return - - p.append("") - p.append("Confused? React with \N{INFORMATION SOURCE} for more info.") - self.embed.description = "\n".join(p) - self.message = await self.channel.send(embed=self.embed) - - await self.message.add_reaction("🔣") - - async def add_rest_reactions(self): - await self.message.remove_reaction("🔣", self.message.guild.me) - for (reaction, _) in self.reaction_emojis: - if self.maximum_pages == 2 and reaction in ("\u23ed", "\u23ee"): - # no |<< or >>| buttons if we only have two pages - # we can't forbid it if someone ends up using it but remove - # it from the default set - continue - - await self.message.add_reaction(reaction) - - async def checked_show_page(self, page): - if page != 0 and page <= self.maximum_pages: - await self.show_page(page) - - async def first_page(self): - """goes to the first page""" - await self.show_page(1) - - async def last_page(self): - """goes to the last page""" - await self.show_page(self.maximum_pages) - - async def next_page(self): - """goes to the next page""" - await self.checked_show_page(self.current_page + 1) - - async def previous_page(self): - """goes to the previous page""" - await self.checked_show_page(self.current_page - 1) - - async def show_current_page(self): - if self.paginating: - await self.show_page(self.current_page) - - async def numbered_page(self): - """lets you type a page number to go to""" - # noinspection PyListCreation - to_delete = [] - to_delete.append(await self.channel.send("What page do you want to go to?")) - - def message_check(m): - return ( - m.author == self.author - and self.channel == m.channel - and m.content.isdigit() - ) - - try: - msg = await self.bot.wait_for("message", check=message_check, timeout=30.0) - except asyncio.TimeoutError: - to_delete.append(await self.channel.send("Took too long.")) - await asyncio.sleep(5) - else: - page = int(msg.content) - to_delete.append(msg) - if page != 0 and page <= self.maximum_pages: - await self.show_page(page) - else: - to_delete.append( - await self.channel.send( - f"Invalid page given. ({page}/{self.maximum_pages})" - ) - ) - await asyncio.sleep(5) - - # noinspection PyBroadException - try: - await self.channel.delete_messages(to_delete) - except Exception: - pass - - async def show_help(self): - """shows this message""" - messages = [ - "Welcome to the interactive paginator!\n", - "This interactively allows you to see pages " - "of text by navigating with " - "reactions. They are as follows:\n", - ] - - for (emoji, func) in self.reaction_emojis: - messages.append(f"{emoji} {func.__doc__}") - - self.embed.description = "\n".join(messages) - self.embed.clear_fields() - self.embed.set_footer( - text=f"We were on page {self.current_page} before this message." - ) - await self.message.edit(embed=self.embed) - - async def go_back_to_current_page(): - await asyncio.sleep(60.0) - await self.show_current_page() - - self.bot.loop.create_task(go_back_to_current_page()) - - async def stop_pages(self): - """stops the interactive pagination session""" - await self.message.delete() - self.paginating = False - - def react_check(self, reaction, user): - if user is None or user.id != self.author.id: - return False - - if reaction.message.id != self.message.id: - return False - - if reaction.emoji == "🔣": - self.match = self.add_rest_reactions - return True - - for (emoji, func) in self.reaction_emojis: - if reaction.emoji == emoji: - # noinspection PyAttributeOutsideInit - self.match = func - return True - return False - - async def paginate(self): - """Actually paginate the entries and - run the interactive loop if necessary.""" - if not self.entries and not self.hide_no_results: - # I just say no results found because that's my most common use - # case. - return await self.channel.send("No results found.") - - first_page = self.show_page(1, first=True) - if not self.paginating: - await first_page - else: - # allow us to react to reactions right away if we're paginating - self.bot.loop.create_task(first_page) - - while self.paginating: - try: - reaction, user = await self.bot.wait_for( - "reaction_add", check=self.react_check, timeout=120.0 - ) - except asyncio.TimeoutError: - self.paginating = False - # noinspection PyBroadException - try: - await self.message.clear_reactions() - except Exception: - pass - finally: - break - - # noinspection PyBroadException - try: - await self.message.remove_reaction(reaction, user) - except Exception: - pass # can't remove it so don't bother doing so - - await self.match() - - -class EmbedPages: - """Similar to Pages, but you use [`discord.Embed`]""" - - def __init__(self, ctx, *, embeds): - self.bot = ctx.bot - self.embeds = embeds - self.message = ctx.message - self.channel = ctx.channel - self.author = ctx.author - pages = len(self.embeds) - self.maximum_pages = pages - self.paginating = len(embeds) > 1 - self.reaction_emojis = [ - ( - "\N{BLACK LEFT-POINTING DOUBLE TRIANGLE WITH VERTICAL BAR}", - self.first_page, - ), - ("\N{BLACK LEFT-POINTING TRIANGLE}", self.previous_page), - ("\N{BLACK RIGHT-POINTING TRIANGLE}", self.next_page), - ( - "\N{BLACK RIGHT-POINTING DOUBLE TRIANGLE WITH VERTICAL BAR}", - self.last_page, - ), - ("\N{INPUT SYMBOL FOR NUMBERS}", self.numbered_page), - ("\N{BLACK SQUARE FOR STOP}", self.stop_pages), - ("\N{INFORMATION SOURCE}", self.show_help), - ] - - if ctx.guild is not None: - self.permissions = self.channel.permissions_for(ctx.guild.me) - else: - self.permissions = self.channel.permissions_for(ctx.bot.user) - - if not self.permissions.embed_links: - raise CannotPaginate("Bot does not have embed links permission.") - - if not self.permissions.send_messages: - raise CannotPaginate("Bot cannot send messages.") - - if self.paginating: - # verify we can actually use the pagination session - if not self.permissions.add_reactions: - raise CannotPaginate("Bot does not have add reactions permission.") - - if not self.permissions.read_message_history: - raise CannotPaginate( - "Bot does not have Read Message History permission." - ) - - async def show_page(self, page, *, first=False): - # noinspection PyAttributeOutsideInit - self.current_page = page - embed = copy.copy(self.embeds[page - 1]) - p = [] - - if self.maximum_pages > 1: - text = f"Page {page}/{self.maximum_pages}" - - embed.set_footer(text=text) - - if not self.paginating: - return await self.channel.send(embed=embed) - - if not first: - return await self.message.edit(embed=embed) - - p.append("") - p.append("Confused? React with \N{INFORMATION SOURCE} for more info.") - embed.description = ( - "" if embed.description == discord.Embed.Empty else embed.description - ) - embed.description += "\n".join(p) - self.message = await self.channel.send(embed=embed) - - await self.message.add_reaction("🔣") - - async def add_rest_reactions(self): - await self.message.remove_reaction("🔣", self.message.guild.me) - for (reaction, _) in self.reaction_emojis: - if self.maximum_pages == 2 and reaction in ("\u23ed", "\u23ee"): - # no |<< or >>| buttons if we only have two pages - # we can't forbid it if someone ends up using it but remove - # it from the default set - continue - - await self.message.add_reaction(reaction) - - async def checked_show_page(self, page): - if page != 0 and page <= self.maximum_pages: - await self.show_page(page) - - async def first_page(self): - """goes to the first page""" - await self.show_page(1) - - async def last_page(self): - """goes to the last page""" - await self.show_page(self.maximum_pages) - - async def next_page(self): - """goes to the next page""" - await self.checked_show_page(self.current_page + 1) - - async def previous_page(self): - """goes to the previous page""" - await self.checked_show_page(self.current_page - 1) - - async def show_current_page(self): - if self.paginating: - await self.show_page(self.current_page) - - async def numbered_page(self): - """lets you type a page number to go to""" - # noinspection PyListCreation - to_delete = [] - to_delete.append(await self.channel.send("What page do you want to go to?")) - - def message_check(m): - return ( - m.author == self.author - and self.channel == m.channel - and m.content.isdigit() - ) - - try: - msg = await self.bot.wait_for("message", check=message_check, timeout=30.0) - except asyncio.TimeoutError: - to_delete.append(await self.channel.send("Took too long.")) - await asyncio.sleep(5) - else: - page = int(msg.content) - to_delete.append(msg) - if page != 0 and page <= self.maximum_pages: - await self.show_page(page) - else: - to_delete.append( - await self.channel.send( - f"Invalid page given. ({page}/{self.maximum_pages})" - ) - ) - await asyncio.sleep(5) - - # noinspection PyBroadException - try: - await self.channel.delete_messages(to_delete) - except Exception: - pass - - async def show_help(self): - """shows this message""" - messages = [ - "Welcome to the interactive paginator!\n", - "This interactively allows you to see pages " - "of text by navigating with " - "reactions. They are as follows:\n", - ] - - for (emoji, func) in self.reaction_emojis: - messages.append(f"{emoji} {func.__doc__}") - - embed = discord.Embed() - - embed.description = "\n".join(messages) - embed.clear_fields() - embed.set_footer( - text=f"We were on page {self.current_page} before this message." - ) - await self.message.edit(embed=embed) - - async def go_back_to_current_page(): - await asyncio.sleep(60.0) - await self.show_current_page() - - self.bot.loop.create_task(go_back_to_current_page()) - - async def stop_pages(self): - """stops the interactive pagination session""" - await self.message.delete() - self.paginating = False - - def react_check(self, reaction, user): - if user is None or user.id != self.author.id: - return False - - if reaction.message.id != self.message.id: - return False - - if reaction.emoji == "🔣": - self.match = self.add_rest_reactions - return True - - for (emoji, func) in self.reaction_emojis: - if reaction.emoji == emoji: - # noinspection PyAttributeOutsideInit - self.match = func - return True - return False - - async def paginate(self): - """Actually paginate the entries and - run the interactive loop if necessary.""" - first_page = self.show_page(1, first=True) - if not self.paginating: - await first_page - else: - # allow us to react to reactions right away if we're paginating - self.bot.loop.create_task(first_page) - - while self.paginating: - try: - reaction, user = await self.bot.wait_for( - "reaction_add", check=self.react_check, timeout=120.0 - ) - except asyncio.TimeoutError: - self.paginating = False - # noinspection PyBroadException - try: - await self.message.clear_reactions() - except Exception: - pass - finally: - break - - # noinspection PyBroadException - try: - await self.message.remove_reaction(reaction, user) - except Exception: - pass # can't remove it so don't bother doing so - - await self.match() - - -class FieldPages(Pages): - """Similar to Pages except entries should be a list of - tuples having (key, value) to show as embed fields instead. - """ - - async def show_page(self, page, *, first=False): - # noinspection PyAttributeOutsideInit - self.current_page = page - entries = self.get_page(page) - - self.embed.clear_fields() - self.embed.description = discord.Embed.Empty - - for key, value in entries: - self.embed.add_field(name=key, value=value, inline=False) - - if self.maximum_pages > 1: - if self.show_entry_count: - text = ( - f"Page {page}/{self.maximum_pages} " - f"({len(self.entries)} entries)" - ) - else: - text = f"Page {page}/{self.maximum_pages}" - - self.embed.set_footer(text=text) - - if not self.paginating: - return await self.channel.send(embed=self.embed) - - if not first: - await self.message.edit(embed=self.embed) - return - - self.message = await self.channel.send(embed=self.embed) - for (reaction, _) in self.reaction_emojis: - if self.maximum_pages == 2 and reaction in ("\u23ed", "\u23ee"): - # no |<< or >>| buttons if we only have two pages - # we can't forbid it if someone ends up using it but remove - # it from the default set - continue - - await self.message.add_reaction(reaction) - - -# ?help -# ?help Cog -# ?help command -# -> could be a subcommand - -_mention = re.compile(r"<@!?([0-9]{1,19})>") - - -def cleanup_prefix(bot, prefix): - m = _mention.match(prefix) - if m: - user = bot.get_user(int(m.group(1))) - if user: - return f"@{user.name} " - return prefix - - -async def _can_run(cmd, ctx): - # noinspection PyBroadException - try: - return await cmd.can_run(ctx) - except Exception: - return False - - -def _command_signature(cmd): - # this is modified from discord.py source - # which I wrote myself - - result = [cmd.qualified_name] - if cmd.usage: - result.append(cmd.usage) - return " ".join(result) - - params = cmd.clean_params - if not params: - return " ".join(result) - - for name, param in params.items(): - if param.default is not param.empty: - # We don't want None or '' to trigger the [name=value] case and - # instead it should do [name] since [name=None] or [name=] are - # not exactly useful for the user. - should_print = ( - param.default - if isinstance(param.default, str) - else param.default is not None - ) - if should_print: - result.append(f"[{name}={param.default!r}]") - else: - result.append(f"[{name}]") - elif param.kind == param.VAR_POSITIONAL: - result.append(f"[{name}...]") - else: - result.append(f"<{name}>") - - return " ".join(result) - - -class HelpPaginator(Pages): - def __init__(self, ctx, entries, *, per_page=4): - super().__init__(ctx, entries=entries, per_page=per_page, hide_no_results=True) - self.reaction_emojis.append( - ("\N{WHITE QUESTION MARK ORNAMENT}", self.show_bot_help) - ) - self.total = len(entries) - - @classmethod - async def from_cog(cls, ctx, cog): - cog_name = cog.__class__.__name__ - - # get the commands - entries = sorted(cog.get_commands(), key=lambda c: c.name) - - # remove the ones we can't run - entries = [ - cmd for cmd in entries if (await _can_run(cmd, ctx)) and not cmd.hidden - ] - - self = cls(ctx, entries) - self.title = f"{cog_name} Commands".upper() - self.description = inspect.getdoc(cog) - self.prefix = cleanup_prefix(ctx.bot, ctx.prefix) - - # no longer need the database - - return self - - @classmethod - async def from_command(cls, ctx, command): - try: - entries = sorted(command.commands, key=lambda c: c.name) - except AttributeError: - entries = [] - else: - entries = [ - cmd for cmd in entries if (await _can_run(cmd, ctx)) and not cmd.hidden - ] - - self = cls(ctx, entries) - self.title = command.signature - - if command.description: - self.description = f"{command.description}\n\n{command.help}" - else: - self.description = command.help or "No help given." - - self.prefix = cleanup_prefix(ctx.bot, ctx.prefix) - return self - - @classmethod - async def from_bot(cls, ctx): - def key(c): - return c.cog_name or "\u200bMisc" - - entries = sorted(ctx.bot.commands, key=key) - nested_pages = [] - per_page = 9 - - # 0: (cog, desc, commands) (max len == 9) - # 1: (cog, desc, commands) (max len == 9) - # ... - - for cog, commands in itertools.groupby(entries, key=key): - plausible = [ - cmd for cmd in commands if (await _can_run(cmd, ctx)) and not cmd.hidden - ] - if len(plausible) == 0: - continue - - description = ctx.bot.get_cog(cog) - if description is None: - description = discord.Embed.Empty - else: - description = inspect.getdoc(description) or discord.Embed.Empty - - nested_pages.extend( - (cog, description, plausible[i : i + per_page]) - for i in range(0, len(plausible), per_page) - ) - - self = cls(ctx, nested_pages, per_page=1) # this forces the pagination session - self.prefix = cleanup_prefix(ctx.bot, ctx.prefix) - - # swap the get_page implementation with - # one that supports our style of pagination - self.get_page = self.get_bot_page - self._is_bot = True - - # replace the actual total - self.total = sum(len(o) for _, _, o in nested_pages) - return self - - # noinspection PyAttributeOutsideInit - def get_bot_page(self, page): - cog, description, commands = self.entries[page - 1] - self.title = f"{cog} Commands" - self.description = description - return commands - - async def show_page(self, page, *, first=False): - # noinspection PyAttributeOutsideInit - self.current_page = page - entries = self.get_page(page) - - self.embed.clear_fields() - self.embed.description = self.description - self.embed.title = self.title - - # noinspection PyUnresolvedReferences - self.embed.set_footer( - text=f'Use "{self.prefix}help command" for more info on a command.' - ) - - signature = _command_signature - - for entry in entries: - self.embed.add_field( - name=signature(entry), - value=entry.short_doc or "No help given", - inline=False, - ) - - if self.maximum_pages: - self.embed.set_author( - name=f"Page {page}/{self.maximum_pages} ({self.total} commands)" - ) - - if not self.paginating: - return await self.channel.send(embed=self.embed) - - if not first: - await self.message.edit(embed=self.embed) - return - - self.message = await self.channel.send(embed=self.embed) - for (reaction, _) in self.reaction_emojis: - if self.maximum_pages == 2 and reaction in ("\u23ed", "\u23ee"): - # no |<< or >>| buttons if we only have two pages - # we can't forbid it if someone ends up using it but remove - # it from the default set - continue - - await self.message.add_reaction(reaction) - - async def show_help(self): - """shows this message""" - - self.embed.title = "Paginator help" - self.embed.description = "Hello! Welcome to the help page." - - messages = [f"{emoji} {func.__doc__}" for emoji, func in self.reaction_emojis] - self.embed.clear_fields() - self.embed.add_field( - name="What are these reactions for?", - value="\n".join(messages), - inline=False, - ) - - self.embed.set_footer( - text=f"We were on page {self.current_page} before this message." - ) - await self.message.edit(embed=self.embed) - - async def go_back_to_current_page(): - await asyncio.sleep(30.0) - await self.show_current_page() - - self.bot.loop.create_task(go_back_to_current_page()) - - async def show_bot_help(self): - """shows how to use the bot""" - - self.embed.title = "Using the bot" - self.embed.description = "Hello! Welcome to the help page." - self.embed.clear_fields() - - entries = ( - ("", "This means the argument is __**required**__."), - ("[argument]", "This means the argument is __**optional**__."), - ("[A|B]", "This means the it can be __**either A or B**__."), - ( - "[argument...]", - "This means you can have multiple arguments.\n" - "Now that you know the basics, it should be " - "noted that...\n" - "__**You do not type in the brackets!**__", - ), - ) - - self.embed.add_field( - name="How do I use this bot?", - value="Reading the bot signature is pretty simple.", - ) - - for name, value in entries: - self.embed.add_field(name=name, value=value, inline=False) - - self.embed.set_footer( - text=f"We were on page {self.current_page} before this message." - ) - await self.message.edit(embed=self.embed) - - async def go_back_to_current_page(): - await asyncio.sleep(30.0) - await self.show_current_page() - - self.bot.loop.create_task(go_back_to_current_page())