1
0
Fork 0
mirror of synced 2024-06-09 22:24:32 +12:00
Bonfire/cogs/owner.py
2016-08-21 15:11:18 -05:00

221 lines
8.9 KiB
Python

from discord.ext import commands
from .utils import config
from .utils import checks
import re
import os
import glob
import sys
import discord
import inspect
import aiohttp
getter = re.compile(r'`(?!`)(.*?)`')
multi = re.compile(r'```(.*?)```', re.DOTALL)
class Owner:
"""Commands that can only be used by Phantom, bot management commands"""
def __init__(self, bot):
self.bot = bot
@commands.command()
@commands.check(checks.is_owner)
async def running(self):
servers_playing_music = len([server_id for server_id, state in self.bot.get_cog('Music').voice_states.items() if
state.is_playing()])
hm_games = len([server_id for server_id, game in self.bot.get_cog('Hangman').games.items()])
ttt_games = len([server_id for server_id, game in self.bot.get_cog('TicTacToe').boards.items()])
battles = 0
for battles in self.bot.get_cog('Interaction').battles:
battles += len(battles)
fmt = ""
if servers_playing_music:
fmt += "Playing songs in {} different servers\n".format(servers_playing_music)
if hm_games:
fmt += "{} different hangman games running\n".format(hm_games)
if ttt_games:
fmt += "{} different TicTacToe games running\n".format(ttt_games)
if battles:
fmt += "{} different battles going on\n".format(battles)
if not fmt:
fmt = "Nothing currently running!"
await self.bot.say(fmt)
@commands.command(pass_context=True)
@commands.check(checks.is_owner)
async def saferestart(self, ctx):
"""This commands is used to check if there is anything playing in any servers at the moment
If there is, I'll tell you not to restart, if not I'll just go ahead and restart"""
# I do not want to restart the bot if someone is playing music
# This gets all the exiting VoiceStates that are playing music right now
# If we are, say which server it
servers_playing_music = [server_id for server_id, state in self.bot.get_cog('Music').voice_states.items() if
state.is_playing()]
if len(servers_playing_music) > 0:
await self.bot.say("Sorry, it's not safe to restart. I am currently playing a song on {} servers".format(
len(servers_playing_music)))
else:
config.save_content('restart_server', ctx.message.channel.id)
await self.bot.say("Restarting; see you in the next life {0}!".format(ctx.message.author.mention))
python = sys.executable
os.execl(python, python, *sys.argv)
@commands.command(pass_context=True)
@commands.check(checks.is_owner)
async def restart(self, ctx):
"""Forces the bot to restart"""
# This command is left in so that we can invoke it from saferestart, or we need a restart no matter what
config.save_content('restart_server', ctx.message.channel.id)
await self.bot.say("Restarting; see you in the next life {0}!".format(ctx.message.author.mention))
python = sys.executable
os.execl(python, python, *sys.argv)
@commands.command()
@commands.check(checks.is_owner)
async def adddoggo(self, url: str):
"""Saves a URL as an image to add for the doggo command"""
# Save the local path based on how many images there currently are
local_path = 'images/doggo{}.jpg'.format(len(glob.glob('images/doggo*')))
# "read" the image and save as bytes
with aiohttp.ClientSession() as s:
async with s.get(url) as r:
val = await r.read()
with open(local_path, "wb") as f:
f.write(val)
await self.bot.say(
"Just saved a new doggo image! I now have {} doggo images!".format(len(glob.glob('images/doggo*'))))
@commands.command()
@commands.check(checks.is_owner)
async def addsnek(self, url: str):
"""Saves a URL as an image to add for the snek command"""
# Save the local path based on how many images there currently are
local_path = 'images/snek{}.jpg'.format(len(glob.glob('images/snek*')))
# "read" the image and save as bytes
with aiohttp.ClientSession() as s:
async with s.get(url) as r:
val = await r.read()
with open(local_path, "wb") as f:
f.write(val)
await self.bot.say(
"Just saved a new snek image! I now have {} snek images!".format(len(glob.glob('images/snek*'))))
@commands.command(pass_context=True)
@commands.check(checks.is_owner)
async def debug(self, ctx):
"""Executes code"""
# Eval and exec have different useful purposes, so use both
try:
# `Get all content in this format`
match_single = getter.findall(ctx.message.content)
# ```\nGet all content in this format```
match_multi = multi.findall(ctx.message.content)
if match_single:
result = eval(match_single[0])
# In case the result needs to be awaited, handle that
if inspect.isawaitable(result):
result = await result
await self.bot.say("```\n{0}```".format(result))
elif match_multi:
# Internal method to send the message to the channel, of whatever is passed
def r(v):
self.bot.loop.create_task(self.bot.say("```\n{}```".format(v)))
exec(match_multi[0])
except Exception as error:
fmt = 'An error occurred while processing this request: ```py\n{}: {}\n```'
await self.bot.say(fmt.format(type(error).__name__, error))
@commands.command(pass_context=True)
@commands.check(checks.is_owner)
async def shutdown(self, ctx):
"""Shuts the bot down"""
fmt = 'Shutting down, I will miss you {0.author.name}'
await self.bot.say(fmt.format(ctx.message))
await self.bot.logout()
await self.bot.close()
@commands.command()
@commands.check(checks.is_owner)
async def avatar(self, content: str):
"""Changes the avatar for the bot to the filename following the command"""
file = 'images/' + content
with open(file, 'rb') as fp:
await self.bot.edit_profile(avatar=fp.read())
@commands.command()
@commands.check(checks.is_owner)
async def name(self, newNick: str):
"""Changes the bot's name"""
await self.bot.edit_profile(username=newNick)
await self.bot.say('Changed username to ' + newNick)
@commands.command()
@commands.check(checks.is_owner)
async def status(self, *, status: str):
"""Changes the bot's 'playing' status"""
await self.bot.change_status(discord.Game(name=status, type=0))
await self.bot.say("Just changed my status to '{0}'!".format(status))
@commands.command()
@commands.check(checks.is_owner)
async def load(self, *, module: str):
"""Loads a module"""
# Do this because I'm too lazy to type cogs.module
module = module.lower()
if not module.startswith("cogs"):
module = "cogs.{}".format(module)
# This try catch will catch errors such as syntax errors in the module we are loading
try:
self.bot.load_extension(module)
await self.bot.say("I have just loaded the {} module".format(module))
except Exception as error:
fmt = 'An error occurred while processing this request: ```py\n{}: {}\n```'
await self.bot.say(fmt.format(type(error).__name__, error))
@commands.command()
@commands.check(checks.is_owner)
async def unload(self, *, module: str):
"""Unloads a module"""
# Do this because I'm too lazy to type cogs.module
module = module.lower()
if not module.startswith("cogs"):
module = "cogs.{}".format(module)
self.bot.unload_extension(module)
await self.bot.say("I have just unloaded the {} module".format(module))
@commands.command()
@commands.check(checks.is_owner)
async def reload(self, *, module: str):
"""Reloads a module"""
# Do this because I'm too lazy to type cogs.module
module = module.lower()
if not module.startswith("cogs"):
module = "cogs.{}".format(module)
self.bot.unload_extension(module)
# This try block will catch errors such as syntax errors in the module we are loading
try:
self.bot.load_extension(module)
await self.bot.say("I have just reloaded the {} module".format(module))
except Exception as error:
fmt = 'An error occurred while processing this request: ```py\n{}: {}\n```'
await self.bot.say(fmt.format(type(error).__name__, error))
def setup(bot):
bot.add_cog(Owner(bot))