TheLovinator/discord-reminder-bot
1
Fork 0
Code Issues Pull Requests Actions Packages Releases Activity

Add better error handling

Browse Source
This commit is contained in:
Joakim Hellsén
2025-01-14 00:22:52 +01:00
parent f1695848ef
commit e379afe108
4 changed files with 189 additions and 56 deletions
Download Patch File Download Diff File Expand all files Collapse all files

68
discord_reminder_bot/interaction_responses.py Normal file
View File

@ -0,0 +1,68 @@
from __future__ import annotations
import logging
import discord
logger: logging.Logger = logging.getLogger(__name__)
async def send_error_response(interaction: discord.Interaction, msg: str, *, ephemeral: bool = False) -> None:
"""Handle the error.
Args:
interaction (discord.Interaction): The interaction object for the command.
msg (str): The message to send.
ephemeral (bool, optional): Whether the message should be ephemeral. Defaults to False.
"""
logger.exception(msg)
await interaction.response.send_message(msg, ephemeral=ephemeral)
async def followup_msg(
interaction: discord.Interaction,
*, # So that the following arguments are keyword-only
msg: str | None = None,
ephemeral: bool = False,
embed: discord.Embed | None = None,
view: discord.ui.View | None = None,
) -> None:
"""Send a followup message to the interaction.
Handles the exceptions that may occur when sending a followup message.
Args:
interaction (discord.Interaction): The interaction object for the command.
msg (str): The message to send.
ephemeral (bool, optional): Whether the message should be ephemeral. Defaults to False.
embed (discord.Embed | None, optional): The embed to send. Defaults to no embed.
view (discord.ui.View | None, optional): The view to send. Defaults to no view.
"""
if not msg:
msg = "No message was provided."
try:
if embed and view:
log_msg: str = f"Sending followup message with embed and view to the interaction.\n{msg=}.\n{ephemeral=}\n{embed.to_dict()=}" # noqa: E501
for view_item in view.children:
log_msg += f"\n{view_item=}"
logger.info(log_msg)
await interaction.followup.send(embed=embed, ephemeral=ephemeral, view=view)
else:
logger.info("Sending followup message to the interaction.\nMessage: %s.\nEphemeral: %s", msg, ephemeral)
await interaction.followup.send(content=msg, ephemeral=ephemeral)
except (discord.NotFound, discord.Forbidden, discord.HTTPException, TypeError, ValueError) as e:
error_messages: dict[type[discord.HTTPException | TypeError | ValueError], str] = {
discord.NotFound: "The original message was not found.",
discord.Forbidden: "The authorization token for the webhook is incorrect.",
discord.HTTPException: "Failed to send followup message.",
TypeError: "We specified both embed and embeds or file and files, or thread and threads.",
ValueError: (
"The length of embeds was invalid, there was no token associated with this webhook or "
"ephemeral was passed with the improper webhook type or there was no state attached with "
"this webhook when giving it a view."
),
}
assert_msg: str = error_messages[type(e)]
await send_error_response(interaction=interaction, msg=assert_msg, ephemeral=ephemeral)

114
discord_reminder_bot/main.py
View File

