Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Feat/help command #30

Merged
merged 9 commits into from
Oct 25, 2023
78 changes: 78 additions & 0 deletions src/bot/custom_help_command.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,78 @@
import discord
from discord.ext.commands import HelpCommand

# pylint: disable=arguments-differ
class RootPythiaHelpCommand(HelpCommand):
"""
Implementation of HelpCommand

HelpCommand.send_bot_help(mapping) that gets called with <prefix>help
HelpCommand.send_command_help(command) that gets called with <prefix>help <command>
HelpCommand.send_group_help(group) that gets called with <prefix>help <group>
HelpCommand.send_cog_help(cog) that gets called with <prefix>help <cog>
"""

def get_command_signature(self, command):
return f"{command.qualified_name} {command.signature}"

async def send_bot_help(self, mapping):
# color attribute sets color of the border on the left
embed = discord.Embed(title="Help", color=discord.Color.blurple())

for cog, cmds in mapping.items():
# Commands a user can't use are filtered
filtered_cmds = await self.filter_commands(cmds, sort=True)
cmd_signatures = [self.get_command_signature(cmd) for cmd in filtered_cmds]

if cmd_signatures:
cog_name = getattr(cog, "qualified_name", "No Category")
embed.add_field(name=cog_name, value="\n".join(cmd_signatures), inline=False)

await self.get_destination().send(embed=embed)

async def send_command_help(self, command):
embed = discord.Embed(
title=self.get_command_signature(command),
color=discord.Color.blurple())

if command.help:
embed.description = command.help
if alias := command.aliases:
atxr marked this conversation as resolved.
Show resolved Hide resolved
embed.add_field(name="Aliases", value=", ".join(alias), inline=False)

await self.get_destination().send(embed=embed)

async def send_group_help(self, group):
embed = discord.Embed(
title=self.get_command_signature(group),
description=group.help,
color=discord.Color.blurple())

if filtered_commands := await self.filter_commands(group.commands):
for command in filtered_commands:
embed.add_field(
name=self.get_command_signature(command),
value=command.help or "No Help Message Found... ")

await self.get_destination().send(embed=embed)

async def send_cog_help(self, cog):
embed = discord.Embed(
title=cog.qualified_name or "No Category",
description=cog.description,
color=discord.Color.blurple())

if filtered_commands := await self.filter_commands(cog.get_commands()):
for command in filtered_commands:
embed.add_field(
name=self.get_command_signature(command),
value=command.help or "No Help Message Found... ",
inline=False)

await self.get_destination().send(embed=embed)

async def send_error_message(self, error):
embed = discord.Embed(title="Error", description=error, color=discord.Color.red())
channel = self.get_destination()

await channel.send(embed=embed)
46 changes: 31 additions & 15 deletions src/bot/root_pythia_bot.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@

from api.rootme_api import RootMeAPIManager
from api.rate_limiter import RateLimiter
from bot.custom_help_command import RootPythiaHelpCommand
from bot.root_pythia_cogs import RootPythiaCommands
from bot.dummy_db_manager import DummyDBManager

Expand All @@ -17,33 +18,42 @@


def craft_intents():
# Intents int: 19456 means:
# - Send messages
# - Embed Links
# - Read messages/View Channels
# configured in the discord dev portal https://discord.com/developers/applications
# Notice that the bot doesn't required the intent "Send messages in threads"
intents = discord.Intents(value=19456)

# Disable privilegied and enbale message_content privilegied intent to enable commands
"""Function that enables necessary intents for the bot"""

# Disable everything
intents = discord.Intents.none()
# enable guild related events
# More info: https://docs.pycord.dev/en/stable/api/data_classes.html#discord.Intents.guilds
intents.guilds = True

# Warning: message_content is a privileged intents
# you must authorize it in the discord dev portal https://discord.com/developers/applications
# enbale message_content privilegied intent to enable commands
# More info below:
# https://docs.pycord.dev/en/stable/api/data_classes.html#discord.Intents.message_content
intents.message_content = True
intents.messages = True
intents.typing = False
intents.guild_typing = False
intents.presences = False

# enable guild messages related events
# More info below:
# https://docs.pycord.dev/en/stable/api/data_classes.html#discord.Intents.guild_messages
intents.guild_messages = True

return intents


########### Create bot object #################
_DESCRIPTION = (
"RootPythia is a Discord bot fetching RootMe API to notify everyone"
"RootPythia is a Discord bot fetching RootMe API to notify everyone "
"when a user solves a new challenge!"
)
_PREFIX = "!"
_INTENTS = craft_intents()

BOT = commands.Bot(command_prefix=_PREFIX, description=_DESCRIPTION, intents=_INTENTS)
BOT = commands.Bot(
command_prefix=_PREFIX,
description=_DESCRIPTION,
intents=_INTENTS,
help_command=RootPythiaHelpCommand())

# Create Bot own logger, each Cog will also have its own
BOT.logger = logging.getLogger(__name__)
Expand Down Expand Up @@ -84,3 +94,9 @@ async def on_error(event, *args, **kwargs):
await BOT.channel.send(f"{event} event failed, please check logs for more details")

BOT.logger.exception("Unhandled exception in '%s' event", event)


@BOT.event
async def on_command(ctx):
"""Add logging when a command is triggered by a user"""
BOT.logger.info("'%s' command triggered by '%s'", ctx.command, ctx.author)
ctmbl marked this conversation as resolved.
Show resolved Hide resolved
61 changes: 34 additions & 27 deletions src/bot/root_pythia_cogs.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,8 +14,8 @@

