feat(monitoring): add http server for monitoring
This commit is contained in:
parent
c442fd55fe
commit
64b092dff2
5 changed files with 6 additions and 751 deletions
5
bot.py
5
bot.py
|
|
@ -28,7 +28,6 @@ l_extensions: List[str] = [
|
||||||
'cogs.utility',
|
'cogs.utility',
|
||||||
'cogs.logs',
|
'cogs.logs',
|
||||||
'cogs.poll',
|
'cogs.poll',
|
||||||
'cogs.help',
|
|
||||||
'jishaku',
|
'jishaku',
|
||||||
'cogs.monitoring'
|
'cogs.monitoring'
|
||||||
]
|
]
|
||||||
|
|
@ -41,7 +40,7 @@ async def _prefix_callable(bot, message: discord.message) -> list:
|
||||||
if str(message.guild.id) in bot.prefixes:
|
if str(message.guild.id) in bot.prefixes:
|
||||||
extras.extend(
|
extras.extend(
|
||||||
bot.prefixes.get(str(message.guild.id), "prefixes").split(
|
bot.prefixes.get(str(message.guild.id), "prefixes").split(
|
||||||
bot.config.get("misc", "separator")
|
bot.config.get("misc", "Separator")
|
||||||
)
|
)
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
@ -87,7 +86,7 @@ class TuxBot(commands.AutoShardedBot):
|
||||||
+ extension, exc_info=e)
|
+ extension, exc_info=e)
|
||||||
|
|
||||||
async def is_owner(self, user: discord.User) -> bool:
|
async def is_owner(self, user: discord.User) -> bool:
|
||||||
return str(user.id) in self.config.get("permissions", "owners").split(
|
return str(user.id) in self.config.get("permissions", "Owners").split(
|
||||||
', ')
|
', ')
|
||||||
|
|
||||||
async def on_socket_response(self, msg):
|
async def on_socket_response(self, msg):
|
||||||
|
|
|
||||||
621
cogs/help.py
621
cogs/help.py
|
|
@ -1,621 +0,0 @@
|
||||||
from discord.ext import commands
|
|
||||||
from .utils import checks, formats, time
|
|
||||||
from .utils.paginator import Pages
|
|
||||||
import discord
|
|
||||||
from collections import OrderedDict, deque, Counter
|
|
||||||
import os, datetime
|
|
||||||
import asyncio
|
|
||||||
import copy
|
|
||||||
import unicodedata
|
|
||||||
import inspect
|
|
||||||
import itertools
|
|
||||||
from typing import Union
|
|
||||||
|
|
||||||
class Prefix(commands.Converter):
|
|
||||||
async def convert(self, ctx, argument):
|
|
||||||
user_id = ctx.bot.user.id
|
|
||||||
if argument.startswith((f'<@{user_id}>', f'<@!{user_id}>')):
|
|
||||||
raise commands.BadArgument('That is a reserved prefix already in use.')
|
|
||||||
return argument
|
|
||||||
|
|
||||||
class FetchedUser(commands.Converter):
|
|
||||||
async def convert(self, ctx, argument):
|
|
||||||
if not argument.isdigit():
|
|
||||||
raise commands.BadArgument('Not a valid user ID.')
|
|
||||||
try:
|
|
||||||
return await ctx.bot.fetch_user(argument)
|
|
||||||
except discord.NotFound:
|
|
||||||
raise commands.BadArgument('User not found.') from None
|
|
||||||
except discord.HTTPException:
|
|
||||||
raise commands.BadArgument('An error occurred while fetching the user.') from None
|
|
||||||
|
|
||||||
class HelpPaginator(Pages):
|
|
||||||
def __init__(self, help_command, ctx, entries, *, per_page=4):
|
|
||||||
super().__init__(ctx, entries=entries, per_page=per_page)
|
|
||||||
self.reaction_emojis.append(('\N{WHITE QUESTION MARK ORNAMENT}', self.show_bot_help))
|
|
||||||
self.total = len(entries)
|
|
||||||
self.help_command = help_command
|
|
||||||
self.prefix = help_command.clean_prefix
|
|
||||||
self.is_bot = False
|
|
||||||
|
|
||||||
def get_bot_page(self, page):
|
|
||||||
cog, description, commands = self.entries[page - 1]
|
|
||||||
self.title = f'{cog} Commands'
|
|
||||||
self.description = description
|
|
||||||
return commands
|
|
||||||
|
|
||||||
def prepare_embed(self, entries, page, *, first=False):
|
|
||||||
self.embed.clear_fields()
|
|
||||||
self.embed.description = self.description
|
|
||||||
self.embed.title = self.title
|
|
||||||
|
|
||||||
if self.is_bot:
|
|
||||||
value ='For more help, join the official bot support server: https://discord.gg/DWEaqMy'
|
|
||||||
self.embed.add_field(name='Support', value=value, inline=False)
|
|
||||||
|
|
||||||
self.embed.set_footer(text=f'Use "{self.prefix}help command" for more info on a command.')
|
|
||||||
|
|
||||||
for entry in entries:
|
|
||||||
signature = f'{entry.qualified_name} {entry.signature}'
|
|
||||||
self.embed.add_field(name=signature, 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)')
|
|
||||||
|
|
||||||
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())
|
|
||||||
|
|
||||||
class PaginatedHelpCommand(commands.HelpCommand):
|
|
||||||
def __init__(self):
|
|
||||||
super().__init__(command_attrs={
|
|
||||||
'cooldown': commands.Cooldown(1, 3.0, commands.BucketType.member),
|
|
||||||
'help': 'Shows help about the bot, a command, or a category'
|
|
||||||
})
|
|
||||||
|
|
||||||
async def on_help_command_error(self, ctx, error):
|
|
||||||
if isinstance(error, commands.CommandInvokeError):
|
|
||||||
await ctx.send(str(error.original))
|
|
||||||
|
|
||||||
def get_command_signature(self, command):
|
|
||||||
parent = command.full_parent_name
|
|
||||||
if len(command.aliases) > 0:
|
|
||||||
aliases = '|'.join(command.aliases)
|
|
||||||
fmt = f'[{command.name}|{aliases}]'
|
|
||||||
if parent:
|
|
||||||
fmt = f'{parent} {fmt}'
|
|
||||||
alias = fmt
|
|
||||||
else:
|
|
||||||
alias = command.name if not parent else f'{parent} {command.name}'
|
|
||||||
return f'{alias} {command.signature}'
|
|
||||||
|
|
||||||
async def send_bot_help(self, mapping):
|
|
||||||
def key(c):
|
|
||||||
return c.cog_name or '\u200bNo Category'
|
|
||||||
|
|
||||||
bot = self.context.bot
|
|
||||||
entries = await self.filter_commands(bot.commands, sort=True, key=key)
|
|
||||||
nested_pages = []
|
|
||||||
per_page = 9
|
|
||||||
total = 0
|
|
||||||
|
|
||||||
for cog, commands in itertools.groupby(entries, key=key):
|
|
||||||
commands = sorted(commands, key=lambda c: c.name)
|
|
||||||
if len(commands) == 0:
|
|
||||||
continue
|
|
||||||
|
|
||||||
total += len(commands)
|
|
||||||
actual_cog = bot.get_cog(cog)
|
|
||||||
# get the description if it exists (and the cog is valid) or return Empty embed.
|
|
||||||
description = (actual_cog and actual_cog.description) or discord.Embed.Empty
|
|
||||||
nested_pages.extend((cog, description, commands[i:i + per_page]) for i in range(0, len(commands), per_page))
|
|
||||||
|
|
||||||
# a value of 1 forces the pagination session
|
|
||||||
pages = HelpPaginator(self, self.context, nested_pages, per_page=1)
|
|
||||||
|
|
||||||
# swap the get_page implementation to work with our nested pages.
|
|
||||||
pages.get_page = pages.get_bot_page
|
|
||||||
pages.is_bot = True
|
|
||||||
pages.total = total
|
|
||||||
await self.context.release()
|
|
||||||
await pages.paginate()
|
|
||||||
|
|
||||||
async def send_cog_help(self, cog):
|
|
||||||
entries = await self.filter_commands(cog.get_commands(), sort=True)
|
|
||||||
pages = HelpPaginator(self, self.context, entries)
|
|
||||||
pages.title = f'{cog.qualified_name} Commands'
|
|
||||||
pages.description = cog.description
|
|
||||||
|
|
||||||
await self.context.release()
|
|
||||||
await pages.paginate()
|
|
||||||
|
|
||||||
def common_command_formatting(self, page_or_embed, command):
|
|
||||||
page_or_embed.title = self.get_command_signature(command)
|
|
||||||
if command.description:
|
|
||||||
page_or_embed.description = f'{command.description}\n\n{command.help}'
|
|
||||||
else:
|
|
||||||
page_or_embed.description = command.help or 'No help found...'
|
|
||||||
|
|
||||||
async def send_command_help(self, command):
|
|
||||||
# No pagination necessary for a single command.
|
|
||||||
embed = discord.Embed(colour=discord.Colour.blurple())
|
|
||||||
self.common_command_formatting(embed, command)
|
|
||||||
await self.context.send(embed=embed)
|
|
||||||
|
|
||||||
async def send_group_help(self, group):
|
|
||||||
subcommands = group.commands
|
|
||||||
if len(subcommands) == 0:
|
|
||||||
return await self.send_command_help(group)
|
|
||||||
|
|
||||||
entries = await self.filter_commands(subcommands, sort=True)
|
|
||||||
pages = HelpPaginator(self, self.context, entries)
|
|
||||||
self.common_command_formatting(pages, group)
|
|
||||||
|
|
||||||
await self.context.release()
|
|
||||||
await pages.paginate()
|
|
||||||
|
|
||||||
class Meta(commands.Cog):
|
|
||||||
"""Commands for utilities related to Discord or the Bot itself."""
|
|
||||||
|
|
||||||
def __init__(self, bot):
|
|
||||||
self.bot = bot
|
|
||||||
self.old_help_command = bot.help_command
|
|
||||||
bot.help_command = PaginatedHelpCommand()
|
|
||||||
bot.help_command.cog = self
|
|
||||||
|
|
||||||
def cog_unload(self):
|
|
||||||
self.bot.help_command = self.old_help_command
|
|
||||||
|
|
||||||
async def cog_command_error(self, ctx, error):
|
|
||||||
if isinstance(error, commands.BadArgument):
|
|
||||||
await ctx.send(error)
|
|
||||||
|
|
||||||
@commands.command(hidden=True)
|
|
||||||
async def hello(self, ctx):
|
|
||||||
"""Displays my intro message."""
|
|
||||||
await ctx.send('Hello! I\'m a robot! Danny#0007 made me.')
|
|
||||||
|
|
||||||
@commands.command()
|
|
||||||
async def charinfo(self, ctx, *, characters: str):
|
|
||||||
"""Shows you information about a number of characters.
|
|
||||||
|
|
||||||
Only up to 25 characters at a time.
|
|
||||||
"""
|
|
||||||
|
|
||||||
def to_string(c):
|
|
||||||
digit = f'{ord(c):x}'
|
|
||||||
name = unicodedata.name(c, 'Name not found.')
|
|
||||||
return f'`\\U{digit:>08}`: {name} - {c} \N{EM DASH} <http://www.fileformat.info/info/unicode/char/{digit}>'
|
|
||||||
msg = '\n'.join(map(to_string, characters))
|
|
||||||
if len(msg) > 2000:
|
|
||||||
return await ctx.send('Output too long to display.')
|
|
||||||
await ctx.send(msg)
|
|
||||||
|
|
||||||
@commands.group(name='prefix', invoke_without_command=True)
|
|
||||||
async def prefix(self, ctx):
|
|
||||||
"""Manages the server's custom prefixes.
|
|
||||||
|
|
||||||
If called without a subcommand, this will list the currently set
|
|
||||||
prefixes.
|
|
||||||
"""
|
|
||||||
|
|
||||||
prefixes = self.bot.get_guild_prefixes(ctx.guild)
|
|
||||||
|
|
||||||
# we want to remove prefix #2, because it's the 2nd form of the mention
|
|
||||||
# and to the end user, this would end up making them confused why the
|
|
||||||
# mention is there twice
|
|
||||||
del prefixes[1]
|
|
||||||
|
|
||||||
e = discord.Embed(title='Prefixes', colour=discord.Colour.blurple())
|
|
||||||
e.set_footer(text=f'{len(prefixes)} prefixes')
|
|
||||||
e.description = '\n'.join(f'{index}. {elem}' for index, elem in enumerate(prefixes, 1))
|
|
||||||
await ctx.send(embed=e)
|
|
||||||
|
|
||||||
@prefix.command(name='add', ignore_extra=False)
|
|
||||||
@checks.is_mod()
|
|
||||||
async def prefix_add(self, ctx, prefix: Prefix):
|
|
||||||
"""Appends a prefix to the list of custom prefixes.
|
|
||||||
|
|
||||||
Previously set prefixes are not overridden.
|
|
||||||
|
|
||||||
To have a word prefix, you should quote it and end it with
|
|
||||||
a space, e.g. "hello " to set the prefix to "hello ". This
|
|
||||||
is because Discord removes spaces when sending messages so
|
|
||||||
the spaces are not preserved.
|
|
||||||
|
|
||||||
Multi-word prefixes must be quoted also.
|
|
||||||
|
|
||||||
You must have Manage Server permission to use this command.
|
|
||||||
"""
|
|
||||||
|
|
||||||
current_prefixes = self.bot.get_raw_guild_prefixes(ctx.guild.id)
|
|
||||||
current_prefixes.append(prefix)
|
|
||||||
try:
|
|
||||||
await self.bot.set_guild_prefixes(ctx.guild, current_prefixes)
|
|
||||||
except Exception as e:
|
|
||||||
await ctx.send(f'{ctx.tick(False)} {e}')
|
|
||||||
else:
|
|
||||||
await ctx.send(ctx.tick(True))
|
|
||||||
|
|
||||||
@prefix_add.error
|
|
||||||
async def prefix_add_error(self, ctx, error):
|
|
||||||
if isinstance(error, commands.TooManyArguments):
|
|
||||||
await ctx.send("You've given too many prefixes. Either quote it or only do it one by one.")
|
|
||||||
|
|
||||||
@prefix.command(name='remove', aliases=['delete'], ignore_extra=False)
|
|
||||||
@checks.is_mod()
|
|
||||||
async def prefix_remove(self, ctx, prefix: Prefix):
|
|
||||||
"""Removes a prefix from the list of custom prefixes.
|
|
||||||
|
|
||||||
This is the inverse of the 'prefix add' command. You can
|
|
||||||
use this to remove prefixes from the default set as well.
|
|
||||||
|
|
||||||
You must have Manage Server permission to use this command.
|
|
||||||
"""
|
|
||||||
|
|
||||||
current_prefixes = self.bot.get_raw_guild_prefixes(ctx.guild.id)
|
|
||||||
|
|
||||||
try:
|
|
||||||
current_prefixes.remove(prefix)
|
|
||||||
except ValueError:
|
|
||||||
return await ctx.send('I do not have this prefix registered.')
|
|
||||||
|
|
||||||
try:
|
|
||||||
await self.bot.set_guild_prefixes(ctx.guild, current_prefixes)
|
|
||||||
except Exception as e:
|
|
||||||
await ctx.send(f'{ctx.tick(False)} {e}')
|
|
||||||
else:
|
|
||||||
await ctx.send(ctx.tick(True))
|
|
||||||
|
|
||||||
@prefix.command(name='clear')
|
|
||||||
@checks.is_mod()
|
|
||||||
async def prefix_clear(self, ctx):
|
|
||||||
"""Removes all custom prefixes.
|
|
||||||
|
|
||||||
After this, the bot will listen to only mention prefixes.
|
|
||||||
|
|
||||||
You must have Manage Server permission to use this command.
|
|
||||||
"""
|
|
||||||
|
|
||||||
await self.bot.set_guild_prefixes(ctx.guild, [])
|
|
||||||
await ctx.send(ctx.tick(True))
|
|
||||||
|
|
||||||
@commands.command()
|
|
||||||
async def source(self, ctx, *, command: str = None):
|
|
||||||
"""Displays my full source code or for a specific command.
|
|
||||||
|
|
||||||
To display the source code of a subcommand you can separate it by
|
|
||||||
periods, e.g. tag.create for the create subcommand of the tag command
|
|
||||||
or by spaces.
|
|
||||||
"""
|
|
||||||
source_url = 'https://github.com/Rapptz/RoboDanny'
|
|
||||||
branch = 'rewrite'
|
|
||||||
if command is None:
|
|
||||||
return await ctx.send(source_url)
|
|
||||||
|
|
||||||
if command == 'help':
|
|
||||||
src = type(self.bot.help_command)
|
|
||||||
module = src.__module__
|
|
||||||
filename = inspect.getsourcefile(src)
|
|
||||||
else:
|
|
||||||
obj = self.bot.get_command(command.replace('.', ' '))
|
|
||||||
if obj is None:
|
|
||||||
return await ctx.send('Could not find command.')
|
|
||||||
|
|
||||||
# since we found the command we're looking for, presumably anyway, let's
|
|
||||||
# try to access the code itself
|
|
||||||
src = obj.callback.__code__
|
|
||||||
module = obj.callback.__module__
|
|
||||||
filename = src.co_filename
|
|
||||||
|
|
||||||
lines, firstlineno = inspect.getsourcelines(src)
|
|
||||||
if not module.startswith('discord'):
|
|
||||||
# not a built-in command
|
|
||||||
location = os.path.relpath(filename).replace('\\', '/')
|
|
||||||
else:
|
|
||||||
location = module.replace('.', '/') + '.py'
|
|
||||||
source_url = 'https://github.com/Rapptz/discord.py'
|
|
||||||
branch = 'master'
|
|
||||||
|
|
||||||
final_url = f'<{source_url}/blob/{branch}/{location}#L{firstlineno}-L{firstlineno + len(lines) - 1}>'
|
|
||||||
await ctx.send(final_url)
|
|
||||||
|
|
||||||
@commands.command(name='quit', hidden=True)
|
|
||||||
@commands.is_owner()
|
|
||||||
async def _quit(self, ctx):
|
|
||||||
"""Quits the bot."""
|
|
||||||
await self.bot.logout()
|
|
||||||
|
|
||||||
@commands.command()
|
|
||||||
async def avatar(self, ctx, *, user: Union[discord.Member, FetchedUser] = None):
|
|
||||||
"""Shows a user's enlarged avatar (if possible)."""
|
|
||||||
embed = discord.Embed()
|
|
||||||
user = user or ctx.author
|
|
||||||
avatar = user.avatar_url_as(static_format='png')
|
|
||||||
embed.set_author(name=str(user), url=avatar)
|
|
||||||
embed.set_image(url=avatar)
|
|
||||||
await ctx.send(embed=embed)
|
|
||||||
|
|
||||||
@commands.command()
|
|
||||||
async def info(self, ctx, *, user: Union[discord.Member, FetchedUser] = None):
|
|
||||||
"""Shows info about a user."""
|
|
||||||
|
|
||||||
user = user or ctx.author
|
|
||||||
if ctx.guild and isinstance(user, discord.User):
|
|
||||||
user = ctx.guild.get_member(user.id) or user
|
|
||||||
|
|
||||||
e = discord.Embed()
|
|
||||||
roles = [role.name.replace('@', '@\u200b') for role in getattr(user, 'roles', [])]
|
|
||||||
shared = sum(g.get_member(user.id) is not None for g in self.bot.guilds)
|
|
||||||
e.set_author(name=str(user))
|
|
||||||
|
|
||||||
def format_date(dt):
|
|
||||||
if dt is None:
|
|
||||||
return 'N/A'
|
|
||||||
return f'{dt:%Y-%m-%d %H:%M} ({time.human_timedelta(dt, accuracy=3)})'
|
|
||||||
|
|
||||||
e.add_field(name='ID', value=user.id, inline=False)
|
|
||||||
e.add_field(name='Servers', value=f'{shared} shared', inline=False)
|
|
||||||
e.add_field(name='Joined', value=format_date(getattr(user, 'joined_at', None)), inline=False)
|
|
||||||
e.add_field(name='Created', value=format_date(user.created_at), inline=False)
|
|
||||||
|
|
||||||
voice = getattr(user, 'voice', None)
|
|
||||||
if voice is not None:
|
|
||||||
vc = voice.channel
|
|
||||||
other_people = len(vc.members) - 1
|
|
||||||
voice = f'{vc.name} with {other_people} others' if other_people else f'{vc.name} by themselves'
|
|
||||||
e.add_field(name='Voice', value=voice, inline=False)
|
|
||||||
|
|
||||||
if roles:
|
|
||||||
e.add_field(name='Roles', value=', '.join(roles) if len(roles) < 10 else f'{len(roles)} roles', inline=False)
|
|
||||||
|
|
||||||
colour = user.colour
|
|
||||||
if colour.value:
|
|
||||||
e.colour = colour
|
|
||||||
|
|
||||||
if user.avatar:
|
|
||||||
e.set_thumbnail(url=user.avatar_url)
|
|
||||||
|
|
||||||
if isinstance(user, discord.User):
|
|
||||||
e.set_footer(text='This member is not in this server.')
|
|
||||||
|
|
||||||
await ctx.send(embed=e)
|
|
||||||
|
|
||||||
@commands.command(aliases=['guildinfo'], usage='')
|
|
||||||
@commands.guild_only()
|
|
||||||
async def serverinfo(self, ctx, *, guild_id: int = None):
|
|
||||||
"""Shows info about the current server."""
|
|
||||||
|
|
||||||
if guild_id is not None and await self.bot.is_owner(ctx.author):
|
|
||||||
guild = self.bot.get_guild(guild_id)
|
|
||||||
if guild is None:
|
|
||||||
return await ctx.send(f'Invalid Guild ID given.')
|
|
||||||
else:
|
|
||||||
guild = ctx.guild
|
|
||||||
|
|
||||||
roles = [role.name.replace('@', '@\u200b') for role in guild.roles]
|
|
||||||
|
|
||||||
# we're going to duck type our way here
|
|
||||||
class Secret:
|
|
||||||
pass
|
|
||||||
|
|
||||||
secret_member = Secret()
|
|
||||||
secret_member.id = 0
|
|
||||||
secret_member.roles = [guild.default_role]
|
|
||||||
|
|
||||||
# figure out what channels are 'secret'
|
|
||||||
secret = Counter()
|
|
||||||
totals = Counter()
|
|
||||||
for channel in guild.channels:
|
|
||||||
perms = channel.permissions_for(secret_member)
|
|
||||||
channel_type = type(channel)
|
|
||||||
totals[channel_type] += 1
|
|
||||||
if not perms.read_messages:
|
|
||||||
secret[channel_type] += 1
|
|
||||||
elif isinstance(channel, discord.VoiceChannel) and (not perms.connect or not perms.speak):
|
|
||||||
secret[channel_type] += 1
|
|
||||||
|
|
||||||
member_by_status = Counter(str(m.status) for m in guild.members)
|
|
||||||
|
|
||||||
e = discord.Embed()
|
|
||||||
e.title = guild.name
|
|
||||||
e.add_field(name='ID', value=guild.id)
|
|
||||||
e.add_field(name='Owner', value=guild.owner)
|
|
||||||
if guild.icon:
|
|
||||||
e.set_thumbnail(url=guild.icon_url)
|
|
||||||
|
|
||||||
channel_info = []
|
|
||||||
key_to_emoji = {
|
|
||||||
discord.TextChannel: '<:text_channel:586339098172850187>',
|
|
||||||
discord.VoiceChannel: '<:voice_channel:586339098524909604>',
|
|
||||||
}
|
|
||||||
for key, total in totals.items():
|
|
||||||
secrets = secret[key]
|
|
||||||
try:
|
|
||||||
emoji = key_to_emoji[key]
|
|
||||||
except KeyError:
|
|
||||||
continue
|
|
||||||
|
|
||||||
if secrets:
|
|
||||||
channel_info.append(f'{emoji} {total} ({secrets} locked)')
|
|
||||||
else:
|
|
||||||
channel_info.append(f'{emoji} {total}')
|
|
||||||
|
|
||||||
info = []
|
|
||||||
features = set(guild.features)
|
|
||||||
all_features = {
|
|
||||||
'PARTNERED': 'Partnered',
|
|
||||||
'VERIFIED': 'Verified',
|
|
||||||
'DISCOVERABLE': 'Server Discovery',
|
|
||||||
'PUBLIC': 'Server Discovery/Public',
|
|
||||||
'INVITE_SPLASH': 'Invite Splash',
|
|
||||||
'VIP_REGIONS': 'VIP Voice Servers',
|
|
||||||
'VANITY_URL': 'Vanity Invite',
|
|
||||||
'MORE_EMOJI': 'More Emoji',
|
|
||||||
'COMMERCE': 'Commerce',
|
|
||||||
'LURKABLE': 'Lurkable',
|
|
||||||
'NEWS': 'News Channels',
|
|
||||||
'ANIMATED_ICON': 'Animated Icon',
|
|
||||||
'BANNER': 'Banner'
|
|
||||||
}
|
|
||||||
|
|
||||||
for feature, label in all_features.items():
|
|
||||||
if feature in features:
|
|
||||||
info.append(f'{ctx.tick(True)}: {label}')
|
|
||||||
|
|
||||||
if info:
|
|
||||||
e.add_field(name='Features', value='\n'.join(info))
|
|
||||||
|
|
||||||
e.add_field(name='Channels', value='\n'.join(channel_info))
|
|
||||||
|
|
||||||
if guild.premium_tier != 0:
|
|
||||||
boosts = f'Level {guild.premium_tier}\n{guild.premium_subscription_count} boosts'
|
|
||||||
last_boost = max(guild.members, key=lambda m: m.premium_since or guild.created_at)
|
|
||||||
if last_boost.premium_since is not None:
|
|
||||||
boosts = f'{boosts}\nLast Boost: {last_boost} ({time.human_timedelta(last_boost.premium_since, accuracy=2)})'
|
|
||||||
e.add_field(name='Boosts', value=boosts, inline=False)
|
|
||||||
|
|
||||||
fmt = f'<:online:316856575413321728> {member_by_status["online"]} ' \
|
|
||||||
f'<:idle:316856575098880002> {member_by_status["idle"]} ' \
|
|
||||||
f'<:dnd:316856574868193281> {member_by_status["dnd"]} ' \
|
|
||||||
f'<:offline:316856575501402112> {member_by_status["offline"]}\n' \
|
|
||||||
f'Total: {guild.member_count}'
|
|
||||||
|
|
||||||
e.add_field(name='Members', value=fmt, inline=False)
|
|
||||||
|
|
||||||
# TODO: maybe chunk and stuff for top role members
|
|
||||||
# requires max-concurrency d.py check to work though.
|
|
||||||
|
|
||||||
e.add_field(name='Roles', value=', '.join(roles) if len(roles) < 10 else f'{len(roles)} roles')
|
|
||||||
e.set_footer(text='Created').timestamp = guild.created_at
|
|
||||||
await ctx.send(embed=e)
|
|
||||||
|
|
||||||
async def say_permissions(self, ctx, member, channel):
|
|
||||||
permissions = channel.permissions_for(member)
|
|
||||||
e = discord.Embed(colour=member.colour)
|
|
||||||
allowed, denied = [], []
|
|
||||||
for name, value in permissions:
|
|
||||||
name = name.replace('_', ' ').replace('guild', 'server').title()
|
|
||||||
if value:
|
|
||||||
allowed.append(name)
|
|
||||||
else:
|
|
||||||
denied.append(name)
|
|
||||||
|
|
||||||
e.add_field(name='Allowed', value='\n'.join(allowed))
|
|
||||||
e.add_field(name='Denied', value='\n'.join(denied))
|
|
||||||
await ctx.send(embed=e)
|
|
||||||
|
|
||||||
@commands.command()
|
|
||||||
@commands.guild_only()
|
|
||||||
async def permissions(self, ctx, member: discord.Member = None, channel: discord.TextChannel = None):
|
|
||||||
"""Shows a member's permissions in a specific channel.
|
|
||||||
|
|
||||||
If no channel is given then it uses the current one.
|
|
||||||
|
|
||||||
You cannot use this in private messages. If no member is given then
|
|
||||||
the info returned will be yours.
|
|
||||||
"""
|
|
||||||
channel = channel or ctx.channel
|
|
||||||
if member is None:
|
|
||||||
member = ctx.author
|
|
||||||
|
|
||||||
await self.say_permissions(ctx, member, channel)
|
|
||||||
|
|
||||||
@commands.command()
|
|
||||||
@commands.guild_only()
|
|
||||||
@checks.admin_or_permissions(manage_roles=True)
|
|
||||||
async def botpermissions(self, ctx, *, channel: discord.TextChannel = None):
|
|
||||||
"""Shows the bot's permissions in a specific channel.
|
|
||||||
|
|
||||||
If no channel is given then it uses the current one.
|
|
||||||
|
|
||||||
This is a good way of checking if the bot has the permissions needed
|
|
||||||
to execute the commands it wants to execute.
|
|
||||||
|
|
||||||
To execute this command you must have Manage Roles permission.
|
|
||||||
You cannot use this in private messages.
|
|
||||||
"""
|
|
||||||
channel = channel or ctx.channel
|
|
||||||
member = ctx.guild.me
|
|
||||||
await self.say_permissions(ctx, member, channel)
|
|
||||||
|
|
||||||
@commands.command(aliases=['invite'])
|
|
||||||
async def join(self, ctx):
|
|
||||||
"""Joins a server."""
|
|
||||||
perms = discord.Permissions.none()
|
|
||||||
perms.read_messages = True
|
|
||||||
perms.external_emojis = True
|
|
||||||
perms.send_messages = True
|
|
||||||
perms.manage_roles = True
|
|
||||||
perms.manage_channels = True
|
|
||||||
perms.ban_members = True
|
|
||||||
perms.kick_members = True
|
|
||||||
perms.manage_messages = True
|
|
||||||
perms.embed_links = True
|
|
||||||
perms.read_message_history = True
|
|
||||||
perms.attach_files = True
|
|
||||||
perms.add_reactions = True
|
|
||||||
await ctx.send(f'<{discord.utils.oauth_url(self.bot.client_id, perms)}>')
|
|
||||||
|
|
||||||
@commands.command(rest_is_raw=True, hidden=True)
|
|
||||||
@commands.is_owner()
|
|
||||||
async def echo(self, ctx, *, content):
|
|
||||||
await ctx.send(content)
|
|
||||||
|
|
||||||
@commands.command(hidden=True)
|
|
||||||
async def cud(self, ctx):
|
|
||||||
"""pls no spam"""
|
|
||||||
|
|
||||||
for i in range(3):
|
|
||||||
await ctx.send(3 - i)
|
|
||||||
await asyncio.sleep(1)
|
|
||||||
|
|
||||||
await ctx.send('go')
|
|
||||||
|
|
||||||
def setup(bot):
|
|
||||||
bot.add_cog(Meta(bot))
|
|
||||||
|
|
@ -1,6 +1,8 @@
|
||||||
import asyncio
|
import asyncio
|
||||||
import threading
|
import threading
|
||||||
|
|
||||||
from aiohttp import web
|
from aiohttp import web
|
||||||
|
from aiohttp.web_request import Request
|
||||||
|
|
||||||
from discord.ext import commands
|
from discord.ext import commands
|
||||||
from bot import TuxBot
|
from bot import TuxBot
|
||||||
|
|
@ -18,7 +20,7 @@ class Monitoring(commands.Cog):
|
||||||
t.start()
|
t.start()
|
||||||
|
|
||||||
def aiohttp_server(self):
|
def aiohttp_server(self):
|
||||||
async def hi(request):
|
async def hi(request: Request):
|
||||||
return web.Response(text="I'm alive !")
|
return web.Response(text="I'm alive !")
|
||||||
|
|
||||||
self.app.add_routes([web.get('/', hi)])
|
self.app.add_routes([web.get('/', hi)])
|
||||||
|
|
@ -31,7 +33,7 @@ class Monitoring(commands.Cog):
|
||||||
loop = asyncio.new_event_loop()
|
loop = asyncio.new_event_loop()
|
||||||
asyncio.set_event_loop(loop)
|
asyncio.set_event_loop(loop)
|
||||||
loop.run_until_complete(runner.setup())
|
loop.run_until_complete(runner.setup())
|
||||||
site = web.TCPSite(runner, '0.0.0.', 3389)
|
site = web.TCPSite(runner, '0.0.0.0', 8080)
|
||||||
loop.run_until_complete(site.start())
|
loop.run_until_complete(site.start())
|
||||||
loop.run_forever()
|
loop.run_forever()
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -1,4 +1,3 @@
|
||||||
from .checks import *
|
|
||||||
from .config import *
|
from .config import *
|
||||||
from .lang import *
|
from .lang import *
|
||||||
from .version import *
|
from .version import *
|
||||||
|
|
|
||||||
|
|
@ -1,124 +0,0 @@
|
||||||
from discord.ext import commands
|
|
||||||
|
|
||||||
|
|
||||||
def is_owner_check(message):
|
|
||||||
return str(message.author.id) in ['171685542553976832',
|
|
||||||
'269156684155453451']
|
|
||||||
|
|
||||||
|
|
||||||
def is_owner(warn=True):
|
|
||||||
def check(ctx, log):
|
|
||||||
owner = is_owner_check(ctx.message)
|
|
||||||
if not owner and log:
|
|
||||||
print(ctx.message.author.name + " à essayer d'executer " + ctx.message.content + " sur le serveur " + ctx.message.guild.name)
|
|
||||||
return owner
|
|
||||||
|
|
||||||
owner = commands.check(lambda ctx: check(ctx, warn))
|
|
||||||
return owner
|
|
||||||
|
|
||||||
|
|
||||||
"""-------------------------------------------------------------------------"""
|
|
||||||
|
|
||||||
|
|
||||||
async def check_permissions(ctx, perms, *, check=all):
|
|
||||||
is_owner = await ctx.bot.is_owner(ctx.author)
|
|
||||||
if is_owner or is_owner_check(ctx.message) is True:
|
|
||||||
return True
|
|
||||||
|
|
||||||
resolved = ctx.channel.permissions_for(ctx.author)
|
|
||||||
return check(getattr(resolved, name, None) == value for name, value in
|
|
||||||
perms.items())
|
|
||||||
|
|
||||||
|
|
||||||
def has_permissions(*, check=all, **perms):
|
|
||||||
async def pred(ctx):
|
|
||||||
return await check_permissions(ctx, perms, check=check)
|
|
||||||
|
|
||||||
return commands.check(pred)
|
|
||||||
|
|
||||||
|
|
||||||
async def check_guild_permissions(ctx, perms, *, check=all):
|
|
||||||
is_owner = await ctx.bot.is_owner(ctx.author)
|
|
||||||
if is_owner:
|
|
||||||
return True
|
|
||||||
|
|
||||||
if ctx.guild is None:
|
|
||||||
return False
|
|
||||||
|
|
||||||
resolved = ctx.author.guild_permissions
|
|
||||||
return check(getattr(resolved, name, None) == value for name, value in
|
|
||||||
perms.items())
|
|
||||||
|
|
||||||
|
|
||||||
def has_guild_permissions(*, check=all, **perms):
|
|
||||||
async def pred(ctx):
|
|
||||||
return await check_guild_permissions(ctx, perms, check=check)
|
|
||||||
|
|
||||||
return commands.check(pred)
|
|
||||||
|
|
||||||
|
|
||||||
# These do not take channel overrides into account
|
|
||||||
|
|
||||||
|
|
||||||
def is_mod():
|
|
||||||
async def pred(ctx):
|
|
||||||
return await check_guild_permissions(ctx, {'manage_guild': True})
|
|
||||||
|
|
||||||
return commands.check(pred)
|
|
||||||
|
|
||||||
|
|
||||||
def is_admin():
|
|
||||||
async def pred(ctx):
|
|
||||||
return await check_guild_permissions(ctx, {'administrator': True})
|
|
||||||
|
|
||||||
return commands.check(pred)
|
|
||||||
|
|
||||||
|
|
||||||
def mod_or_permissions(**perms):
|
|
||||||
perms['manage_guild'] = True
|
|
||||||
|
|
||||||
async def predicate(ctx):
|
|
||||||
return await check_guild_permissions(ctx, perms, check=any)
|
|
||||||
|
|
||||||
return commands.check(predicate)
|
|
||||||
|
|
||||||
|
|
||||||
def admin_or_permissions(**perms):
|
|
||||||
perms['administrator'] = True
|
|
||||||
|
|
||||||
async def predicate(ctx):
|
|
||||||
return await check_guild_permissions(ctx, perms, check=any)
|
|
||||||
|
|
||||||
return commands.check(predicate)
|
|
||||||
|
|
||||||
|
|
||||||
def is_in_guilds(*guild_ids):
|
|
||||||
def predicate(ctx):
|
|
||||||
guild = ctx.guild
|
|
||||||
if guild is None:
|
|
||||||
return False
|
|
||||||
return guild.id in guild_ids
|
|
||||||
|
|
||||||
return commands.check(predicate)
|
|
||||||
|
|
||||||
|
|
||||||
def get_user(message, user):
|
|
||||||
try:
|
|
||||||
member = message.mentions[0]
|
|
||||||
except:
|
|
||||||
member = message.guild.get_member_named(user)
|
|
||||||
if not member:
|
|
||||||
try:
|
|
||||||
member = message.guild.get_member(int(user))
|
|
||||||
except ValueError:
|
|
||||||
pass
|
|
||||||
if not member:
|
|
||||||
return None
|
|
||||||
return member
|
|
||||||
|
|
||||||
|
|
||||||
def check_date(date: str):
|
|
||||||
if len(date) == 1:
|
|
||||||
return f"0{date}"
|
|
||||||
else:
|
|
||||||
return date
|
|
||||||
Loading…
Reference in a new issue