Moved the Geoffrey Discord bot to its own repo

master
Joey Hines 2019-01-04 10:54:38 -06:00
parent 2ad56b2ede
commit 29f2bab6d4
12 changed files with 748 additions and 1 deletions

2
.gitignore vendored
View File

@ -114,3 +114,5 @@ dmypy.json
# Pyre type checker
.pyre/
.idea/*

View File

@ -0,0 +1,29 @@
from itertools import zip_longest
def get_name(args):
if len(args) > 0:
name = ' '.join(args)
else:
name = None
return name
def get_nickname(discord_user, special_users):
if discord_user.nick is None:
name = discord_user.display_name
else:
name = discord_user.nick
if name in special_users:
return special_users[name]
else:
return name
def get_args_dict(args):
if len(args) != 0:
return dict(zip_longest(*[iter(args)] * 2, fillvalue=" "))
else:
return {}

View File

@ -0,0 +1,3 @@
def formatted_item_listing(item):
return "{} {} for {}D".format(item["amount"], item["item_name"], item["price"])

View File

151
GeoffreyBot/bot.py 100644
View File

@ -0,0 +1,151 @@
import asyncio
import logging
import time
from discord import Game
from discord.ext import commands
from discord.utils import oauth_url
from sys import stdout
import requests
logger = logging.getLogger(__name__)
description = '''
Geoffrey (pronounced JOFF-ree) started his life as an inside joke none of you will understand.
At some point, she was to become an airhorn discord_bot. Now, they know where your stuff is.
Please respect Geoffrey, the discord_bot is very sensitive.
All commands must be prefaced with '?'
If have a suggestion or if something is borked, you can PM my ding dong of a creator BirbHD.
*You must use ?register before adding things to Geoffrey*
For a better a explanation on how this discord_bot works go the following link:
https://github.com/joeyahines/Geoffrey/blob/master/README.md
'''
bad_error_message = 'OOPSIE WOOPSIE!! Uwu We made a fucky wucky!! A wittle fucko boingo! The admins at our ' \
'headquarters are working VEWY HAWD to fix this! (Error in command {})'
extensions = [
#'GeoffreyApp.cogs.Add_Commands',
#'GeoffreyApp.cogs.Delete_Commands',
#'GeoffreyApp.cogs.Edit_Commands',
'GeoffreyApp.assets.bots.discord_bot.cogs.Search_Commands',
#'GeoffreyApp.cogs.Admin_Commands'
]
class GeoffreyBot(commands.Bot):
def __init__(self):
URL = 'http://127.0.0.1:8000/api/settings/'
setting = {}
while True:
try:
setting = requests.get(url=URL).json()
break
except Exception:
time.sleep(1)
super().__init__(command_prefix=setting['BOT_PREFIX'], description=description, pm_help=True,
case_insensitive=True)
self.error_users = setting['ERROR_USERS']
self.admin_users = setting['MOD_RANK']
self.special_users = []
self.default_status = setting['DEFAULT_STATUS']
for extension in extensions:
try:
self.load_extension(extension)
except Exception as e:
logger.info('Failed to load extension {}'.format(extension))
raise e
async def on_ready(self):
logger.info("%s Online, ID: %s", self.user.name, self.user.id)
info = await self.application_info()
url = oauth_url(info.id)
logger.info("Bot url: %s", url)
await self.change_presence(activity=Game(self.default_status))
async def on_command(self, ctx):
URL = 'http://127.0.0.1:8000/api/command/{}'
if ctx.invoked_subcommand is None:
subcommand = ""
else:
subcommand = ":" + ctx.invoked_subcommand.__str__()
logger.info("User %s, used command %s%s with context: %s", ctx.message.author, ctx.command.name, subcommand,
ctx.args)
if ctx.invoked_with.lower() == 'help' and ctx.message.guild is not None:
await ctx.send("{}, I sent you some help in the DMs.".format(ctx.message.author.mention))
async def on_command_error(self, ctx, error):
error_str = ''
if error_str is '':
await self.send_error_message(
'Geoffrey encountered unhandled exception: {} Command: **{}** Context: {}'.format(error,
ctx.command.name,
ctx.args))
error_str = bad_error_message.format(ctx.invoked_with)
logger.error("Geoffrey encountered exception: %s", error)
await ctx.message.channel.send('{} **Error Running Command:** {}'.format(
ctx.message.author.mention, error_str))
async def send_error_message(self, msg):
for user_id in self.error_users:
user = await self.get_user_info(user_id)
await user.send(msg)
def run_command(self, command, args):
URL = 'http://127.0.0.1:8000/api/command/{}'
return requests.get(url=URL.format(command), params=args).json()
def setup_logging():
discord_logger = logging.getLogger('discord')
discord_logger.setLevel(logging.INFO)
bot_info_logger = logging.getLogger('GeoffreyApp.api.discord_bot.discord_bot')
bot_info_logger.setLevel(logging.INFO)
console = logging.StreamHandler(stdout)
console.setFormatter(logging.Formatter('%(asctime)s:%(levelname)s:%(name)s: %(message)s'))
bot_info_logger.addHandler(console)
def start_bot():
asyncio.set_event_loop(asyncio.new_event_loop())
bot = None
try:
bot = GeoffreyBot()
@bot.command(pass_context=True)
async def test(ctx):
"""
Checks if the discord_bot is alive.
"""
await ctx.send('I\'m here you ding dong')
await ctx.send(bot.get)
setup_logging()
bot.run("MTgzMDMyMDE5MTk2ODM3ODg4.DeSPKw.7FnBb4fu2b3CVL2Ls9PZZMDBQQc")
except KeyboardInterrupt:
logger.info("Bot received keyboard interrupt")
except Exception as e:
print(e)
logger.info('Bot encountered the following unhandled exception %s', e)
finally:
if bot is not None:
bot.loop.stop()
logger.info("Bot shutting down...")

View File

@ -0,0 +1,145 @@
from discord.ext import commands
from GeoffreyApp.assets.bots.discord_bot.DiscordHelperFunctions import *
@commands.cooldown(5, 60, commands.BucketType.user)
class Add_Commands:
"""
Commands for adding things to Geoffrey.
*You must use ?register before using any of these commands!*
"""
def __init__(self, bot):
self.bot = bot
@commands.command(pass_context=True)
@commands.cooldown(5, 60, commands.BucketType.user)
async def register(self, ctx):
"""
Registers your Discord and Minecraft account with the the database
You must do this before adding entries to the database.
"""
try:
player_name = get_nickname(ctx.message.author, self.bot.special_users)
self.bot.bot_commands.register(player_name, ctx.message.author.id)
await ctx.send('{}, you have been added to the database. Use ?help to see all the commands this discord_bot can do.'
.format(ctx.message.author.mention))
except AttributeError:
raise NotOnServerError
except PlayerInDBError:
await ctx.send('{}, you are already in the database. Ding dong.'.format(ctx.message.author.mention))
@commands.command(pass_context=True)
@commands.cooldown(5, 60, commands.BucketType.user)
async def add_base(self, ctx, x_pos: int, z_pos: int, *args):
"""
Adds your base to the database. The base name is optional if this is your first base
?add_base [X Coordinate] [Z Coordinate] [Base Name]
"""
name = get_name(args)
try:
base = self.bot.bot_commands.add_base(x_pos, z_pos, base_name=name, discord_uuid=ctx.message.author.id)
await ctx.send(
'{}, your base has been added to the database: \n\n{}'.format(ctx.message.author.mention, base))
except LocationInitError:
raise commands.UserInputError
except EntryNameNotUniqueError:
if name is None:
await ctx.send('{}, you already have one base in the database, you need to specify a base'
' name'.format(ctx.message.author.mention))
else:
await ctx.send(
'{}, a base called **{}** already exists. You need to specify a different name.'.format(
ctx.message.author.mention, name))
@commands.command(pass_context=True)
@commands.cooldown(5, 60, commands.BucketType.user)
async def add_shop(self, ctx, x_pos: int, z_pos: int, *args):
"""
Adds your shop to the database. The name is optional if this is your first shop
?add_shop [X Coordinate] [Z Coordinate] [Shop Name]
"""
name = get_name(args)
try:
shop = self.bot.bot_commands.add_shop(x_pos, z_pos, shop_name=name, discord_uuid=ctx.message.author.id)
await ctx.send(
'{}, your shop has been added to the database: \n\n{}'.format(ctx.message.author.mention, shop))
except LocationInitError:
raise commands.UserInputError
except EntryNameNotUniqueError:
if name is None:
await ctx.send(
'{}, you already have one shop in the database, you need to specify a shop name'.format(
ctx.message.author.mention))
else:
await ctx.send(
'{}, a shop called **{}** already exists. You need to specify a different name.'.format(
ctx.message.author.mention, name))
@commands.command(pass_context=True)
@commands.cooldown(5, 60, commands.BucketType.user)
async def add_tunnel(self, ctx, tunnel_direction: str, tunnel_number: int, *args):
"""
Adds your tunnel to the database. If you only have one location, you do not need to specify a location name
Directions: North South East West
?tunnel [Tunnel Direction] [Tunnel Number] [Location Name]
"""
loc_name = get_name(args)
try:
self.bot.bot_commands.add_tunnel(tunnel_direction, tunnel_number, discord_uuid=ctx.message.author.id,
location_name=loc_name)
await ctx.send('{}, your tunnel has been added to the database'.format(ctx.message.author.mention))
except LocationLookUpError:
await ctx.send('{}, you do not have a location called **{}**.'.format(
ctx.message.author.mention, loc_name))
except NoLocationsInDatabase:
await ctx.send('{}, you do not have a location in the database.'.format(
ctx.message.author.mention, loc_name))
except LocationHasTunnelError:
await ctx.send('{}, **{}** already has a tunnel.'.format(ctx.message.author.mention, loc_name))
except TunnelInitError:
await ctx.send('{}, invalid tunnel direction.'.format(ctx.message.author.mention))
except EntryNameNotUniqueError:
await ctx.send('{}, you have more than one location, you need to specify a location.'
.format(ctx.message.author.mention))
except InvalidTunnelError:
await ctx.send(
'{}, **{}** is an invalid tunnel direction.'.format(ctx.message.author.mention, tunnel_direction))
@commands.command(pass_context=True)
@commands.cooldown(5, 60, commands.BucketType.user)
async def add_item(self, ctx, item_name: str, quantity: int, diamond_price: int, *args):
"""
Adds an item to a shop's inventory. If you have one shop, the shop name is not required
Quantity for Diamond Price. eg. 32 Dirt for 1D. If the item name has spaces in wrap in in quotes eg "Silk Touch"
?add_item [Item Name] [Quantity] [Price] [Shop name]
"""
shop_name = get_name(args)
try:
self.bot.bot_commands.add_item(item_name, quantity, diamond_price, shop_name=shop_name,
discord_uuid=ctx.message.author.id)
await ctx.send(
'{}, **{}** has been added to the inventory of your shop.'.format(ctx.message.author.mention,
item_name))
except NoLocationsInDatabase:
await ctx.send('{}, you don\'t have any shops in the database.'.format(ctx.message.author.mention))
except EntryNameNotUniqueError:
await ctx.send('{}, you have more than one shop in the database, please specify a shop name.'
.format(ctx.message.author.mention))
except LocationLookUpError:
await ctx.send(
'{}, you don\'t have any shops named **{}** in the database.'.format(ctx.message.author.mention,
shop_name))
def setup(bot):
bot.add_cog(Add_Commands(bot))

View File

@ -0,0 +1,164 @@
from discord import Game
from discord.ext import commands
from GeoffreyApp.assets.bots.discord_bot.DiscordHelperFunctions import *
def check_mod(user, admin_users):
try:
for role in user.roles:
if str(role.id) in admin_users:
return True
except AttributeError:
raise NotOnServerError
return False
class Admin_Commands:
"""
Commands for cool people only.
"""
def __init__(self, bot):
self.bot = bot
async def error(self, ctx, error):
error_str = ""
if hasattr(error, "original"):
if isinstance(error.original, PlayerNotFound):
error_str = 'that player is not in the database.'
elif isinstance(error.original, DeleteEntryError) or isinstance(error.original, LocationLookUpError):
error_str = 'that player does not have a location by that name.'
if error_str is "":
error_str = 'the discord_bot encountered the following error: {}'.format(error.__str__())
await ctx.send('{}, {}'.format(ctx.message.author.mention, error_str))
@commands.command(pass_context=True)
async def test(self, ctx):
"""
Checks if the discord_bot is alive.
"""
if check_mod(ctx.message.author, self.bot.admin_users):
await ctx.send('I\'m here you ding dong')
else:
raise NoPermissionError
@commands.group(pass_context=True)
async def mod(self, ctx):
"""
Bot moderation tools.
"""
if check_mod(ctx.message.author, self.bot.admin_users):
if ctx.invoked_subcommand is None:
await ctx.send('{}, invalid sub-command for command **mod**.'.format(ctx.message.author.mention))
else:
raise NoPermissionError
@mod.command(pass_context=True)
async def delete(self, ctx, discord_uuid: str, location_name: str):
"""
Deletes a location in the database.
"""
self.bot.bot_commands.delete(location_name, discord_uuid=discord_uuid)
await ctx.send('{}, **{}** has been deleted.'.format(ctx.message.author.mention, location_name))
@delete.error
async def delete_error(self, ctx, error):
await self.error(ctx, error)
@mod.command(pass_context=True)
async def edit_name(self, ctx, discord_uuid: str, new_name: str, current_name: str):
"""
Edits the name of a location in the database.
"""
self.bot.bot_commands.edit_name(new_name, current_name, discord_uuid=discord_uuid)
await ctx.send('{}, **{}** has been rename to **{}**.'.format(ctx.message.author.mention, current_name,
new_name))
@edit_name.error
async def edit_error(self, ctx, error):
await self.error(ctx, error)
@mod.command(pass_context=True)
async def update_mc_uuid(self, ctx, discord_uuid: str, mc_uuid: str):
"""
Updates a user's MC UUID
"""
self.bot.bot_commands.update_mc_uuid(discord_uuid, mc_uuid)
await ctx.send('{}, **{}** has been updated.'.format(ctx.message.author.mention, discord_uuid))
@update_mc_uuid.error
async def update_mc_uuid_error(self, ctx, error):
await self.error(ctx, error)
@mod.command(pass_context=True)
async def update_discord_uuid(self, ctx, new_discord_uuid: str, current_discord_uuid: str):
"""
Updates a user's Discord UUID
"""
self.bot.bot_commands.update_mc_uuid(current_discord_uuid, new_discord_uuid)
await ctx.send('{}, user **{}** has been updated.'.format(ctx.message.author.mention, current_discord_uuid))
@update_discord_uuid.error
async def update_discord_uuid_error(self, ctx, error):
await self.error(ctx, error)
@mod.command(pass_context=True)
async def update_mc_name(self, ctx, discord_uuid: str):
"""
Updates a user's MC name to the current name on the MC UUID
"""
self.bot.bot_commands.update_mc_name(discord_uuid)
await ctx.send('{}, user **{}**\'s MC name has update.'.format(ctx.message.author.mention, discord_uuid))
@update_mc_name.error
async def update_mc_name_error(self, ctx, error):
await self.error(ctx, error)
@mod.command(pass_context=True)
async def status(self, ctx, *args):
"""
Updates "playing [game]" status of the discord_bot
"""
status = get_name(args)
await self.bot.change_presence(activity=Game(status))
await ctx.send('{}, status has been changed'.format(ctx.message.author.mention))
@mod.command(pass_context=True)
async def add_player(self, ctx, discord_uuid: str, mc_name: str):
"""
Manually add a player to the database
"""
try:
db_id = self.bot.bot_commands.add_player(discord_uuid, mc_name)
await ctx.send('{}, user **{}** been added to the data base with id {}.'.format(ctx.message.author.mention,
mc_name, db_id))
except PlayerInDBError:
await ctx.send('{}, user **{}** is already in the database.'.format(ctx.message.author.mention, mc_name))
@add_player.error
async def add_player_error(self, ctx, error):
await self.error(ctx, error)
@mod.command(pass_context=True)
async def find_player(self, ctx, discord_uuid: str):
"""
Finds a player in the database
"""
try:
db_id, username, discord_uuid, minecraft_uuid = self.bot.bot_commands.find_player(discord_uuid)
await ctx.send('Username: {}, id: {}, Discord UUID: {}, Minecraft UUID: {}'
.format(username, db_id, discord_uuid, minecraft_uuid))
except PlayerNotFound:
await ctx.send('That player is not in the database...')
@find_player.error
async def find_player_error(self, ctx, error):
await self.error(ctx, error)
def setup(bot):
bot.add_cog(Admin_Commands(bot))

View File

@ -0,0 +1,63 @@
from discord.ext import commands
from GeoffreyApp.assets.bots.discord_bot.DiscordHelperFunctions import *
class Delete_Commands:
"""
Commands to help Geoffrey forget.
*You must use ?register before using any of these commands!*
"""
def __init__(self, bot):
self.bot = bot
@commands.command(pass_context=True)
async def delete(self, ctx, *args):
"""
Deletes a location from the database
?delete [Location name]
"""
loc = get_name(args)
try:
if loc is None:
raise commands.UserInputError
self.bot.bot_commands.delete(loc, discord_uuid=ctx.message.author.id)
await ctx.send(
'{}, your location named **{}** has been deleted.'.format(ctx.message.author.mention, loc))
except (DeleteEntryError, PlayerNotFound):
await ctx.send('{}, you do not have a location named **{}**.'.format(ctx.message.author.mention, loc))
@commands.command(pass_context=True)
async def delete_item(self, ctx, item: str, *args):
"""
Deletes an item listing from a shop inventory
The item name must be wrapped in quotes if it has a space in it
?delete_name [Item] [Shop Name]
"""
shop = get_name(args)
try:
shop_name = self.bot.bot_commands.delete_item(item, shop, discord_uuid=ctx.message.author.id)
await ctx.send('{}, **{}** has been removed from the inventory of **{}**.'.
format(ctx.message.author.mention, item, shop_name))
except LocationLookUpError:
await ctx.send('{}, you do not have a shop called **{}**.'.format(ctx.message.author.mention, shop))
except NoLocationsInDatabase:
await ctx.send('{}, you do have any shops in the database.'.format(ctx.message.author.mention))
except EntryNameNotUniqueError:
await ctx.send('{}, you have more than one shop in the database, please specify a shop name.'
.format(ctx.message.author.mention))
except DeleteEntryError:
if shop is not None:
await ctx.send('{}, **{}** does not sell **{}**.'.format(ctx.message.author.mention, shop, item))
else:
await ctx.send('{}, your shop does not sell **{}**.'.format(ctx.message.author.mention, item))
def setup(bot):
bot.add_cog(Delete_Commands(bot))

View File

@ -0,0 +1,74 @@
from discord.ext import commands
from GeoffreyApp.assets.bots.discord_bot.DiscordHelperFunctions import *
class Edit_Commands:
"""
Commands for editing your stuff in Geoffrey.
"""
def __init__(self, bot):
self.bot = bot
@commands.command(pass_context=True)
@commands.cooldown(5, 60, commands.BucketType.user)
async def edit_pos(self, ctx, x_pos: int, z_pos: int, *args):
"""
Edits the position of a location
?edit_pos [X Coordinate] [Z Coordinate] [Location Name]
"""
loc = get_name(args)
try:
loc_str = self.bot.bot_commands.edit_pos(x_pos, z_pos, loc, discord_uuid=ctx.message.author.id)
await ctx.send(
'{}, the following location has been updated: \n\n{}'.format(ctx.message.author.mention, loc_str))
except LocationLookUpError:
await ctx.send('{}, you do not have a location called **{}**.'.format(
ctx.message.author.mention, loc))
@commands.command(pass_context=True)
@commands.cooldown(5, 60, commands.BucketType.user)
async def edit_tunnel(self, ctx, tunnel_direction: str, tunnel_number: int, *args):
"""
Edits the tunnel of a location
Directions: North South East West
?edit_tunnel [Tunnel Direction] [Tunnel Number] [Location Name]
"""
loc = get_name(args)
try:
loc_str = self.bot.bot_commands.edit_tunnel(tunnel_direction, tunnel_number, loc,
discord_uuid=ctx.message.author.id)
await ctx.send(
'{}, the following location has been updated: \n\n{}'.format(ctx.message.author.mention, loc_str))
except LocationLookUpError:
await ctx.send('{}, you do not have a location called **{}**.'.format(
ctx.message.author.mention, loc))
except InvalidTunnelError:
await ctx.send(
'{}, **{}** is an invalid tunnel direction.'.format(ctx.message.author.mention, tunnel_direction))
@commands.command(pass_context=True)
@commands.cooldown(5, 60, commands.BucketType.user)
async def edit_name(self, ctx, new_name: str, current_name: str):
"""
Edits the name of a location
IF A NAME HAS SPACES IN IT YOU NEED TO WRAP IT IN QUOTATION MARKS. eg. "Cool Shop 123"
?edit_name [New Name] [Current Name]
"""
try:
loc_str = self.bot.bot_commands.edit_name(new_name, current_name, discord_uuid=ctx.message.author.id)
await ctx.send(
'{}, the following location has been updated: \n\n{}'.format(ctx.message.author.mention, loc_str))
except LocationLookUpError:
await ctx.send('{}, you do not have a location called **{}**.'.format(
ctx.message.author.mention, current_name))
def setup(bot):
bot.add_cog(Edit_Commands(bot))

View File

@ -0,0 +1,115 @@
from discord.ext import commands
from GeoffreyApp.assets.bots.discord_bot.DiscordHelperFunctions import *
class Search_Commands:
"""
Commands to find stuff.
"""
def __init__(self, bot):
self.bot = bot
@commands.command(pass_context=True)
@commands.cooldown(5, 60, commands.BucketType.user)
async def find(self, ctx, *args):
"""
Finds all the locations matching the search term
?find [Search]
"""
search = get_name(args)
if search is None:
raise commands.UserInputError
result = self.bot.run_command("find_location", {"search": search})
await ctx.send(
'{}, The following entries match **{}**:\n{}'.format(ctx.message.author.mention, search, result))
@commands.command(pass_context=True)
@commands.cooldown(5, 60, commands.BucketType.user)
async def tunnel(self, ctx, player: str):
"""
Finds all the tunnels a player owns
?tunnel [Player]
"""
result = self.bot.run_command("tunnel", {"player_name": player})
await ctx.send(
'{}, **{}** owns the following tunnel(s): \n{}'.format(ctx.message.author.mention, player, result))
@commands.command(pass_context=True)
@commands.cooldown(5, 60, commands.BucketType.user)
async def find_around(self, ctx, x_pos: int, z_pos: int, *args):
"""
Finds all the locations around a certain point.
The radius defaults to 200 blocks if no value is given
Default dimension is the overworld
?find_around [X Coordinate] [Z Coordinate] [Radius]
"""
radius = 200
dimension = 'Overworld'
if len(args) > 0:
radius = int(args[0])
base_string = self.bot.run_command("find_around", {"x_pos": x_pos, "z_pos": z_pos})
if len(base_string) != 0:
await ctx.send('{}, the following locations(s) are within **{}** blocks of that point: \n {}'.format(
ctx.message.author.mention, radius, base_string))
else:
await ctx.send('{}, there are no locations within {} blocks of that point'
.format(ctx.message.author.mention, radius))
@commands.command(pass_context=True)
@commands.cooldown(5, 60, commands.BucketType.user)
async def selling(self, ctx, *args):
"""
Lists all the shops selling an item
?selling [item]
"""
item_name = get_name(args)
result = self.bot.run_command("selling", {"item_name": item_name})
item_listings = []
for item in result:
item_str = "{} {} for {}D".format(item["amount"], item["item_name"], item["price"])
item_listings.append(item_str)
await ctx.send(
'{}, the following shop(s) sell **{}**: \n{}'.format(ctx.message.author.mention, item_name, item_listings))
@commands.command(pass_context=True)
@commands.cooldown(5, 60, commands.BucketType.user)
async def info(self, ctx, *args):
"""
Displays info about a location.
If the location is a shop, it displays the shop's inventory
?info [Location Name]
"""
loc = get_name(args)
info_str = self.bot.run_command("info", {"location_name": loc})
await ctx.send(info_str)
@commands.command(pass_context=True)
@commands.cooldown(5, 60, commands.BucketType.user)
async def me(self, ctx):
"""
Displays all your locations in the database
"""
loc_str = self.bot.run_command("info", {"discord_uuid": ctx.message.author.id})
await ctx.send('{}, here are your location(s) in the database: \n {}'.format(ctx.message.author.mention,
loc_str))
def setup(bot):
bot.add_cog(Search_Commands(bot))

View File

@ -0,0 +1 @@

View File

@ -1,6 +1,6 @@
MIT License
Copyright (c) <year> <copyright holders>
Copyright (c) 2019 ZeroHD
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal