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

I give up :sadge:

Browse Source
This commit is contained in:
Joakim Hellsén
2025-05-21 05:05:43 +02:00
parent 67d5c5e478
commit 1f4b7490f7
8 changed files with 393 additions and 429 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
.vscode/settings.json vendored
View File

@ -24,6 +24,7 @@
"jobstores",
"Jocke",
"levelname",
"levelno",
"loguru",
"Lovinator",
"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:
"""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:
interaction: The discord interaction object that triggered the command.
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}**.",
)

420
discord_reminder_bot/main.py
View File

@ -1,46 +1,60 @@
from __future__ import annotations
import datetime
import logging
import os
import sys
from typing import TYPE_CHECKING, Any
import discord
import sentry_sdk
from apscheduler import events
from apscheduler.job import Job
from discord.abc import PrivateChannel
from discord_webhook import DiscordWebhook
from dotenv import load_dotenv
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.cron import cron_reminder_job
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.pause import pause_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.unpause import unpause_reminder
from discord_reminder_bot.parsers import calculate, parse_time
from discord_reminder_bot.settings import scheduler
if TYPE_CHECKING:
from apscheduler.events import JobExecutionEvent
from apscheduler.job import Job
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from discord.interactions import InteractionChannel
from requests import Response
logger.remove()
logger.add(
sys.stdout,
format="<green>{time:YYYY-MM-DD at HH:mm:ss}</green> | <cyan>{name}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> - <level>{message}</level>",
level=logging.DEBUG,
)
def my_listener(event: JobExecutionEvent) -> None:
"""Listener for job events.
Args:
event: The event that occurred.
"""
logger.debug(f"Job event: {event=}")
# 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")
msg: str = f"Job {event.job_id} was missed! Was scheduled at {scheduled_time}"
send_webhook(message=msg)
if event.exception:
logger.error(f"discord-reminder-bot failed to send message to Discord\n{event}")
with sentry_sdk.push_scope() as scope:
@ -114,11 +128,6 @@ class RemindBotClient(discord.Client):
"""Log when the bot is ready."""
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, events.EVENT_JOB_MISSED | events.EVENT_JOB_ERROR)
jobs: list[Job] = scheduler.get_jobs()
if not jobs:
logger.info("No jobs available.")
@ -135,9 +144,340 @@ class RemindBotClient(discord.Client):
except (AttributeError, LookupError):
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()
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):
"""Base class for the /remind commands."""
@ -449,7 +789,7 @@ class RemindGroup(discord.app_commands.Group):
# 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 = discord.Intents.all()
intents: discord.Intents = discord.Intents.all()
bot = RemindBotClient(intents=intents)
# Add the group to the bot
@ -501,32 +841,20 @@ async def send_to_discord(channel_id: int, message: str, author_id: int) -> None
channels = list(bot.get_all_channels())
logger.debug(f"We are in {len(channels)} channels.")
channel = None
try:
if not bot.is_closed():
channel: discord.guild.GuildChannel | discord.threads.Thread | discord.abc.PrivateChannel | None = bot.get_channel(channel_id)
else:
logger.info("Bot is closed, skipping...")
except AttributeError as e:
e.add_note(str(bot))
logger.exception(e)
if channel is None:
try:
await bot.wait_until_ready()
channel = await bot.fetch_channel(channel_id)
except discord.NotFound:
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
channel = await bot.fetch_channel(int(channel_id))
except discord.NotFound:
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
if isinstance(channel, discord.ForumChannel | discord.CategoryChannel | PrivateChannel):
@ -576,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}'")
if __name__ == "__main__":
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.")
logger.info("Starting bot.")
bot.run(bot_token)
if __name__ == "__main__":
# 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
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
elif isinstance(job.trigger, CronTrigger):

24
discord_reminder_bot/settings.py
View File

@ -2,29 +2,15 @@ from __future__ import annotations
import logging
import os
import platform
from functools import lru_cache
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
import pytz
import sentry_sdk
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from dotenv import load_dotenv
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",
include_local_variables=True,
traces_sample_rate=1.0,
profiles_sample_rate=1.0,
send_default_pii=True,
)
config_timezone: str = os.getenv("TIMEZONE", default="")
if not config_timezone:
@ -52,7 +38,15 @@ def get_scheduler() -> AsyncIOScheduler:
AsyncIOScheduler: The scheduler instance.
"""
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)
return AsyncIOScheduler(jobstores=jobstores, timezone=timezone, job_defaults=job_defaults)