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

Compare commits

base: TheLovinator:45a9d2655c936911c9c96a32281dfcf7f33ae01e
Branches Tags
TheLovinator:master TheLovinator:kewlrewrite TheLovinator:pls-kill-me
..
compare: TheLovinator:1f4b7490f7516af85fb28fac89e72b64b7c7178c
Branches Tags
TheLovinator:kewlrewrite TheLovinator:pls-kill-me TheLovinator:master

4 Commits
45a9d2655c ... 1f4b7490f7

Author SHA1 Message Date
Joakim Hellsén
1f4b7490f7 I give up :sadge: 2025-05-21 05:05:43 +02:00
Joakim Hellsén
67d5c5e478 Update ruff-pre-commit to version v0.11.10 2025-05-17 16:36:56 +02:00
Joakim Hellsén
49771ddde9 Reintegrate Sentry for error logging 2025-05-17 16:35:09 +02:00
Joakim Hellsén
e2c7a08188 Enhance Sentry SDK configuration with profiling support 2025-04-28 17:13:17 +02:00
9 changed files with 462 additions and 495 deletions
Expand all files Collapse all files

2
.pre-commit-config.yaml
View File

@ -29,7 +29,7 @@ repos:
args: ["--py310-plus"] args: ["--py310-plus"]
- repo: https://github.com/astral-sh/ruff-pre-commit - repo: https://github.com/astral-sh/ruff-pre-commit
rev: v0.11.6 rev: v0.11.10
hooks: hooks:
- id: ruff-format - id: ruff-format
- id: ruff - id: ruff

1
.vscode/settings.json vendored
View File

@ -24,6 +24,7 @@
"jobstores", "jobstores",
"Jocke", "Jocke",
"levelname", "levelname",
"levelno",
"loguru", "loguru",
"Lovinator", "Lovinator",
"pycodestyle", "pycodestyle",

91
discord_reminder_bot/commands/add.py
View File

@ -1,91 +0,0 @@
from __future__ import annotations
from typing import TYPE_CHECKING
from loguru import logger
from discord_reminder_bot.parsers import calculate, parse_time
from discord_reminder_bot.settings import scheduler
if TYPE_CHECKING:
import datetime
from apscheduler.job import Job
def add_reminder_job(
message: str,
time: str,
channel_id: int,
author_id: int,
user_id: int | None = None,
guild_id: int | None = None,
dm_and_current_channel: bool | None = None,
) -> str:
"""Adds a reminder job to the scheduler based on user input.
Schedules a message to be sent at a specified time either to a specific channel,
a specific user via direct message, or both. It handles permission checks,
time parsing, and job creation using the APScheduler instance.
Args:
message: The content of the reminder message to be sent.
time: A string representing the date and time for the reminder.
This string will be parsed to a datetime object.
channel_id: The ID of the channel where the reminder will be sent.
user_id: The Discord ID of the user to send a DM to. If None, no DM is sent.
guild_id: The ID of the guild (server) where the reminder is set.
author_id: The ID of the user who created the reminder.
dm_and_current_channel: If True and a user is specified, sends the
reminder to both the user's DM and the target channel. If False
and a user is specified, only sends the DM. Defaults to None,
behaving like False if only a user is specified, or sending only
to the channel if no user is specified.
Returns:
The response message indicating the status of the reminder job creation.
"""
dm_message: str = ""
if user_id:
parsed_time: datetime.datetime | None = parse_time(date_to_parse=time)
if not parsed_time:
return f"Failed to parse time: {time}."
user_reminder: Job = scheduler.add_job(
func="discord_reminder_bot.main:send_to_user",
trigger="date",
run_date=parsed_time,
kwargs={
"user_id": user_id,
"guild_id": guild_id,
"message": message,
},
)
logger.info(f"User reminder job created: {user_reminder} for {user_id} at {parsed_time}")
dm_message = f" and a DM to <@{user_id}>"
if not dm_and_current_channel:
return (
f"Hello <@{author_id}>,\n"
f"I will send a DM to <@{user_id}> at:\n"
f"First run in {calculate(user_reminder)} with the message:\n**{message}**."
)
# Create channel reminder job
channel_job: Job = scheduler.add_job(
func="discord_reminder_bot.main:send_to_discord",
trigger="date",
run_date=parse_time(date_to_parse=time),
kwargs={
"channel_id": channel_id,
"message": message,
"author_id": author_id,
},
)
logger.info(f"Channel reminder job created: {channel_job} for {channel_id}")
return (
f"Hello <@{author_id}>,\n"
f"I will notify you in <#{channel_id}>{dm_message}.\n"
f"First run in {calculate(channel_job)} with the message:\n**{message}**."
)

8
discord_reminder_bot/commands/backup.py
View File

