Remove CommandError, return error embeds instead
This commit is contained in:
Ske
parent
99e2fad2b2
commit
2ae8fd5f34
@ -6,7 +6,7 @@ import discord
|
||||
|
||||
import pluralkit
|
||||
from pluralkit import db
|
||||
from pluralkit.bot import utils
|
||||
from pluralkit.bot import utils, embeds
|
||||
|
||||
logger = logging.getLogger("pluralkit.bot.commands")
|
||||
|
||||
@ -17,10 +17,6 @@ class InvalidCommandSyntax(Exception):
|
||||
|
||||
class NoSystemRegistered(Exception):
|
||||
pass
|
||||
|
||||
class CommandError(Exception):
|
||||
def __init__(self, message):
|
||||
self.message = message
|
||||
|
||||
class CommandContext(namedtuple("CommandContext", ["client", "conn", "message", "system"])):
|
||||
client: discord.Client
|
||||
@ -62,9 +58,6 @@ def command(cmd, usage=None, description=None, category=None, system_required=Tr
|
||||
except InvalidCommandSyntax:
|
||||
usage_str = "**Usage:** pk;{} {}".format(cmd, usage or "")
|
||||
await client.send_message(message.channel, embed=utils.make_default_embed(usage_str))
|
||||
except CommandError as e:
|
||||
embed = e.message if isinstance(e.message, discord.Embed) else utils.make_error_embed(e.message)
|
||||
await client.send_message(message.channel, embed=embed)
|
||||
except Exception:
|
||||
logger.exception("Exception while handling command {} (args={}, system={})".format(cmd, args, system.hid if system else "(none)"))
|
||||
|
||||
@ -86,7 +79,7 @@ def member_command(cmd, usage=None, description=None, category=None, system_only
|
||||
member = await utils.get_member_fuzzy(ctx.conn, system_id=system_id, key=args[0], system_only=system_only)
|
||||
|
||||
if member is None:
|
||||
raise CommandError("Can't find member \"{}\".".format(args[0]))
|
||||
return embeds.error("Can't find member \"{}\".".format(args[0]))
|
||||
|
||||
ctx = MemberCommandContext(client=ctx.client, conn=ctx.conn, message=ctx.message, system=ctx.system, member=member)
|
||||
return await func(ctx, args[1:])
|
||||
|
||||
@ -1,10 +1,8 @@
|
||||
import asyncio
|
||||
import re
|
||||
from datetime import datetime
|
||||
import logging
|
||||
from typing import List
|
||||
|
||||
from pluralkit.bot import utils
|
||||
from pluralkit.bot.commands import *
|
||||
|
||||
logger = logging.getLogger("pluralkit.commands")
|
||||
@ -16,7 +14,7 @@ async def import_tupperware(ctx: CommandContext, args: List[str]):
|
||||
|
||||
# Check if there's any Tupperware bot on the server
|
||||
if not tupperware_members:
|
||||
raise CommandError("This command only works in a server where the Tupperware bot is also present.")
|
||||
return embeds.error("This command only works in a server where the Tupperware bot is also present.")
|
||||
|
||||
# Make sure at least one of the bts have send/read permissions here
|
||||
for bot_member in tupperware_members:
|
||||
@ -26,7 +24,7 @@ async def import_tupperware(ctx: CommandContext, args: List[str]):
|
||||
break
|
||||
else:
|
||||
# If no bots have permission (ie. loop doesn't break), throw error
|
||||
raise CommandError("This command only works in a channel where the Tupperware bot has read/send access.")
|
||||
return embeds.error("This command only works in a channel where the Tupperware bot has read/send access.")
|
||||
|
||||
await ctx.reply(embed=utils.make_default_embed("Please reply to this message with `tul!list` (or the server equivalent)."))
|
||||
|
||||
@ -43,12 +41,12 @@ async def import_tupperware(ctx: CommandContext, args: List[str]):
|
||||
|
||||
return tw_msg.embeds[0]["title"].startswith("{}#{}".format(ctx.message.author.name, ctx.message.author.discriminator))
|
||||
|
||||
embeds = []
|
||||
tupperware_page_embeds = []
|
||||
|
||||
tw_msg: discord.Message = await ctx.client.wait_for_message(channel=ctx.message.channel, timeout=60.0, check=ensure_account)
|
||||
if not tw_msg:
|
||||
raise CommandError("Tupperware import timed out.")
|
||||
embeds.append(tw_msg.embeds[0])
|
||||
return embeds.error("Tupperware import timed out.")
|
||||
tupperware_page_embeds.append(tw_msg.embeds[0])
|
||||
|
||||
# Handle Tupperware pagination
|
||||
def match_pagination():
|
||||
@ -84,11 +82,11 @@ async def import_tupperware(ctx: CommandContext, args: List[str]):
|
||||
|
||||
# Make sure it doesn't spin here for too long, time out after 30 seconds since last new page
|
||||
if (datetime.utcnow() - last_found_time).seconds > 30:
|
||||
raise CommandError("Pagination scan timed out.")
|
||||
return embeds.error("Pagination scan timed out.")
|
||||
|
||||
# Now that we've got all the pages, put them in the embeds list
|
||||
# Make sure to erase the original one we put in above too
|
||||
embeds = list([embed for page, embed in sorted(pages_found.items(), key=lambda x: x[0])])
|
||||
tupperware_page_embeds = list([embed for page, embed in sorted(pages_found.items(), key=lambda x: x[0])])
|
||||
|
||||
# Also edit the status message to indicate we're now importing, and it may take a while because there's probably a lot of members
|
||||
await ctx.client.edit_message(status_msg, "All pages read. Now importing...")
|
||||
@ -103,7 +101,7 @@ async def import_tupperware(ctx: CommandContext, args: List[str]):
|
||||
system = await db.create_system(ctx.conn, system_name=None, system_hid=hid)
|
||||
await db.link_account(ctx.conn, system_id=system.id, account_id=ctx.message.author.id)
|
||||
|
||||
for embed in embeds:
|
||||
for embed in tupperware_page_embeds:
|
||||
for field in embed["fields"]:
|
||||
name = field["name"]
|
||||
lines = field["value"].split("\n")
|
||||
@ -150,4 +148,4 @@ async def import_tupperware(ctx: CommandContext, args: List[str]):
|
||||
await db.update_member_field(ctx.conn, member_id=existing_member.id, field="birthday", value=member_birthdate)
|
||||
await db.update_member_field(ctx.conn, member_id=existing_member.id, field="description", value=member_description)
|
||||
|
||||
return "System information imported. Try using `pk;system` now.\nYou should probably remove your members from Tupperware to avoid double-posting."
|
||||
return embeds.success("System information imported. Try using `pk;system` now.\nYou should probably remove your members from Tupperware to avoid double-posting.")
|
||||
|
||||
@ -4,7 +4,7 @@ from datetime import datetime
|
||||
from typing import List
|
||||
from urllib.parse import urlparse
|
||||
|
||||
from pluralkit.bot import utils
|
||||
from pluralkit.bot import utils, embeds
|
||||
from pluralkit.bot.commands import *
|
||||
|
||||
logger = logging.getLogger("pluralkit.commands")
|
||||
@ -21,14 +21,14 @@ async def new_member(ctx: MemberCommandContext, args: List[str]):
|
||||
name = " ".join(args)
|
||||
bounds_error = utils.bounds_check_member_name(name, ctx.system.tag)
|
||||
if bounds_error:
|
||||
raise CommandError(bounds_error)
|
||||
return embeds.error(bounds_error)
|
||||
|
||||
# TODO: figure out what to do if this errors out on collision on generate_hid
|
||||
hid = utils.generate_hid()
|
||||
|
||||
# Insert member row
|
||||
await db.create_member(ctx.conn, system_id=ctx.system.id, member_name=name, member_hid=hid)
|
||||
return "Member \"{}\" (`{}`) registered!".format(name, hid)
|
||||
return embeds.success("Member \"{}\" (`{}`) registered!".format(name, hid))
|
||||
|
||||
|
||||
@member_command(cmd="member set", usage="<name|description|color|pronouns|birthdate|avatar> [value]", description="Edits a member property. Leave [value] blank to clear.", category="Member commands")
|
||||
@ -48,7 +48,7 @@ async def member_set(ctx: MemberCommandContext, args: List[str]):
|
||||
|
||||
prop = args[0]
|
||||
if prop not in allowed_properties:
|
||||
raise CommandError("Unknown property {}. Allowed properties are {}.".format(prop, ", ".join(allowed_properties)))
|
||||
return embeds.error("Unknown property {}. Allowed properties are {}.".format(prop, ", ".join(allowed_properties)))
|
||||
|
||||
if len(args) >= 2:
|
||||
value = " ".join(args[1:])
|
||||
@ -57,12 +57,12 @@ async def member_set(ctx: MemberCommandContext, args: List[str]):
|
||||
if prop == "name":
|
||||
bounds_error = utils.bounds_check_member_name(value, ctx.system.tag)
|
||||
if bounds_error:
|
||||
raise CommandError(bounds_error)
|
||||
return embeds.error(bounds_error)
|
||||
|
||||
if prop == "color":
|
||||
match = re.fullmatch("#?([0-9A-Fa-f]{6})", value)
|
||||
if not match:
|
||||
raise CommandError("Color must be a valid hex color (eg. #ff0000)")
|
||||
return embeds.error("Color must be a valid hex color (eg. #ff0000)")
|
||||
|
||||
value = match.group(1).lower()
|
||||
|
||||
@ -76,7 +76,7 @@ async def member_set(ctx: MemberCommandContext, args: List[str]):
|
||||
# Useful if you want your birthday to be displayed yearless.
|
||||
value = datetime.strptime("0001-" + value, "%Y-%m-%d").date()
|
||||
except ValueError:
|
||||
raise CommandError("Invalid date. Date must be in ISO-8601 format (eg. 1999-07-25).")
|
||||
return embeds.error("Invalid date. Date must be in ISO-8601 format (eg. 1999-07-25).")
|
||||
|
||||
if prop == "avatar":
|
||||
user = await utils.parse_mention(ctx.client, value)
|
||||
@ -90,11 +90,11 @@ async def member_set(ctx: MemberCommandContext, args: List[str]):
|
||||
if u.scheme in ["http", "https"] and u.netloc and u.path:
|
||||
value = value
|
||||
else:
|
||||
raise CommandError("Invalid URL.")
|
||||
return embeds.error("Invalid URL.")
|
||||
else:
|
||||
# Can't clear member name
|
||||
if prop == "name":
|
||||
raise CommandError("Can't clear member name.")
|
||||
return embeds.error("Can't clear member name.")
|
||||
|
||||
# Clear from DB
|
||||
value = None
|
||||
@ -102,7 +102,7 @@ async def member_set(ctx: MemberCommandContext, args: List[str]):
|
||||
db_prop = db_properties[prop]
|
||||
await db.update_member_field(ctx.conn, member_id=ctx.member.id, field=db_prop, value=value)
|
||||
|
||||
response = utils.make_default_embed("{} {}'s {}.".format("Updated" if value else "Cleared", ctx.member.name, prop))
|
||||
response = embeds.success("{} {}'s {}.".format("Updated" if value else "Cleared", ctx.member.name, prop))
|
||||
if prop == "avatar" and value:
|
||||
response.set_image(url=value)
|
||||
if prop == "color" and value:
|
||||
@ -117,10 +117,10 @@ async def member_proxy(ctx: MemberCommandContext, args: List[str]):
|
||||
# Sanity checking
|
||||
example = " ".join(args)
|
||||
if "text" not in example:
|
||||
raise CommandError("Example proxy message must contain the string 'text'.")
|
||||
return embeds.error("Example proxy message must contain the string 'text'.")
|
||||
|
||||
if example.count("text") != 1:
|
||||
raise CommandError("Example proxy message must contain the string 'text' exactly once.")
|
||||
return embeds.error("Example proxy message must contain the string 'text' exactly once.")
|
||||
|
||||
# Extract prefix and suffix
|
||||
prefix = example[:example.index("text")].strip()
|
||||
@ -136,7 +136,7 @@ async def member_proxy(ctx: MemberCommandContext, args: List[str]):
|
||||
async with ctx.conn.transaction():
|
||||
await db.update_member_field(ctx.conn, member_id=ctx.member.id, field="prefix", value=prefix)
|
||||
await db.update_member_field(ctx.conn, member_id=ctx.member.id, field="suffix", value=suffix)
|
||||
return "Proxy settings updated." if prefix or suffix else "Proxy settings cleared."
|
||||
return embeds.success("Proxy settings updated." if prefix or suffix else "Proxy settings cleared.")
|
||||
|
||||
@member_command("member delete", description="Deletes a member from your system ***permanently***.", category="Member commands")
|
||||
async def member_delete(ctx: MemberCommandContext, args: List[str]):
|
||||
@ -145,6 +145,6 @@ async def member_delete(ctx: MemberCommandContext, args: List[str]):
|
||||
msg = await ctx.client.wait_for_message(author=ctx.message.author, channel=ctx.message.channel, timeout=60.0)
|
||||
if msg and msg.content.lower() == ctx.member.hid.lower():
|
||||
await db.delete_member(ctx.conn, member_id=ctx.member.id)
|
||||
return "Member deleted."
|
||||
return embeds.success("Member deleted.")
|
||||
else:
|
||||
return "Member deletion cancelled."
|
||||
return embeds.success("Member deletion cancelled.")
|
||||
@ -1,7 +1,7 @@
|
||||
import logging
|
||||
from typing import List
|
||||
|
||||
from pluralkit.bot import utils
|
||||
from pluralkit.bot import utils, embeds
|
||||
from pluralkit.bot.commands import *
|
||||
|
||||
logger = logging.getLogger("pluralkit.commands")
|
||||
@ -21,7 +21,7 @@ async def message_info(ctx: CommandContext, args: List[str]):
|
||||
# Find the message in the DB
|
||||
message = await db.get_message(ctx.conn, str(mid))
|
||||
if not message:
|
||||
raise CommandError("Message not found.")
|
||||
raise embeds.error("Message with ID '{}' not found.".format(args[0]))
|
||||
|
||||
# Get the original sender of the messages
|
||||
try:
|
||||
|
||||
@ -7,7 +7,7 @@ from typing import List
|
||||
from discord.utils import oauth_url
|
||||
|
||||
import pluralkit.utils
|
||||
from pluralkit.bot import utils
|
||||
from pluralkit.bot import utils, embeds
|
||||
from pluralkit.bot.commands import *
|
||||
|
||||
logger = logging.getLogger("pluralkit.commands")
|
||||
@ -47,7 +47,7 @@ async def invite_link(ctx: CommandContext, args: List[str]):
|
||||
|
||||
url = oauth_url(client_id, permissions)
|
||||
logger.debug("Sending invite URL: {}".format(url))
|
||||
return url
|
||||
return embeds.success("Use this link to add PluralKit to your server: {}".format(url))
|
||||
|
||||
@command(cmd="export", description="Exports system data to a machine-readable format.")
|
||||
async def export(ctx: CommandContext, args: List[str]):
|
||||
|
||||
@ -1,7 +1,7 @@
|
||||
import logging
|
||||
from typing import List
|
||||
|
||||
from pluralkit.bot import utils
|
||||
from pluralkit.bot import utils, embeds
|
||||
from pluralkit.bot.commands import *
|
||||
|
||||
logger = logging.getLogger("pluralkit.commands")
|
||||
@ -9,7 +9,7 @@ logger = logging.getLogger("pluralkit.commands")
|
||||
@command(cmd="mod log", usage="[channel]", description="Sets the bot to log events to a specified channel. Leave blank to disable.", category="Moderation commands", system_required=False)
|
||||
async def set_log(ctx: CommandContext, args: List[str]):
|
||||
if not ctx.message.author.server_permissions.administrator:
|
||||
raise CommandError("You must be a server administrator to use this command.")
|
||||
return embeds.error("You must be a server administrator to use this command.")
|
||||
|
||||
server = ctx.message.server
|
||||
if len(args) == 0:
|
||||
@ -17,8 +17,8 @@ async def set_log(ctx: CommandContext, args: List[str]):
|
||||
else:
|
||||
channel = utils.parse_channel_mention(args[0], server=server)
|
||||
if not channel:
|
||||
raise CommandError("Channel not found.")
|
||||
return embeds.error("Channel not found.")
|
||||
channel_id = channel.id
|
||||
|
||||
await db.update_server(ctx.conn, server.id, logging_channel_id=channel_id)
|
||||
return "Updated logging channel." if channel_id else "Cleared logging channel."
|
||||
return embeds.success("Updated logging channel." if channel_id else "Cleared logging channel.")
|
||||
|
||||
@ -7,7 +7,7 @@ import humanize
|
||||
|
||||
import pluralkit.utils
|
||||
from pluralkit import Member
|
||||
from pluralkit.bot import utils
|
||||
from pluralkit.bot import utils, embeds
|
||||
from pluralkit.bot.commands import *
|
||||
|
||||
logger = logging.getLogger("pluralkit.commands")
|
||||
@ -22,7 +22,7 @@ async def switch_member(ctx: MemberCommandContext, args: List[str]):
|
||||
# Find the member
|
||||
member = await utils.get_member_fuzzy(ctx.conn, ctx.system.id, member_name)
|
||||
if not member:
|
||||
raise CommandError("Couldn't find member \"{}\".".format(member_name))
|
||||
return embeds.error("Couldn't find member \"{}\".".format(member_name))
|
||||
members.append(member)
|
||||
|
||||
# Compare requested switch IDs and existing fronter IDs to check for existing switches
|
||||
@ -31,12 +31,12 @@ async def switch_member(ctx: MemberCommandContext, args: List[str]):
|
||||
fronter_ids = (await pluralkit.utils.get_fronter_ids(ctx.conn, ctx.system.id))[0]
|
||||
if member_ids == fronter_ids:
|
||||
if len(members) == 1:
|
||||
raise CommandError("{} is already fronting.".format(members[0].name))
|
||||
raise CommandError("Members {} are already fronting.".format(", ".join([m.name for m in members])))
|
||||
return embeds.error("{} is already fronting.".format(members[0].name))
|
||||
return embeds.error("Members {} are already fronting.".format(", ".join([m.name for m in members])))
|
||||
|
||||
# Also make sure there aren't any duplicates
|
||||
if len(set(member_ids)) != len(member_ids):
|
||||
raise CommandError("Duplicate members in switch list.")
|
||||
return embeds.error("Duplicate members in switch list.")
|
||||
|
||||
# Log the switch
|
||||
async with ctx.conn.transaction():
|
||||
@ -45,20 +45,20 @@ async def switch_member(ctx: MemberCommandContext, args: List[str]):
|
||||
await db.add_switch_member(ctx.conn, switch_id=switch_id, member_id=member.id)
|
||||
|
||||
if len(members) == 1:
|
||||
return "Switch registered. Current fronter is now {}.".format(members[0].name)
|
||||
return embeds.success("Switch registered. Current fronter is now {}.".format(members[0].name))
|
||||
else:
|
||||
return "Switch registered. Current fronters are now {}.".format(", ".join([m.name for m in members]))
|
||||
return embeds.success("Switch registered. Current fronters are now {}.".format(", ".join([m.name for m in members])))
|
||||
|
||||
@command(cmd="switch out", description="Registers a switch with no one in front.", category="Switching commands")
|
||||
async def switch_out(ctx: MemberCommandContext, args: List[str]):
|
||||
# Get current fronters
|
||||
fronters, _ = await pluralkit.utils.get_fronter_ids(ctx.conn, system_id=ctx.system.id)
|
||||
if not fronters:
|
||||
raise CommandError("There's already no one in front.")
|
||||
raise embeds.error("There's already no one in front.")
|
||||
|
||||
# Log it, and don't log any members
|
||||
await db.add_switch(ctx.conn, system_id=ctx.system.id)
|
||||
return "Switch-out registered."
|
||||
return embeds.success("Switch-out registered.")
|
||||
|
||||
@command(cmd="switch move", usage="<time>", description="Moves the most recent switch to a different point in time.", category="Switching commands")
|
||||
async def switch_move(ctx: MemberCommandContext, args: List[str]):
|
||||
@ -71,18 +71,18 @@ async def switch_move(ctx: MemberCommandContext, args: List[str]):
|
||||
"RETURN_AS_TIMEZONE_AWARE": False
|
||||
})
|
||||
if not new_time:
|
||||
raise CommandError("{} can't be parsed as a valid time.".format(" ".join(args)))
|
||||
return embeds.error("{} can't be parsed as a valid time.".format(" ".join(args)))
|
||||
|
||||
# Make sure the time isn't in the future
|
||||
if new_time > datetime.now():
|
||||
raise CommandError("Can't move switch to a time in the future.")
|
||||
return embeds.error("Can't move switch to a time in the future.")
|
||||
|
||||
# Make sure it all runs in a big transaction for atomicity
|
||||
async with ctx.conn.transaction():
|
||||
# Get the last two switches to make sure the switch to move isn't before the second-last switch
|
||||
last_two_switches = await pluralkit.utils.get_front_history(ctx.conn, ctx.system.id, count=2)
|
||||
if len(last_two_switches) == 0:
|
||||
raise CommandError("There are no registered switches for this system.")
|
||||
return embeds.error("There are no registered switches for this system.")
|
||||
|
||||
last_timestamp, last_fronters = last_two_switches[0]
|
||||
if len(last_two_switches) > 1:
|
||||
@ -90,7 +90,7 @@ async def switch_move(ctx: MemberCommandContext, args: List[str]):
|
||||
|
||||
if new_time < second_last_timestamp:
|
||||
time_str = humanize.naturaltime(second_last_timestamp)
|
||||
raise CommandError("Can't move switch to before last switch time ({}), as it would cause conflicts.".format(time_str))
|
||||
return embeds.error("Can't move switch to before last switch time ({}), as it would cause conflicts.".format(time_str))
|
||||
|
||||
# Display the confirmation message w/ humanized times
|
||||
members = ", ".join([member.name for member in last_fronters]) or "nobody"
|
||||
@ -107,14 +107,14 @@ async def switch_move(ctx: MemberCommandContext, args: List[str]):
|
||||
|
||||
reaction = await ctx.client.wait_for_reaction(emoji=["✅", "❌"], message=confirm_msg, user=ctx.message.author, timeout=60.0)
|
||||
if not reaction:
|
||||
raise CommandError("Switch move timed out.")
|
||||
return embeds.error("Switch move timed out.")
|
||||
|
||||
if reaction.reaction.emoji == "❌":
|
||||
raise CommandError("Switch move cancelled.")
|
||||
return embeds.error("Switch move cancelled.")
|
||||
|
||||
# DB requires the actual switch ID which our utility method above doesn't return, do this manually
|
||||
switch_id = (await db.front_history(ctx.conn, ctx.system.id, count=1))[0]["id"]
|
||||
|
||||
# Change the switch in the DB
|
||||
await db.move_last_switch(ctx.conn, ctx.system.id, switch_id, new_time)
|
||||
return "Switch moved."
|
||||
return embeds.success("Switch moved.")
|
||||
|
||||
@ -6,6 +6,7 @@ import dateparser
|
||||
import humanize
|
||||
|
||||
import pluralkit.utils
|
||||
from pluralkit.bot import embeds
|
||||
from pluralkit.bot.commands import *
|
||||
|
||||
logger = logging.getLogger("pluralkit.commands")
|
||||
@ -21,14 +22,14 @@ async def system_info(ctx: CommandContext, args: List[str]):
|
||||
system = await utils.get_system_fuzzy(ctx.conn, ctx.client, args[0])
|
||||
|
||||
if system is None:
|
||||
raise CommandError("Unable to find system \"{}\".".format(args[0]))
|
||||
return embeds.error("Unable to find system \"{}\".".format(args[0]))
|
||||
|
||||
await ctx.reply(embed=await utils.generate_system_info_card(ctx.conn, ctx.client, system))
|
||||
|
||||
@command(cmd="system new", usage="[name]", description="Registers a new system to this account.", category="System commands", system_required=False)
|
||||
async def new_system(ctx: CommandContext, args: List[str]):
|
||||
if ctx.system:
|
||||
raise CommandError("You already have a system registered. To delete your system, use `pk;system delete`, or to unlink your system from this account, use `pk;system unlink`.")
|
||||
return embeds.error("You already have a system registered. To delete your system, use `pk;system delete`, or to unlink your system from this account, use `pk;system unlink`.")
|
||||
|
||||
system_name = None
|
||||
if len(args) > 0:
|
||||
@ -42,7 +43,7 @@ async def new_system(ctx: CommandContext, args: List[str]):
|
||||
|
||||
# Link account
|
||||
await db.link_account(ctx.conn, system_id=system.id, account_id=ctx.message.author.id)
|
||||
return "System registered! To begin adding members, use `pk;member new <name>`."
|
||||
return embeds.success("System registered! To begin adding members, use `pk;member new <name>`.")
|
||||
|
||||
@command(cmd="system set", usage="<name|description|tag|avatar> [value]", description="Edits a system property. Leave [value] blank to clear.", category="System commands")
|
||||
async def system_set(ctx: CommandContext, args: List[str]):
|
||||
@ -59,14 +60,14 @@ async def system_set(ctx: CommandContext, args: List[str]):
|
||||
|
||||
prop = args[0]
|
||||
if prop not in allowed_properties:
|
||||
raise CommandError("Unknown property {}. Allowed properties are {}.".format(prop, ", ".join(allowed_properties)))
|
||||
raise embeds.error("Unknown property {}. Allowed properties are {}.".format(prop, ", ".join(allowed_properties)))
|
||||
|
||||
if len(args) >= 2:
|
||||
value = " ".join(args[1:])
|
||||
# Sanity checking
|
||||
if prop == "tag":
|
||||
if len(value) > 32:
|
||||
raise CommandError("Can't have system tag longer than 32 characters.")
|
||||
raise embeds.error("Can't have system tag longer than 32 characters.")
|
||||
|
||||
# Make sure there are no members which would make the combined length exceed 32
|
||||
members_exceeding = await db.get_members_exceeding(ctx.conn, system_id=ctx.system.id, length=32 - len(value) - 1)
|
||||
@ -75,7 +76,7 @@ async def system_set(ctx: CommandContext, args: List[str]):
|
||||
member_names = ", ".join([member.name
|
||||
for member in members_exceeding])
|
||||
logger.debug("Members exceeding combined length with tag '{}': {}".format(value, member_names))
|
||||
raise CommandError("The maximum length of a name plus the system tag is 32 characters. The following members would exceed the limit: {}. Please reduce the length of the tag, or rename the members.".format(member_names))
|
||||
raise embeds.error("The maximum length of a name plus the system tag is 32 characters. The following members would exceed the limit: {}. Please reduce the length of the tag, or rename the members.".format(member_names))
|
||||
|
||||
if prop == "avatar":
|
||||
user = await utils.parse_mention(ctx.client, value)
|
||||
@ -89,7 +90,7 @@ async def system_set(ctx: CommandContext, args: List[str]):
|
||||
if u.scheme in ["http", "https"] and u.netloc and u.path:
|
||||
value = value
|
||||
else:
|
||||
raise CommandError("Invalid URL.")
|
||||
raise embeds.error("Invalid URL.")
|
||||
else:
|
||||
# Clear from DB
|
||||
value = None
|
||||
@ -97,7 +98,7 @@ async def system_set(ctx: CommandContext, args: List[str]):
|
||||
db_prop = db_properties[prop]
|
||||
await db.update_system_field(ctx.conn, system_id=ctx.system.id, field=db_prop, value=value)
|
||||
|
||||
response = utils.make_default_embed("{} system {}.".format("Updated" if value else "Cleared", prop))
|
||||
response = embeds.success("{} system {}.".format("Updated" if value else "Cleared", prop))
|
||||
if prop == "avatar" and value:
|
||||
response.set_image(url=value)
|
||||
return response
|
||||
@ -110,12 +111,12 @@ async def system_link(ctx: CommandContext, args: List[str]):
|
||||
# Find account to link
|
||||
linkee = await utils.parse_mention(ctx.client, args[0])
|
||||
if not linkee:
|
||||
raise CommandError("Account not found.")
|
||||
return embeds.error("Account not found.")
|
||||
|
||||
# Make sure account doesn't already have a system
|
||||
account_system = await db.get_system_by_account(ctx.conn, linkee.id)
|
||||
if account_system:
|
||||
raise CommandError("Account is already linked to a system (`{}`)".format(account_system.hid))
|
||||
return embeds.error("Account is already linked to a system (`{}`)".format(account_system.hid))
|
||||
|
||||
# Send confirmation message
|
||||
msg = await ctx.reply("{}, please confirm the link by clicking the ✅ reaction on this message.".format(linkee.mention))
|
||||
@ -125,22 +126,22 @@ async def system_link(ctx: CommandContext, args: List[str]):
|
||||
reaction = await ctx.client.wait_for_reaction(emoji=["✅", "❌"], message=msg, user=linkee, timeout=60.0)
|
||||
# If account to be linked confirms...
|
||||
if not reaction:
|
||||
raise CommandError("Account link timed out.")
|
||||
return embeds.error("Account link timed out.")
|
||||
if not reaction.reaction.emoji == "✅":
|
||||
raise CommandError("Account link cancelled.")
|
||||
return embeds.error("Account link cancelled.")
|
||||
|
||||
await db.link_account(ctx.conn, system_id=ctx.system.id, account_id=linkee.id)
|
||||
return "Account linked to system."
|
||||
return embeds.success("Account linked to system.")
|
||||
|
||||
@command(cmd="system unlink", description="Unlinks your system from this account. There must be at least one other account linked.", category="System commands")
|
||||
async def system_unlink(ctx: CommandContext, args: List[str]):
|
||||
# Make sure you can't unlink every account
|
||||
linked_accounts = await db.get_linked_accounts(ctx.conn, system_id=ctx.system.id)
|
||||
if len(linked_accounts) == 1:
|
||||
raise CommandError("This is the only account on your system, so you can't unlink it.")
|
||||
return embeds.error("This is the only account on your system, so you can't unlink it.")
|
||||
|
||||
await db.unlink_account(ctx.conn, system_id=ctx.system.id, account_id=ctx.message.author.id)
|
||||
return "Account unlinked."
|
||||
return embeds.success("Account unlinked.")
|
||||
|
||||
@command(cmd="system fronter", usage="[system]", description="Gets the current fronter(s) in the system.", category="Switching commands", system_required=False)
|
||||
async def system_fronter(ctx: CommandContext, args: List[str]):
|
||||
@ -152,7 +153,7 @@ async def system_fronter(ctx: CommandContext, args: List[str]):
|
||||
system = await utils.get_system_fuzzy(ctx.conn, ctx.client, args[0])
|
||||
|
||||
if system is None:
|
||||
raise CommandError("Can't find system \"{}\".".format(args[0]))
|
||||
return embeds.error("Can't find system \"{}\".".format(args[0]))
|
||||
|
||||
fronters, timestamp = await pluralkit.utils.get_fronters(ctx.conn, system_id=system.id)
|
||||
fronter_names = [member.name for member in fronters]
|
||||
@ -180,7 +181,7 @@ async def system_fronthistory(ctx: CommandContext, args: List[str]):
|
||||
system = await utils.get_system_fuzzy(ctx.conn, ctx.client, args[0])
|
||||
|
||||
if system is None:
|
||||
raise CommandError("Can't find system \"{}\".".format(args[0]))
|
||||
raise embeds.error("Can't find system \"{}\".".format(args[0]))
|
||||
|
||||
lines = []
|
||||
front_history = await pluralkit.utils.get_front_history(ctx.conn, system.id, count=10)
|
||||
@ -213,9 +214,9 @@ async def system_delete(ctx: CommandContext, args: List[str]):
|
||||
msg = await ctx.client.wait_for_message(author=ctx.message.author, channel=ctx.message.channel, timeout=60.0)
|
||||
if msg and msg.content.lower() == ctx.system.hid.lower():
|
||||
await db.remove_system(ctx.conn, system_id=ctx.system.id)
|
||||
return "System deleted."
|
||||
return embeds.success("System deleted.")
|
||||
else:
|
||||
return "System deletion cancelled."
|
||||
return embeds.error("System deletion cancelled.")
|
||||
|
||||
|
||||
@command(cmd="system frontpercent", usage="[time]",
|
||||
@ -235,7 +236,7 @@ async def system_frontpercent(ctx: CommandContext, args: List[str]):
|
||||
# Fetch list of switches
|
||||
all_switches = await pluralkit.utils.get_front_history(ctx.conn, ctx.system.id, 99999)
|
||||
if not all_switches:
|
||||
raise CommandError("No switches registered to this system.")
|
||||
return embeds.error("No switches registered to this system.")
|
||||
|
||||
# Cull the switches *ending* before the limit, if given
|
||||
# We'll need to find the first switch starting before the limit, then cut off every switch *before* that
|
||||
@ -284,7 +285,7 @@ async def system_frontpercent(ctx: CommandContext, args: List[str]):
|
||||
span_start = max(start_times[-1], before) if before else start_times[-1]
|
||||
total_time = datetime.utcnow() - span_start
|
||||
|
||||
embed = utils.make_default_embed(None)
|
||||
embed = embeds.status("")
|
||||
for member_id, front_time in sorted(member_times.items(), key=lambda x: x[1], reverse=True):
|
||||
member = members_by_id[member_id] if member_id else None
|
||||
|
||||
|
||||
29
src/pluralkit/bot/embeds.py
Normal file
29
src/pluralkit/bot/embeds.py
Normal file
@ -0,0 +1,29 @@
|
||||
from typing import Tuple
|
||||
|
||||
import discord
|
||||
|
||||
|
||||
def success(text: str):
|
||||
embed = discord.Embed()
|
||||
embed.description = text
|
||||
embed.colour = discord.Colour.green()
|
||||
return embed
|
||||
|
||||
|
||||
def error(text: str, help: Tuple[str, str] = None):
|
||||
embed = discord.Embed()
|
||||
embed.description = text
|
||||
embed.colour = discord.Colour.dark_red()
|
||||
|
||||
if help:
|
||||
help_title, help_text = help
|
||||
embed.add_field(name=help_title, value=help_text)
|
||||
|
||||
return embed
|
||||
|
||||
|
||||
def status(text: str):
|
||||
embed = discord.Embed()
|
||||
embed.description = text
|
||||
embed.colour = discord.Colour.blue()
|
||||
return embed
|
||||
Loading…
Reference in New Issue
Block a user