mirror of
https://github.com/Matthww/TuneBot.git
synced 2026-09-21 21:27:48 +00:00
niku is gonna work on this
This commit is contained in:
+133
@@ -0,0 +1,133 @@
|
||||
# Byte-compiled / optimized / DLL files
|
||||
__pycache__/
|
||||
*.py[cod]
|
||||
*$py.class
|
||||
|
||||
# C extensions
|
||||
*.so
|
||||
|
||||
# Distribution / packaging
|
||||
.Python
|
||||
build/
|
||||
develop-eggs/
|
||||
dist/
|
||||
downloads/
|
||||
eggs/
|
||||
.eggs/
|
||||
lib/
|
||||
lib64/
|
||||
parts/
|
||||
sdist/
|
||||
var/
|
||||
wheels/
|
||||
pip-wheel-metadata/
|
||||
share/python-wheels/
|
||||
*.egg-info/
|
||||
.installed.cfg
|
||||
*.egg
|
||||
MANIFEST
|
||||
|
||||
# PyInstaller
|
||||
# Usually these files are written by a python script from a template
|
||||
# before PyInstaller builds the exe, so as to inject date/other infos into it.
|
||||
*.manifest
|
||||
*.spec
|
||||
|
||||
# Installer logs
|
||||
pip-log.txt
|
||||
pip-delete-this-directory.txt
|
||||
|
||||
# Unit test / coverage reports
|
||||
htmlcov/
|
||||
.tox/
|
||||
.nox/
|
||||
.coverage
|
||||
.coverage.*
|
||||
.cache
|
||||
nosetests.xml
|
||||
coverage.xml
|
||||
*.cover
|
||||
*.py,cover
|
||||
.hypothesis/
|
||||
.pytest_cache/
|
||||
|
||||
# Translations
|
||||
*.mo
|
||||
*.pot
|
||||
|
||||
# Django stuff:
|
||||
*.log
|
||||
local_settings.py
|
||||
db.sqlite3
|
||||
db.sqlite3-journal
|
||||
|
||||
# Flask stuff:
|
||||
instance/
|
||||
.webassets-cache
|
||||
|
||||
# Scrapy stuff:
|
||||
.scrapy
|
||||
|
||||
# Sphinx documentation
|
||||
docs/_build/
|
||||
|
||||
# PyBuilder
|
||||
target/
|
||||
|
||||
# Jupyter Notebook
|
||||
.ipynb_checkpoints
|
||||
|
||||
# IPython
|
||||
profile_default/
|
||||
ipython_config.py
|
||||
|
||||
# pyenv
|
||||
.python-version
|
||||
|
||||
# pipenv
|
||||
# According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
|
||||
# However, in case of collaboration, if having platform-specific dependencies or dependencies
|
||||
# having no cross-platform support, pipenv may install dependencies that don't work, or not
|
||||
# install all needed dependencies.
|
||||
#Pipfile.lock
|
||||
|
||||
# PEP 582; used by e.g. github.com/David-OConnor/pyflow
|
||||
__pypackages__/
|
||||
|
||||
# Celery stuff
|
||||
celerybeat-schedule
|
||||
celerybeat.pid
|
||||
|
||||
# SageMath parsed files
|
||||
*.sage.py
|
||||
|
||||
# Environments
|
||||
.env
|
||||
.venv
|
||||
env/
|
||||
venv/
|
||||
ENV/
|
||||
env.bak/
|
||||
venv.bak/
|
||||
|
||||
# Spyder project settings
|
||||
.spyderproject
|
||||
.spyproject
|
||||
|
||||
# Rope project settings
|
||||
.ropeproject
|
||||
|
||||
# mkdocs documentation
|
||||
/site
|
||||
|
||||
# mypy
|
||||
.mypy_cache/
|
||||
.dmypy.json
|
||||
dmypy.json
|
||||
|
||||
# Pyre type checker
|
||||
.pyre/
|
||||
|
||||
# CloudKid
|
||||
config.json
|
||||
Pipfile.lock
|
||||
Vendored
+3
@@ -0,0 +1,3 @@
|
||||
{
|
||||
"python.formatting.provider": "yapf"
|
||||
}
|
||||
@@ -0,0 +1,17 @@
|
||||
[[source]]
|
||||
name = "pypi"
|
||||
url = "https://pypi.org/simple"
|
||||
verify_ssl = true
|
||||
|
||||
[dev-packages]
|
||||
yapf = "*"
|
||||
|
||||
[packages]
|
||||
humanize = "*"
|
||||
"discord.py" = {extras = ["voice"], version = "*"}
|
||||
sqlalchemy = "*"
|
||||
aiomysql = "*"
|
||||
lavalink = "*"
|
||||
|
||||
[requires]
|
||||
python_version = "3.8"
|
||||
@@ -0,0 +1,87 @@
|
||||
import discord
|
||||
from discord.ext import commands
|
||||
import sys
|
||||
import json
|
||||
import asyncio
|
||||
import sqlalchemy as sa
|
||||
from aiomysql.sa import Engine
|
||||
from tables import get_tables
|
||||
from typing import Sequence
|
||||
from aiomysql.sa import create_engine
|
||||
from discord import ActivityType
|
||||
from discord import Status
|
||||
from typing import Mapping
|
||||
|
||||
|
||||
class Database():
|
||||
def __init__(self, engine: Engine):
|
||||
self.loop = asyncio.get_event_loop()
|
||||
self.metadata = sa.MetaData()
|
||||
self.tables = get_tables(self.metadata)
|
||||
self.engine = engine
|
||||
|
||||
@staticmethod
|
||||
async def init(dbconf: dict):
|
||||
engine = await create_engine(**dbconf, autocommit=True)
|
||||
return Database(engine)
|
||||
|
||||
|
||||
class CloudKid(commands.Bot):
|
||||
INITIAL_EXTENSIONS = [
|
||||
'cogs.owner', 'cogs.settings', 'cogs.information', 'cogs.music'
|
||||
]
|
||||
|
||||
def __init__(self, config: dict):
|
||||
intents: discord.Intents = discord.Intents.none()
|
||||
intents.voice_states = True
|
||||
intents.guild_messages = True
|
||||
|
||||
self.config = config
|
||||
self.colors = self.process_colors(config["colors"])
|
||||
|
||||
super().__init__(command_prefix=self.prefix_callable,
|
||||
description='CloudKid Radio',
|
||||
case_insensitive=True,
|
||||
fetch_offline_members=False,
|
||||
intents=intents)
|
||||
|
||||
self.database = None
|
||||
self.loop.create_task(self.async_init())
|
||||
|
||||
async def async_init(self):
|
||||
dbconf = self.config["database"]
|
||||
self.database = await Database.init(dbconf)
|
||||
|
||||
async def prefix_callable(self, _, msg):
|
||||
return commands.when_mentioned_or(*self.config["prefixes"])(self, msg)
|
||||
|
||||
async def load_cogs(self, cog_names: Sequence[str]):
|
||||
for cog in cog_names:
|
||||
try:
|
||||
self.load_extension(cog)
|
||||
print(f'Succesfully loaded extension {cog}.')
|
||||
except Exception as e:
|
||||
print(f'Failed to load extension {cog}.\n\t{e}',
|
||||
file=sys.stderr)
|
||||
|
||||
async def on_ready(self):
|
||||
print(f"Logged in as: {self.user}")
|
||||
print(f"Version: {discord.__version__}")
|
||||
print(
|
||||
f"Invite: https://discord.com/oauth2/authorize?scope=bot&client_id=257816631072391168&permissions=70642768"
|
||||
)
|
||||
text = "ck!connect | ck!help"
|
||||
await self.change_presence(activity=discord.Game(text))
|
||||
await self.load_cogs(self.INITIAL_EXTENSIONS)
|
||||
|
||||
def process_colors(
|
||||
self, colors: Mapping[str, str]) -> Mapping[str, discord.Color]:
|
||||
for name, color in colors.items():
|
||||
colors[name] = discord.Color(int(color, 16))
|
||||
return colors
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
config = json.load(open('config.json', 'r', encoding='utf-8'))
|
||||
token = config.pop("token")
|
||||
CloudKid(config).run(token, reconnect=True)
|
||||
@@ -0,0 +1,135 @@
|
||||
import discord
|
||||
from discord.ext import tasks, commands
|
||||
|
||||
import time
|
||||
import humanize
|
||||
import datetime
|
||||
import lavalink
|
||||
from utils import metadata
|
||||
from utils.paginator import HelpPaginator
|
||||
from discord import Status
|
||||
|
||||
|
||||
class InformationCog(commands.Cog, name="Information"):
|
||||
def __init__(self, bot: commands.Bot):
|
||||
self.bot = bot
|
||||
self.is_help_msg = True
|
||||
self.update_status.start()
|
||||
|
||||
@tasks.loop(seconds=30.0)
|
||||
async def update_status(self):
|
||||
await self.bot.wait_until_ready()
|
||||
title = "ck!connect | ck!help"
|
||||
if self.is_help_msg:
|
||||
title = (await metadata.fetch_metadata())["name"]
|
||||
self.is_help_msg = not self.is_help_msg
|
||||
await self.bot.change_presence(activity=discord.Game(title))
|
||||
|
||||
@commands.cooldown(rate=1, per=5, type=commands.BucketType.user)
|
||||
@commands.command(description='PONG!', aliases=['pong'])
|
||||
async def ping(self, ctx):
|
||||
"""Test the latency"""
|
||||
avatar = ctx.author.avatar_url_as(static_format='jpeg')
|
||||
emoji = discord.utils.get(ctx.bot.emojis, name='loading')
|
||||
start = time.monotonic()
|
||||
msg = await ctx.send(embed=discord.Embed(
|
||||
description=f'{emoji} Calculating ping'))
|
||||
millis = (time.monotonic() - start) * 1000
|
||||
heartbeat = ctx.bot.latency * 1000
|
||||
embed = discord.Embed(color=discord.Color.blue())
|
||||
embed.add_field(name=':heartbeat: Heartbeat',
|
||||
value=f'`{heartbeat:,.2f}ms`',
|
||||
inline=True)
|
||||
embed.add_field(name=':file_cabinet: ACK',
|
||||
value=f'`{millis:,.2f}ms`',
|
||||
inline=True)
|
||||
embed.set_footer(text=f"Requested by: {ctx.author}",
|
||||
icon_url=f"{avatar}")
|
||||
await msg.edit(embed=embed)
|
||||
|
||||
@commands.command(name='invite')
|
||||
@commands.cooldown(rate=1, per=5, type=commands.BucketType.user)
|
||||
async def invite(self, ctx):
|
||||
"""Gets the invite link!"""
|
||||
color = self.bot.colors["embed"]
|
||||
avatar = ctx.author.avatar_url_as(static_format='jpeg')
|
||||
client_id = ctx.bot.user.id
|
||||
link = f"https://discord.com/oauth2/authorize?scope=bot&client_id={client_id}&permissions=70642768"
|
||||
embed = discord.Embed(color=color)
|
||||
embed.add_field(name='Add our bot to your server:', value=link)
|
||||
embed.set_footer(text=f"Requested by: {ctx.author}",
|
||||
icon_url=f"{avatar}")
|
||||
await ctx.send(embed=embed)
|
||||
|
||||
@commands.cooldown(rate=1, per=5, type=commands.BucketType.user)
|
||||
@commands.command()
|
||||
async def wlinfo(self, ctx):
|
||||
"""Retrieve various Node/Server/Player information."""
|
||||
player = self.bot.lavalink.player_manager.get(ctx.guild.id)
|
||||
node = player.node
|
||||
|
||||
used = humanize.naturalsize(node.stats.memory_used)
|
||||
total = humanize.naturalsize(node.stats.memory_allocated)
|
||||
free = humanize.naturalsize(node.stats.memory_free)
|
||||
cpu = node.stats.cpu_cores
|
||||
|
||||
fmt = f'**WaveLink:** `{lavalink.__version__}`\n\n' \
|
||||
f'Connected to `{len(self.bot.lavalink.nodes)}` nodes.\n' \
|
||||
f'Best available Node `{self.bot.lavalink.get_best_node().__repr__()}`\n' \
|
||||
f'`{len(self.bot.lavalink.players)}` players are distributed on nodes.\n' \
|
||||
f'`{node.stats.players}` players are distributed on server.\n' \
|
||||
f'`{node.stats.playing_players}` players are playing on server.\n\n' \
|
||||
f'Server Memory: `{used}/{total}` | `({free} free)`\n' \
|
||||
f'Server CPU: `{cpu}`\n\n' \
|
||||
f'Server Uptime: `{datetime.timedelta(milliseconds=node.stats.uptime)}`'
|
||||
await ctx.send(fmt)
|
||||
|
||||
@commands.command(name="help", aliases=["about", "info"])
|
||||
@commands.cooldown(1, 1, commands.BucketType.user)
|
||||
async def about(self, ctx, command: str = None):
|
||||
"""Exobot command list"""
|
||||
if command:
|
||||
entity = self.bot.get_cog(command) or self.bot.get_command(command)
|
||||
|
||||
if entity is None:
|
||||
clean = command.replace('@', '@\u200b')
|
||||
return await ctx.send(
|
||||
f'Command or category "{clean}" not found.')
|
||||
elif isinstance(entity, discord.ext.commands.Command):
|
||||
p = await HelpPaginator.from_command(ctx, entity)
|
||||
else:
|
||||
p = await HelpPaginator.from_cog(ctx, entity)
|
||||
return await p.paginate()
|
||||
|
||||
info = self.bot.config["info"]
|
||||
title = info["name"] + " Help"
|
||||
descr = info["description"]
|
||||
color = self.bot.colors["embed"]
|
||||
|
||||
embed = discord.Embed(color=color, title=title, description=descr)
|
||||
|
||||
display_cogs = {
|
||||
"Information": ":information_source:",
|
||||
"Music": ":musical_note:",
|
||||
"Settings": ":gear:"
|
||||
}
|
||||
|
||||
for cog_name, cog_icon in display_cogs.items():
|
||||
cog = self.bot.cogs.get(cog_name)
|
||||
if not cog:
|
||||
continue
|
||||
cogname_str = f"{cog_icon} {cog_name}"
|
||||
commands = [
|
||||
f"`{cmd.name}`" for cmd in cog.get_commands() if not cmd.hidden
|
||||
]
|
||||
commands_str = ", ".join(commands)
|
||||
embed.add_field(name=cogname_str, value=commands_str, inline=False)
|
||||
|
||||
avatar = ctx.author.avatar_url_as(static_format='jpeg')
|
||||
embed.set_footer(text=f"Requested by: {ctx.author}", icon_url=avatar)
|
||||
await ctx.send(embed=embed)
|
||||
|
||||
|
||||
def setup(bot: commands.Bot):
|
||||
bot.remove_command("help")
|
||||
bot.add_cog(InformationCog(bot))
|
||||
+147
@@ -0,0 +1,147 @@
|
||||
import discord
|
||||
from discord.ext import commands
|
||||
|
||||
import time
|
||||
import lavalink
|
||||
import re
|
||||
import random
|
||||
import asyncio
|
||||
from utils.database import AutoJoin
|
||||
from utils import metadata
|
||||
from utils.EmbedGenerator import EmbedGenerator
|
||||
|
||||
|
||||
class MusicCog(commands.Cog, name="Music"):
|
||||
def __init__(self, bot: commands.Bot):
|
||||
self.bot = bot
|
||||
self.stream = "https://azuracast.exobot.site/radio/8000/radio.opus"
|
||||
|
||||
if not hasattr(bot, 'lavalink'):
|
||||
bot.lavalink = lavalink.Client(bot.user.id)
|
||||
bot.lavalink.add_node('de-1.rivalmc.net', 2333, '12345', 'eu',
|
||||
'poggers')
|
||||
bot.add_listener(self.bot.lavalink.voice_update_handler,
|
||||
'on_socket_response')
|
||||
|
||||
lavalink.add_event_hook(self.track_hook)
|
||||
|
||||
bot.loop.create_task(self.async_init())
|
||||
|
||||
async def async_init(self):
|
||||
await self.bot.wait_until_ready()
|
||||
channels = await AutoJoin.get_channels(self.bot)
|
||||
|
||||
# We startup to fast #NOTPOGGERS
|
||||
await asyncio.sleep(5)
|
||||
for x in channels:
|
||||
guild = self.bot.get_guild(x[0])
|
||||
player = self.bot.lavalink.player_manager.create(x[0],
|
||||
endpoint=str(
|
||||
guild.region))
|
||||
track = await player.node.get_tracks(self.stream)
|
||||
if not player.is_playing:
|
||||
await player.play(track["tracks"][0])
|
||||
await self.connect_to(x[0], x[1])
|
||||
|
||||
def cog_unload(self):
|
||||
""" Cog unload handler. This removes any event hooks that were registered. """
|
||||
self.bot.lavalink._event_hooks.clear()
|
||||
|
||||
async def cog_before_invoke(self, ctx):
|
||||
""" Command before-invoke handler. """
|
||||
guild_check = ctx.guild is not None
|
||||
if guild_check:
|
||||
await self.ensure_voice(ctx)
|
||||
return guild_check
|
||||
|
||||
async def ensure_voice(self, ctx):
|
||||
""" This check ensures that the bot and command author are in the same voicechannel. """
|
||||
player = self.bot.lavalink.player_manager.create(ctx.guild.id,
|
||||
endpoint=str(
|
||||
ctx.guild.region))
|
||||
should_connect = ctx.command.name in ('connect', )
|
||||
|
||||
if not ctx.author.voice or not ctx.author.voice.channel:
|
||||
raise commands.CommandInvokeError('Join a voicechannel first.')
|
||||
|
||||
if not player.is_connected:
|
||||
if not should_connect:
|
||||
raise commands.CommandInvokeError('Not connected.')
|
||||
|
||||
permissions = ctx.author.voice.channel.permissions_for(ctx.me)
|
||||
|
||||
if not permissions.connect or not permissions.speak: # Check user limit too?
|
||||
raise commands.CommandInvokeError(
|
||||
'I need the `CONNECT` and `SPEAK` permissions.')
|
||||
|
||||
player.store('channel', ctx.channel.id)
|
||||
await self.connect_to(ctx.guild.id,
|
||||
str(ctx.author.voice.channel.id))
|
||||
else:
|
||||
if int(player.channel_id) != ctx.author.voice.channel.id:
|
||||
raise commands.CommandInvokeError(
|
||||
'You need to be in my voicechannel.')
|
||||
|
||||
async def track_hook(self, event):
|
||||
if isinstance(event, lavalink.events.QueueEndEvent):
|
||||
guild_id = int(event.player.guild_id)
|
||||
await self.connect_to(guild_id, None)
|
||||
|
||||
async def connect_to(self, guild_id: int, channel_id: str):
|
||||
""" Connects to the given voicechannel ID. A channel_id of `None` means disconnect. """
|
||||
ws = self.bot._connection._get_websocket(guild_id)
|
||||
await ws.voice_state(str(guild_id), channel_id)
|
||||
|
||||
@commands.command(name='connect')
|
||||
async def connect(self, ctx):
|
||||
"""Starts vibing."""
|
||||
player = self.bot.lavalink.player_manager.get(ctx.guild.id)
|
||||
results = await player.node.get_tracks(self.stream)
|
||||
|
||||
if not results or not results['tracks']:
|
||||
return await ctx.send('Nothing found!')
|
||||
|
||||
if results['loadType'] == 'PLAYLIST_LOADED':
|
||||
tracks = results['tracks']
|
||||
|
||||
for track in tracks:
|
||||
player.add(requester=ctx.author.id, track=track)
|
||||
else:
|
||||
track = results['tracks'][0]
|
||||
track = lavalink.models.AudioTrack(track,
|
||||
ctx.author.id,
|
||||
recommended=True)
|
||||
player.add(requester=ctx.author.id, track=track)
|
||||
|
||||
if not player.is_playing:
|
||||
await player.play()
|
||||
|
||||
@commands.command(aliases=['dc'])
|
||||
async def disconnect(self, ctx):
|
||||
""" Disconnects the player from the voice channel and clears its queue. """
|
||||
player = self.bot.lavalink.player_manager.get(ctx.guild.id)
|
||||
|
||||
if not player.is_connected:
|
||||
return await ctx.send('Not connected.')
|
||||
|
||||
if not ctx.author.voice or (
|
||||
player.is_connected
|
||||
and ctx.author.voice.channel.id != int(player.channel_id)):
|
||||
return await ctx.send('You\'re not in my voicechannel!')
|
||||
|
||||
player.queue.clear()
|
||||
await player.stop()
|
||||
await self.connect_to(ctx.guild.id, None)
|
||||
|
||||
@commands.command(name='now', aliases=['playing'])
|
||||
async def now_playing(self, ctx):
|
||||
"""Stop and disconnect the player and controller."""
|
||||
np = await metadata.fetch_metadata()
|
||||
em = discord.Embed(color=self.bot.colors["embed"])
|
||||
em.set_thumbnail(url=np["thumbnail"])
|
||||
em.add_field(name="Currently playing:", value=np["name"])
|
||||
await EmbedGenerator.SendWithFooter(ctx, em)
|
||||
|
||||
|
||||
def setup(bot):
|
||||
bot.add_cog(MusicCog(bot))
|
||||
+195
@@ -0,0 +1,195 @@
|
||||
import discord
|
||||
from discord.ext import commands
|
||||
|
||||
import textwrap
|
||||
import io
|
||||
import traceback
|
||||
import asyncio
|
||||
import time
|
||||
from asyncio.subprocess import PIPE
|
||||
from io import BytesIO
|
||||
from platform import python_version
|
||||
from contextlib import redirect_stdout
|
||||
|
||||
|
||||
class OwnerCog(commands.Cog):
|
||||
def __init__(self, bot):
|
||||
self.bot = bot
|
||||
self._last_result = None
|
||||
|
||||
@staticmethod
|
||||
def cleanup_code(content):
|
||||
if content.startswith('```') and content.endswith('```'):
|
||||
return '\n'.join(content.split('\n')[1:-1])
|
||||
return content.strip('` \n')
|
||||
|
||||
# Hidden means it won't show up on the default help.
|
||||
@commands.command(name='load', hidden=True)
|
||||
@commands.is_owner()
|
||||
async def _cog_load(self, ctx, *, cog: str):
|
||||
"""Command which Loads a Module."""
|
||||
|
||||
try:
|
||||
self.bot.load_extension(cog)
|
||||
except Exception as e:
|
||||
await ctx.send(f'**`ERROR:`** {type(e).__name__} - {e}')
|
||||
else:
|
||||
await ctx.send('**`SUCCESS`**')
|
||||
|
||||
@commands.command(name='unload', hidden=True)
|
||||
@commands.is_owner()
|
||||
async def _cog_unload(self, ctx, *, cog: str):
|
||||
"""Command which Unloads a Module."""
|
||||
|
||||
try:
|
||||
self.bot.unload_extension(cog)
|
||||
except Exception as e:
|
||||
await ctx.send(f'**`ERROR:`** {type(e).__name__} - {e}')
|
||||
else:
|
||||
await ctx.send('**`SUCCESS`**')
|
||||
|
||||
@commands.command(name='reload', hidden=True)
|
||||
@commands.is_owner()
|
||||
async def _cog_reload(self, ctx, *, cog: str):
|
||||
"""Command which Reloads a Module."""
|
||||
|
||||
try:
|
||||
self.bot.unload_extension(cog)
|
||||
self.bot.load_extension(cog)
|
||||
except Exception as e:
|
||||
await ctx.send(f'**`ERROR:`** {type(e).__name__} - {e}')
|
||||
else:
|
||||
await ctx.send('**`SUCCESS`**')
|
||||
|
||||
@commands.command(name='shutdown', hidden=True)
|
||||
@commands.is_owner()
|
||||
async def shutdown(self, ctx):
|
||||
"""Command which shutdowns the bot."""
|
||||
await ctx.bot.logout()
|
||||
|
||||
@commands.is_owner()
|
||||
@commands.command(pass_context=True,
|
||||
hidden=True,
|
||||
name='eval',
|
||||
aliases=['evaluate'])
|
||||
async def _eval(self, ctx, *, body: str):
|
||||
env = {
|
||||
'bot': self.bot,
|
||||
'ctx': ctx,
|
||||
'channel': ctx.channel,
|
||||
'author': ctx.author,
|
||||
'guild': ctx.guild,
|
||||
'message': ctx.message,
|
||||
'_': self._last_result
|
||||
}
|
||||
|
||||
if "import os" in body and ctx.author.id == 190875175460405249:
|
||||
return await ctx.send("Ah ah ah, you didn't say the magic word.")
|
||||
|
||||
env.update(globals())
|
||||
|
||||
body = self.cleanup_code(body)
|
||||
stdout = io.StringIO()
|
||||
|
||||
to_compile = f'async def func():\n{textwrap.indent(body, " ")}'
|
||||
# await ctx.message.add_reaction('a:loading:452489773396000778')
|
||||
|
||||
try:
|
||||
exec(to_compile, env)
|
||||
except Exception as e:
|
||||
# await ctx.message.add_reaction('naokoerror:447495055603662849')
|
||||
fooem = discord.Embed(color=0xff0000)
|
||||
fooem.add_field(name="Code evaluation was not successful.",
|
||||
value=f'```\n{e.__class__.__name__}: {e}\n```')
|
||||
fooem.set_footer(text=f"Evaluated using Python {python_version()}",
|
||||
icon_url="http://i.imgur.com/9EftiVK.png")
|
||||
await ctx.send(embed=fooem)
|
||||
# await ctx.message.remove_reaction('a:loading:452489773396000778', member=ctx.me)
|
||||
|
||||
func = env['func']
|
||||
try:
|
||||
with redirect_stdout(stdout):
|
||||
ret = await func()
|
||||
except Exception as e:
|
||||
value = stdout.getvalue()
|
||||
# await ctx.message.add_reaction('naokoerror:447495055603662849')
|
||||
fooem = discord.Embed(color=0xff0000)
|
||||
fooem.add_field(name="Code evaluation was not successful.",
|
||||
value=f'```\n{value}{traceback.format_exc()}\n```')
|
||||
fooem.set_footer(text=f"Evaluated using Python {python_version()}",
|
||||
icon_url="http://i.imgur.com/9EftiVK.png")
|
||||
await ctx.send(embed=fooem)
|
||||
try:
|
||||
# await ctx.message.remove_reaction('a:loading:452489773396000778', member=ctx.me)
|
||||
# await ctx.message.add_reaction('naokotick:447494238872141827')
|
||||
pass
|
||||
except Exception:
|
||||
pass
|
||||
else:
|
||||
value = stdout.getvalue()
|
||||
try:
|
||||
await ctx.message.remove_reaction(
|
||||
'a:loading:452489773396000778', member=ctx.me)
|
||||
# await ctx.message.add_reaction('naokotick:447494238872141827')
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
if ret is None:
|
||||
if value:
|
||||
sfooem = discord.Embed(color=0x170041)
|
||||
sfooem.add_field(name="Code evaluation was successful!",
|
||||
value=f'```\n{value}\n```')
|
||||
sfooem.set_footer(
|
||||
text=f"Evaluated using Python {python_version()}",
|
||||
icon_url="http://i.imgur.com/9EftiVK.png")
|
||||
await ctx.send(embed=sfooem)
|
||||
else:
|
||||
self._last_result = ret
|
||||
ssfooem = discord.Embed(color=0x170041)
|
||||
ssfooem.add_field(name="Code evaluation was successful!",
|
||||
value=f'```\n{value}{ret}\n```')
|
||||
ssfooem.set_footer(
|
||||
text=f"Evaluated using Python {python_version()}",
|
||||
icon_url="http://i.imgur.com/9EftiVK.png")
|
||||
await ctx.send(embed=ssfooem)
|
||||
|
||||
@commands.is_owner()
|
||||
@commands.command(hidden=True, aliases=['exec'])
|
||||
async def execute(self, ctx, *, text: str):
|
||||
""" Do a shell command. """
|
||||
message = await ctx.send(f"Loading...")
|
||||
proc = await asyncio.create_subprocess_shell(text,
|
||||
stdin=None,
|
||||
stderr=PIPE,
|
||||
stdout=PIPE)
|
||||
out = (await proc.stdout.read()).decode('utf-8').strip()
|
||||
err = (await proc.stderr.read()).decode('utf-8').strip()
|
||||
|
||||
if not out and not err:
|
||||
await message.delete()
|
||||
return await ctx.message.add_reaction('👌')
|
||||
|
||||
content = ""
|
||||
|
||||
if err:
|
||||
content += f"Error:\r\n{err}\r\n{'-' * 30}\r\n"
|
||||
if out:
|
||||
content += out
|
||||
|
||||
if len(content) > 1500:
|
||||
try:
|
||||
data = BytesIO(content.encode('utf-8'))
|
||||
await message.delete()
|
||||
await ctx.send(content=f"The result was a bit too long..",
|
||||
file=discord.File(
|
||||
data,
|
||||
filename=f"result_{int(time.time())}.txt"))
|
||||
except asyncio.TimeoutError as e:
|
||||
await message.delete()
|
||||
return await ctx.send(e)
|
||||
else:
|
||||
await message.edit(content=f"```fix\n{content}\n```")
|
||||
|
||||
|
||||
def setup(bot):
|
||||
bot.add_cog(OwnerCog(bot))
|
||||
@@ -0,0 +1,45 @@
|
||||
from discord.ext import commands
|
||||
from utils.database import AutoJoin
|
||||
import discord
|
||||
from utils.EmbedGenerator import EmbedGenerator
|
||||
from discord.ext.commands import Context
|
||||
|
||||
|
||||
class SettingsCog(commands.Cog, name="Settings"):
|
||||
def __init__(self, bot):
|
||||
self.bot = bot
|
||||
|
||||
@commands.command(name='placeholder', hidden=True)
|
||||
@commands.is_owner()
|
||||
async def placeholder(self, ctx):
|
||||
"""Placeholder Command"""
|
||||
|
||||
await ctx.send("Pong!")
|
||||
|
||||
@commands.group(aliases=["aj"], invoke_without_command=True)
|
||||
@commands.cooldown(rate=1, per=5, type=commands.BucketType.user)
|
||||
async def autojoin(self, ctx):
|
||||
await EmbedGenerator.Message(
|
||||
ctx, "Autojoin",
|
||||
f"Usage:\n\n`{ctx.prefix}autojoin set`\n`{ctx.prefix}autojoin unset`"
|
||||
)
|
||||
|
||||
@autojoin.command(name="set")
|
||||
@commands.has_permissions(manage_channels=True)
|
||||
@commands.cooldown(rate=1, per=5, type=commands.BucketType.user)
|
||||
async def autojoin_set(self, ctx):
|
||||
vc = ctx.author.voice.channel
|
||||
await AutoJoin.update_channel(self.bot, ctx.guild.id, vc.id)
|
||||
await EmbedGenerator.Message(ctx, "Autojoin", "`enabled`")
|
||||
|
||||
@autojoin.command(name="unset")
|
||||
@commands.has_permissions(manage_channels=True)
|
||||
@commands.cooldown(rate=1, per=5, type=commands.BucketType.user)
|
||||
async def autojoin_del(self, ctx):
|
||||
vc = ctx.author.voice.channel
|
||||
await AutoJoin.del_channel(self.bot, ctx.guild.id)
|
||||
await EmbedGenerator.Message(ctx, "Autojoin", "`disabled`")
|
||||
|
||||
|
||||
def setup(bot):
|
||||
bot.add_cog(SettingsCog(bot))
|
||||
@@ -0,0 +1,26 @@
|
||||
{
|
||||
"token": "",
|
||||
"owner_ids": [194545408960102400, 190875175460405249],
|
||||
"prefixes": ["ck!"],
|
||||
"database": {
|
||||
"host": "",
|
||||
"port": 3306,
|
||||
"user": "",
|
||||
"password": "",
|
||||
"db": ""
|
||||
},
|
||||
"lavalink": {
|
||||
"host": "",
|
||||
"port": 2333,
|
||||
"password": "",
|
||||
"region": "",
|
||||
"name": ""
|
||||
},
|
||||
"colors": {
|
||||
"embed": "7289da"
|
||||
},
|
||||
"info": {
|
||||
"name": "CloudKid Radio",
|
||||
"description": "A sample bot description"
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,4 @@
|
||||
[yapf]
|
||||
based_on_style = pep8
|
||||
spaces_before_comment = 4
|
||||
split_before_logical_operator = true
|
||||
@@ -0,0 +1,7 @@
|
||||
import sqlalchemy as sa
|
||||
from .autojoin import table as autojoin
|
||||
from typing import Dict
|
||||
|
||||
|
||||
def get_tables(metadata: sa.MetaData) -> Dict[str, sa.Table]:
|
||||
return {name: table for name, table in [autojoin(metadata)]}
|
||||
@@ -0,0 +1,12 @@
|
||||
import sqlalchemy as sa
|
||||
|
||||
|
||||
def table(metadata: sa.MetaData) -> sa.Table:
|
||||
tablename = 'autojoin'
|
||||
|
||||
tableobject = sa.Table(
|
||||
tablename, metadata,
|
||||
sa.Column('guild_id', sa.BigInteger, primary_key=True),
|
||||
sa.Column('channel_id', sa.BigInteger))
|
||||
|
||||
return (tablename, tableobject)
|
||||
@@ -0,0 +1,48 @@
|
||||
from typing import Optional
|
||||
|
||||
import discord
|
||||
from discord import Embed
|
||||
from discord.ext.commands import Context
|
||||
|
||||
|
||||
class EmbedGenerator:
|
||||
@staticmethod
|
||||
async def Error(ctx: Context, message: str, **kwargs):
|
||||
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):
|
||||
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):
|
||||
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):
|
||||
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) -> discord.Message:
|
||||
avatar = ctx.author.avatar_url_as()
|
||||
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)
|
||||
@@ -0,0 +1,32 @@
|
||||
import aiomysql
|
||||
import asyncio
|
||||
from bot import CloudKid
|
||||
from aiomysql.sa import SAConnection
|
||||
from sqlalchemy import Table
|
||||
from typing import Sequence
|
||||
|
||||
|
||||
class AutoJoin:
|
||||
@staticmethod
|
||||
async def get_channels(bot: CloudKid) -> Sequence[tuple]:
|
||||
autojoin: Table = bot.database.tables["autojoin"]
|
||||
conn: SAConnection = await bot.database.engine.acquire()
|
||||
result = await conn.execute(autojoin.select())
|
||||
channels: Sequence[tuple] = await result.fetchall()
|
||||
return channels
|
||||
|
||||
@staticmethod
|
||||
async def update_channel(bot, guild_id, channel_id):
|
||||
autojoin: Table = bot.database.tables["autojoin"]
|
||||
conn: SAConnection = await bot.database.engine.acquire()
|
||||
await conn.execute(autojoin.insert().values({
|
||||
"guild_id": guild_id,
|
||||
"channel_id": channel_id
|
||||
}))
|
||||
|
||||
@staticmethod
|
||||
async def del_channel(bot, guild_id):
|
||||
autojoin: Table = bot.database.tables["autojoin"]
|
||||
conn: SAConnection = await bot.database.engine.acquire()
|
||||
await conn.execute(
|
||||
autojoin.delete().where(autojoin.c.guild_id == guild_id))
|
||||
@@ -0,0 +1,60 @@
|
||||
import re
|
||||
import struct
|
||||
import sys
|
||||
import urllib.request as urllib2
|
||||
import aiohttp
|
||||
|
||||
key = "b9803d262fab977f:04c791f69496359d638fc634e60c08d0"
|
||||
|
||||
|
||||
def hasNumbers(inputString):
|
||||
return any(char.isdigit() for char in inputString)
|
||||
|
||||
|
||||
def process_stream_title(title: str) -> list:
|
||||
l = title.split(' - ', 1)
|
||||
if l[0].isdigit():
|
||||
return l[1].rsplit(" - ", 1)
|
||||
return title.rsplit(" - ", 1)
|
||||
|
||||
|
||||
async def fetch_metadata():
|
||||
url = "https://azuracast.exobot.site/api/nowplaying"
|
||||
async with aiohttp.ClientSession() as session:
|
||||
async with session.get(url) as response:
|
||||
data = await response.json()
|
||||
source = data["playing_next"]["text"]
|
||||
name, vid = process_stream_title(source["title"])
|
||||
return {"name": name, "thumbnail": data["playing_next"]["art"]}
|
||||
|
||||
|
||||
async def get_metadata():
|
||||
"""Deprecated: use fetch_metadata()"""
|
||||
url = 'https://icecast.exobot.site/stream' # radio stream
|
||||
encoding = 'iso-8859-1' # default: iso-8859-1 for mp3 and utf-8 for ogg streams
|
||||
request = urllib2.Request(url, headers={'Icy-MetaData':
|
||||
1}) # request metadata
|
||||
response = urllib2.urlopen(request)
|
||||
# print(response.headers, file=sys.stderr)
|
||||
metaint = int(response.headers['icy-metaint'])
|
||||
for _ in range(10): # title may be empty initially, try several times
|
||||
response.read(metaint) # skip to metadata
|
||||
metadata_length = struct.unpack(
|
||||
'B', response.read(1))[0] * 16 # length byte
|
||||
metadata = response.read(metadata_length).rstrip(b'\0')
|
||||
# print(metadata, file=sys.stderr)
|
||||
# extract title from the metadata
|
||||
m = re.search(br"StreamTitle='([^']*)';", metadata)
|
||||
if m:
|
||||
title = m.group(1)
|
||||
if title:
|
||||
break
|
||||
else:
|
||||
sys.exit('no title found')
|
||||
|
||||
title = title.decode(encoding, errors='replace')
|
||||
#print(f"DEBUG: {title}")
|
||||
if hasNumbers(str(title[:4])):
|
||||
return title[7:][:-14]
|
||||
else:
|
||||
return title[:-14]
|
||||
@@ -0,0 +1,821 @@
|
||||
# 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())
|
||||
Reference in New Issue
Block a user