@ -16,14 +16,6 @@ if TYPE_CHECKING:
async def backup_reminder_job(interaction: discord.Interaction, scheduler: AsyncIOScheduler, all_servers: bool) -> None: async def backup_reminder_job(interaction: discord.Interaction, scheduler: AsyncIOScheduler, all_servers: bool) -> None:
"""Backs up reminder jobs from the scheduler to a JSON file. """Backs up reminder jobs from the scheduler to a JSON file.
Exports jobs from the provided AsyncIOScheduler to a temporary JSON file.
If `all_servers` is False, it filters the jobs to include only those
associated with the guild where the command was invoked. This requires
the invoking user to have administrator permissions in that guild.
The resulting list of jobs (either all or filtered) is then written
to another temporary JSON file and sent as an attachment via the
interaction followup, along with a confirmation message.
Args: Args:
interaction: The discord interaction object that triggered the command. interaction: The discord interaction object that triggered the command.
scheduler: The AsyncIOScheduler instance containing the jobs to back up. scheduler: The AsyncIOScheduler instance containing the jobs to back up.

139
discord_reminder_bot/commands/cron.py
View File

@ -1,139 +0,0 @@
from __future__ import annotations
import logging
from typing import TYPE_CHECKING
import discord
from discord_reminder_bot.parsers import calculate
if TYPE_CHECKING:
import discord
from apscheduler.job import Job
from apscheduler.schedulers.asyncio import AsyncIOScheduler
logger: logging.Logger = logging.getLogger("discord_reminder_bot")
async def cron_reminder_job(
interaction: discord.Interaction,
scheduler: AsyncIOScheduler,
message: str,
year: str | None = None,
month: str | None = None,
day: str | None = None,
week: str | None = None,
day_of_week: str | None = None,
hour: str | None = None,
minute: str | None = None,
second: str | None = None,
start_date: str | None = None,
end_date: str | None = None,
timezone: str | None = None,
jitter: int | None = None,
channel: discord.TextChannel | None = None,
user: discord.User | None = None,
dm_and_current_channel: bool | None = None,
) -> None:
"""Create a new cron job.
Args that are None will be defaulted to *.
Args:
interaction (discord.Interaction): The interaction object for the command.
scheduler (AsyncIOScheduler): The scheduler to add the job to.
message (str): The content of the reminder.
year (str): 4-digit year. Defaults to *.
month (str): Month (1-12). Defaults to *.
day (str): Day of the month (1-31). Defaults to *.
week (str): ISO Week of the year (1-53). Defaults to *.
day_of_week (str): Number or name of weekday (0-6 or mon,tue,wed,thu,fri,sat,sun).
hour (str): Hour (0-23). Defaults to *.
minute (str): Minute (0-59). Defaults to *.
second (str): Second (0-59). Defaults to *.
start_date (str): Earliest possible date/time to trigger on (inclusive). Will get parsed.
end_date (str): Latest possible date/time to trigger on (inclusive). Will get parsed.
timezone (str): Time zone to use for the date/time calculations Defaults to scheduler timezone.
jitter (int, optional): Delay the job execution by jitter seconds at most.
channel (discord.TextChannel, optional): The channel to send the reminder to. Defaults to current channel.
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.
"""
# Log kwargs
logger.info("New cron job from %s (%s) in %s", interaction.user, interaction.user.id, interaction.channel)
logger.info("Cron job arguments: %s", locals())
# Get the channel ID
channel_id: int | None = channel.id if channel else (interaction.channel.id if interaction.channel else None)
if not channel_id:
await interaction.followup.send(content="Failed to get channel.", ephemeral=True)
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.", ephemeral=True)
return
# Create user DM reminder job if user is specified
dm_message: str = ""
if user:
user_reminder: Job = scheduler.add_job(
func="discord_reminder_bot.main:send_to_user",
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={
"user_id": user.id,
"guild_id": guild.id,
"message": message,
},
)
dm_message = f" and a DM to {user.display_name}"
if not dm_and_current_channel:
await interaction.followup.send(
content=f"Hello {interaction.user.display_name},\n"
f"I will send a DM to {user.display_name} at:\n"
f"First run in {calculate(user_reminder)} with the message:\n**{message}**.",
)
return
# Create channel reminder job
channel_job: Job = scheduler.add_job(
func="discord_reminder_bot.main:send_to_discord",
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={
"channel_id": channel_id,
"message": message,
"author_id": interaction.user.id,
},
)
await interaction.followup.send(
content=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}**.",
)

135
discord_reminder_bot/commands/interval.py
View File

