mirror of
https://github.com/Cog-Creators/Red-DiscordBot.git
synced 2025-11-06 11:18:54 -05:00
Owner cog / Version command / doc link update / OAUTH (#203)
* Version command (#183) * reword. replace link with link to our doc * Move all owner commands to a separate plugin (#188) * Move all owner commands to their own plugin. * Move all initial cog loading to Owner plugin * Final fix of initial cog loading * don't allow people to unload the owner plugin without reloading it * make sure we modify the cog registry like we're supposed to * log the functionname too Message updates, grammar, politeness etc Get right names in logging formatter Ignore cogs.owner if we find it * add version back in add reload docstring Heh, woops...little security bug here * Add in globals and bot to locals * pass exception to the logger * Bot will now generate OAUTH URL from the supplied endpoint (#196) * Formatting changes, revert uptime * Store the oauth_url internally, provide it if the owner uses `!join`
This commit is contained in:
Will
parent
81409271f5
commit
5b764c41c3
395
cogs/owner.py
Normal file
395
cogs/owner.py
Normal file
@ -0,0 +1,395 @@
|
||||
import discord
|
||||
from discord.ext import commands
|
||||
from cogs.utils import checks
|
||||
from __main__ import set_cog, send_cmd_help, settings
|
||||
|
||||
import importlib
|
||||
import traceback
|
||||
import logging
|
||||
import asyncio
|
||||
import threading
|
||||
import datetime
|
||||
import glob
|
||||
import os
|
||||
import time
|
||||
|
||||
log = logging.getLogger("red.owner")
|
||||
|
||||
|
||||
class CogNotFoundError(Exception):
|
||||
pass
|
||||
|
||||
|
||||
class CogLoadError(Exception):
|
||||
pass
|
||||
|
||||
|
||||
class NoSetupError(CogLoadError):
|
||||
pass
|
||||
|
||||
|
||||
class CogUnloadError(Exception):
|
||||
pass
|
||||
|
||||
|
||||
class OwnerUnloadWithoutReloadError(CogUnloadError):
|
||||
pass
|
||||
|
||||
|
||||
class Owner:
|
||||
"""All owner-only commands that relate to debug bot operations.
|
||||
"""
|
||||
|
||||
def __init__(self, bot):
|
||||
self.bot = bot
|
||||
self.setowner_lock = False
|
||||
|
||||
@commands.command()
|
||||
@checks.is_owner()
|
||||
async def load(self, *, module: str):
|
||||
"""Loads a module
|
||||
|
||||
Example: load mod"""
|
||||
module = module.strip()
|
||||
if "cogs." not in module:
|
||||
module = "cogs." + module
|
||||
try:
|
||||
self._load_cog(module)
|
||||
except CogNotFoundError:
|
||||
await self.bot.say("That module could not be found.")
|
||||
except CogLoadError as e:
|
||||
log.exception(e)
|
||||
traceback.print_exc()
|
||||
await self.bot.say("There was an issue loading the module."
|
||||
" Check your logs for more information.")
|
||||
except Exception as e:
|
||||
log.exception(e)
|
||||
traceback.print_exc()
|
||||
await self.bot.say('Module was found and possibly loaded but '
|
||||
'something went wrong.'
|
||||
' Check your logs for more information.')
|
||||
else:
|
||||
set_cog(module, True)
|
||||
await self.bot.say("Module enabled.")
|
||||
|
||||
@commands.command()
|
||||
@checks.is_owner()
|
||||
async def unload(self, *, module: str):
|
||||
"""Unloads a module
|
||||
|
||||
Example: unload mod"""
|
||||
module = module.strip()
|
||||
if "cogs." not in module:
|
||||
module = "cogs." + module
|
||||
if not self._does_cogfile_exist(module):
|
||||
await self.bot.say("That module file doesn't exist. I will not"
|
||||
" turn off autoloading at start just in case"
|
||||
" this isn't supposed to happen.")
|
||||
else:
|
||||
set_cog(module, False)
|
||||
try: # No matter what we should try to unload it
|
||||
self._unload_cog(module)
|
||||
except OwnerUnloadWithoutReloadError:
|
||||
await self.bot.say("I cannot allow you to unload the Owner plugin"
|
||||
" unless you are in the process of reloading.")
|
||||
except CogUnloadError as e:
|
||||
log.exception(e)
|
||||
traceback.print_exc()
|
||||
await self.bot.say('Unable to safely disable that module.')
|
||||
else:
|
||||
await self.bot.say("Module disabled.")
|
||||
|
||||
@checks.is_owner()
|
||||
@commands.command(name="reload")
|
||||
async def _reload(self, module):
|
||||
"""Reloads a module
|
||||
|
||||
Example: reload audio"""
|
||||
if "cogs." not in module:
|
||||
module = "cogs." + module
|
||||
|
||||
try:
|
||||
self._unload_cog(module, reloading=True)
|
||||
except:
|
||||
pass
|
||||
|
||||
try:
|
||||
self._load_cog(module)
|
||||
except CogNotFoundError:
|
||||
await self.bot.say("That module cannot be found.")
|
||||
except NoSetupError:
|
||||
await self.bot.say("That module does not have a setup function.")
|
||||
except CogLoadError as e:
|
||||
log.exception(e)
|
||||
traceback.print_exc()
|
||||
await self.bot.say("That module could not be loaded. Check your"
|
||||
" logs for more information.")
|
||||
else:
|
||||
set_cog(module, True)
|
||||
await self.bot.say("Module reloaded.")
|
||||
|
||||
@commands.command(pass_context=True, hidden=True)
|
||||
@checks.is_owner()
|
||||
async def debug(self, ctx, *, code):
|
||||
"""Evaluates code
|
||||
|
||||
Modified function, originally made by Rapptz"""
|
||||
code = code.strip('` ')
|
||||
python = '```py\n{}\n```'
|
||||
result = None
|
||||
|
||||
local_vars = locals().copy()
|
||||
local_vars['bot'] = self.bot
|
||||
|
||||
try:
|
||||
result = eval(code, globals(), local_vars)
|
||||
except Exception as e:
|
||||
await self.bot.say(python.format(type(e).__name__ + ': ' + str(e)))
|
||||
return
|
||||
|
||||
if asyncio.iscoroutine(result):
|
||||
result = await result
|
||||
|
||||
result = python.format(result)
|
||||
if not ctx.message.channel.is_private:
|
||||
censor = (settings.email, settings.password)
|
||||
r = "[EXPUNGED]"
|
||||
for w in censor:
|
||||
if w != "":
|
||||
result = result.replace(w, r)
|
||||
result = result.replace(w.lower(), r)
|
||||
result = result.replace(w.upper(), r)
|
||||
await self.bot.say(result)
|
||||
|
||||
@commands.group(name="set", pass_context=True)
|
||||
async def _set(self, ctx):
|
||||
"""Changes Red's global settings."""
|
||||
if ctx.invoked_subcommand is None:
|
||||
await send_cmd_help(ctx)
|
||||
return
|
||||
|
||||
@_set.command(pass_context=True)
|
||||
async def owner(self, ctx):
|
||||
"""Sets owner"""
|
||||
if settings.owner != "id_here":
|
||||
await self.bot.say("Owner ID has already been set.")
|
||||
return
|
||||
|
||||
if self.setowner_lock:
|
||||
await self.bot.say("A set owner command is already pending.")
|
||||
return
|
||||
|
||||
await self.bot.say("Confirm in the console that you're the owner.")
|
||||
self.setowner_lock = True
|
||||
t = threading.Thread(target=self._wait_for_answer,
|
||||
args=(ctx.message.author,))
|
||||
t.start()
|
||||
|
||||
@_set.command()
|
||||
@checks.is_owner()
|
||||
async def prefix(self, *prefixes):
|
||||
"""Sets prefixes
|
||||
|
||||
Must be separated by a space. Enclose in double
|
||||
quotes if a prefix contains spaces."""
|
||||
if prefixes == ():
|
||||
await self.bot.say("Example: setprefix [ ! ^ .")
|
||||
return
|
||||
|
||||
self.bot.command_prefix = sorted(prefixes, reverse=True)
|
||||
settings.prefixes = sorted(prefixes, reverse=True)
|
||||
log.debug("Setting prefixes to:\n\t{}".format(settings.prefixes))
|
||||
|
||||
if len(prefixes) > 1:
|
||||
await self.bot.say("Prefixes set")
|
||||
else:
|
||||
await self.bot.say("Prefix set")
|
||||
|
||||
@_set.command(pass_context=True)
|
||||
@checks.is_owner()
|
||||
async def name(self, ctx, *, name):
|
||||
"""Sets Red's name"""
|
||||
name = name.strip()
|
||||
if name == "":
|
||||
await send_cmd_help(ctx)
|
||||
await self.bot.edit_profile(settings.password, username=name)
|
||||
await self.bot.say("Done.")
|
||||
|
||||
@_set.command(pass_context=True)
|
||||
@checks.is_owner()
|
||||
async def status(self, ctx, *, status=None):
|
||||
"""Sets Red's status
|
||||
|
||||
Leaving this empty will clear it."""
|
||||
|
||||
if status:
|
||||
status = status.strip()
|
||||
await self.bot.change_status(discord.Game(name=status))
|
||||
log.debug('Status set to "{}" by owner'.format(status))
|
||||
else:
|
||||
await self.bot.change_status(None)
|
||||
log.debug('status cleared by owner')
|
||||
await self.bot.say("Done.")
|
||||
|
||||
@_set.command()
|
||||
@checks.is_owner()
|
||||
async def avatar(self, url):
|
||||
"""Sets Red's avatar"""
|
||||
try:
|
||||
async with self.bot.session.get(url) as r:
|
||||
data = await r.read()
|
||||
await self.bot.edit_profile(settings.password, avatar=data)
|
||||
await self.bot.say("Done.")
|
||||
log.debug("changed avatar")
|
||||
except Exception as e:
|
||||
await self.bot.say("Error, check your logs for more information.")
|
||||
log.exception(e)
|
||||
traceback.print_exc()
|
||||
|
||||
@_set.command(name="token")
|
||||
@checks.is_owner()
|
||||
async def _token(self, token):
|
||||
"""Sets Red's login token"""
|
||||
if len(token) < 50:
|
||||
await self.bot.say("Invalid token.")
|
||||
else:
|
||||
settings.login_type = "token"
|
||||
settings.email = token
|
||||
settings.password = ""
|
||||
await self.bot.say("Token set. Restart me.")
|
||||
log.debug("Just converted to a bot account.")
|
||||
|
||||
@commands.command()
|
||||
@checks.is_owner()
|
||||
async def shutdown(self):
|
||||
"""Shuts down Red"""
|
||||
await self.bot.logout()
|
||||
|
||||
@commands.command()
|
||||
@checks.is_owner()
|
||||
async def join(self, invite_url: discord.Invite=None):
|
||||
"""Joins new server"""
|
||||
if hasattr(self.bot.user, 'bot') and self.bot.user.bot is True:
|
||||
# Check to ensure they're using updated discord.py
|
||||
msg = ("I have a **BOT** tag, so I must be invited with an OAuth2"
|
||||
" link:\nFor more information: "
|
||||
"https://twentysix26.github.io/"
|
||||
"Red-Docs/red_guide_bot_accounts/#bot-invites")
|
||||
await self.bot.say(msg)
|
||||
if hasattr(self.bot, 'oauth_url'):
|
||||
await self.bot.whisper("Here's my OAUTH2 link:\n{}".format(
|
||||
self.bot.oauth_url))
|
||||
return
|
||||
|
||||
if invite_url is None:
|
||||
await self.bot.say("I need a Discord Invite link for the "
|
||||
"server you want me to join.")
|
||||
return
|
||||
|
||||
try:
|
||||
await self.bot.accept_invite(invite_url)
|
||||
await self.bot.say("Server joined.")
|
||||
log.debug("We just joined {}".format(invite_url))
|
||||
except discord.NotFound:
|
||||
await self.bot.say("The invite was invalid or expired.")
|
||||
except discord.HTTPException:
|
||||
await self.bot.say("I wasn't able to accept the invite."
|
||||
" Try again.")
|
||||
|
||||
@commands.command(pass_context=True)
|
||||
@checks.is_owner()
|
||||
async def leave(self, ctx):
|
||||
"""Leaves server"""
|
||||
message = ctx.message
|
||||
|
||||
await self.bot.say("Are you sure you want me to leave this server?"
|
||||
" Type yes to confirm.")
|
||||
response = await self.bot.wait_for_message(author=message.author)
|
||||
|
||||
if response.content.lower().strip() == "yes":
|
||||
await self.bot.say("Alright. Bye :wave:")
|
||||
log.debug('Leaving "{}"'.format(message.server.name))
|
||||
await self.bot.leave_server(message.server)
|
||||
else:
|
||||
await self.bot.say("Ok I'll stay here then.")
|
||||
|
||||
@commands.command()
|
||||
async def uptime(self):
|
||||
"""Shows Red's uptime"""
|
||||
up = abs(self.bot.uptime - int(time.perf_counter()))
|
||||
up = str(datetime.timedelta(seconds=up))
|
||||
await self.bot.say("`Uptime: {}`".format(up))
|
||||
|
||||
@commands.command()
|
||||
async def version(self):
|
||||
"""Shows Red's current version"""
|
||||
response = self.bot.loop.run_in_executor(None, self._get_version)
|
||||
result = await asyncio.wait_for(response, timeout=10)
|
||||
await self.bot.say(result)
|
||||
|
||||
def _load_cog(self, cogname):
|
||||
if not self._does_cogfile_exist(cogname):
|
||||
raise CogNotFoundError(cogname)
|
||||
try:
|
||||
mod_obj = importlib.import_module(cogname)
|
||||
importlib.reload(mod_obj)
|
||||
self.bot.load_extension(mod_obj.__name__)
|
||||
except discord.ClientException:
|
||||
raise NoSetupError
|
||||
except SyntaxError as e:
|
||||
raise CogLoadError(*e.args)
|
||||
|
||||
def _unload_cog(self, cogname, reloading=False):
|
||||
if not reloading and cogname == "cogs.owner":
|
||||
raise OwnerUnloadWithoutReloadError(
|
||||
"Can't unload the owner plugin :P")
|
||||
try:
|
||||
self.bot.unload_extension(cogname)
|
||||
except:
|
||||
raise CogUnloadError
|
||||
|
||||
def _list_cogs(self):
|
||||
cogs = glob.glob("cogs/*.py")
|
||||
clean = []
|
||||
for c in cogs:
|
||||
c = c.replace("/", "\\") # Linux fix
|
||||
clean.append("cogs." + c.split("\\")[1].replace(".py", ""))
|
||||
return clean
|
||||
|
||||
def _does_cogfile_exist(self, module):
|
||||
if "cogs." not in module:
|
||||
module = "cogs." + module
|
||||
if module not in self._list_cogs():
|
||||
return False
|
||||
return True
|
||||
|
||||
def _wait_for_answer(self, author):
|
||||
print(author.name + " requested to be set as owner. If this is you, "
|
||||
"type 'yes'. Otherwise press enter.")
|
||||
print()
|
||||
print("*DO NOT* set anyone else as owner.")
|
||||
|
||||
choice = "None"
|
||||
while choice.lower() != "yes" and choice == "None":
|
||||
choice = input("> ")
|
||||
|
||||
if choice == "yes":
|
||||
settings.owner = author.id
|
||||
print(author.name + " has been set as owner.")
|
||||
self.setowner_lock = False
|
||||
self.owner.hidden = True
|
||||
else:
|
||||
print("setowner request has been ignored.")
|
||||
self.setowner_lock = False
|
||||
|
||||
def _get_version(self):
|
||||
getversion = os.popen(r'git show -s HEAD --format="%cr|%s|%h"')
|
||||
getversion = getversion.read()
|
||||
version = getversion.split('|')
|
||||
return 'Last updated: ``{}``\nCommit: ``{}``\nHash: ``{}``'.format(
|
||||
*version)
|
||||
|
||||
|
||||
def setup(bot):
|
||||
n = Owner(bot)
|
||||
bot.add_cog(n)
|
||||
453
red.py
453
red.py
@ -1,18 +1,12 @@
|
||||
from discord.ext import commands
|
||||
import discord
|
||||
from cogs.utils.settings import Settings
|
||||
from random import choice as rndchoice
|
||||
import threading
|
||||
import datetime, re
|
||||
import json, asyncio
|
||||
import copy
|
||||
import glob
|
||||
import json
|
||||
import asyncio
|
||||
import os
|
||||
import time
|
||||
import sys
|
||||
import logging
|
||||
import aiohttp
|
||||
import importlib
|
||||
import shutil
|
||||
import traceback
|
||||
|
||||
@ -38,7 +32,6 @@ settings = Settings()
|
||||
|
||||
from cogs.utils import checks
|
||||
|
||||
lock = False
|
||||
|
||||
@bot.event
|
||||
async def on_ready():
|
||||
@ -53,17 +46,28 @@ async def on_ready():
|
||||
print(servers + " servers")
|
||||
print(channels + " channels")
|
||||
print(users + " users")
|
||||
print("\n{0} active cogs with {1} commands\n".format(str(len(bot.cogs)), str(len(bot.commands))))
|
||||
print("\n{0} active cogs with {1} commands\n".format(
|
||||
str(len(bot.cogs)), str(len(bot.commands))))
|
||||
if settings.login_type == "token":
|
||||
print("------")
|
||||
print("Use this url to bring your bot to a server:")
|
||||
url = await get_oauth_url()
|
||||
bot.oauth_url = url
|
||||
print(url)
|
||||
print("------")
|
||||
|
||||
|
||||
@bot.event
|
||||
async def on_command(command, ctx):
|
||||
pass
|
||||
|
||||
|
||||
@bot.event
|
||||
async def on_message(message):
|
||||
if user_allowed(message):
|
||||
await bot.process_commands(message)
|
||||
|
||||
|
||||
@bot.event
|
||||
async def on_command_error(error, ctx):
|
||||
if isinstance(error, commands.MissingRequiredArgument):
|
||||
@ -81,236 +85,6 @@ async def send_cmd_help(ctx):
|
||||
for page in pages:
|
||||
await bot.send_message(ctx.message.channel, page)
|
||||
|
||||
@bot.command()
|
||||
@checks.is_owner()
|
||||
async def load(*, module : str):
|
||||
"""Loads a module
|
||||
|
||||
Example: load mod"""
|
||||
module = module.strip()
|
||||
if "cogs." not in module: module = "cogs." + module
|
||||
if not module in list_cogs():
|
||||
await bot.say("That module doesn't exist.")
|
||||
return
|
||||
set_cog(module, True)
|
||||
try:
|
||||
mod_obj = importlib.import_module(module)
|
||||
importlib.reload(mod_obj)
|
||||
bot.load_extension(mod_obj.__name__)
|
||||
except Exception as e:
|
||||
await bot.say('{}: {}'.format(type(e).__name__, e))
|
||||
else:
|
||||
await bot.say("Module enabled.")
|
||||
|
||||
@bot.command()
|
||||
@checks.is_owner()
|
||||
async def unload(*, module : str):
|
||||
"""Unloads a module
|
||||
|
||||
Example: unload mod"""
|
||||
module = module.strip()
|
||||
if "cogs." not in module: module = "cogs." + module
|
||||
if not module in list_cogs():
|
||||
await bot.say("That module doesn't exist.")
|
||||
return
|
||||
set_cog(module, False)
|
||||
try:
|
||||
bot.unload_extension(module)
|
||||
except Exception as e:
|
||||
await bot.say('{}: {}'.format(type(e).__name__, e))
|
||||
else:
|
||||
await bot.say("Module disabled.")
|
||||
|
||||
@bot.command(name="reload")
|
||||
@checks.is_owner()
|
||||
async def _reload(*, module : str):
|
||||
"""Reloads a module
|
||||
|
||||
Example: reload mod"""
|
||||
module = module.strip()
|
||||
if "cogs." not in module: module = "cogs." + module
|
||||
if not module in list_cogs():
|
||||
await bot.say("That module doesn't exist.")
|
||||
return
|
||||
set_cog(module, True)
|
||||
try:
|
||||
bot.unload_extension(module)
|
||||
mod_obj = importlib.import_module(module)
|
||||
importlib.reload(mod_obj)
|
||||
bot.load_extension(mod_obj.__name__)
|
||||
except Exception as e:
|
||||
await bot.say('\U0001f52b')
|
||||
await bot.say('{}: {}'.format(type(e).__name__, e))
|
||||
else:
|
||||
await bot.say("Module reloaded.")
|
||||
|
||||
|
||||
@bot.command(pass_context=True, hidden=True) # Modified function, originally made by Rapptz
|
||||
@checks.is_owner()
|
||||
async def debug(ctx, *, code : str):
|
||||
"""Evaluates code"""
|
||||
code = code.strip('` ')
|
||||
python = '```py\n{}\n```'
|
||||
result = None
|
||||
|
||||
try:
|
||||
result = eval(code)
|
||||
except Exception as e:
|
||||
await bot.say(python.format(type(e).__name__ + ': ' + str(e)))
|
||||
return
|
||||
|
||||
if asyncio.iscoroutine(result):
|
||||
result = await result
|
||||
|
||||
result = python.format(result)
|
||||
if not ctx.message.channel.is_private:
|
||||
censor = (settings.email, settings.password)
|
||||
r = "[EXPUNGED]"
|
||||
for w in censor:
|
||||
if w != "":
|
||||
result = result.replace(w, r)
|
||||
result = result.replace(w.lower(), r)
|
||||
result = result.replace(w.upper(), r)
|
||||
await bot.say(result)
|
||||
|
||||
@bot.group(name="set", pass_context=True)
|
||||
async def _set(ctx):
|
||||
"""Changes settings"""
|
||||
if ctx.invoked_subcommand is None:
|
||||
await send_cmd_help(ctx)
|
||||
|
||||
@_set.command(pass_context=True)
|
||||
async def owner(ctx):
|
||||
"""Sets owner"""
|
||||
global lock
|
||||
msg = ctx.message
|
||||
if settings.owner != "id_here":
|
||||
await bot.say("Owner ID has already been set.")
|
||||
return
|
||||
if lock:
|
||||
await bot.say("A setowner request is already pending. Check the console.")
|
||||
return
|
||||
await bot.say("Confirm in the console that you're the owner.")
|
||||
lock = True
|
||||
t = threading.Thread(target=wait_for_answer, args=(ctx.message.author,))
|
||||
t.start()
|
||||
|
||||
@_set.command()
|
||||
@checks.is_owner()
|
||||
async def prefix(*prefixes):
|
||||
"""Sets prefixes
|
||||
|
||||
Must be separated by a space. Enclose in double
|
||||
quotes if a prefix contains spaces."""
|
||||
if prefixes == ():
|
||||
await bot.say("Example: setprefix [ ! ^ .")
|
||||
return
|
||||
bot.command_prefix = sorted(prefixes, reverse=True)
|
||||
settings.prefixes = sorted(prefixes, reverse=True)
|
||||
if len(prefixes) > 1:
|
||||
await bot.say("Prefixes set")
|
||||
else:
|
||||
await bot.say("Prefix set")
|
||||
|
||||
@_set.command(pass_context=True)
|
||||
@checks.is_owner()
|
||||
async def name(ctx, *name : str):
|
||||
"""Sets Red's name"""
|
||||
if name == ():
|
||||
await send_cmd_help(ctx)
|
||||
await bot.edit_profile(settings.password, username=" ".join(name))
|
||||
await bot.say("Done.")
|
||||
|
||||
@_set.command(pass_context=True)
|
||||
@checks.is_owner()
|
||||
async def status(ctx, *status : str):
|
||||
"""Sets Red's status"""
|
||||
if status != ():
|
||||
await bot.change_status(discord.Game(name=" ".join(status)))
|
||||
else:
|
||||
await bot.change_status(None)
|
||||
await bot.say("Done.")
|
||||
|
||||
@_set.command()
|
||||
@checks.is_owner()
|
||||
async def avatar(url : str):
|
||||
"""Sets Red's avatar"""
|
||||
try:
|
||||
async with aiohttp.get(url) as r:
|
||||
data = await r.read()
|
||||
await bot.edit_profile(settings.password, avatar=data)
|
||||
await bot.say("Done.")
|
||||
except:
|
||||
await bot.say("Error.")
|
||||
|
||||
@_set.command(name="token")
|
||||
@checks.is_owner()
|
||||
async def _token(token : str):
|
||||
"""Sets Red's login token"""
|
||||
if len(token) < 50:
|
||||
await bot.say("Invalid token.")
|
||||
else:
|
||||
settings.login_type = "token"
|
||||
settings.email = token
|
||||
settings.password = ""
|
||||
await bot.say("Token set. Restart me.")
|
||||
|
||||
@bot.command()
|
||||
@checks.is_owner()
|
||||
async def shutdown():
|
||||
"""Shuts down Red"""
|
||||
await bot.logout()
|
||||
|
||||
@bot.command()
|
||||
@checks.is_owner()
|
||||
async def join(invite_url : discord.Invite):
|
||||
"""Joins new server"""
|
||||
if bot.user.bot == True:
|
||||
msg = "I have a **BOT** tag, so I must be invited with an OAuth2 link:\n"
|
||||
msg += "`https://discordapp.com/oauth2/authorize?&client_id=`__**`MY_CLIENT_ID_HERE`**__`&scope=bot`\n"
|
||||
msg += "For more information: https://twentysix26.github.io/Red-Docs/red_guide_bot_accounts/#bot-invites"
|
||||
await bot.say(msg)
|
||||
else:
|
||||
try:
|
||||
await bot.accept_invite(invite_url)
|
||||
await bot.say("Server joined.")
|
||||
except discord.NotFound:
|
||||
await bot.say("The invite was invalid or expired.")
|
||||
except discord.HTTPException:
|
||||
await bot.say("I wasn't able to accept the invite. Try again.")
|
||||
|
||||
@bot.command(pass_context=True)
|
||||
@checks.is_owner()
|
||||
async def leave(ctx):
|
||||
"""Leaves server"""
|
||||
message = ctx.message
|
||||
await bot.say("Are you sure you want me to leave this server? Type yes to confirm")
|
||||
response = await bot.wait_for_message(author=message.author)
|
||||
if response.content.lower().strip() == "yes":
|
||||
await bot.say("Alright. Bye :wave:")
|
||||
await bot.leave_server(message.server)
|
||||
else:
|
||||
await bot.say("Ok I'll stay here then.")
|
||||
|
||||
@bot.command(name="uptime")
|
||||
async def _uptime():
|
||||
"""Shows Red's uptime"""
|
||||
up = abs(bot.uptime - int(time.perf_counter()))
|
||||
up = str(datetime.timedelta(seconds=up))
|
||||
await bot.say("`Uptime: {}`".format(up))
|
||||
|
||||
@bot.command()
|
||||
async def version():
|
||||
"""Shows Red's current version"""
|
||||
loop = asyncio.get_event_loop()
|
||||
response = loop.run_in_executor(None, get_version)
|
||||
result = await asyncio.wait_for(response, timeout=10)
|
||||
await bot.say(result)
|
||||
|
||||
def get_version():
|
||||
getversion = os.popen(r'git show -s HEAD --format="%cr|%s|%h"').read()
|
||||
version = getversion.split('|')
|
||||
return 'Last updated: ``{}``\nCommit: ``{}``\nHash: ``{}``'.format(*version)
|
||||
|
||||
def user_allowed(message):
|
||||
|
||||
@ -323,10 +97,12 @@ def user_allowed(message):
|
||||
return True
|
||||
if not message.channel.is_private:
|
||||
server = message.server
|
||||
names = (settings.get_server_admin(server),settings.get_server_mod(server))
|
||||
results = map(lambda name: discord.utils.get(author.roles,name=name),names)
|
||||
names = (settings.get_server_admin(
|
||||
server), settings.get_server_mod(server))
|
||||
results = map(
|
||||
lambda name: discord.utils.get(author.roles, name=name), names)
|
||||
for r in results:
|
||||
if r != None:
|
||||
if r is not None:
|
||||
return True
|
||||
|
||||
if author.id in mod.blacklist_list:
|
||||
@ -346,29 +122,17 @@ def user_allowed(message):
|
||||
else:
|
||||
return True
|
||||
|
||||
def wait_for_answer(author):
|
||||
global lock
|
||||
print(author.name + " requested to be set as owner. If this is you, type 'yes'. Otherwise press enter.")
|
||||
print("*DO NOT* set anyone else as owner.")
|
||||
choice = "None"
|
||||
while choice.lower() != "yes" and choice == "None":
|
||||
choice = input("> ")
|
||||
if choice == "yes":
|
||||
settings.owner = author.id
|
||||
print(author.name + " has been set as owner.")
|
||||
lock = False
|
||||
owner.hidden = True
|
||||
else:
|
||||
print("setowner request has been ignored.")
|
||||
lock = False
|
||||
|
||||
def list_cogs():
|
||||
cogs = glob.glob("cogs/*.py")
|
||||
clean = []
|
||||
for c in cogs:
|
||||
c = c.replace("/", "\\") # Linux fix
|
||||
clean.append("cogs." + c.split("\\")[1].replace(".py", ""))
|
||||
return clean
|
||||
async def get_oauth_url():
|
||||
endpoint = "https://discordapp.com/api/oauth2/applications/@me"
|
||||
if bot.headers.get('authorization') is None:
|
||||
bot.headers['authorization'] = "Bot {}".format(settings.email)
|
||||
|
||||
async with bot.session.get(endpoint, headers=bot.headers) as resp:
|
||||
data = await resp.json()
|
||||
|
||||
return discord.utils.oauth_url(data.get('id'))
|
||||
|
||||
|
||||
def check_folders():
|
||||
folders = ("data", "data/red", "cogs", "cogs/utils")
|
||||
@ -377,18 +141,25 @@ def check_folders():
|
||||
print("Creating " + folder + " folder...")
|
||||
os.makedirs(folder)
|
||||
|
||||
|
||||
def check_configs():
|
||||
if settings.bot_settings == settings.default_settings:
|
||||
print("Red - First run configuration\n")
|
||||
print("You either need a normal account or a bot account to use Red. *Do not* use your own.")
|
||||
print("For more information on bot accounts see: https://discordapp.com/developers/docs/topics/oauth2#bot-vs-user-accounts")
|
||||
print("If you're not interested in a bot account, create a normal account on https://discordapp.com")
|
||||
print("Otherwise make one and copy the token from https://discordapp.com/developers/applications/me")
|
||||
print("You either need a normal account or a bot account to use Red. "
|
||||
"*Do not* use your own.")
|
||||
print("For more information on bot accounts see: https://twentysix26."
|
||||
"github.io/Red-Docs/red_guide_bot_accounts/"
|
||||
"#creating-a-new-bot-account")
|
||||
print("If you decide to use a normal account, create an account for "
|
||||
"your bot on https://discordapp.com then enter your email here.")
|
||||
print("Otherwise make a bot account and copy the token from "
|
||||
"https://discordapp.com/developers/applications/me then enter "
|
||||
"your token here.")
|
||||
print("\nType your email or token:")
|
||||
|
||||
choice = input("> ")
|
||||
|
||||
if "@" not in choice and len(choice) >= 50: #Assuming token
|
||||
if "@" not in choice and len(choice) >= 50: # Assuming token
|
||||
settings.login_type = "token"
|
||||
settings.email = choice
|
||||
elif "@" in choice:
|
||||
@ -397,10 +168,12 @@ def check_configs():
|
||||
settings.password = input("\nPassword> ")
|
||||
else:
|
||||
os.remove('data/red/settings.json')
|
||||
input("Invalid input. Restart Red and repeat the configuration process.")
|
||||
input("Invalid input. Restart Red and repeat the configuration "
|
||||
"process.")
|
||||
exit(1)
|
||||
|
||||
print("\nChoose a prefix (or multiple ones, one at once) for the commands. Type exit when you're done. Example prefix: !")
|
||||
print("\nChoose a prefix (or multiple ones, one at once) for the "
|
||||
"commands. Type exit when you're done. Example prefix: !")
|
||||
prefixes = []
|
||||
new_prefix = ""
|
||||
while new_prefix.lower() != "exit" or prefixes == []:
|
||||
@ -410,24 +183,32 @@ def check_configs():
|
||||
# Remember we're using property's here, oh well...
|
||||
settings.prefixes = sorted(prefixes, reverse=True)
|
||||
|
||||
print("\nIf you know what an User ID is, input *your own* now and press enter.")
|
||||
print("Otherwise you can just set yourself as owner later with '[prefix]set owner'. Leave empty and press enter in this case.")
|
||||
print("\nIf you know what an User ID is, input *your own* now and"
|
||||
" press enter.")
|
||||
print("Otherwise you can just set yourself as owner later with "
|
||||
"'[prefix]set owner'. Leave empty and press enter in this case.")
|
||||
settings.owner = input("\nID> ")
|
||||
if settings.owner == "": settings.owner = "id_here"
|
||||
if settings.owner == "":
|
||||
settings.owner = "id_here"
|
||||
if not settings.owner.isdigit() or len(settings.owner) < 17:
|
||||
if settings.owner != "id_here":
|
||||
print("\nERROR: What you entered is not a valid ID. Set yourself as owner later with [prefix]set owner")
|
||||
print("\nERROR: What you entered is not a valid ID. Set "
|
||||
"yourself as owner later with [prefix]set owner")
|
||||
settings.owner = "id_here"
|
||||
|
||||
print("\nInput the admin role's name. Anyone with this role will be able to use the bot's admin commands")
|
||||
print("\nInput the admin role's name. Anyone with this role will be "
|
||||
"able to use the bot's admin commands")
|
||||
print("Leave blank for default name (Transistor)")
|
||||
settings.default_admin = input("\nAdmin role> ")
|
||||
if settings.default_admin == "": settings.default_admin = "Transistor"
|
||||
if settings.default_admin == "":
|
||||
settings.default_admin = "Transistor"
|
||||
|
||||
print("\nInput the moderator role's name. Anyone with this role will be able to use the bot's mod commands")
|
||||
print("\nInput the moderator role's name. Anyone with this role will "
|
||||
"be able to use the bot's mod commands")
|
||||
print("Leave blank for default name (Process)")
|
||||
settings.default_mod = input("\nModerator role> ")
|
||||
if settings.default_mod == "": settings.default_mod = "Process"
|
||||
if settings.default_mod == "":
|
||||
settings.default_mod = "Process"
|
||||
|
||||
cogs_s_path = "data/red/cogs.json"
|
||||
cogs = {}
|
||||
@ -436,20 +217,30 @@ def check_configs():
|
||||
with open(cogs_s_path, "w") as f:
|
||||
f.write(json.dumps(cogs))
|
||||
|
||||
|
||||
def set_logger():
|
||||
global logger
|
||||
logger = logging.getLogger("discord")
|
||||
logger.setLevel(logging.WARNING)
|
||||
handler = logging.FileHandler(filename='data/red/discord.log', encoding='utf-8', mode='a')
|
||||
handler.setFormatter(logging.Formatter('%(asctime)s %(module)s %(lineno)d %(message)s', datefmt="[%d/%m/%Y %H:%M]"))
|
||||
handler = logging.FileHandler(
|
||||
filename='data/red/discord.log', encoding='utf-8', mode='a')
|
||||
handler.setFormatter(logging.Formatter(
|
||||
'%(asctime)s %(levelname)s %(module)s %(funcName)s %(lineno)d: '
|
||||
'%(message)s',
|
||||
datefmt="[%d/%m/%Y %H:%M]"))
|
||||
logger.addHandler(handler)
|
||||
|
||||
logger = logging.getLogger("red")
|
||||
logger.setLevel(logging.WARNING)
|
||||
handler = logging.FileHandler(filename='data/red/red.log', encoding='utf-8', mode='a')
|
||||
handler.setFormatter(logging.Formatter('%(asctime)s %(module)s %(lineno)d %(message)s', datefmt="[%d/%m/%Y %H:%M]"))
|
||||
handler = logging.FileHandler(
|
||||
filename='data/red/red.log', encoding='utf-8', mode='a')
|
||||
handler.setFormatter(logging.Formatter(
|
||||
'%(asctime)s %(levelname)s %(module)s %(funcName)s %(lineno)d: '
|
||||
'%(message)s',
|
||||
datefmt="[%d/%m/%Y %H:%M]"))
|
||||
logger.addHandler(handler)
|
||||
|
||||
|
||||
def get_answer():
|
||||
choices = ("yes", "y", "no", "n")
|
||||
c = ""
|
||||
@ -460,6 +251,7 @@ def get_answer():
|
||||
else:
|
||||
return False
|
||||
|
||||
|
||||
def set_cog(cog, value):
|
||||
with open('data/red/cogs.json', "r") as f:
|
||||
data = json.load(f)
|
||||
@ -467,6 +259,7 @@ def set_cog(cog, value):
|
||||
with open('data/red/cogs.json', "w") as f:
|
||||
f.write(json.dumps(data))
|
||||
|
||||
|
||||
def load_cogs():
|
||||
try:
|
||||
if sys.argv[1] == "--no-prompt":
|
||||
@ -476,39 +269,47 @@ def load_cogs():
|
||||
except:
|
||||
no_prompt = False
|
||||
|
||||
with open('data/red/cogs.json', "r") as f:
|
||||
data = json.load(f)
|
||||
register = tuple(data.keys()) #known cogs
|
||||
extensions = list_cogs()
|
||||
try:
|
||||
with open('data/red/cogs.json', "r") as f:
|
||||
registry = json.load(f)
|
||||
except:
|
||||
registry = {}
|
||||
|
||||
if extensions: print("\nLoading cogs...\n")
|
||||
bot.load_extension('cogs.owner')
|
||||
owner_cog = bot.get_cog('Owner')
|
||||
if owner_cog is None:
|
||||
print("You got rid of the damn OWNER cog, it has special functions"
|
||||
" that I require to run.\n\n"
|
||||
"I can't start without it!")
|
||||
print()
|
||||
print("Go here to find a new copy:\n{}".format(
|
||||
"https://github.com/Twentysix26/Red-DiscordBot"))
|
||||
exit(1)
|
||||
|
||||
failed = []
|
||||
extensions = owner_cog._list_cogs()
|
||||
for extension in extensions:
|
||||
if extension in register:
|
||||
if data[extension]:
|
||||
try:
|
||||
bot.load_extension(extension)
|
||||
except Exception as e:
|
||||
print(e)
|
||||
failed.append(extension)
|
||||
else:
|
||||
if not no_prompt:
|
||||
print("\nNew extension: " + extension)
|
||||
print("Load it?(y/n)")
|
||||
if get_answer():
|
||||
data[extension] = True
|
||||
try:
|
||||
bot.load_extension(extension)
|
||||
except Exception as e:
|
||||
print(e)
|
||||
failed.append(extension)
|
||||
else:
|
||||
data[extension] = False
|
||||
if extension.lower() == "cogs.owner":
|
||||
continue
|
||||
in_reg = extension in registry
|
||||
if not (in_reg or no_prompt):
|
||||
print("\nNew extension: {}".format(extension))
|
||||
print("Load it?(y/n)")
|
||||
if not get_answer():
|
||||
registry[extension] = False
|
||||
continue
|
||||
registry[extension] = True
|
||||
try:
|
||||
owner_cog._load_cog(extension)
|
||||
except Exception as e:
|
||||
print("{}: {}".format(e.__class__.__name__, str(e)))
|
||||
logger.exception(e)
|
||||
failed.append(extension)
|
||||
registry[extension] = False
|
||||
|
||||
if extensions:
|
||||
with open('data/red/cogs.json', "w") as f:
|
||||
f.write(json.dumps(data))
|
||||
f.write(json.dumps(registry))
|
||||
|
||||
if failed:
|
||||
print("\nFailed to load: ", end="")
|
||||
@ -516,6 +317,9 @@ def load_cogs():
|
||||
print(m + " ", end="")
|
||||
print("\n")
|
||||
|
||||
return owner_cog
|
||||
|
||||
|
||||
def main():
|
||||
global settings
|
||||
global checks
|
||||
@ -523,7 +327,7 @@ def main():
|
||||
check_folders()
|
||||
check_configs()
|
||||
set_logger()
|
||||
load_cogs()
|
||||
owner_cog = load_cogs()
|
||||
if settings.prefixes != []:
|
||||
bot.command_prefix = settings.prefixes
|
||||
else:
|
||||
@ -534,22 +338,24 @@ def main():
|
||||
else:
|
||||
print("Once you're owner use !set prefix to set it.")
|
||||
if settings.owner == "id_here":
|
||||
print("Owner has not been set yet. Do '{}set owner' in chat to set yourself as owner.".format(bot.command_prefix[0]))
|
||||
print("Owner has not been set yet. Do '{}set owner' in chat to set "
|
||||
"yourself as owner.".format(bot.command_prefix[0]))
|
||||
else:
|
||||
owner.hidden = True # Hides the set owner command from help
|
||||
owner_cog.owner.hidden = True # Hides the set owner command from help
|
||||
print("-- Logging in.. --")
|
||||
print("Make sure to keep your bot updated by using: git pull")
|
||||
print("and: pip3 install --upgrade git+https://github.com/Rapptz/discord.py@async")
|
||||
print("and: pip3 install --upgrade git+https://github.com/Rapptz/"
|
||||
"discord.py@async")
|
||||
if settings.login_type == "token":
|
||||
_token.hidden = True
|
||||
owner_cog._token.hidden = True
|
||||
try:
|
||||
yield from bot.login(settings.email)
|
||||
except TypeError as e:
|
||||
print(e)
|
||||
msg = "\n"
|
||||
msg += "You are using an outdated discord.py.\n"
|
||||
msg += "update your discord.py with by running this in your cmd prompt/terminal.\n"
|
||||
msg += "pip3 install --upgrade git+https://github.com/Rapptz/discord.py@async"
|
||||
msg = ("\nYou are using an outdated discord.py.\n"
|
||||
"update your discord.py with by running this in your cmd "
|
||||
"prompt/terminal.\npip3 install --upgrade git+https://"
|
||||
"github.com/Rapptz/discord.py@async")
|
||||
sys.exit(msg)
|
||||
else:
|
||||
yield from bot.login(settings.email, settings.password)
|
||||
@ -561,9 +367,12 @@ if __name__ == '__main__':
|
||||
loop.run_until_complete(main())
|
||||
except discord.LoginFailure:
|
||||
logger.error(traceback.format_exc())
|
||||
print("Invalid login credentials. Restart Red and configure it properly.")
|
||||
shutil.copy('data/red/settings.json', 'data/red/settings-{}.bak'.format(int(time.time())))
|
||||
os.remove('data/red/settings.json') # Hopefully this won't backfire in case of discord servers' problems
|
||||
print("Invalid login credentials. Restart Red and configure it"
|
||||
" properly.")
|
||||
shutil.copy('data/red/settings.json',
|
||||
'data/red/settings-{}.bak'.format(int(time.time())))
|
||||
# Hopefully this won't backfire in case of discord servers' problems
|
||||
os.remove('data/red/settings.json')
|
||||
except:
|
||||
logger.error(traceback.format_exc())
|
||||
loop.run_until_complete(bot.logout())
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user