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="{time:YYYY-MM-DD at HH:mm:ss} | {name}:{function}:{line} - {message}", 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.")