@ -1,135 +0,0 @@
from __future__ import annotations
import logging
from typing import TYPE_CHECKING
from discord_reminder_bot.parsers import calculate
from discord_reminder_bot.settings import scheduler
if TYPE_CHECKING:
import discord
from apscheduler.job import Job
from discord.interactions import InteractionChannel
logger: logging.Logger = logging.getLogger("discord_reminder_bot")
async def interval_reminder_job(
interaction: discord.Interaction,
message: str,
weeks: int = 0,
days: int = 0,
hours: int = 0,
minutes: int = 0,
seconds: int = 0,
start_date: str | None = None,
end_date: str | None = None,
timezone: str | None = None,
jitter: int | None = None,
channel: discord.TextChannel | None = None,
user: discord.User | None = None,
dm_and_current_channel: bool | None = None,
) -> None:
"""Create a new reminder that triggers based on an interval.
Args:
interaction (discord.Interaction): The interaction object for the command.
message (str): The content of the reminder.
weeks (int, optional): Number of weeks between each run. Defaults to 0.
days (int, optional): Number of days between each run. Defaults to 0.
hours (int, optional): Number of hours between each run. Defaults to 0.
minutes (int, optional): Number of minutes between each run. Defaults to 0.
seconds (int, optional): Number of seconds between each run. Defaults to 0.
start_date (str, optional): Earliest possible date/time to trigger on (inclusive). Will get parsed.
end_date (str, optional): Latest possible date/time to trigger on (inclusive). Will get parsed.
timezone (str, optional): Time zone to use for the date/time calculations Defaults to scheduler timezone.
jitter (int, optional): Delay the job execution by jitter seconds at most.
channel (discord.TextChannel, optional): The channel to send the reminder to. Defaults to current channel.
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.
"""
logger.info("New interval job from %s (%s) in %s", interaction.user, interaction.user.id, interaction.channel)
logger.info("Arguments: %s", locals())
# Only allow intervals of 30 seconds or more so we don't spam the channel
if weeks == days == hours == minutes == 0 and seconds < 30:
await interaction.followup.send(content="Interval must be at least 30 seconds.", ephemeral=True)
return
# Check if we have access to the specified channel or the current channel
target_channel: InteractionChannel | None = channel or interaction.channel
if target_channel and interaction.guild and not target_channel.permissions_for(interaction.guild.me).send_messages:
await interaction.followup.send(
content=f"I don't have permission to send messages in <#{target_channel.id}>.",
ephemeral=True,
)
# Get the channel ID
channel_id: int | None = channel.id if channel else (interaction.channel.id if interaction.channel else None)
if not channel_id:
await interaction.followup.send(content="Failed to get channel.", ephemeral=True)
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.", ephemeral=True)
return
# Create user DM reminder job if user is specified
dm_message: str = ""
if user:
dm_job: Job = scheduler.add_job(
func="discord_reminder_bot.main:send_to_user",
trigger="interval",
weeks=weeks,
days=days,
hours=hours,
minutes=minutes,
seconds=seconds,
start_date=start_date,
end_date=end_date,
timezone=timezone,
jitter=jitter,
kwargs={
"user_id": user.id,
"guild_id": guild.id,
"message": message,
},
)
dm_message = f" and a DM to {user.display_name} "
if not dm_and_current_channel:
await interaction.followup.send(
content=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}**.",
)
# Create channel reminder job
# TODO(TheLovinator): Test that "discord_reminder_bot.main:send_to_discord" is always there # noqa: TD003
channel_job: Job = scheduler.add_job(
func="discord_reminder_bot.main:send_to_discord",
trigger="interval",
weeks=weeks,
days=days,
hours=hours,
minutes=minutes,
seconds=seconds,
start_date=start_date,
end_date=end_date,
timezone=timezone,
jitter=jitter,
kwargs={
"channel_id": channel_id,
"message": message,
"author_id": interaction.user.id,
},
)
await interaction.followup.send(
content=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}**.",
)

555
discord_reminder_bot/main.py
View File

