Author SHA1 Message Date
matthew f81c89119c Gotta remove shit to be compliant yaknow 2022-12-07 00:13:52 +01:00
matthew d2658578b5 Merge pull request #48 from strNophix/improved-track-embed
Rework of player embed
2022-11-14 18:56:24 +01:00
matthew d831251f13 Merge pull request #47 from strNophix/customize-status-loop
Bot status is now more customizable
2022-11-14 18:56:17 +01:00
niku 80e699cb25 Updated config.json.sample 2022-11-13 19:01:37 +01:00
niku e0472bb545 Merge remote-tracking branch 'upstream/dev-2' into improved-track-embed 2022-11-13 18:57:50 +01:00
niku 62cb4d9f94 Some ordering magic 2022-11-13 18:55:42 +01:00
niku 7fd7e5ffa7 Added history support and changed track fmting 2022-11-13 18:08:44 +01:00
niku 765b23e8fe Removed controls 2022-11-13 16:07:01 +01:00
niku 74f2cf6017 Bot status is now more customizable 2022-11-13 15:57:44 +01:00
niku ffb7dc999f New embed with queue and image 2022-11-13 15:20:47 +01:00
niku 1b837c16a3 Removed unused utils 2022-11-13 14:00:54 +01:00
13 changed files with 145 additions and 957 deletions
+11 -15
View File
@@ -5,6 +5,7 @@ from typing import Dict
from typing import List
from typing import Sequence
from typing import TYPE_CHECKING
import itertools
import aioredis
import discord
@@ -28,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:
@@ -46,19 +48,20 @@ if len(sys.argv) > 1:
config = json.load(open(config_path, "r", encoding="utf-8"))
redis_prefix = config["redis_prefix"]
colors: Dict[str, Color] = process_colours(config["colors"])
status_messages: Sequence[str] = config["info"]["status"]["messages"]
status_interval: int = config["info"]["status"]["interval"]
class TuneBot(commands.Bot):
lavalink: "lavalink.Client"
invite_link: str
status_cycle = itertools.cycle(status_messages)
def __init__(self, config: Dict[Any, Any]):
intents = discord.Intents(
voice_states=True, guild_messages=True, guilds=True, messages=True
)
self.rpc_is_help_message = True
self.config = config
self.initial_cog_names: List[str] = self.config.get("cogs", [])
@@ -98,8 +101,8 @@ class TuneBot(commands.Bot):
self.update_status.start()
async def prefix_callable(self, _, msg: Message):
logger.info(f"{self.config['prefixes']=}")
return commands.when_mentioned_or(*self.config["prefixes"])(self, msg)
logger.info(f"{self.config['prefix']=}")
return commands.when_mentioned_or(self.config["prefix"])(self, msg)
async def load_cogs(self, cog_names: Sequence[str]):
for cog in cog_names:
@@ -127,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"],
@@ -146,18 +149,11 @@ class TuneBot(commands.Bot):
def playlist_source_context(self, ctx: "ContextLike") -> "PlaylistSource":
return RedisPlaylistSource(ctx)
@tasks.loop(seconds=30)
@tasks.loop(seconds=status_interval)
async def update_status(self):
await self.wait_until_ready()
bot_prefix = self.config["prefixes"][0]
if self.rpc_is_help_message:
title = f"for {bot_prefix}connect | {bot_prefix}help"
activity = discord.Activity(name=title, type=ActivityType.watching)
else:
activity = discord.Activity(name="Some song", type=ActivityType.playing)
self.rpc_is_help_message = not self.rpc_is_help_message
name = next(self.status_cycle)
activity = discord.Activity(name=name, type=ActivityType.listening)
await self.change_presence(activity=activity)
+14
View File
@@ -48,3 +48,17 @@ class InfoCommands(app_commands.Group):
f"Server CPU: `{cpu}`\n\n"
)
await ctx.response.send_message(embed=embed)
@app_commands.command(name="tos", description="Terms of Service")
async def tos(self, ctx: Interaction):
embed = create_embed(ctx.user)
embed.title = "Terms of Service (ToS)"
embed.description = "[https://exobot.site/static/tos.html](https://exobot.site/static/tos.html)"
await ctx.response.send_message(embed=embed)
@app_commands.command(name="privacy", description="Privacy Policy")
async def privacy(self, ctx: Interaction):
embed = create_embed(ctx.user)
embed.title = "Privacy Policy"
embed.description = "[https://exobot.site/static/privacy.html](https://exobot.site/static/privacy.html)"
await ctx.response.send_message(embed=embed)
+16 -10
View File
@@ -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
@@ -26,6 +26,7 @@ class MusicCog(BaseCog, name="Music"):
self.bot.lavalink.add_event_hook(self.track_hook)
"""
@commands.Cog.listener()
async def on_ready(self):
while not self.is_lavalink_ready():
@@ -41,25 +42,30 @@ 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"):
+3 -20
View File
@@ -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")
+6 -3
View File
@@ -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")
+60
View File
@@ -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
+2 -2
View File
@@ -1,7 +1,7 @@
from discord.ext import commands
from bot import TuneBot
from cogs.settings.interactions import SettingCommands
#from cogs.settings.interactions import SettingCommands
from cogs.settings.interactions import SourceCommands
from context import CustomContext
from utils.classes import BaseCog
@@ -13,5 +13,5 @@ class SettingsCog(BaseCog, name="Settings"):
async def setup(bot: TuneBot):
await bot.add_cog(SettingsCog(bot))
bot.tree.add_command(SettingCommands(bot), override=True)
#bot.tree.add_command(SettingCommands(bot), override=True)
bot.tree.add_command(SourceCommands(bot), override=True)
+2 -1
View File
@@ -105,7 +105,7 @@ class SourceCommands(app_commands.Group):
await ctx.response.send_message(embed=embed)
"""
class SettingCommands(app_commands.Group):
def __init__(self, bot: "TuneBot"):
super().__init__(name="settings", description="Tweak and customize")
@@ -144,3 +144,4 @@ class SettingCommands(app_commands.Group):
await autojoin.disable()
embed.title = "AutoJoin disabled"
await ctx.response.send_message(embed=embed)
"""
+10 -4
View File
@@ -5,9 +5,7 @@
190875175460405250
],
"manager_ids": [],
"prefixes": [
"ck!"
],
"prefix": "ck!",
"redis_url": "",
"redis_prefix": "",
"lavalink": {
@@ -22,7 +20,14 @@
},
"info": {
"name": "CloudKid Radio",
"description": "A sample bot description"
"description": "A sample bot description",
"status": {
"messages": [
"bangers",
"noises 🤖"
],
"interval": 12
}
},
"cogs": [
"jishaku",
@@ -31,5 +36,6 @@
"cogs.music"
],
"queue_buffer_size": 5,
"history_size": 2,
"slash_descriptions": {}
}
+21
View File
@@ -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)
-47
View File
@@ -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)
-12
View File
@@ -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)
-843
View File
@@ -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 = (
("<argument>", "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())