153 lines
5.2 KiB
Python
153 lines
5.2 KiB
Python
import logging
|
|
|
|
from discord import ApplicationContext, Cog, option
|
|
from discord.ext import commands
|
|
|
|
from classes.enums import MemberStatus
|
|
from classes.errors import (
|
|
CheckChallengeNotFoundError,
|
|
CheckNotFoundError,
|
|
GuildNotFoundError,
|
|
MemberNotFoundError,
|
|
)
|
|
from classes.pycord_bot import PycordBot
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class CogVerifySetup(Cog):
|
|
def __init__(self, client: PycordBot) -> None:
|
|
super().__init__()
|
|
self.client = client
|
|
|
|
@commands.slash_command(
|
|
name="verify", description="Submit verification check answer"
|
|
)
|
|
@option("answer", description="Answer to the challenge")
|
|
async def command_verify(self, ctx: ApplicationContext, answer: str):
|
|
# Verify that this command has been called in a guild
|
|
if ctx.guild is None:
|
|
await ctx.respond(
|
|
"This command can only be used in a guild.", ephemeral=True
|
|
)
|
|
return
|
|
|
|
# Verify that the guild exists and is configured
|
|
try:
|
|
pycord_guild = await self.client.find_guild(ctx.guild.id)
|
|
if not pycord_guild.is_valid():
|
|
raise GuildNotFoundError(pycord_guild.id)
|
|
except GuildNotFoundError:
|
|
await ctx.respond(
|
|
"This guild has not been set up for checks.", ephemeral=True
|
|
)
|
|
return
|
|
|
|
# Verify that the member is already known
|
|
try:
|
|
pycord_member = await self.client.find_member(ctx.user.id, pycord_guild._id)
|
|
except MemberNotFoundError:
|
|
await ctx.respond("There's no check on your record.", ephemeral=True)
|
|
return
|
|
|
|
# Verify that the member has an active check
|
|
try:
|
|
check = await pycord_member.get_check()
|
|
except CheckNotFoundError:
|
|
await ctx.respond("You do not have an active check.", ephemeral=True)
|
|
return
|
|
|
|
# Verify that the check has a valid challenge
|
|
try:
|
|
challenge = await check.get_challenge()
|
|
except CheckChallengeNotFoundError:
|
|
await ctx.respond(
|
|
"Your check does not have a valid channel.", ephemeral=True
|
|
)
|
|
return
|
|
|
|
# If the answer provided is incorrect
|
|
if not challenge.solve(answer):
|
|
logger.info(
|
|
"Member %s of guild %s has failed a check %s (challenge %s) with answer '%s'",
|
|
pycord_member.id,
|
|
pycord_guild.id,
|
|
check._id,
|
|
challenge._id,
|
|
answer,
|
|
)
|
|
|
|
# Update member's status
|
|
await pycord_member.set_status(MemberStatus.ADDITIONAL)
|
|
|
|
await ctx.respond("Your answer to the challenge is incorrect.")
|
|
|
|
# Assign the member an additional verification role
|
|
try:
|
|
member = ctx.guild.get_member(ctx.user.id)
|
|
additional_role = ctx.guild.get_role(pycord_guild.roles.additional) # type: ignore
|
|
await member.add_roles(additional_role)
|
|
except Exception as exc:
|
|
logger.error(
|
|
"Could not give role %s to %s of guild %s after passing the check due to %s",
|
|
pycord_guild.roles.verified,
|
|
pycord_member.id,
|
|
pycord_guild.id,
|
|
exc,
|
|
)
|
|
|
|
moderator_role = ctx.guild.get_role(pycord_guild.roles.moderator) # type: ignore
|
|
|
|
# Notify moderators (if the thread is available)
|
|
if thread := ctx.guild.get_thread(check.thread_id):
|
|
await thread.send(f"{moderator_role.mention} ACTION REQUIRED")
|
|
|
|
return
|
|
|
|
logger.info(
|
|
"Member %s of guild %s has passed a check %s (challenge %s) with answer '%s'",
|
|
pycord_member.id,
|
|
pycord_guild.id,
|
|
check._id,
|
|
challenge._id,
|
|
answer,
|
|
)
|
|
|
|
# Deactivate the check and update member's status
|
|
await check.deactivate()
|
|
await pycord_member.set_status(MemberStatus.VERIFIED)
|
|
|
|
await ctx.respond("You have passed the challenge!")
|
|
|
|
# Assign the member a verified role
|
|
try:
|
|
member = ctx.guild.get_member(ctx.user.id)
|
|
verified_role = ctx.guild.get_role(pycord_guild.roles.verified) # type: ignore
|
|
|
|
await member.add_roles(verified_role)
|
|
except Exception as exc:
|
|
logger.error(
|
|
"Could not give role %s to %s of guild %s after passing the check due to %s",
|
|
pycord_guild.roles.verified,
|
|
pycord_member.id,
|
|
pycord_guild.id,
|
|
exc,
|
|
)
|
|
|
|
# Delete the thread (if possible)
|
|
if thread := ctx.guild.get_thread(check.thread_id):
|
|
try:
|
|
await thread.delete()
|
|
except Exception as exc:
|
|
logger.error(
|
|
"Could not delete thread %s after the check of %s in guild %s due to %s",
|
|
thread.id,
|
|
pycord_member.id,
|
|
pycord_guild.id,
|
|
exc,
|
|
)
|
|
|
|
|
|
def setup(client: PycordBot):
|
|
client.add_cog(CogVerifySetup(client))
|