@ -1,66 +1,79 @@
from __future__ import annotations from __future__ import annotations
import datetime import datetime
import logging
import os import os
import sys
from typing import TYPE_CHECKING, Any from typing import TYPE_CHECKING, Any
import discord import discord
import sentry_sdk
from apscheduler import events
from apscheduler.job import Job
from discord.abc import PrivateChannel from discord.abc import PrivateChannel
from discord_webhook import DiscordWebhook from discord_webhook import DiscordWebhook
from dotenv import load_dotenv
from loguru import logger from loguru import logger
from discord_reminder_bot.commands.add import add_reminder_job
from discord_reminder_bot.commands.backup import backup_reminder_job from discord_reminder_bot.commands.backup import backup_reminder_job
from discord_reminder_bot.commands.cron import cron_reminder_job
from discord_reminder_bot.commands.event import add_discord_event from discord_reminder_bot.commands.event import add_discord_event
from discord_reminder_bot.commands.interval import interval_reminder_job
from discord_reminder_bot.commands.list import list_reminder_job from discord_reminder_bot.commands.list import list_reminder_job
from discord_reminder_bot.commands.pause import pause_reminder_job from discord_reminder_bot.commands.pause import pause_reminder_job
from discord_reminder_bot.commands.remove import remove_reminder_job from discord_reminder_bot.commands.remove import remove_reminder_job
from discord_reminder_bot.commands.restore import restore_reminder_job from discord_reminder_bot.commands.restore import restore_reminder_job
from discord_reminder_bot.commands.unpause import unpause_reminder from discord_reminder_bot.commands.unpause import unpause_reminder
from discord_reminder_bot.parsers import calculate, parse_time
from discord_reminder_bot.settings import scheduler from discord_reminder_bot.settings import scheduler
if TYPE_CHECKING: if TYPE_CHECKING:
from apscheduler.events import JobExecutionEvent
from apscheduler.job import Job from apscheduler.job import Job
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from discord.interactions import InteractionChannel from discord.interactions import InteractionChannel
from requests import Response from requests import Response
# def my_listener(event: JobExecutionEvent) -> None: logger.remove()
# """Listener for job events. logger.add(
# sys.stdout,
# Args: format="<green>{time:YYYY-MM-DD at HH:mm:ss}</green> | <cyan>{name}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> - <level>{message}</level>",
# event: The event that occurred. level=logging.DEBUG,
# """ )
# # TODO(TheLovinator): We should save the job state to a file and send it to Discord. # noqa: TD003
# if event.code == events.EVENT_JOB_MISSED:
# scheduled_time: str = event.scheduled_run_time.strftime("%Y-%m-%d %H:%M:%S") def my_listener(event: JobExecutionEvent) -> None:
# msg: str = f"Job {event.job_id} was missed! Was scheduled at {scheduled_time}" """Listener for job events.
# send_webhook(message=msg)
# Args:
# if event.exception: event: The event that occurred.
# logger.error(f"discord-reminder-bot failed to send message to Discord\n{event}") """
# with sentry_sdk.push_scope() as scope: logger.debug(f"Job event: {event=}")
# scope.set_extra("job_id", event.job_id)
# scope.set_extra("scheduled_run_time", event.scheduled_run_time.isoformat() if event.scheduled_run_time else "None") # TODO(TheLovinator): We should save the job state to a file and send it to Discord. # noqa: TD003
# scope.set_extra("event_code", event.code) if event.code == events.EVENT_JOB_MISSED:
# sentry_sdk.capture_exception(event.exception) scheduled_time: str = event.scheduled_run_time.strftime("%Y-%m-%d %H:%M:%S")
# msg: str = f"Job {event.job_id} was missed! Was scheduled at {scheduled_time}"
# err_msg: str = ( send_webhook(message=msg)
# f"discord-reminder-bot failed to send message to Discord\n"
# f"Job ID: {event.job_id}\n" if event.exception:
# f"Scheduled run time: {event.scheduled_run_time.isoformat() if event.scheduled_run_time else 'None'}\n" logger.error(f"discord-reminder-bot failed to send message to Discord\n{event}")
# f"Event code: {event.code}\n" with sentry_sdk.push_scope() as scope:
# f"Exception: {event.exception}\n" scope.set_extra("job_id", event.job_id)
# "```python\n" scope.set_extra("scheduled_run_time", event.scheduled_run_time.isoformat() if event.scheduled_run_time else "None")
# f"{event.traceback}\n" scope.set_extra("event_code", event.code)
# "```" sentry_sdk.capture_exception(event.exception)
# ) err_msg: str = (
# f"discord-reminder-bot failed to send message to Discord\n"
# send_webhook(message=err_msg) f"Job ID: {event.job_id}\n"
# f"Scheduled run time: {event.scheduled_run_time.isoformat() if event.scheduled_run_time else 'None'}\n"
# raise event.exception f"Event code: {event.code}\n"
f"Exception: {event.exception}\n"
"```python\n"
f"{event.traceback}\n"
"```"
)
send_webhook(message=err_msg)
raise event.exception
class RemindBotClient(discord.Client): class RemindBotClient(discord.Client):
@ -75,49 +88,46 @@ class RemindBotClient(discord.Client):
super().__init__(intents=intents) super().__init__(intents=intents)
self.tree = discord.app_commands.CommandTree(self) self.tree = discord.app_commands.CommandTree(self)
# async def on_error(self, event_method: str, *args: list[Any], **kwargs: dict[str, Any]) -> None: async def on_error(self, event_method: str, *args: list[Any], **kwargs: dict[str, Any]) -> None:
# """Log errors that occur in the bot.""" """Log errors that occur in the bot."""
# logger.exception(f"An error occurred in {event_method} with args: {args} and kwargs: {kwargs}") logger.exception(f"An error occurred in {event_method} with args: {args} and kwargs: {kwargs}")
# with sentry_sdk.push_scope() as scope: with sentry_sdk.push_scope() as scope:
# # Add event details # Add event details
# scope.set_tag("event_method", event_method) scope.set_tag("event_method", event_method)
# scope.set_extra("args", args) scope.set_extra("args", args)
# scope.set_extra("kwargs", kwargs) scope.set_extra("kwargs", kwargs)
# # Add bot state # Add bot state
# scope.set_tag("bot_user_id", self.user.id if self.user else "Unknown") scope.set_tag("bot_user_id", self.user.id if self.user else "Unknown")
# scope.set_tag("bot_user_name", str(self.user) if self.user else "Unknown") scope.set_tag("bot_user_name", str(self.user) if self.user else "Unknown")
# scope.set_tag("bot_latency", self.latency) scope.set_tag("bot_latency", self.latency)
# # If specific arguments are available, extract and add details # If specific arguments are available, extract and add details
# if args: if args:
# interaction = next((arg for arg in args if isinstance(arg, discord.Interaction)), None) interaction = next((arg for arg in args if isinstance(arg, discord.Interaction)), None)
# if interaction: if interaction:
# scope.set_extra("interaction_id", interaction.id) scope.set_extra("interaction_id", interaction.id)
# scope.set_extra("interaction_user", interaction.user.id) scope.set_extra("interaction_user", interaction.user.id)
# scope.set_extra("interaction_user_tag", str(interaction.user)) scope.set_extra("interaction_user_tag", str(interaction.user))
# scope.set_extra("interaction_command", interaction.command.name if interaction.command else None) scope.set_extra("interaction_command", interaction.command.name if interaction.command else None)
# scope.set_extra("interaction_channel", str(interaction.channel)) scope.set_extra("interaction_channel", str(interaction.channel))
# scope.set_extra("interaction_guild", str(interaction.guild) if interaction.guild else None) scope.set_extra("interaction_guild", str(interaction.guild) if interaction.guild else None)
# # Add Sentry tags for interaction details # Add Sentry tags for interaction details
# scope.set_tag("interaction_id", interaction.id) scope.set_tag("interaction_id", interaction.id)
# scope.set_tag("interaction_user_id", interaction.user.id) scope.set_tag("interaction_user_id", interaction.user.id)
# scope.set_tag("interaction_user_tag", str(interaction.user)) scope.set_tag("interaction_user_tag", str(interaction.user))
# scope.set_tag("interaction_command", interaction.command.name if interaction.command else "None") scope.set_tag("interaction_command", interaction.command.name if interaction.command else "None")
# scope.set_tag("interaction_channel_id", interaction.channel.id if interaction.channel else "None") scope.set_tag("interaction_channel_id", interaction.channel.id if interaction.channel else "None")
# scope.set_tag("interaction_channel_name", str(interaction.channel)) scope.set_tag("interaction_channel_name", str(interaction.channel))
# scope.set_tag("interaction_guild_id", interaction.guild.id if interaction.guild else "None") scope.set_tag("interaction_guild_id", interaction.guild.id if interaction.guild else "None")
# scope.set_tag("interaction_guild_name", str(interaction.guild) if interaction.guild else "None") scope.set_tag("interaction_guild_name", str(interaction.guild) if interaction.guild else "None")
# # Add APScheduler context
# scope.set_extra("scheduler_jobs", [job.id for job in scheduler.get_jobs()]) # Add APScheduler context
# sentry_sdk.capture_exception() scope.set_extra("scheduler_jobs", [job.id for job in scheduler.get_jobs()])
sentry_sdk.capture_exception()
async def on_ready(self) -> None: async def on_ready(self) -> None:
"""Log when the bot is ready.""" """Log when the bot is ready."""
logger.info(f"Logged in as {self.user} ({self.user.id if self.user else 'Unknown'})") logger.info(f"Logged in as {self.user} ({self.user.id if self.user else 'Unknown'})")
async def setup_hook(self) -> None:
"""Setup the bot."""
scheduler.start()
# scheduler.add_listener(my_listener, EVENT_JOB_MISSED | EVENT_JOB_ERROR)
jobs: list[Job] = scheduler.get_jobs() jobs: list[Job] = scheduler.get_jobs()
if not jobs: if not jobs:
logger.info("No jobs available.") logger.info("No jobs available.")
@ -134,9 +144,340 @@ class RemindBotClient(discord.Client):
except (AttributeError, LookupError): except (AttributeError, LookupError):
logger.exception("Failed to loop through jobs") logger.exception("Failed to loop through jobs")
scheduler.start()
scheduler.add_listener(my_listener)
async def setup_hook(self) -> None:
"""Setup the bot."""
await self.tree.sync() await self.tree.sync()
def on_shutdown(self) -> None:
"""Log when the bot is shutting down."""
logger.info("Shutting down...")
scheduler.shutdown()
def add_reminder_job(
message: str,
time: str,
channel_id: int,
author_id: int,
user_id: int | None = None,
guild_id: int | None = None,
dm_and_current_channel: bool | None = None,
) -> str:
"""Adds a reminder job to the scheduler based on user input.
Schedules a message to be sent at a specified time either to a specific channel,
a specific user via direct message, or both. It handles permission checks,
time parsing, and job creation using the APScheduler instance.
Args:
message: The content of the reminder message to be sent.
time: A string representing the date and time for the reminder.
This string will be parsed to a datetime object.
channel_id: The ID of the channel where the reminder will be sent.
user_id: The Discord ID of the user to send a DM to. If None, no DM is sent.
guild_id: The ID of the guild (server) where the reminder is set.
author_id: The ID of the user who created the reminder.
dm_and_current_channel: If True and a user is specified, sends the
reminder to both the user's DM and the target channel. If False
and a user is specified, only sends the DM. Defaults to None,
behaving like False if only a user is specified, or sending only
to the channel if no user is specified.
Returns:
The response message indicating the status of the reminder job creation.
"""
dm_message: str = ""
if user_id:
parsed_time: datetime.datetime | None = parse_time(date_to_parse=time)
if not parsed_time:
return f"Failed to parse time: {time}."
user_reminder: Job = scheduler.add_job(
func="discord_reminder_bot.main:send_to_user",
trigger="date",
run_date=parsed_time,
kwargs={
"user_id": user_id,
"guild_id": guild_id,
"message": message,
},
)
logger.info(f"User reminder job created: {user_reminder} for {user_id} at {parsed_time}")
dm_message = f" and a DM to <@{user_id}>"
if not dm_and_current_channel:
return (
f"Hello <@{author_id}>,\n"
f"I will send a DM to <@{user_id}> at:\n"
f"First run in {calculate(user_reminder)} with the message:\n**{message}**."
)
# Create channel reminder job
channel_job: Job = scheduler.add_job(
func="discord_reminder_bot.main:send_to_discord",
trigger="date",
run_date=parse_time(date_to_parse=time),
kwargs={
"channel_id": channel_id,
"message": message,
"author_id": author_id,
},
)
logger.info(f"Channel reminder job created: {channel_job} for {channel_id}")
return (
f"Hello <@{author_id}>,\n"
f"I will notify you in <#{channel_id}>{dm_message}.\n"
f"First run in {calculate(channel_job)} with the message:\n**{message}**."
)
async def cron_reminder_job(
interaction: discord.Interaction,
scheduler: AsyncIOScheduler,
message: str,
year: str | None = None,
month: str | None = None,
day: str | None = None,
week: str | None = None,
day_of_week: str | None = None,
hour: str | None = None,
minute: str | None = None,
second: str | None = None,
start_date: str | None = None,
end_date: str | None = None,
timezone: str | None = None,
jitter: int | None = None,
channel: discord.TextChannel | None = None,
user: discord.User | None = None,
dm_and_current_channel: bool | None = None,
) -> None:
"""Create a new cron job.
Args that are None will be defaulted to *.
Args:
interaction (discord.Interaction): The interaction object for the command.
scheduler (AsyncIOScheduler): The scheduler to add the job to.
message (str): The content of the reminder.
year (str): 4-digit year. Defaults to *.
month (str): Month (1-12). Defaults to *.
day (str): Day of the month (1-31). Defaults to *.
week (str): ISO Week of the year (1-53). Defaults to *.
day_of_week (str): Number or name of weekday (0-6 or mon,tue,wed,thu,fri,sat,sun).
hour (str): Hour (0-23). Defaults to *.
minute (str): Minute (0-59). Defaults to *.
second (str): Second (0-59). Defaults to *.
start_date (str): Earliest possible date/time to trigger on (inclusive). Will get parsed.
end_date (str): Latest possible date/time to trigger on (inclusive). Will get parsed.
timezone (str): Time zone to use for the date/time calculations Defaults to scheduler timezone.
jitter (int, optional): Delay the job execution by jitter seconds at most.
channel (discord.TextChannel, optional): The channel to send the reminder to. Defaults to current channel.
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.
"""
# Log kwargs
logger.info("New cron job from %s (%s) in %s", interaction.user, interaction.user.id, interaction.channel)
logger.info("Cron job arguments: %s", locals())
# Get the channel ID
channel_id: int | None = channel.id if channel else (interaction.channel.id if interaction.channel else None)
if not channel_id:
await interaction.followup.send(content="Failed to get channel.", ephemeral=True)
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.", ephemeral=True)
return
# Create user DM reminder job if user is specified
dm_message: str = ""
if user:
user_reminder: Job = scheduler.add_job(
func="discord_reminder_bot.main:send_to_user",
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={
"user_id": user.id,
"guild_id": guild.id,
"message": message,
},
)
dm_message = f" and a DM to {user.display_name}"
if not dm_and_current_channel:
await interaction.followup.send(
content=f"Hello {interaction.user.display_name},\n"
f"I will send a DM to {user.display_name} at:\n"
f"First run in {calculate(user_reminder)} with the message:\n**{message}**.",
)
return
# Create channel reminder job
channel_job: Job = scheduler.add_job(
func="discord_reminder_bot.main:send_to_discord",
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={
"channel_id": channel_id,
"message": message,
"author_id": interaction.user.id,
},
)
await interaction.followup.send(
content=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}**.",
)
async def interval_reminder_job(
interaction: discord.Interaction,
message: str,
weeks: int = 0,
days: int = 0,
hours: int = 0,
minutes: int = 0,
seconds: int = 0,
start_date: str | None = None,
end_date: str | None = None,
timezone: str | None = None,
jitter: int | None = None,
channel: discord.TextChannel | None = None,
user: discord.User | None = None,
dm_and_current_channel: bool | None = None,
) -> None:
"""Create a new reminder that triggers based on an interval.
Args:
interaction (discord.Interaction): The interaction object for the command.
message (str): The content of the reminder.
weeks (int, optional): Number of weeks between each run. Defaults to 0.
days (int, optional): Number of days between each run. Defaults to 0.
hours (int, optional): Number of hours between each run. Defaults to 0.
minutes (int, optional): Number of minutes between each run. Defaults to 0.
seconds (int, optional): Number of seconds between each run. Defaults to 0.
start_date (str, optional): Earliest possible date/time to trigger on (inclusive). Will get parsed.
end_date (str, optional): Latest possible date/time to trigger on (inclusive). Will get parsed.
timezone (str, optional): Time zone to use for the date/time calculations Defaults to scheduler timezone.
jitter (int, optional): Delay the job execution by jitter seconds at most.
channel (discord.TextChannel, optional): The channel to send the reminder to. Defaults to current channel.
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.
"""
logger.info("New interval job from %s (%s) in %s", interaction.user, interaction.user.id, interaction.channel)
logger.info("Arguments: %s", locals())
# Only allow intervals of 30 seconds or more so we don't spam the channel
if weeks == days == hours == minutes == 0 and seconds < 30:
await interaction.followup.send(content="Interval must be at least 30 seconds.", ephemeral=True)
return
# Check if we have access to the specified channel or the current channel
target_channel: InteractionChannel | None = channel or interaction.channel
if target_channel and interaction.guild and not target_channel.permissions_for(interaction.guild.me).send_messages:
await interaction.followup.send(
content=f"I don't have permission to send messages in <#{target_channel.id}>.",
ephemeral=True,
)
# Get the channel ID
channel_id: int | None = channel.id if channel else (interaction.channel.id if interaction.channel else None)
if not channel_id:
await interaction.followup.send(content="Failed to get channel.", ephemeral=True)
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.", ephemeral=True)
return
# Create user DM reminder job if user is specified
dm_message: str = ""
if user:
dm_job: Job = scheduler.add_job(
func="discord_reminder_bot.main:send_to_user",
trigger="interval",
weeks=weeks,
days=days,
hours=hours,
minutes=minutes,
seconds=seconds,
start_date=start_date,
end_date=end_date,
timezone=timezone,
jitter=jitter,
kwargs={
"user_id": user.id,
"guild_id": guild.id,
"message": message,
},
)
dm_message = f" and a DM to {user.display_name} "
if not dm_and_current_channel:
await interaction.followup.send(
content=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}**.",
)
# Create channel reminder job
# TODO(TheLovinator): Test that "discord_reminder_bot.main:send_to_discord" is always there # noqa: TD003
channel_job: Job = scheduler.add_job(
func="discord_reminder_bot.main:send_to_discord",
trigger="interval",
weeks=weeks,
days=days,
hours=hours,
minutes=minutes,
seconds=seconds,
start_date=start_date,
end_date=end_date,
timezone=timezone,
jitter=jitter,
kwargs={
"channel_id": channel_id,
"message": message,
"author_id": interaction.user.id,
},
)
await interaction.followup.send(
content=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}**.",
)
class RemindGroup(discord.app_commands.Group): class RemindGroup(discord.app_commands.Group):
"""Base class for the /remind commands.""" """Base class for the /remind commands."""
@ -448,7 +789,7 @@ class RemindGroup(discord.app_commands.Group):
# intents.guilds = True # For getting the channel to send the reminder to # intents.guilds = True # For getting the channel to send the reminder to
# intents.members = True # For getting the user to send the reminder to # intents.members = True # For getting the user to send the reminder to
intents = discord.Intents.all() intents: discord.Intents = discord.Intents.all()
bot = RemindBotClient(intents=intents) bot = RemindBotClient(intents=intents)
# Add the group to the bot # Add the group to the bot
@ -500,32 +841,20 @@ async def send_to_discord(channel_id: int, message: str, author_id: int) -> None
channels = list(bot.get_all_channels()) channels = list(bot.get_all_channels())
logger.debug(f"We are in {len(channels)} channels.") logger.debug(f"We are in {len(channels)} channels.")
channel = None
try: try:
if not bot.is_closed(): channel = await bot.fetch_channel(int(channel_id))
channel: discord.guild.GuildChannel | discord.threads.Thread | discord.abc.PrivateChannel | None = bot.get_channel(channel_id) except discord.NotFound:
else: logger.exception(f"Channel not found. Current channels: {bot.get_all_channels()}")
logger.info("Bot is closed, skipping...") return
except AttributeError as e: except discord.Forbidden:
e.add_note(str(bot)) logger.exception(f"We do not have access to the channel. Channel: {channel_id}")
logger.exception(e) return
except discord.HTTPException:
if channel is None: logger.exception(f"Fetching the channel failed. Channel: {channel_id}")
try: return
await bot.wait_until_ready() except discord.InvalidData:
channel = await bot.fetch_channel(channel_id) logger.exception(f"Invalid data. Channel: {channel_id}")
except discord.NotFound: return
logger.exception(f"Channel not found. Current channels: {bot.get_all_channels()}")
return
except discord.Forbidden:
logger.exception(f"We do not have access to the channel. Channel: {channel_id}")
return
except discord.HTTPException:
logger.exception(f"Fetching the channel failed. Channel: {channel_id}")
return
except discord.InvalidData:
logger.exception(f"Invalid data. Channel: {channel_id}")
return
# Channels we can't send messages to # Channels we can't send messages to
if isinstance(channel, discord.ForumChannel | discord.CategoryChannel | PrivateChannel): if isinstance(channel, discord.ForumChannel | discord.CategoryChannel | PrivateChannel):
@ -575,11 +904,21 @@ async def send_to_user(user_id: int, guild_id: int, message: str) -> None:
logger.exception(f"Failed to send message '{message}' to user '{user_id}' in guild '{guild_id}'") logger.exception(f"Failed to send message '{message}' to user '{user_id}' in guild '{guild_id}'")
if __name__ == "__main__": logger.info("Starting bot.")
bot_token: str = os.getenv("BOT_TOKEN", default="")
if not bot_token:
msg = "Missing bot token. Please set the BOT_TOKEN environment variable. Read the README for more information."
raise ValueError(msg)
logger.info("Starting bot.") if __name__ == "__main__":
bot.run(bot_token) # Load environment variables
load_dotenv()
default_sentry_dsn: str = "https://c4c61a52838be9b5042144420fba5aaa@o4505228040339456.ingest.us.sentry.io/4508707268984832"
sentry_sdk.init(
dsn=os.getenv("SENTRY_DSN", default_sentry_dsn),
include_local_variables=True,
traces_sample_rate=1.0,
profiles_sample_rate=1.0,
send_default_pii=True,
)
# Start the bot
bot.run(os.getenv("BOT_TOKEN", default=""))
logger.info("Bot stopped.")

