Author SHA1 Message Date
niku 5692d04952 Update config.json.sample 2022-11-14 19:02:42 +01:00
niku 1f5c541f81 Merge pull request #46 from Matthww/dev-2
push to prod
2022-11-13 19:04:29 +01:00
13 changed files with 957 additions and 144 deletions
+15 -11
View File
@@ -5,7 +5,6 @@ from typing import Dict
from typing import List from typing import List
from typing import Sequence from typing import Sequence
from typing import TYPE_CHECKING from typing import TYPE_CHECKING
import itertools
import aioredis import aioredis
import discord import discord
@@ -29,7 +28,6 @@ from tunebot.redis import GlobalRedisPlaylistSource
from tunebot.redis import RedisAutoJoin from tunebot.redis import RedisAutoJoin
from tunebot.redis import RedisPlaylistSource from tunebot.redis import RedisPlaylistSource
from utils.assets import process_colours from utils.assets import process_colours
from lavalink_player import CustomPlayer
from utils.log import logger from utils.log import logger
if TYPE_CHECKING: if TYPE_CHECKING:
@@ -48,20 +46,19 @@ if len(sys.argv) > 1:
config = json.load(open(config_path, "r", encoding="utf-8")) config = json.load(open(config_path, "r", encoding="utf-8"))
redis_prefix = config["redis_prefix"] redis_prefix = config["redis_prefix"]
colors: Dict[str, Color] = process_colours(config["colors"]) 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): class TuneBot(commands.Bot):
lavalink: "lavalink.Client" lavalink: "lavalink.Client"
invite_link: str invite_link: str
status_cycle = itertools.cycle(status_messages)
def __init__(self, config: Dict[Any, Any]): def __init__(self, config: Dict[Any, Any]):
intents = discord.Intents( intents = discord.Intents(
voice_states=True, guild_messages=True, guilds=True, messages=True voice_states=True, guild_messages=True, guilds=True, messages=True
) )
self.rpc_is_help_message = True
self.config = config self.config = config
self.initial_cog_names: List[str] = self.config.get("cogs", []) self.initial_cog_names: List[str] = self.config.get("cogs", [])
@@ -101,8 +98,8 @@ class TuneBot(commands.Bot):
self.update_status.start() self.update_status.start()
async def prefix_callable(self, _, msg: Message): async def prefix_callable(self, _, msg: Message):
logger.info(f"{self.config['prefix']=}") logger.info(f"{self.config['prefixes']=}")
return commands.when_mentioned_or(self.config["prefix"])(self, msg) return commands.when_mentioned_or(*self.config["prefixes"])(self, msg)
async def load_cogs(self, cog_names: Sequence[str]): async def load_cogs(self, cog_names: Sequence[str]):
for cog in cog_names: for cog in cog_names:
@@ -130,7 +127,7 @@ class TuneBot(commands.Bot):
def create_lavalink(self, user_id: int) -> "lavalink.Client": def create_lavalink(self, user_id: int) -> "lavalink.Client":
cfg = self.config["lavalink"] cfg = self.config["lavalink"]
client: lavalink.Client = lavalink.Client(user_id, player=CustomPlayer) client: lavalink.Client = lavalink.Client(user_id)
client.add_node( client.add_node(
cfg["host"], cfg["host"],
cfg["port"], cfg["port"],
@@ -149,11 +146,18 @@ class TuneBot(commands.Bot):
def playlist_source_context(self, ctx: "ContextLike") -> "PlaylistSource": def playlist_source_context(self, ctx: "ContextLike") -> "PlaylistSource":
return RedisPlaylistSource(ctx) return RedisPlaylistSource(ctx)
@tasks.loop(seconds=status_interval) @tasks.loop(seconds=30)
async def update_status(self): async def update_status(self):
await self.wait_until_ready() await self.wait_until_ready()
name = next(self.status_cycle)
activity = discord.Activity(name=name, type=ActivityType.listening) 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
await self.change_presence(activity=activity) await self.change_presence(activity=activity)
-14
View File
@@ -48,17 +48,3 @@ class InfoCommands(app_commands.Group):
f"Server CPU: `{cpu}`\n\n" f"Server CPU: `{cpu}`\n\n"
) )
await ctx.response.send_message(embed=embed) 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)
+9 -15
View File
@@ -7,13 +7,13 @@ from discord.ext import commands
from bot import TuneBot from bot import TuneBot
from cogs.music import helper from cogs.music import helper
from cogs.music import track_embed
from cogs.music.interactions import MusicCommands from cogs.music.interactions import MusicCommands
from cogs.music.interactions import QUEUE_SIZE from cogs.music.interactions import QUEUE_SIZE
from cogs.music.voice_client import LavalinkVoiceClient from cogs.music.voice_client import LavalinkVoiceClient
from utils.classes import BaseCog from utils.classes import BaseCog
from utils.log import logger from utils.log import logger
if typing.TYPE_CHECKING: if typing.TYPE_CHECKING:
from bot import TuneBot from bot import TuneBot
@@ -26,7 +26,6 @@ class MusicCog(BaseCog, name="Music"):
self.bot.lavalink.add_event_hook(self.track_hook) self.bot.lavalink.add_event_hook(self.track_hook)
"""
@commands.Cog.listener() @commands.Cog.listener()
async def on_ready(self): async def on_ready(self):
while not self.is_lavalink_ready(): while not self.is_lavalink_ready():
@@ -44,28 +43,23 @@ class MusicCog(BaseCog, name="Music"):
await player.play() await player.play()
except: except:
logger.error(f"Failed to autojoin guild {guild_id}.") logger.error(f"Failed to autojoin guild {guild_id}.")
"""
async def track_hook(self, event: lavalink.Event): async def track_hook(self, event: lavalink.Event):
if isinstance(event, lavalink.events.QueueEndEvent): 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_id = event.player.guild_id
guild = self.bot.get_guild(guild_id) guild = self.bot.get_guild(guild_id)
await guild.voice_client.disconnect(force=True) await guild.voice_client.disconnect(force=True)
elif isinstance(event, lavalink.events.TrackEndEvent): elif isinstance(event, lavalink.events.TrackStartEvent):
channel_id = int(event.player.fetch("channel")) channel_id = int(event.player.fetch("channel"))
channel: TextChannel = self.bot.get_channel(channel_id) channel: TextChannel = self.bot.get_channel(channel_id)
await helper.fill_player_queue(self.bot, event.player, 1) embed = helper.create_track_embed(event.player.current)
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) await channel.send(embed=embed)
elif isinstance(event, lavalink.events.TrackEndEvent):
await helper.fill_player_queue(self.bot, event.player, 1)
async def setup(bot: "TuneBot"): async def setup(bot: "TuneBot"):
+20 -3
View File
@@ -10,11 +10,10 @@ from utils.log import logger
if typing.TYPE_CHECKING: if typing.TYPE_CHECKING:
from lavalink import DefaultPlayer from lavalink import DefaultPlayer
from lavalink_player import CustomPlayer
from bot import TuneBot from bot import TuneBot
def get_player(bot: "TuneBot", guild_id: int) -> "CustomPlayer": def get_player(bot: "TuneBot", guild_id: int) -> "DefaultPlayer":
if player := bot.lavalink.player_manager.get(guild_id=guild_id): if player := bot.lavalink.player_manager.get(guild_id=guild_id):
return player return player
@@ -38,6 +37,24 @@ async def fill_player_queue(
player.add(requester=bot.user.id, track=track) 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( async def ensure_voice(
permissions: discord.Permissions, permissions: discord.Permissions,
player: "DefaultPlayer", player: "DefaultPlayer",
@@ -60,4 +77,4 @@ async def ensure_voice(
raise Exception("You need to be in my voicechannel") raise Exception("You need to be in my voicechannel")
__all__ = ("fill_player_queue", "get_player", "ensure_voice") __all__ = ("fill_player_queue", "create_track_embed", "get_player", "ensure_voice")
+3 -6
View File
@@ -6,12 +6,11 @@ from discord import Interaction
from bot import config from bot import config
from cogs.music import helper from cogs.music import helper
from utils.embed import create_embed from utils.embed import create_embed
from cogs.music import track_embed
if typing.TYPE_CHECKING: if typing.TYPE_CHECKING:
from bot import TuneBot from bot import TuneBot
QUEUE_SIZE = config["queue_buffer_size"] QUEUE_SIZE = config["queue_buffer_size"] + 1
class MusicCommands(app_commands.Group): class MusicCommands(app_commands.Group):
@@ -47,7 +46,7 @@ class MusicCommands(app_commands.Group):
if not player.is_playing: if not player.is_playing:
await player.play() await player.play()
embed = track_embed.create_track_embed(player.current, player.queue, []) embed.title = "*⃣ | Connected."
await ctx.response.send_message(embed=embed) await ctx.response.send_message(embed=embed)
@app_commands.command( @app_commands.command(
@@ -85,9 +84,7 @@ class MusicCommands(app_commands.Group):
await ctx.response.send_message(embed=embed) await ctx.response.send_message(embed=embed)
return return
embed = track_embed.create_track_embed( embed = helper.create_track_embed(player.current)
player.current, player.queue, player.history
)
await ctx.response.send_message(embed=embed) await ctx.response.send_message(embed=embed)
@app_commands.command(name="queue", description="See what's ahead") @app_commands.command(name="queue", description="See what's ahead")
-60
View File
@@ -1,60 +0,0 @@
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 discord.ext import commands
from bot import TuneBot from bot import TuneBot
#from cogs.settings.interactions import SettingCommands from cogs.settings.interactions import SettingCommands
from cogs.settings.interactions import SourceCommands from cogs.settings.interactions import SourceCommands
from context import CustomContext from context import CustomContext
from utils.classes import BaseCog from utils.classes import BaseCog
@@ -13,5 +13,5 @@ class SettingsCog(BaseCog, name="Settings"):
async def setup(bot: TuneBot): async def setup(bot: TuneBot):
await bot.add_cog(SettingsCog(bot)) 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) bot.tree.add_command(SourceCommands(bot), override=True)
+1 -2
View File
@@ -105,7 +105,7 @@ class SourceCommands(app_commands.Group):
await ctx.response.send_message(embed=embed) await ctx.response.send_message(embed=embed)
"""
class SettingCommands(app_commands.Group): class SettingCommands(app_commands.Group):
def __init__(self, bot: "TuneBot"): def __init__(self, bot: "TuneBot"):
super().__init__(name="settings", description="Tweak and customize") super().__init__(name="settings", description="Tweak and customize")
@@ -144,4 +144,3 @@ class SettingCommands(app_commands.Group):
await autojoin.disable() await autojoin.disable()
embed.title = "AutoJoin disabled" embed.title = "AutoJoin disabled"
await ctx.response.send_message(embed=embed) await ctx.response.send_message(embed=embed)
"""
+4 -9
View File
@@ -5,7 +5,9 @@
190875175460405250 190875175460405250
], ],
"manager_ids": [], "manager_ids": [],
"prefix": "ck!", "prefixes": [
"ck!"
],
"redis_url": "", "redis_url": "",
"redis_prefix": "", "redis_prefix": "",
"lavalink": { "lavalink": {
@@ -20,14 +22,7 @@
}, },
"info": { "info": {
"name": "CloudKid Radio", "name": "CloudKid Radio",
"description": "A sample bot description", "description": "A sample bot description"
"status": {
"messages": [
"bangers",
"noises 🤖"
],
"interval": 12
}
}, },
"cogs": [ "cogs": [
"jishaku", "jishaku",
-21
View File
@@ -1,21 +0,0 @@
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
@@ -0,0 +1,47 @@
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
@@ -0,0 +1,12 @@
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
@@ -0,0 +1,843 @@
# 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())