925 lines
37 KiB
Python
925 lines
37 KiB
Python
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.backup import backup_reminder_job
|
|
from discord_reminder_bot.commands.event import add_discord_event
|
|
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:
|
|
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")
|
|
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"
|
|
f"Job ID: {event.job_id}\n"
|
|
f"Scheduled run time: {event.scheduled_run_time.isoformat() if event.scheduled_run_time else 'None'}\n"
|
|
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):
|
|
"""The bot client for the Discord Reminder Bot."""
|
|
|
|
def __init__(self, *, intents: discord.Intents) -> None:
|
|
"""Initialize the bot client and command tree.
|
|
|
|
Args:
|
|
intents: The intents to use.
|
|
"""
|
|
super().__init__(intents=intents)
|
|
self.tree = discord.app_commands.CommandTree(self)
|
|
|
|
async def on_error(self, event_method: str, *args: list[Any], **kwargs: dict[str, Any]) -> None:
|
|
"""Log errors that occur in the bot."""
|
|
logger.exception(f"An error occurred in {event_method} with args: {args} and kwargs: {kwargs}")
|
|
with sentry_sdk.push_scope() as scope:
|
|
# Add event details
|
|
scope.set_tag("event_method", event_method)
|
|
scope.set_extra("args", args)
|
|
scope.set_extra("kwargs", kwargs)
|
|
# Add bot state
|
|
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_latency", self.latency)
|
|
# If specific arguments are available, extract and add details
|
|
if args:
|
|
interaction = next((arg for arg in args if isinstance(arg, discord.Interaction)), None)
|
|
if interaction:
|
|
scope.set_extra("interaction_id", interaction.id)
|
|
scope.set_extra("interaction_user", interaction.user.id)
|
|
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_channel", str(interaction.channel))
|
|
scope.set_extra("interaction_guild", str(interaction.guild) if interaction.guild else None)
|
|
# Add Sentry tags for interaction details
|
|
scope.set_tag("interaction_id", interaction.id)
|
|
scope.set_tag("interaction_user_id", interaction.user.id)
|
|
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_channel_id", interaction.channel.id if interaction.channel else "None")
|
|
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_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()])
|
|
sentry_sdk.capture_exception()
|
|
|
|
async def on_ready(self) -> None:
|
|
"""Log when the bot is ready."""
|
|
logger.info(f"Logged in as {self.user} ({self.user.id if self.user else 'Unknown'})")
|
|
|
|
jobs: list[Job] = scheduler.get_jobs()
|
|
if not jobs:
|
|
logger.info("No jobs available.")
|
|
return
|
|
|
|
logger.info("Jobs available:")
|
|
try:
|
|
for job in jobs:
|
|
msg: str = job.kwargs.get("message", "") if (job.kwargs and isinstance(job.kwargs, dict)) else ""
|
|
time: str = "Paused"
|
|
if hasattr(job, "next_run_time") and job.next_run_time and isinstance(job.next_run_time, datetime.datetime):
|
|
time = job.next_run_time.strftime("%Y-%m-%d %H:%M:%S")
|
|
logger.info(f"\t{job.id}: {job.name} - {time} - {msg}")
|
|
|
|
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."""
|
|
|
|
def __init__(self) -> None:
|
|
"""Initialize the remind group."""
|
|
super().__init__(name="remind", description="Group for remind commands")
|
|
"""Group for remind commands."""
|
|
|
|
# /remind add
|
|
@discord.app_commands.command(name="add", description="Add a new reminder")
|
|
async def add(
|
|
self,
|
|
interaction: discord.Interaction,
|
|
message: str,
|
|
time: str,
|
|
channel: discord.TextChannel | None = None,
|
|
user: discord.User | None = None,
|
|
dm_and_current_channel: bool | None = None,
|
|
) -> None:
|
|
"""Add a new reminder.
|
|
|
|
Args:
|
|
interaction (discord.Interaction): The interaction object for the command.
|
|
message (str): The content of the reminder.
|
|
time (str): The time of the reminder. (e.g. Friday at 3 PM)
|
|
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): Send reminder as a DM to the user and in this channel. Defaults to False.
|
|
"""
|
|
logger.info(f"New reminder from {interaction.user} ({interaction.user.id}) in {interaction.channel}")
|
|
logger.debug(f"Arguments: {locals()}")
|
|
|
|
if not interaction.guild:
|
|
await interaction.response.send_message(content="Failed to get guild.", ephemeral=True)
|
|
return
|
|
|
|
# Check if we should send the reminder to the specified channel or the current channel
|
|
# TODO(TheLovinator): Check if we have access to the specified channel or the current channel # noqa: TD003
|
|
target_channel: InteractionChannel | None = channel or interaction.channel
|
|
channel_id: int | None = target_channel.id if target_channel else None
|
|
if not channel_id:
|
|
await interaction.response.send_message(content="Failed to get channel.", ephemeral=True)
|
|
return
|
|
|
|
msg: str = add_reminder_job(
|
|
author_id=interaction.user.id,
|
|
message=message,
|
|
time=time,
|
|
channel_id=channel_id,
|
|
guild_id=interaction.guild.id,
|
|
user_id=user.id if user else None,
|
|
dm_and_current_channel=dm_and_current_channel,
|
|
)
|
|
await interaction.response.send_message(content=msg)
|
|
|
|
# /remind event
|
|
@discord.app_commands.command(name="event", description="Add a new Discord event.")
|
|
async def add_event(
|
|
self,
|
|
interaction: discord.Interaction,
|
|
message: str,
|
|
event_start: str,
|
|
event_end: str,
|
|
location: str,
|
|
reason: str | None = None,
|
|
) -> None:
|
|
"""Add a new reminder.
|
|
|
|
Args:
|
|
interaction (discord.Interaction): The interaction object for the command.
|
|
message (str): The description of the scheduled event.
|
|
event_start (str): The scheduled start time of the scheduled event. Will get parsed.
|
|
event_end (str, optional): The scheduled end time of the scheduled event. Will get parsed.
|
|
reason (str, optional): The reason for creating this scheduled event. Shows up on the audit log.
|
|
location (str, optional): The location of the scheduled event.
|
|
"""
|
|
await interaction.response.defer()
|
|
return await add_discord_event(
|
|
interaction=interaction,
|
|
message=message,
|
|
event_start=event_start,
|
|
event_end=event_end,
|
|
location=location,
|
|
reason=reason,
|
|
)
|
|
|
|
# /remind list
|
|
@discord.app_commands.command(name="list", description="List, pause, unpause, and remove reminders.")
|
|
async def list(self, interaction: discord.Interaction) -> None:
|
|
"""List all reminders with pagination and buttons for deleting and modifying jobs.
|
|
|
|
Args:
|
|
interaction(discord.Interaction): The interaction object for the command.
|
|
"""
|
|
await interaction.response.defer()
|
|
return await list_reminder_job(
|
|
interaction=interaction,
|
|
scheduler=scheduler,
|
|
)
|
|
|
|
# /remind cron
|
|
@discord.app_commands.command(name="cron", description="Create new cron job. Works like UNIX cron.")
|
|
async def cron(
|
|
self,
|
|
interaction: discord.Interaction,
|
|
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.
|
|
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.
|
|
"""
|
|
await interaction.response.defer()
|
|
return await cron_reminder_job(
|
|
interaction=interaction,
|
|
scheduler=scheduler,
|
|
message=message,
|
|
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,
|
|
channel=channel,
|
|
user=user,
|
|
dm_and_current_channel=dm_and_current_channel,
|
|
)
|
|
|
|
# /remind interval
|
|
@discord.app_commands.command(
|
|
name="interval",
|
|
description="Create a new reminder that triggers based on an interval.",
|
|
)
|
|
async def interval(
|
|
self,
|
|
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.
|
|
"""
|
|
await interaction.response.defer()
|
|
return await interval_reminder_job(
|
|
interaction=interaction,
|
|
message=message,
|
|
weeks=weeks,
|
|
days=days,
|
|
hours=hours,
|
|
minutes=minutes,
|
|
seconds=seconds,
|
|
start_date=start_date,
|
|
end_date=end_date,
|
|
timezone=timezone,
|
|
jitter=jitter,
|
|
channel=channel,
|
|
user=user,
|
|
dm_and_current_channel=dm_and_current_channel,
|
|
)
|
|
|
|
# /remind backup
|
|
@discord.app_commands.command(name="backup", description="Backup all reminders to a file.")
|
|
async def backup(self, interaction: discord.Interaction, all_servers: bool = False) -> None:
|
|
"""Backup all reminders to a file.
|
|
|
|
Args:
|
|
interaction (discord.Interaction): The interaction object for the command.
|
|
all_servers (bool): Backup all servers or just the current server. Defaults to only the current server.
|
|
"""
|
|
await interaction.response.defer()
|
|
return await backup_reminder_job(
|
|
interaction=interaction,
|
|
scheduler=scheduler,
|
|
all_servers=all_servers,
|
|
)
|
|
|
|
# /remind restore
|
|
@discord.app_commands.command(name="restore", description="Restore reminders from a file.")
|
|
async def restore(self, interaction: discord.Interaction) -> None:
|
|
"""Restore reminders from a file.
|
|
|
|
Args:
|
|
interaction (discord.Interaction): The interaction object for the command.
|
|
"""
|
|
await interaction.response.defer()
|
|
return await restore_reminder_job(
|
|
bot=bot,
|
|
interaction=interaction,
|
|
scheduler=scheduler,
|
|
)
|
|
|
|
# /remind remove
|
|
@discord.app_commands.command(name="remove", description="Remove a reminder")
|
|
async def remove(self, interaction: discord.Interaction, job_id: str) -> None:
|
|
"""Remove a scheduled reminder.
|
|
|
|
Args:
|
|
interaction (discord.Interaction): The interaction object for the command.
|
|
job_id (str): The identifier of the job to remove.
|
|
"""
|
|
await interaction.response.defer()
|
|
return await remove_reminder_job(
|
|
interaction=interaction,
|
|
job_id=job_id,
|
|
scheduler=scheduler,
|
|
)
|
|
|
|
# /remind pause
|
|
@discord.app_commands.command(name="pause", description="Pause a reminder")
|
|
async def pause(self, interaction: discord.Interaction, job_id: str) -> None:
|
|
"""Pause a scheduled reminder.
|
|
|
|
Args:
|
|
interaction (discord.Interaction): The interaction object for the command.
|
|
job_id (str): The identifier of the job to pause.
|
|
"""
|
|
await interaction.response.defer()
|
|
return await pause_reminder_job(
|
|
interaction=interaction,
|
|
job_id=job_id,
|
|
scheduler=scheduler,
|
|
)
|
|
|
|
# /remind unpause
|
|
@discord.app_commands.command(name="unpause", description="Unpause a reminder")
|
|
async def unpause(self, interaction: discord.Interaction, job_id: str) -> None:
|
|
"""Unpause a scheduled reminder.
|
|
|
|
Args:
|
|
interaction (discord.Interaction): The interaction object for the command.
|
|
job_id (str): The identifier of the job to unpause.
|
|
"""
|
|
await interaction.response.defer()
|
|
return await unpause_reminder(
|
|
interaction=interaction,
|
|
job_id=job_id,
|
|
scheduler=scheduler,
|
|
)
|
|
|
|
|
|
# intents: discord.Intents = discord.Intents.none()
|
|
# intents.guild_scheduled_events = True # For creating events
|
|
# 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 = discord.Intents.all()
|
|
bot = RemindBotClient(intents=intents)
|
|
|
|
# Add the group to the bot
|
|
remind_group = RemindGroup()
|
|
bot.tree.add_command(remind_group)
|
|
|
|
|
|
def send_webhook(custom_url: str = "", message: str = "") -> None:
|
|
"""Send a webhook to Discord.
|
|
|
|
Args:
|
|
custom_url: The custom webhook URL to send the message to. If not provided, uses the WEBHOOK_URL environment variable.
|
|
message: The message that will be sent to Discord.
|
|
"""
|
|
webhook_url: str = os.getenv("WEBHOOK_URL", default="")
|
|
url: str = custom_url or webhook_url
|
|
|
|
logger.info(f"Sending webhook to '{url}' with message: '{message}'")
|
|
|
|
if not message:
|
|
logger.error("No message provided.")
|
|
message = "No message provided."
|
|
|
|
if not url:
|
|
logger.error("No webhook URL provided.")
|
|
return
|
|
|
|
webhook: DiscordWebhook = DiscordWebhook(url=url, content=message, rate_limit_retry=True)
|
|
webhook_response: Response = webhook.execute()
|
|
|
|
if not webhook_response.ok:
|
|
webhook_json: dict[str, Any] = webhook.json
|
|
logger.error(f"Failed to send {webhook_json}.\nStatus code: {webhook_response.status_code}.\n{webhook_response.text}.\n")
|
|
return
|
|
|
|
logger.info(f"Webhook sent successfully. Status code: {webhook_response.status_code}.")
|
|
|
|
|
|
async def send_to_discord(channel_id: int, message: str, author_id: int) -> None:
|
|
"""Send a message to Discord.
|
|
|
|
Args:
|
|
channel_id: The Discord channel ID.
|
|
message: The reminder message.
|
|
author_id: User we should ping.
|
|
"""
|
|
logger.info(f"Sending message to '{channel_id=}' with '{message=}' and '{author_id=}'")
|
|
|
|
channels = list(bot.get_all_channels())
|
|
logger.debug(f"We are in {len(channels)} channels.")
|
|
|
|
try:
|
|
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):
|
|
logger.error(f"We haven't implemented sending messages to this channel type {type(channel)}")
|
|
return
|
|
|
|
await channel.send(f"<@{author_id}>\n{message}")
|
|
|
|
|
|
async def send_to_user(user_id: int, guild_id: int, message: str) -> None:
|
|
"""Send a message to a user via DM.
|
|
|
|
Args:
|
|
user_id: The user ID to send the message to.
|
|
guild_id: The guild ID to get the user from.
|
|
message: The message to send.
|
|
"""
|
|
logger.info(f"Sending message to user {user_id} in guild {guild_id}")
|
|
try:
|
|
guild: discord.Guild | None = bot.get_guild(guild_id)
|
|
if guild is None:
|
|
guild = await bot.fetch_guild(guild_id)
|
|
except discord.NotFound:
|
|
logger.exception(f"Guild not found. Current guilds: {bot.guilds}")
|
|
return
|
|
except discord.HTTPException:
|
|
logger.exception(f"Failed to fetch guild {guild_id}")
|
|
return
|
|
|
|
try:
|
|
member: discord.Member | None = guild.get_member(user_id)
|
|
if member is None:
|
|
member = await guild.fetch_member(user_id)
|
|
except discord.Forbidden:
|
|
logger.exception(f"We do not have access to the guild. Guild: {guild_id}, User: {user_id}")
|
|
return
|
|
except discord.NotFound:
|
|
logger.exception(f"Member not found. Guild: {guild_id}, User: {user_id}")
|
|
return
|
|
except discord.HTTPException:
|
|
logger.exception(f"Fetching the member failed. Guild: {guild_id}, User: {user_id}")
|
|
return
|
|
|
|
try:
|
|
await member.send(message)
|
|
except discord.HTTPException:
|
|
logger.exception(f"Failed to send message '{message}' to user '{user_id}' in guild '{guild_id}'")
|
|
|
|
|
|
logger.info("Starting bot.")
|
|
|
|
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.")
|