4
discord_reminder_bot/parsers.py
View File

@ -67,6 +67,10 @@ def calculate(job: Job) -> str:
""" """
trigger_time = None trigger_time = None
if isinstance(job.trigger, DateTrigger | IntervalTrigger): if isinstance(job.trigger, DateTrigger | IntervalTrigger):
if not hasattr(job, "next_run_time"):
logger.debug("No next run time found for '%s'", job.id)
logger.debug("%s", job.__getstate__())
return "Paused"
trigger_time = job.next_run_time or None trigger_time = job.next_run_time or None
elif isinstance(job.trigger, CronTrigger): elif isinstance(job.trigger, CronTrigger):

22
discord_reminder_bot/settings.py
View File

@ -2,27 +2,15 @@ from __future__ import annotations
import logging import logging
import os import os
import platform
from functools import lru_cache from functools import lru_cache
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
import pytz import pytz
import sentry_sdk
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
from apscheduler.schedulers.asyncio import AsyncIOScheduler from apscheduler.schedulers.asyncio import AsyncIOScheduler
from dotenv import load_dotenv
logger: logging.Logger = logging.getLogger("discord_reminder_bot") logger: logging.Logger = logging.getLogger("discord_reminder_bot")
load_dotenv()
default_sentry_dsn: str = "https://c4c61a52838be9b5042144420fba5aaa@o4505228040339456.ingest.us.sentry.io/4508707268984832"
sentry_sdk.init(
dsn=os.getenv("SENTRY_DSN", default_sentry_dsn),
environment=platform.node() or "Unknown",
traces_sample_rate=1.0,
send_default_pii=True,
)
config_timezone: str = os.getenv("TIMEZONE", default="") config_timezone: str = os.getenv("TIMEZONE", default="")
if not config_timezone: if not config_timezone:
@ -50,7 +38,15 @@ def get_scheduler() -> AsyncIOScheduler:
AsyncIOScheduler: The scheduler instance. AsyncIOScheduler: The scheduler instance.
""" """
jobstores: dict[str, SQLAlchemyJobStore] = {"default": SQLAlchemyJobStore(url=f"sqlite://{sqlite_location}")} jobstores: dict[str, SQLAlchemyJobStore] = {"default": SQLAlchemyJobStore(url=f"sqlite://{sqlite_location}")}
job_defaults: dict[str, bool] = {"coalesce": True} job_defaults: dict[str, bool | int] = {
# If coalesce is True, when the scheduler recovers from downtime,
# missed executions will be rolled into a single execution rather than
# executing each missed run separately. This prevents a backlog of
# accumulated executions when the scheduler is down for a period of time.
"coalesce": True,
# The time (in seconds) how much this job's execution is allowed to be late
"misfire_grace_time": 60,
}
timezone = pytz.timezone(config_timezone) timezone = pytz.timezone(config_timezone)
return AsyncIOScheduler(jobstores=jobstores, timezone=timezone, job_defaults=job_defaults) return AsyncIOScheduler(jobstores=jobstores, timezone=timezone, job_defaults=job_defaults)