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

Use Ruff and fix all its warnings and errors

Browse Source
This commit is contained in:
Joakim Hellsén
2023-03-19 00:54:21 +01:00
parent ec2978b529
commit 11d9f2c2a5
10 changed files with 687 additions and 363 deletions
Download Patch File Download Diff File Expand all files Collapse all files

20
discord_reminder_bot/countdown.py
View File

@ -1,7 +1,4 @@
"""
calculate(job) - Calculates how many days, hours and minutes till trigger.
"""
import datetime
from datetime import datetime, timedelta
import pytz
from apscheduler.job import Job
@ -11,8 +8,7 @@ from discord_reminder_bot.settings import config_timezone
def calculate(job: Job) -> str:
"""Get trigger time from a reminder and calculate how many days,
hours and minutes till trigger.
"""Get trigger time from a reminder and calculate how many days, hours and minutes till trigger.
Days/Minutes will not be included if 0.
@ -25,10 +21,7 @@ def calculate(job: Job) -> str:
# TODO: This "breaks" when only seconds are left.
# If we use (in {calc_countdown(job)}) it will show (in )
if type(job.trigger) is DateTrigger:
trigger_time = job.trigger.run_date
else:
trigger_time = job.next_run_time
trigger_time: datetime | None = job.trigger.run_date if type(job.trigger) is DateTrigger else job.next_run_time
# Get_job() returns None when it can't find a job with that ID.
if trigger_time is None:
@ -41,8 +34,7 @@ def calculate(job: Job) -> str:
def countdown(trigger_time: datetime) -> str:
"""
Calculate days, hours and minutes to a date.
"""Calculate days, hours and minutes to a date.
Args:
trigger_time: The date.
@ -50,7 +42,7 @@ def countdown(trigger_time: datetime) -> str:
Returns:
A string with the days, hours and minutes.
"""
countdown_time = trigger_time - datetime.datetime.now(tz=pytz.timezone(config_timezone))
countdown_time: timedelta = trigger_time - datetime.now(tz=pytz.timezone(config_timezone))
days, hours, minutes = (
countdown_time.days,
@ -60,7 +52,7 @@ def countdown(trigger_time: datetime) -> str:
# Return seconds if only seconds are left.
if days == 0 and hours == 0 and minutes == 0:
seconds = countdown_time.seconds % 60
seconds: int = countdown_time.seconds % 60
return f"{seconds} second" + ("s" if seconds != 1 else "")
# TODO: Explain this.

123
discord_reminder_bot/create_pages.py
View File

@ -1,17 +1,23 @@
"""This module creates the pages for the paginator."""
from typing import List
from typing import TYPE_CHECKING, Literal
import interactions
from apscheduler.job import Job
from apscheduler.triggers.date import DateTrigger
from interactions import ActionRow, CommandContext, ComponentContext
from interactions import ActionRow, Button, CommandContext, ComponentContext, Embed, Message, Modal, TextInput
from interactions.ext.paginator import Page, Paginator, RowPosition
from discord_reminder_bot.countdown import calculate
from discord_reminder_bot.settings import scheduler
if TYPE_CHECKING:
from datetime import datetime
def create_pages(ctx: CommandContext) -> list[Page]:
from apscheduler.job import Job
max_message_length: Literal[1010] = 1010
max_title_length: Literal[90] = 90
async def create_pages(ctx: CommandContext) -> list[Page]:
"""Create pages for the paginator.
Args:
@ -20,65 +26,69 @@ def create_pages(ctx: CommandContext) -> list[Page]:
Returns:
list[Page]: A list of pages.
"""
pages = []
pages: list[Page] = []
jobs: List[Job] = scheduler.get_jobs()
jobs: list[Job] = scheduler.get_jobs()
for job in jobs:
channel_id = job.kwargs.get("channel_id")
guild_id = job.kwargs.get("guild_id")
channel_id: int = job.kwargs.get("channel_id")
guild_id: int = job.kwargs.get("guild_id")
if ctx.guild is None:
await ctx.send("I can't find the server you're in. Are you sure you're in a server?", ephemeral=True)
return pages
if ctx.guild.channels is None:
await ctx.send("I can't find the channel you're in.", ephemeral=True)
return pages
# Only add reminders from channels in the server we run "/reminder list" in
# Check if channel is in the Discord server, if not, skip it.
for channel in ctx.guild.channels:
if int(channel.id) == channel_id or ctx.guild_id == guild_id:
if type(job.trigger) is DateTrigger:
# Get trigger time for normal reminders
trigger_time = job.trigger.run_date
else:
# Get trigger time for cron and interval jobs
trigger_time = job.next_run_time
trigger_time: datetime | None = (
job.trigger.run_date if type(job.trigger) is DateTrigger else job.next_run_time
)
# Paused reminders returns None
if trigger_time is None:
trigger_value = None
trigger_text = "Paused"
trigger_value: str | None = None
trigger_text: str = "Paused"
else:
trigger_value = f'{trigger_time.strftime("%Y-%m-%d %H:%M")} (in {calculate(job)})'
trigger_text = trigger_value
message = job.kwargs.get("message")
message = f"{message[:1000]}..." if len(message) > 1010 else message
message: str = job.kwargs.get("message")
message = f"{message[:1000]}..." if len(message) > max_message_length else message
edit_button = interactions.Button(
edit_button: Button = interactions.Button(
label="Edit",
style=interactions.ButtonStyle.PRIMARY,
custom_id="edit",
)
pause_button = interactions.Button(
pause_button: Button = interactions.Button(
label="Pause",
style=interactions.ButtonStyle.PRIMARY,
custom_id="pause",
)
unpause_button = interactions.Button(
unpause_button: Button = interactions.Button(
label="Unpause",
style=interactions.ButtonStyle.PRIMARY,
custom_id="unpause",
)
remove_button = interactions.Button(
remove_button: Button = interactions.Button(
label="Remove",
style=interactions.ButtonStyle.DANGER,
custom_id="remove",
)
embed = interactions.Embed(
embed: Embed = interactions.Embed(
title=f"{job.id}",
fields=[
interactions.EmbedField(
name=f"**Channel:**",
name="**Channel:**",
value=f"#{channel.name}",
),
interactions.EmbedField(
name=f"**Message:**",
name="**Message:**",
value=f"{message}",
),
],
@ -86,16 +96,16 @@ def create_pages(ctx: CommandContext) -> list[Page]:
if trigger_value is not None:
embed.add_field(
name=f"**Trigger:**",
name="**Trigger:**",
value=f"{trigger_text}",
)
else:
embed.add_field(
name=f"**Trigger:**",
value=f"_Paused_",
name="**Trigger:**",
value="_Paused_",
)
components = [
components: list[Button] = [
edit_button,
remove_button,
]
@ -103,39 +113,42 @@ def create_pages(ctx: CommandContext) -> list[Page]:
if type(job.trigger) is not DateTrigger:
# Get trigger time for cron and interval jobs
trigger_time = job.next_run_time
if trigger_time is None:
pause_or_unpause_button = unpause_button
else:
pause_or_unpause_button = pause_button
pause_or_unpause_button: Button = unpause_button if trigger_time is None else pause_button
components.insert(1, pause_or_unpause_button)
# Add a page to pages list
title = f"{message[:87]}..." if len(message) > 90 else message
title: str = f"{message[:87]}..." if len(message) > max_title_length else message
pages.append(
Page(
embeds=embed,
title=title,
components=ActionRow(components=components),
components=ActionRow(components=components), # type: ignore # noqa: PGH003
callback=callback,
position=RowPosition.BOTTOM,
)
),
)
return pages
async def callback(self: Paginator, ctx: ComponentContext):
async def callback(self: Paginator, ctx: ComponentContext) -> Message | None: # noqa: PLR0911
"""Callback for the paginator."""
job_id = self.component_ctx.message.embeds[0].title
job = scheduler.get_job(job_id)
if self.component_ctx is None:
return await ctx.send("Something went wrong.", ephemeral=True)
if self.component_ctx.message is None:
return await ctx.send("Something went wrong.", ephemeral=True)
job_id: str | None = self.component_ctx.message.embeds[0].title
job: Job | None = scheduler.get_job(job_id)
if job is None:
return await ctx.send("Job not found.", ephemeral=True)
channel_id = job.kwargs.get("channel_id")
old_message = job.kwargs.get("message")
channel_id: int = job.kwargs.get("channel_id")
old_message: str = job.kwargs.get("message")
components = [
components: list[TextInput] = [
interactions.TextInput(
style=interactions.TextStyleType.PARAGRAPH,
label="New message",
@ -147,8 +160,8 @@ async def callback(self: Paginator, ctx: ComponentContext):
if type(job.trigger) is DateTrigger:
# Get trigger time for normal reminders
trigger_time = job.trigger.run_date
job_type = "normal"
trigger_time: datetime | None = job.trigger.run_date
job_type: str = "normal"
components.append(
interactions.TextInput(
style=interactions.TextStyleType.SHORT,
@ -165,29 +178,31 @@ async def callback(self: Paginator, ctx: ComponentContext):
job_type = "cron/interval"
if ctx.custom_id == "edit":
modal = interactions.Modal(
modal: Modal = interactions.Modal(
title=f"Edit {job_type} reminder.",
custom_id="edit_modal",
components=components,
components=components, # type: ignore # noqa: PGH003
)
await ctx.popup(modal)
return None
elif ctx.custom_id == "pause":
# TODO: Add unpause button if user paused the wrong job
if ctx.custom_id == "pause":
scheduler.pause_job(job_id)
await ctx.send(f"Job {job_id} paused.")
return None
elif ctx.custom_id == "unpause":
# TODO: Add pause button if user unpauses the wrong job
if ctx.custom_id == "unpause":
scheduler.resume_job(job_id)
await ctx.send(f"Job {job_id} unpaused.")
return None
elif ctx.custom_id == "remove":
# TODO: Add recreate button if user removed the wrong job
if ctx.custom_id == "remove":
scheduler.remove_job(job_id)
await ctx.send(
f"Job {job_id} removed.\n"
f"**Message:** {old_message}\n"
f"**Channel:** {channel_id}\n"
f"**Time:** {trigger_time}"
f"**Time:** {trigger_time}",
)
return None
return None

638
discord_reminder_bot/main.py
View File

@ -1,71 +1,83 @@
import logging
from typing import List
from typing import TYPE_CHECKING
import interactions
from apscheduler import events
from apscheduler.events import EVENT_JOB_ERROR, EVENT_JOB_MISSED
from apscheduler.events import EVENT_JOB_ERROR, EVENT_JOB_MISSED, JobExecutionEvent
from apscheduler.jobstores.base import JobLookupError
from apscheduler.triggers.date import DateTrigger
from discord_webhook import DiscordWebhook
from interactions import CommandContext, Embed, OptionType, autodefer
from interactions.ext.paginator import Paginator
from interactions import Channel, Client, CommandContext, Embed, Member, Message, OptionType, autodefer
from interactions.ext.paginator import Page, Paginator
from discord_reminder_bot.countdown import calculate
from discord_reminder_bot.create_pages import create_pages
from discord_reminder_bot.parse import parse_time
from discord_reminder_bot.parse import ParsedTime, parse_time
from discord_reminder_bot.settings import (
bot_token,
config_timezone,
log_level,
scheduler,
sqlite_location,
webhook_url
webhook_url,
)
bot = interactions.Client(token=bot_token)
if TYPE_CHECKING:
from datetime import datetime
from apscheduler.job import Job
bot: Client = interactions.Client(token=bot_token)
def send_webhook(url=webhook_url, message: str = "discord-reminder-bot: Empty message."):
"""
Send a webhook to Discord.
def send_webhook(
url: str = webhook_url,
message: str = "discord-reminder-bot: Empty message.",
) -> None:
"""Send a webhook to Discord.
Args:
url: Our webhook url, defaults to the one from settings.
message: The message that will be sent to Discord.
"""
if not url:
print("ERROR: Tried to send a webhook but you have no webhook url configured.")
logging.error("ERROR: Tried to send a webhook but you have no webhook url configured.")
return
webhook = DiscordWebhook(url=url, content=message, rate_limit_retry=True)
webhook: DiscordWebhook = DiscordWebhook(url=url, content=message, rate_limit_retry=True)
webhook.execute()
@bot.command(name="remind")
async def base_command(ctx: interactions.CommandContext):
async def base_command(ctx: interactions.CommandContext) -> None: # noqa: ARG001
"""This is the base command for the reminder bot."""
pass
@bot.modal("edit_modal")
async def modal_response_edit(ctx: CommandContext, *response: str):
"""This is what gets triggerd when the user clicks the Edit button in /reminder list.
async def modal_response_edit(ctx: CommandContext, *response: str) -> Message: # noqa: C901, PLR0912, PLR0911
"""This is what gets triggered when the user clicks the Edit button in /reminder list.
Args:
ctx: Context of the slash command. Contains the guild, author and message and more.
response: The response from the modal.
Returns:
A Discord message with changes.
"""
job_id = ctx.message.embeds[0].title
old_date = None
old_message = None
if not ctx.message:
return await ctx.send(
"The message that triggered this modal is missing. Or something else went wrong.",
ephemeral=True,
)
job_id: str | None = ctx.message.embeds[0].title
old_date: str | None = None
old_message: str | None = None
try:
job = scheduler.get_job(job_id)
job: Job | None = scheduler.get_job(job_id)
except JobLookupError as e:
return await ctx.send(
f"Failed to get the job after the modal.\n"
f"Job ID: {job_id}\n"
f"Error: {e}",
f"Failed to get the job after the modal.\nJob ID: {job_id}\nError: {e}",
ephemeral=True,
)
@ -76,50 +88,54 @@ async def modal_response_edit(ctx: CommandContext, *response: str):
return await ctx.send("No changes made.", ephemeral=True)
if type(job.trigger) is DateTrigger:
new_message = response[0]
new_date = response[1]
new_message: str | None = response[0]
new_date: str | None = response[1]
else:
new_message = response[0]
new_date = None
message_embeds: List[Embed] = ctx.message.embeds
message_embeds: list[Embed] = ctx.message.embeds
for embeds in message_embeds:
if embeds.fields is None:
return await ctx.send("No fields found in the embed.", ephemeral=True)
for field in embeds.fields:
if field.name == "**Channel:**":
continue
elif field.name == "**Message:**":
if field.name == "**Message:**":
old_message = field.value
elif field.name == "**Trigger:**":
if field.name == "**Trigger:**":
old_date = field.value
else:
return await ctx.send(
f"Unknown field name ({field.name}).", ephemeral=True
f"Unknown field name ({field.name}).",
ephemeral=True,
)
msg = f"Modified job {job_id}.\n"
if old_date is not None:
if new_date:
# Parse the time/date we got from the command.
parsed = parse_time(date_to_parse=new_date)
if parsed.err:
return await ctx.send(parsed.err_msg)
parsed_date = parsed.parsed_time
date_new = parsed_date.strftime("%Y-%m-%d %H:%M:%S")
msg: str = f"Modified job {job_id}.\n"
if old_date is not None and new_date:
# Parse the time/date we got from the command.
parsed: ParsedTime = parse_time(date_to_parse=new_date)
if parsed.err:
return await ctx.send(parsed.err_msg)
parsed_date: datetime | None = parsed.parsed_time
new_job = scheduler.reschedule_job(job.id, run_date=date_new)
new_time = calculate(new_job)
if parsed_date is None:
return await ctx.send(f"Failed to parse the date. ({new_date})")
# TODO: old_date and date_new has different precision.
# Old date: 2032-09-18 00:07
# New date: 2032-09-18 00:07:13
msg += (
f"**Old date**: {old_date}\n"
f"**New date**: {date_new} (in {new_time})\n"
)
date_new: str = parsed_date.strftime("%Y-%m-%d %H:%M:%S")
new_job: Job = scheduler.reschedule_job(job.id, run_date=date_new)
new_time: str = calculate(new_job)
# TODO: old_date and date_new has different precision.
# Old date: 2032-09-18 00:07
# New date: 2032-09-18 00:07:13
msg += f"**Old date**: {old_date}\n**New date**: {date_new} (in {new_time})\n"
if old_message is not None:
channel_id = job.kwargs.get("channel_id")
job_author_id = job.kwargs.get("author_id")
channel_id: int = job.kwargs.get("channel_id")
job_author_id: int = job.kwargs.get("author_id")
try:
scheduler.modify_job(
job.id,
@ -131,9 +147,7 @@ async def modal_response_edit(ctx: CommandContext, *response: str):
)
except JobLookupError as e:
return await ctx.send(
f"Failed to modify the job.\n"
f"Job ID: {job_id}\n"
f"Error: {e}",
f"Failed to modify the job.\nJob ID: {job_id}\nError: {e}",
ephemeral=True,
)
msg += f"**Old message**: {old_message}\n**New message**: {new_message}\n"
@ -142,44 +156,65 @@ async def modal_response_edit(ctx: CommandContext, *response: str):
@autodefer()
@bot.command(name="parse", description="Parse the time from a string")
@interactions.option(name="time_to_parse", description="The string you want to parse.", type=OptionType.STRING, required=True)
@interactions.option(name="optional_timezone", description="Optional time zone, for example Europe/Stockholm", type=OptionType.STRING, required=False)
async def parse_command(ctx: interactions.CommandContext, time_to_parse: str, optional_timezone: str | None = None):
"""
Find the date and time from a string.
@bot.command(name="parse", description="Parse the time from a string") # type: ignore # noqa: PGH003
@interactions.option(
name="time_to_parse",
description="The string you want to parse.",
type=OptionType.STRING,
required=True,
)
@interactions.option(
name="optional_timezone",
description="Optional time zone, for example Europe/Stockholm",
type=OptionType.STRING,
required=False,
)
async def parse_command(
ctx: interactions.CommandContext,
time_to_parse: str,
optional_timezone: str | None = None,
) -> Message:
"""Find the date and time from a string.
Args:
ctx: Context of the slash command. Contains the guild, author and message and more.
time_to_parse: The string you want to parse.
optional_timezone: Optional time zone, for example Europe/Stockholm.
"""
if optional_timezone:
parsed = parse_time(date_to_parse=time_to_parse, timezone=optional_timezone)
parsed: ParsedTime = parse_time(date_to_parse=time_to_parse, timezone=optional_timezone)
else:
parsed = parse_time(date_to_parse=time_to_parse)
if parsed.err:
return await ctx.send(parsed.err_msg)
parsed_date = parsed.parsed_time
parsed_date: datetime | None = parsed.parsed_time
# Locale’s appropriate date and time representation.
locale_time = parsed_date.strftime("%c")
run_date = parsed_date.strftime("%Y-%m-%d %H:%M:%S")
return await ctx.send(f"**String**: {time_to_parse}\n"
f"**Parsed date**: {parsed_date}\n"
f"**Formatted**: {run_date}\n"
f"**Locale time**: {locale_time}\n")
if parsed_date is None:
return await ctx.send(f"Failed to parse the date. ({time_to_parse})")
# Locale`s appropriate date and time representation.
locale_time: str = parsed_date.strftime("%c")
run_date: str = parsed_date.strftime("%Y-%m-%d %H:%M:%S")
return await ctx.send(
f"**String**: {time_to_parse}\n"
f"**Parsed date**: {parsed_date}\n"
f"**Formatted**: {run_date}\n"
f"**Locale time**: {locale_time}\n",
)
@autodefer()
@base_command.subcommand(name="list", description="List, pause, unpause, and remove reminders.")
async def list_command(ctx: interactions.CommandContext):
@base_command.subcommand(
name="list",
description="List, pause, unpause, and remove reminders.",
)
async def list_command(ctx: interactions.CommandContext) -> Message | None:
"""List, pause, unpause, and remove reminders.
Args:
ctx: Context of the slash command. Contains the guild, author and message and more.
"""
pages = create_pages(ctx)
pages: list[Page] = await create_pages(ctx)
if not pages:
return await ctx.send("No reminders found.", ephemeral=True)
@ -187,7 +222,7 @@ async def list_command(ctx: interactions.CommandContext):
for page in pages:
return await ctx.send(
content="I haven't added support for buttons if there is only one reminder, "
"so you need to add another one to edit/delete this one 🙃",
"so you need to add another one to edit/delete this one 🙃",
embeds=page.embeds,
)
@ -202,23 +237,49 @@ async def list_command(ctx: interactions.CommandContext):
)
await paginator.run()
return None
@autodefer()
@base_command.subcommand(name="add", description="Set a reminder.")
@interactions.option(name="message_reason", description="The message I'm going to send you.", type=OptionType.STRING, required=True)
@interactions.option(name="message_date", description="The date to send the message.", type=OptionType.STRING, required=True)
@interactions.option(name="different_channel", description="The channel to send the message to.", type=OptionType.CHANNEL, required=False)
@interactions.option(name="send_dm_to_user", description="Send message to a user via DM instead of a channel. Set both_dm_and_channel to send both.", type=OptionType.USER, required=False) # noqa
@interactions.option(name="both_dm_and_channel", description="Send both DM and message to the channel, needs send_dm_to_user to be set if you want both.", type=OptionType.BOOLEAN, required=False) # noqa
async def command_add(
ctx: interactions.CommandContext,
message_reason: str,
message_date: str,
different_channel: interactions.Channel | None = None,
send_dm_to_user: interactions.User | None = None,
both_dm_and_channel: bool | None = None,
):
@interactions.option(
name="message_reason",
description="The message I'm going to send you.",
type=OptionType.STRING,
required=True,
)
@interactions.option(
name="message_date",
description="The date to send the message.",
type=OptionType.STRING,
required=True,
)
@interactions.option(
name="different_channel",
description="The channel to send the message to.",
type=OptionType.CHANNEL,
required=False,
)
@interactions.option(
name="send_dm_to_user",
description="Send message to a user via DM instead of a channel. Set both_dm_and_channel to send both.",
type=OptionType.USER,
required=False,
)
@interactions.option(
name="both_dm_and_channel",
description="Send both DM and message to the channel, needs send_dm_to_user to be set if you want both.",
type=OptionType.BOOLEAN,
required=False,
)
async def command_add( # noqa: PLR0913
ctx: interactions.CommandContext,
message_reason: str,
message_date: str,
different_channel: interactions.Channel | None = None,
send_dm_to_user: interactions.User | None = None,
both_dm_and_channel: bool | None = None,
) -> Message | None:
"""Add a new reminder. You can add a date and message.
Args:
@ -230,23 +291,27 @@ async def command_add(
both_dm_and_channel: If we should send both a DM and a message to the channel. Works with different_channel.
"""
# Parse the time/date we got from the command.
parsed = parse_time(date_to_parse=message_date)
parsed: ParsedTime = parse_time(date_to_parse=message_date)
if parsed.err:
return await ctx.send(parsed.err_msg)
parsed_date = parsed.parsed_time
run_date = parsed_date.strftime("%Y-%m-%d %H:%M:%S")
parsed_date: datetime | None = parsed.parsed_time
if parsed_date is None:
return await ctx.send(f"Failed to parse the date. ({message_date})")
run_date: str = parsed_date.strftime("%Y-%m-%d %H:%M:%S")
# If we should send the message to a different channel
channel_id = int(ctx.channel_id)
if different_channel:
channel_id = int(different_channel.id)
dm_message = ""
dm_message: str = ""
where_and_when = "You should never see this message. Please report this to the bot owner if you do. :-)"
should_send_channel_reminder = True
try:
if send_dm_to_user:
dm_reminder = scheduler.add_job(
dm_reminder: Job = scheduler.add_job(
send_to_user,
run_date=run_date,
kwargs={
@ -259,11 +324,15 @@ async def command_add(
if not both_dm_and_channel:
# If we should send the message to the channel too instead of just a DM.
should_send_channel_reminder = False
where_and_when = (f"I will send a DM to {send_dm_to_user.username} at:\n"
f"**{run_date}** (in {calculate(dm_reminder)})\n")
where_and_when: str = (
f"I will send a DM to {send_dm_to_user.username} at:\n"
f"**{run_date}** (in {calculate(dm_reminder)})\n"
)
if ctx.member is None:
return await ctx.send("Something went wrong when grabbing the member, are you in a guild?", ephemeral=True)
if should_send_channel_reminder:
reminder = scheduler.add_job(
reminder: Job = scheduler.add_job(
send_to_discord,
run_date=run_date,
kwargs={
@ -272,23 +341,20 @@ async def command_add(
"author_id": ctx.member.id,
},
)
where_and_when = (f"I will notify you in <#{channel_id}> {dm_message}at:\n"
f"**{run_date}** (in {calculate(reminder)})\n")
where_and_when = (
f"I will notify you in <#{channel_id}> {dm_message}at:\n**{run_date}** (in {calculate(reminder)})\n"
)
except ValueError as e:
await ctx.send(str(e), ephemeral=True)
return
return None
message = (
f"Hello {ctx.member.name}, "
f"{where_and_when}"
f"With the message:\n"
f"**{message_reason}**."
)
message: str = f"Hello {ctx.member.name}, {where_and_when}With the message:\n**{message_reason}**."
await ctx.send(message)
return None
async def send_to_user(user_id: int, guild_id: int, message: str):
async def send_to_user(user_id: int, guild_id: int, message: str) -> None:
"""Send a message to a user via DM.
Args:
@ -296,7 +362,13 @@ async def send_to_user(user_id: int, guild_id: int, message: str):
guild_id: The guild ID to get the user from.
message: The message to send.
"""
member = await interactions.get(bot, interactions.Member, parent_id=guild_id, object_id=user_id, force="http")
member: Member = await interactions.get(
bot,
interactions.Member,
parent_id=guild_id,
object_id=user_id,
force="http",
)
await member.send(message)
@ -305,41 +377,121 @@ async def send_to_user(user_id: int, guild_id: int, message: str):
name="cron",
description="Triggers when current time matches all specified time constraints, similarly to the UNIX cron.",
)
@interactions.option(name="message_reason", description="The message I'm going to send you.", type=OptionType.STRING, required=True)
@interactions.option(name="year", description="4-digit year. (Example: 2042)", type=OptionType.STRING, required=False)
@interactions.option(name="month", description="Month. (1-12)", type=OptionType.STRING, required=False)
@interactions.option(name="day", description="Day of month (1-31)", type=OptionType.STRING, required=False)
@interactions.option(name="week", description="ISO week (1-53)", type=OptionType.STRING, required=False)
@interactions.option(name="day_of_week", description="Number or name of weekday (0-6 or mon,tue,wed,thu,fri,sat,sun).", type=OptionType.STRING, required=False) # noqa
@interactions.option(name="hour", description="Hour (0-23)", type=OptionType.STRING, required=False)
@interactions.option(name="minute", description="Minute (0-59)", type=OptionType.STRING, required=False)
@interactions.option(name="second", description="Second (0-59)", type=OptionType.STRING, required=False)
@interactions.option(name="start_date", description="Earliest possible time to trigger on, in the ISO 8601 format. (Example: 2010-10-10 09:30:00)", type=OptionType.STRING, required=False) # noqa
@interactions.option(name="end_date", description="Latest possible time to trigger on, in the ISO 8601 format. (Example: 2010-10-10 09:30:00)", type=OptionType.STRING, required=False) # noqa
@interactions.option(name="timezone", description="Time zone to use for the date/time calculations (defaults to scheduler timezone)", type=OptionType.STRING, required=False) # noqa
@interactions.option(name="jitter", description="Delay the job execution by x seconds at most. Adds a random component to the execution time.", type=OptionType.INTEGER, required=False) # noqa
@interactions.option(name="different_channel", description="Send the messages to a different channel.", type=OptionType.CHANNEL, required=False) # noqa
@interactions.option(name="send_dm_to_user", description="Send message to a user via DM instead of a channel. Set both_dm_and_channel to send both.", type=OptionType.USER, required=False) # noqa
@interactions.option(name="both_dm_and_channel", description="Send both DM and message to the channel, needs send_dm_to_user to be set if you want both.", type=OptionType.BOOLEAN, required=False) # noqa
async def remind_cron(
ctx: interactions.CommandContext,
message_reason: str,
year: int | None = None,
month: int | None = None,
day: int | None = None,
week: int | None = None,
day_of_week: str | None = None,
hour: int | None = None,
minute: int | None = None,
second: int | None = None,
start_date: str | None = None,
end_date: str | None = None,
timezone: str | None = None,
jitter: int | None = None,
different_channel: interactions.Channel | None = None,
send_dm_to_user: interactions.User | None = None,
both_dm_and_channel: bool | None = None,
):
@interactions.option(
name="message_reason",
description="The message I'm going to send you.",
type=OptionType.STRING,
required=True,
)
@interactions.option(
name="year",
description="4-digit year. (Example: 2042)",
type=OptionType.STRING,
required=False,
)
@interactions.option(
name="month",
description="Month. (1-12)",
type=OptionType.STRING,
required=False,
)
@interactions.option(
name="day",
description="Day of month (1-31)",
type=OptionType.STRING,
required=False,
)
@interactions.option(
name="week",
description="ISO week (1-53)",
type=OptionType.STRING,
required=False,
)
@interactions.option(
name="day_of_week",
description="Number or name of weekday (0-6 or mon,tue,wed,thu,fri,sat,sun).",
type=OptionType.STRING,
required=False,
)
@interactions.option(
name="hour",
description="Hour (0-23)",
type=OptionType.STRING,
required=False,
)
@interactions.option(
name="minute",
description="Minute (0-59)",
type=OptionType.STRING,
required=False,
)
@interactions.option(
name="second",
description="Second (0-59)",
type=OptionType.STRING,
required=False,
)
@interactions.option(
name="start_date",
description="Earliest possible time to trigger on, in the ISO 8601 format. (Example: 2010-10-10 09:30:00)",
type=OptionType.STRING,
required=False,
)
@interactions.option(
name="end_date",
description="Latest possible time to trigger on, in the ISO 8601 format. (Example: 2010-10-10 09:30:00)",
type=OptionType.STRING,
required=False,
)
@interactions.option(
name="timezone",
description="Time zone to use for the date/time calculations (defaults to scheduler timezone)",
type=OptionType.STRING,
required=False,
)
@interactions.option(
name="jitter",
description="Delay the job execution by x seconds at most. Adds a random component to the execution time.",
type=OptionType.INTEGER,
required=False,
)
@interactions.option(
name="different_channel",
description="Send the messages to a different channel.",
type=OptionType.CHANNEL,
required=False,
)
@interactions.option(
name="send_dm_to_user",
description="Send message to a user via DM instead of a channel. Set both_dm_and_channel to send both.",
type=OptionType.USER,
required=False,
)
@interactions.option(
name="both_dm_and_channel",
description="Send both DM and message to the channel, needs send_dm_to_user to be set if you want both.",
type=OptionType.BOOLEAN,
required=False,
)
async def remind_cron( # noqa: PLR0913
ctx: interactions.CommandContext,
message_reason: str,
year: int | None = None,
month: int | None = None,
day: int | None = None,
week: int | None = None,
day_of_week: str | None = None,
hour: int | None = None,
minute: int | None = None,
second: int | None = None,
start_date: str | None = None,
end_date: str | None = None,
timezone: str | None = None,
jitter: int | None = None,
different_channel: interactions.Channel | None = None,
send_dm_to_user: interactions.User | None = None,
both_dm_and_channel: bool | None = None,
) -> None:
"""Create new cron job. Works like UNIX cron.
https://en.wikipedia.org/wiki/Cron
@ -369,12 +521,12 @@ async def remind_cron(
if different_channel:
channel_id = int(different_channel.id)
dm_message = ""
dm_message: str = ""
where_and_when = "You should never see this message. Please report this to the bot owner if you do. :-)"
should_send_channel_reminder = True
try:
if send_dm_to_user:
dm_reminder = scheduler.add_job(
dm_reminder: Job = scheduler.add_job(
send_to_user,
"cron",
year=year,
@ -399,11 +551,15 @@ async def remind_cron(
if not both_dm_and_channel:
# If we should send the message to the channel too instead of just a DM.
should_send_channel_reminder = False
where_and_when = (f"I will send a DM to {send_dm_to_user.username} at:\n"
f"First run in {calculate(dm_reminder)} with the message:\n")
where_and_when: str = (
f"I will send a DM to {send_dm_to_user.username} at:\n"
f"First run in {calculate(dm_reminder)} with the message:\n"
)
if ctx.member is None:
await ctx.send("Failed to get member from context. Are you sure you're in a server?", ephemeral=True)
return
if should_send_channel_reminder:
job = scheduler.add_job(
job: Job = scheduler.add_job(
send_to_discord,
"cron",
year=year,
@ -424,52 +580,118 @@ async def remind_cron(
"author_id": ctx.member.id,
},
)
where_and_when = (f" I will send messages to <#{channel_id}>{dm_message}.\n"
f"First run in {calculate(job)} with the message:\n")
where_and_when = (
f" I will send messages to <#{channel_id}>{dm_message}.\n"
f"First run in {calculate(job)} with the message:\n"
)
except ValueError as e:
await ctx.send(str(e), ephemeral=True)
return
# TODO: Add what arguments we used in the job to the message
message = (
f"Hello {ctx.member.name}, "
f"{where_and_when}"
f"**{message_reason}**."
)
message: str = f"Hello {ctx.member.name}, {where_and_when} **{message_reason}**."
await ctx.send(message)
@base_command.subcommand(name="interval", description="Schedules messages to be run periodically, on selected intervals.")
@interactions.option(name="message_reason", description="The message I'm going to send you.", type=OptionType.STRING, required=True)
@interactions.option(name="weeks", description="Number of weeks to wait", type=OptionType.INTEGER, required=False)
@interactions.option(name="days", description="Number of days to wait", type=OptionType.INTEGER, required=False)
@interactions.option(name="hours", description="Number of hours to wait", type=OptionType.INTEGER, required=False)
@interactions.option(name="minutes", description="Number of minutes to wait", type=OptionType.INTEGER, required=False)
@interactions.option(name="seconds", description="Number of seconds to wait", type=OptionType.INTEGER, required=False)
@interactions.option(name="start_date", description="When to start, in the ISO 8601 format. (Example: 2010-10-10 09:30:00)", type=OptionType.STRING, required=False) # noqa
@interactions.option(name="end_date", description="When to stop, in the ISO 8601 format. (Example: 2014-06-15 11:00:00)", type=OptionType.STRING, required=False) # noqa
@interactions.option(name="timezone", description="Time zone to use for the date/time calculations", type=OptionType.STRING, required=False)
@interactions.option(name="jitter", description="Delay the job execution by x seconds at most. Adds a random component to the execution time.", type=OptionType.INTEGER, required=False) # noqa
@interactions.option(name="different_channel", description="Send the messages to a different channel.", type=OptionType.CHANNEL, required=False)
@interactions.option(name="send_dm_to_user", description="Send message to a user via DM instead of a channel. Set both_dm_and_channel to send both.", type=OptionType.USER, required=False) # noqa
@interactions.option(name="both_dm_and_channel", description="Send both DM and message to the channel, needs send_dm_to_user to be set if you want both.", type=OptionType.BOOLEAN, required=False) # noqa
async def remind_interval(
ctx: interactions.CommandContext,
message_reason: 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,
different_channel: interactions.Channel | None = None,
send_dm_to_user: interactions.User | None = None,
both_dm_and_channel: bool | None = None,
):
@base_command.subcommand(
name="interval",
description="Schedules messages to be run periodically, on selected intervals.",
)
@interactions.option(
name="message_reason",
description="The message I'm going to send you.",
type=OptionType.STRING,
required=True,
)
@interactions.option(
name="weeks",
description="Number of weeks to wait",
type=OptionType.INTEGER,
required=False,
)
@interactions.option(
name="days",
description="Number of days to wait",
type=OptionType.INTEGER,
required=False,
)
@interactions.option(
name="hours",
description="Number of hours to wait",
type=OptionType.INTEGER,
required=False,
)
@interactions.option(
name="minutes",
description="Number of minutes to wait",
type=OptionType.INTEGER,
required=False,
)
@interactions.option(
name="seconds",
description="Number of seconds to wait",
type=OptionType.INTEGER,
required=False,
)
@interactions.option(
name="start_date",
description="When to start, in the ISO 8601 format. (Example: 2010-10-10 09:30:00)",
type=OptionType.STRING,
required=False,
)
@interactions.option(
name="end_date",
description="When to stop, in the ISO 8601 format. (Example: 2014-06-15 11:00:00)",
type=OptionType.STRING,
required=False,
)
@interactions.option(
name="timezone",
description="Time zone to use for the date/time calculations",
type=OptionType.STRING,
required=False,
)
@interactions.option(
name="jitter",
description="Delay the job execution by x seconds at most. Adds a random component to the execution time.",
type=OptionType.INTEGER,
required=False,
)
@interactions.option(
name="different_channel",
description="Send the messages to a different channel.",
type=OptionType.CHANNEL,
required=False,
)
@interactions.option(
name="send_dm_to_user",
description="Send message to a user via DM instead of a channel. Set both_dm_and_channel to send both.",
type=OptionType.USER,
required=False,
)
@interactions.option(
name="both_dm_and_channel",
description="Send both DM and message to the channel, needs send_dm_to_user to be set if you want both.",
type=OptionType.BOOLEAN,
required=False,
)
async def remind_interval( # noqa: PLR0913
ctx: interactions.CommandContext,
message_reason: 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,
different_channel: interactions.Channel | None = None,
send_dm_to_user: interactions.User | None = None,
both_dm_and_channel: bool | None = None,
) -> None:
"""Create a new reminder that triggers based on an interval.
Args:
@ -493,12 +715,12 @@ async def remind_interval(
if different_channel:
channel_id = int(different_channel.id)
dm_message = ""
dm_message: str = ""
where_and_when = "You should never see this message. Please report this to the bot owner if you do. :-)"
should_send_channel_reminder = True
try:
if send_dm_to_user:
dm_reminder = scheduler.add_job(
dm_reminder: Job = scheduler.add_job(
send_to_user,
"interval",
weeks=weeks,
@ -520,10 +742,16 @@ async def remind_interval(
if not both_dm_and_channel:
# If we should send the message to the channel too instead of just a DM.
should_send_channel_reminder = False
where_and_when = (f"I will send a DM to {send_dm_to_user.username} at:\n"
f"First run in {calculate(dm_reminder)} with the message:\n")
where_and_when: str = (
f"I will send a DM to {send_dm_to_user.username} at:\n"
f"First run in {calculate(dm_reminder)} with the message:\n"
)
if ctx.member is None:
await ctx.send("Failed to get the member who sent the command.", ephemeral=True)
return
if should_send_channel_reminder:
job = scheduler.add_job(
job: Job = scheduler.add_job(
send_to_discord,
"interval",
weeks=weeks,
@ -541,37 +769,36 @@ async def remind_interval(
"author_id": ctx.member.id,
},
)
where_and_when = (f" I will send messages to <#{channel_id}>{dm_message}.\n"
f"First run in {calculate(job)} with the message:\n")
where_and_when = (
f" I will send messages to <#{channel_id}>{dm_message}.\n"
f"First run in {calculate(job)} with the message:"
)
except ValueError as e:
await ctx.send(str(e), ephemeral=True)
return
# TODO: Add what arguments we used in the job to the message
message = (
f"Hello {ctx.member.name}\n"
f"{where_and_when}"
f"**{message_reason}**."
)
message: str = f"Hello {ctx.member.name}\n{where_and_when}\n**{message_reason}**."
await ctx.send(message)
def my_listener(event):
def my_listener(event: JobExecutionEvent) -> None:
"""This gets called when something in APScheduler happens."""
if event.code == events.EVENT_JOB_MISSED:
# TODO: Is it possible to get the message?
scheduled_time = event.scheduled_run_time.strftime("%Y-%m-%d %H:%M:%S")
msg = f"Job {event.job_id} was missed! Was scheduled at {scheduled_time}"
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:
send_webhook(f"discord-reminder-bot failed to send message to Discord\n"
f"{event}")
send_webhook(
f"discord-reminder-bot failed to send message to Discord\n{event}",
)
async def send_to_discord(channel_id: int, message: str, author_id: int):
async def send_to_discord(channel_id: int, message: str, author_id: int) -> None:
"""Send a message to Discord.
Args:
@ -579,26 +806,23 @@ async def send_to_discord(channel_id: int, message: str, author_id: int):
message: The message.
author_id: User we should ping.
"""
channel = await interactions.get(
channel: Channel = await interactions.get(
bot,
interactions.Channel,
object_id=int(channel_id),
object_id=channel_id,
force=interactions.Force.HTTP,
)
await channel.send(f"<@{author_id}>\n{message}")
def start():
def start() -> None:
"""Start scheduler and log in to Discord."""
# TODO: Add how many reminders are scheduled.
# TODO: Make backup of jobs.sqlite before running the bot.
logging.basicConfig(level=logging.getLevelName(log_level))
logging.info(
f"\nsqlite_location = {sqlite_location}\n"
f"config_timezone = {config_timezone}\n"
f"log_level = {log_level}"
"\nsqlite_location = %s\nconfig_timezone = %s\nlog_level = %s" % (sqlite_location, config_timezone, log_level),
)
scheduler.start()
scheduler.add_listener(my_listener, EVENT_JOB_MISSED | EVENT_JOB_ERROR)

35
discord_reminder_bot/parse.py
View File

@ -9,8 +9,7 @@ from discord_reminder_bot.settings import config_timezone
@dataclasses.dataclass
class ParsedTime:
"""
This is used when parsing a time or date from a string.
"""This is used when parsing a time or date from a string.
We use this when adding a job with /reminder add.
@ -20,10 +19,11 @@ class ParsedTime:
err_msg: The error message.
parsed_time: The parsed time we got from the string.
"""
date_to_parse: str = None
date_to_parse: str | None = None
err: bool = False
err_msg: str = ""
parsed_time: datetime = None
parsed_time: datetime | None = None
def parse_time(date_to_parse: str, timezone: str = config_timezone) -> ParsedTime:
@ -37,7 +37,7 @@ def parse_time(date_to_parse: str, timezone: str = config_timezone) -> ParsedTim
ParsedTime
"""
try:
parsed_date = dateparser.parse(
parsed_date: datetime | None = dateparser.parse(
f"{date_to_parse}",
settings={
"PREFER_DATES_FROM": "future",
@ -46,12 +46,25 @@ def parse_time(date_to_parse: str, timezone: str = config_timezone) -> ParsedTim
},
)
except SettingValidationError as e:
return ParsedTime(err=True, err_msg=f"Timezone is possible wrong?: {e}", date_to_parse=date_to_parse)
return ParsedTime(
err=True,
err_msg=f"Timezone is possible wrong?: {e}",
date_to_parse=date_to_parse,
)
except ValueError as e:
return ParsedTime(err=True, err_msg=f"Failed to parse date. Unknown language: {e}", date_to_parse=date_to_parse)
return ParsedTime(
err=True,
err_msg=f"Failed to parse date. Unknown language: {e}",
date_to_parse=date_to_parse,
)
except TypeError as e:
return ParsedTime(err=True, err_msg=f"{e}", date_to_parse=date_to_parse)
if not parsed_date:
return ParsedTime(err=True, err_msg=f"Could not parse the date.", date_to_parse=date_to_parse)
return ParsedTime(parsed_time=parsed_date, date_to_parse=date_to_parse)
return (
ParsedTime(parsed_time=parsed_date, date_to_parse=date_to_parse)
if parsed_date
else ParsedTime(
err=True,
err_msg="Could not parse the date.",
date_to_parse=date_to_parse,
)
)

17
discord_reminder_bot/settings.py
View File

@ -6,18 +6,19 @@ from apscheduler.schedulers.asyncio import AsyncIOScheduler
from dotenv import load_dotenv
load_dotenv(verbose=True)
sqlite_location = os.getenv("SQLITE_LOCATION", default="/jobs.sqlite")
config_timezone = os.getenv("TIMEZONE", default="UTC")
bot_token = os.getenv("BOT_TOKEN", default="")
log_level = os.getenv("LOG_LEVEL", default="INFO")
webhook_url = os.getenv("WEBHOOK_URL", default="")
sqlite_location: str = os.getenv("SQLITE_LOCATION", default="/jobs.sqlite")
config_timezone: str = os.getenv("TIMEZONE", default="UTC")
bot_token: str = os.getenv("BOT_TOKEN", default="")
log_level: str = os.getenv("LOG_LEVEL", default="INFO")
webhook_url: str = os.getenv("WEBHOOK_URL", default="")
if not bot_token:
raise ValueError("Missing bot token")
err_msg = "Missing bot token"
raise ValueError(err_msg)
# Advanced Python Scheduler
jobstores = {"default": SQLAlchemyJobStore(url=f"sqlite://{sqlite_location}")}
job_defaults = {"coalesce": True}
jobstores: dict[str, SQLAlchemyJobStore] = {"default": SQLAlchemyJobStore(url=f"sqlite://{sqlite_location}")}
job_defaults: dict[str, bool] = {"coalesce": True}
scheduler = AsyncIOScheduler(
jobstores=jobstores,
timezone=pytz.timezone(config_timezone),