class RootPythiaCommands(commands.Cog, name=NAME):
"""
Define the commands that the bot will respond to, prefixed with the `command_prefix` defined at
the bot init
Define the commands that the bot will respond to, prefixed with '!' \
ctmbl marked this conversation as resolved.
Show resolved Hide resolved
defined at the bot init
"""

def __init__(self, bot, dbmanager):
Expand All @@ -26,16 +26,11 @@ def __init__(self, bot, dbmanager):
# pylint: disable-next=no-member
self.check_new_solves.start()

async def log_command_call(self, ctx):
# Maybe we should use the on_command event
# https://discordpy.readthedocs.io/en/stable/ext/commands/api.html?highlight=cog#discord.discord.ext.commands.on_command
# the logging would be implicit, no need of a redundant @commands.before_invoke(...) on
# each command. But right now I prefer to stick with this explicit solution
self.logger.info("'%s' command triggered by '%s'", ctx.command, ctx.author)

@commands.before_invoke(log_command_call)
@commands.command(name="status")
async def status(self, ctx):
"""
Give info on bot status
"""
rate_limiter = self.dbmanager.api_manager.rate_limiter
check = ":white_check_mark:"
cross = ":x:"
Expand All @@ -55,6 +50,9 @@ async def status(self, ctx):

@commands.command(name="resume")
async def resume(self, ctx):
"""
The bot leaves its idle state
"""
rate_limiter = self.dbmanager.api_manager.rate_limiter
if not rate_limiter.is_idle():
await ctx.message.channel.send("The Rate Limiter isn't idle, no need to resume.")
Expand All @@ -65,27 +63,32 @@ async def resume(self, ctx):
"Resumed successfully from idle state, requests can be sent again."
)

async def base_add_one_user(self, ctx, idx: int):
async def base_add_one_user(self, ctx, user_id: int):
# TODO: except 404 error -> if the request in add_user fails
# try:
# user = await self.dbmanager.add_user(idx)
# except 404Error as 404_err: # Error coming from add_user() method
# pass
user = await self.dbmanager.add_user(idx)
user = await self.dbmanager.add_user(user_id)

if user is None:
await ctx.message.channel.send(f"UserID {idx} already exists in database")
self.logger.warning("UserID '%s' already exists in database", idx)
await ctx.message.channel.send(f"UserID {user_id} already exists in database")
self.logger.warning("UserID '%s' already exists in database", user_id)
return

self.logger.info("Add user '%s'", user)
await ctx.message.channel.send(f"{user} added!\nPoints: {user.score}")

@commands.before_invoke(log_command_call)
@commands.command(name="addusers")
async def addusers(self, ctx, *args):
self.logger.info("command `addusers` received '%s' arguments: '%s'", len(args), args)
for user_id in args:
async def addusers(self, ctx, *user_ids):
"""
Add several rootme users with [user_ids...] to the tracked users
(each id is separated by a whitespace when using the command)
"""
self.logger.info(
"command `addusers` received '%s' arguments: '%s'", len(user_ids), user_ids
)
for user_id in user_ids:
try:
user_id = int(user_id)
except ValueError as value_err: # Error coming from the int() cast
Expand All @@ -99,24 +102,28 @@ async def addusers(self, ctx, *args):
# should continue here (ex: multiple users but only one can be wrong)
await self.base_add_one_user(ctx, user_id)

@commands.before_invoke(log_command_call)
@commands.command(name="adduser")
async def adduser(self, ctx, idx: int):
await self.base_add_one_user(ctx, idx)
async def adduser(self, ctx, user_id: int):
"""
Add rootme user with <user_id> to the tracked users
"""
await self.base_add_one_user(ctx, user_id)

@commands.before_invoke(log_command_call)
@commands.command(name="getuser")
async def getuser(self, ctx, idx: int):
user = self.dbmanager.get_user(idx)
async def getuser(self, ctx, user_id: int):
"""
Get rootme user info with <user_id>
"""
user = self.dbmanager.get_user(user_id)

if user is None:
self.logger.debug("DB Manager returned 'None' for UserID '%s'", idx)
self.logger.debug("DB Manager returned 'None' for UserID '%s'", user_id)
await ctx.message.channel.send(
f"User id '{idx}' isn't in the database, you must add it first"
f"User id '{user_id}' isn't in the database, you must add it first"
)
return

self.logger.debug("Get user '%s' for id=%d", repr(user), idx)
self.logger.debug("Get user '%s' for id=%d", repr(user), user_id)
await ctx.message.channel.send(
f"{user} \nPoints: {user.score}\nRank: {user.rank}\nLast Solves: <TO BE COMPLETED>"
)
Expand Down
14 changes: 14 additions & 0 deletions tests/test_cog.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,8 @@
from data import auteurs_example_data


## NOTE: to debug: print(dpytest.get_message().content)

@pytest.mark.asyncio
async def test_adduser_command(config_bot):
# if the API Manager is not rightly mocked this test should fail, on purpose!
Expand All @@ -29,6 +31,18 @@ async def test_getuser_command(config_bot):
assert dpytest.verify().message().contains().content("Points: 3040")


@pytest.mark.asyncio
async def test_help_command(config_bot):
###
# comments are the same than for test_adduser_command you should check it out
###
await dpytest.message("!help")

embedded_help_words = ["adduser", "addusers", "getuser", "help", "status"]
for word in embedded_help_words:
assert dpytest.verify().message().peek().contains().content(word)


@pytest.mark.asyncio
async def test_getuser_not_found(config_bot):
###
Expand Down