@ -10,6 +10,7 @@ from discord.abc import PrivateChannel
from discord_webhook import DiscordWebhook
from discord_reminder_bot import settings
from discord_reminder_bot.interaction_responses import followup_msg
from discord_reminder_bot.misc import calculate
from discord_reminder_bot.parser import parse_time
from discord_reminder_bot.ui import JobManagementView, create_job_embed
@ -106,7 +107,7 @@ class RemindGroup(discord.app_commands.Group):
run_date: str = parsed_time.strftime("%Y-%m-%d %H:%M:%S")
guild: discord.Guild | None = interaction.guild or None
if not guild:
await interaction.followup.send("Failed to get guild.")
await followup_msg(interaction=interaction, msg="Failed to get guild.")
return
dm_message: str = ""
@ -142,8 +143,9 @@ class RemindGroup(discord.app_commands.Group):
where_and_when = (
f"I will notify you in <#{channel_id}> {dm_message}at:\n**{run_date}** {calculate(reminder)}\n"
)
final_message: str = f"Hello {interaction.user.display_name}, {where_and_when}With the message:\n**{message}**."
await interaction.followup.send(final_message)
await followup_msg(interaction=interaction, msg=final_message)
# /remind list
@discord.app_commands.command(name="list", description="List, pause, unpause, and remove reminders.")
@ -157,21 +159,17 @@ class RemindGroup(discord.app_commands.Group):
jobs: list[Job] = settings.scheduler.get_jobs()
if not jobs:
await interaction.followup.send(content="No jobs available.")
await followup_msg(interaction=interaction, msg="No scheduled jobs found in the database.")
return
first_job: Job | None = jobs[0] if jobs else None
if not first_job:
await interaction.followup.send(content="No jobs available.")
return
embed: discord.Embed = create_job_embed(job=jobs[0])
view = JobManagementView(job=jobs[0], scheduler=settings.scheduler)
embed: discord.Embed = create_job_embed(job=first_job)
view = JobManagementView(job=first_job, scheduler=settings.scheduler)
await interaction.followup.send(embed=embed, view=view)
await followup_msg(interaction=interaction, embed=embed, view=view)
# /remind cron
@discord.app_commands.command(name="cron", description="Create new cron job. Works like UNIX cron.")
async def cron( # noqa: PLR0913, PLR0917
async def cron( # noqa: C901, PLR0911, PLR0913, PLR0917
self,
interaction: discord.Interaction,
message: str,
@ -214,9 +212,14 @@ class RemindGroup(discord.app_commands.Group):
user (discord.User, optional): Send reminder as a DM to this user. Defaults to None.
dm_and_current_channel (bool, optional): If user is provided, send reminder as a DM to the user and in this channel. Defaults to only the user.
""" # noqa: E501
# TODO(TheLovinator): Add try/except for all of these await calls # noqa: TD003
# TODO(TheLovinator): Add a warning if the interval is too short # noqa: TD003
# TODO(TheLovinator): Check if we have access to the channel and user# noqa: TD003
try:
await interaction.response.defer()
except discord.HTTPException as e:
logger.exception("Failed to defer interaction: text=%s, status=%s, code=%s", e.text, e.status, e.code)
return
except discord.InteractionResponded as e:
logger.exception("Interaction already responded to - interaction: %s", e.interaction)
return
# Log kwargs
logger.info("New cron job from %s (%s) in %s", interaction.user, interaction.user.id, interaction.channel)
@ -225,39 +228,43 @@ class RemindGroup(discord.app_commands.Group):
# Get the channel ID
channel_id: int | None = self.get_channel_id(interaction=interaction, channel=channel)
if not channel_id:
await interaction.followup.send(content="Failed to get channel.")
await followup_msg(interaction, msg="Failed to get channel.")
return
# Ensure the guild is valid
guild: discord.Guild | None = interaction.guild or None
if not guild:
await interaction.followup.send(content="Failed to get guild.")
await followup_msg(interaction, msg="Failed to get guild.")
return
# Helper to add a job
def add_job(func: Callable, job_kwargs: dict[str, int | str]) -> Job:
return settings.scheduler.add_job(
func=func,
trigger="cron",
year=year,
month=month,
day=day,
week=week,
day_of_week=day_of_week,
hour=hour,
minute=minute,
second=second,
start_date=start_date,
end_date=end_date,
timezone=timezone,
jitter=jitter,
kwargs=job_kwargs,
)
def add_job(func: Callable, job_kwargs: dict[str, int | str]) -> Job | None:
try:
return settings.scheduler.add_job(
func=func,
trigger="cron",
year=year,
month=month,
day=day,
week=week,
day_of_week=day_of_week,
hour=hour,
minute=minute,
second=second,
start_date=start_date,
end_date=end_date,
timezone=timezone,
jitter=jitter,
kwargs=job_kwargs,
)
except Exception:
logger.exception("Failed to add job: %s")
return None
# Create user DM reminder job if user is specified
dm_message: str = ""
if user:
dm_job: Job = add_job(
dm_job: Job | None = add_job(
func=send_to_user,
job_kwargs={
"user_id": user.id,
@ -265,17 +272,22 @@ class RemindGroup(discord.app_commands.Group):
"message": message,
},
)
if not dm_job:
await followup_msg(interaction=interaction, msg="Failed to create DM reminder job.")
return
dm_message = f" and a DM to {user.display_name}"
if not dm_and_current_channel:
# If only DM is required, notify about the DM job and exit
await interaction.followup.send(
content=f"Hello {interaction.user.display_name},\n"
await followup_msg(
interaction=interaction,
msg=f"Hello {interaction.user.display_name},\n"
f"I will send a DM to {user.display_name} at:\n"
f"First run in {calculate(dm_job)} with the message:\n**{message}**.",
)
return
# Create channel reminder job
channel_job: Job = add_job(
channel_job: Job | None = add_job(
func=send_to_discord,
job_kwargs={
"channel_id": channel_id,
@ -283,10 +295,14 @@ class RemindGroup(discord.app_commands.Group):
"author_id": interaction.user.id,
},
)
if not channel_job:
await followup_msg(interaction, msg="Failed to create channel reminder job.")
return
# Compose the final message
await interaction.followup.send(
content=f"Hello {interaction.user.display_name},\n"
await followup_msg(
interaction=interaction,
msg=f"Hello {interaction.user.display_name},\n"
f"I will notify you in <#{channel_id}>{dm_message}.\n"
f"First run in {calculate(channel_job)} with the message:\n**{message}**.",
)
@ -341,13 +357,13 @@ class RemindGroup(discord.app_commands.Group):
# Get the channel ID
channel_id: int | None = self.get_channel_id(interaction=interaction, channel=channel)
if not channel_id:
await interaction.followup.send(content="Failed to get channel.")
await followup_msg(interaction, msg="Failed to get channel.")
return
# Ensure the guild is valid
guild: discord.Guild | None = interaction.guild or None
if not guild:
await interaction.followup.send(content="Failed to get guild.")
await followup_msg(interaction, msg="Failed to get guild.")
return
# Helper to add a job
@ -381,8 +397,9 @@ class RemindGroup(discord.app_commands.Group):
dm_message = f" and a DM to {user.display_name} "
if not dm_and_current_channel:
# If only DM is required, notify about the DM job and exit
await interaction.followup.send(
content=f"Hello {interaction.user.display_name},\n"
await followup_msg(
interaction=interaction,
msg=f"Hello {interaction.user.display_name},\n"
f"I will send a DM to {user.display_name} at:\n"
f"First run in {calculate(dm_job)} with the message:\n**{message}**.",
)
@ -398,8 +415,9 @@ class RemindGroup(discord.app_commands.Group):
)
# Compose the final message
await interaction.followup.send(
content=f"Hello {interaction.user.display_name},\n"
await followup_msg(
interaction=interaction,
msg=f"Hello {interaction.user.display_name},\n"
f"I will notify you in <#{channel_id}>{dm_message}.\n"
f"First run in {calculate(channel_job)} with the message:\n**{message}**.",
)
@ -444,7 +462,7 @@ class RemindGroup(discord.app_commands.Group):
logger.exception("Error parsing time '%s'", time)
error_during_parsing = e
if not parsed:
await interaction.followup.send(f"Failed to parse time. Error: {error_during_parsing}")
await followup_msg(interaction=interaction, msg=f"Failed to parse time. Error: {error_during_parsing}")
return None
return parsed

30
discord_reminder_bot/misc.py
View File

@ -25,14 +25,42 @@ def calculate(job: Job) -> str:
trigger_time: datetime.datetime | None = (
job.trigger.run_date if isinstance(job.trigger, DateTrigger) else job.next_run_time
)
# Check if the job is paused
if trigger_time is None:
logger.error("Couldn't calculate time for job: %s: %s", job.id, job.name)
logger.error("State: %s", job.__getstate__() if hasattr(job, "__getstate__") else "No state")
return "Couldn't calculate time"
return "Paused"
return f"<t:{int(trigger_time.timestamp())}:R>"
def get_human_time(time: datetime.timedelta) -> str:
"""Convert timedelta to human-readable format.
Args:
time: The timedelta to convert.
Returns:
str: The human-readable time.
"""
days, seconds = divmod(time.total_seconds(), 86400)
hours, seconds = divmod(seconds, 3600)
minutes, seconds = divmod(seconds, 60)
time_str: str = ""
if days:
time_str += f"{int(days)}d"
if hours:
time_str += f"{int(hours)}h"
if minutes:
time_str += f"{int(minutes)}m"
if seconds:
time_str += f"{int(seconds)}s"
return time_str
def calc_time(time: datetime.datetime) -> str:
"""Convert a datetime object to a Discord timestamp.

33
discord_reminder_bot/ui.py
View File

@ -12,7 +12,7 @@ from apscheduler.triggers.interval import IntervalTrigger
from discord.ui import Button, Select
from discord_reminder_bot import settings
from discord_reminder_bot.misc import calc_time, calculate
from discord_reminder_bot.misc import DateTrigger, calc_time, calculate
from discord_reminder_bot.parser import parse_time
if TYPE_CHECKING:
@ -208,9 +208,14 @@ def create_job_embed(job: Job) -> discord.Embed:
author_id: int = job_kwargs.get("author_id", 0)
embed_title: str = textwrap.shorten(f"{message}", width=256, placeholder="...")
# If trigger is IntervalTrigger, show the interval
interval: str = ""
if isinstance(job.trigger, IntervalTrigger):
interval = f"Interval: {job.trigger.interval} seconds"
return discord.Embed(
title=embed_title,
description=f"ID: {job.id}\nNext run: {next_run_time}\nTime left: {calculate(job)}\nChannel: <#{channel_id}>\nAuthor: <@{author_id}>", # noqa: E501
description=f"ID: {job.id}\nNext run: {next_run_time}\nTime left: {calculate(job)}\n{interval}\nChannel: <#{channel_id}>\nAuthor: <@{author_id}>", # noqa: E501
color=discord.Color.blue(),
)
@ -240,9 +245,11 @@ class JobSelector(Select):
label_prefix: str = ""
if job.next_run_time is None:
label_prefix = "Paused: "
# Cron job
elif isinstance(job.trigger, CronTrigger):
label_prefix = "Cron: "
# Interval job
elif isinstance(job.trigger, IntervalTrigger):
label_prefix = "Interval: "
@ -281,6 +288,7 @@ class JobManagementView(discord.ui.View):
self.scheduler: settings.AsyncIOScheduler = scheduler
self.add_item(JobSelector(scheduler))
self.update_buttons()
logger.debug("JobManagementView created for job: %s", job.id)
@discord.ui.button(label="Delete", style=discord.ButtonStyle.danger)
async def delete_button(self, interaction: discord.Interaction, button: Button) -> None: # noqa: ARG002
@ -292,7 +300,7 @@ class JobManagementView(discord.ui.View):
"""
job_kwargs: dict = self.job.kwargs or {}
logger.info("Deleting job: %s", self.job.id)
logger.info("Deleting job: %s because %s clicked the button.", self.job.id, interaction.user.name)
if hasattr(self.job, "__getstate__"):
logger.debug("State: %s", self.job.__getstate__() if hasattr(self.job, "__getstate__") else "No state")
@ -379,6 +387,7 @@ class JobManagementView(discord.ui.View):
if self.job.max_instances:
msg += f"**Max instances**: {self.job.max_instances}\n"
logger.debug("Deletion message: %s", msg)
return msg
@discord.ui.button(label="Modify", style=discord.ButtonStyle.primary)
@ -389,7 +398,7 @@ class JobManagementView(discord.ui.View):
interaction: The interaction object for the command.
button: The button that was clicked.
"""
logger.info("Modifying job: %s", self.job.id)
logger.info("Modifying job: %s. Clicked by %s", self.job.id, interaction.user.name)
if hasattr(self.job, "__getstate__"):
logger.debug("State: %s", self.job.__getstate__() if hasattr(self.job, "__getstate__") else "No state")
@ -422,17 +431,27 @@ class JobManagementView(discord.ui.View):
self.update_buttons()
await interaction.response.edit_message(view=self)
msg: str = f"Job '{self.job.name}' has been {status}."
job_kwargs: dict = self.job.kwargs or {}
job_msg: str = job_kwargs.get("message", "No message found")
job_author: int = job_kwargs.get("author_id", 0)
msg: str = f"Job '{job_msg}' has been {status} by <@{interaction.user.id}>. Job was created by <@{job_author}>."
if hasattr(self.job, "next_run_time"):
msg += f"\nNext run time: {self.job.next_run_time} {calculate(self.job)}"
trigger_time: datetime.datetime | None = (
self.job.trigger.run_date if isinstance(self.job.trigger, DateTrigger) else self.job.next_run_time
)
msg += f"\nNext run time: {trigger_time} {calculate(self.job)}"
await interaction.followup.send(msg)
def update_buttons(self) -> None:
"""Update the visibility of buttons based on job status."""
self.pause_button.disabled = not self.job.next_run_time
logger.debug("Updating buttons for job: %s", self.job.id)
self.pause_button.label = "Resume" if self.job.next_run_time is None else "Pause"
logger.debug("Pause button disabled: %s", self.pause_button.disabled)
logger.debug("Pause button label: %s", self.pause_button.label)
async def interaction_check(self, interaction: discord.Interaction) -> bool: # noqa: ARG002
"""Check the interaction and update buttons before responding.