rewrite: New Pomodoro Timer system.
This commit is contained in:
844
src/modules/pomodoro/cog.py
Normal file
844
src/modules/pomodoro/cog.py
Normal file
@@ -0,0 +1,844 @@
|
||||
from typing import Optional
|
||||
from collections import defaultdict
|
||||
import asyncio
|
||||
|
||||
import discord
|
||||
from discord.ext import commands as cmds
|
||||
from discord import app_commands as appcmds
|
||||
|
||||
from meta import LionCog, LionBot, LionContext
|
||||
from meta.logger import log_wrap
|
||||
from meta.sharding import THIS_SHARD
|
||||
from utils.lib import utc_now
|
||||
|
||||
from wards import low_management
|
||||
|
||||
from . import babel, logger
|
||||
from .data import TimerData
|
||||
from .lib import TimerRole
|
||||
from .settings import TimerSettings
|
||||
from .settingui import TimerConfigUI
|
||||
from .timer import Timer
|
||||
from .options import TimerOptions
|
||||
from .ui import TimerStatusUI
|
||||
from .ui.config import TimerOptionsUI
|
||||
|
||||
_p = babel._p
|
||||
|
||||
_param_options = {
|
||||
'focus_length': (TimerOptions.FocusLength, TimerRole.MANAGER),
|
||||
'break_length': (TimerOptions.BreakLength, TimerRole.MANAGER),
|
||||
'notification_channel': (TimerOptions.NotificationChannel, TimerRole.ADMIN),
|
||||
'inactivity_threshold': (TimerOptions.InactivityThreshold, TimerRole.OWNER),
|
||||
'manager_role': (TimerOptions.ManagerRole, TimerRole.ADMIN),
|
||||
'voice_alerts': (TimerOptions.VoiceAlerts, TimerRole.OWNER),
|
||||
'name': (TimerOptions.BaseName, TimerRole.OWNER),
|
||||
'channel_name': (TimerOptions.ChannelFormat, TimerRole.OWNER),
|
||||
}
|
||||
|
||||
|
||||
class TimerCog(LionCog):
|
||||
def __init__(self, bot: LionBot):
|
||||
self.bot = bot
|
||||
self.data = bot.db.load_registry(TimerData())
|
||||
self.settings = TimerSettings()
|
||||
self.timer_options = TimerOptions()
|
||||
|
||||
self.ready = False
|
||||
self.timers = defaultdict(dict)
|
||||
|
||||
async def cog_load(self):
|
||||
await self.data.init()
|
||||
|
||||
self.bot.core.guild_config.register_model_setting(self.settings.PomodoroChannel)
|
||||
|
||||
configcog = self.bot.get_cog('ConfigCog')
|
||||
self.crossload_group(self.configure_group, configcog.configure_group)
|
||||
|
||||
if self.bot.is_ready():
|
||||
await self.initialise()
|
||||
|
||||
async def cog_unload(self):
|
||||
"""
|
||||
Detach TimerCog and unload components.
|
||||
|
||||
Clears caches and stops run-tasks for each active timer.
|
||||
Does not exist until all timers have completed background tasks.
|
||||
"""
|
||||
timers = (timer for tguild in self.timers.values() for timer in tguild.values())
|
||||
try:
|
||||
await asyncio.gather(*(timer.unload() for timer in timers))
|
||||
except Exception:
|
||||
logger.exception(
|
||||
"Exception encountered while unloading `TimerCog`"
|
||||
)
|
||||
self.timers.clear()
|
||||
|
||||
async def _load_timers(self, timer_data: list[TimerData.Timer]):
|
||||
"""
|
||||
Factored method to load a list of timers from data rows.
|
||||
"""
|
||||
guildids = set()
|
||||
to_delete = []
|
||||
to_create = []
|
||||
for row in timer_data:
|
||||
channel = self.bot.get_channel(row.channelid)
|
||||
if not channel:
|
||||
to_delete.append(row.channelid)
|
||||
else:
|
||||
guildids.add(row.guildid)
|
||||
to_create.append(row)
|
||||
|
||||
if guildids:
|
||||
lguilds = await self.bot.core.lions.fetch_guilds(*guildids)
|
||||
else:
|
||||
lguilds = []
|
||||
|
||||
now = utc_now()
|
||||
to_launch = []
|
||||
to_update = []
|
||||
timer_reg = defaultdict(dict)
|
||||
for row in to_create:
|
||||
timer = Timer(self.bot, row, lguilds[row.guildid])
|
||||
if timer.running:
|
||||
to_launch.append(timer)
|
||||
else:
|
||||
to_update.append(timer)
|
||||
timer_reg[row.guildid][row.channelid] = timer
|
||||
timer.last_seen = {member.id: now for member in timer.members}
|
||||
|
||||
# Delete non-existent timers
|
||||
if to_delete:
|
||||
await self.data.Timer.table.delete_where(channelid=to_delete)
|
||||
idstr = ', '.join(map(str, to_delete))
|
||||
logger.info(
|
||||
f"Destroyed {len(to_delete)} timers with missing voice channels: {idstr}"
|
||||
)
|
||||
|
||||
# Re-launch and update running timers
|
||||
for timer in to_launch:
|
||||
timer.launch()
|
||||
tasks = [
|
||||
asyncio.create_task(timer.update_status_card()) for timer in to_launch
|
||||
]
|
||||
if tasks:
|
||||
try:
|
||||
await asyncio.gather(*tasks)
|
||||
except Exception:
|
||||
logger.exception(
|
||||
"Exception occurred updating timer status for running timers."
|
||||
)
|
||||
logger.info(
|
||||
f"Updated and launched {len(to_launch)} running timers."
|
||||
)
|
||||
|
||||
# Update stopped timers
|
||||
tasks = [
|
||||
asyncio.create_task(timer.update_status_card()) for timer in to_update
|
||||
]
|
||||
if tasks:
|
||||
try:
|
||||
await asyncio.gather(*tasks)
|
||||
except Exception:
|
||||
logger.exception(
|
||||
"Exception occurred updating timer status for stopped timers."
|
||||
)
|
||||
logger.info(
|
||||
f"Updated {len(to_update)} stopped timers."
|
||||
)
|
||||
|
||||
# Update timer registry
|
||||
self.timers.update(timer_reg)
|
||||
|
||||
@LionCog.listener('on_ready')
|
||||
@log_wrap(action='Init Timers')
|
||||
async def initialise(self):
|
||||
"""
|
||||
Restore timers.
|
||||
"""
|
||||
self.ready = False
|
||||
self.timers = defaultdict(dict)
|
||||
|
||||
# Fetch timers in guilds on this shard
|
||||
# TODO: Join with guilds and filter by guilds we are still in
|
||||
timer_data = await self.data.Timer.fetch_where(THIS_SHARD)
|
||||
await self._load_timers(timer_data)
|
||||
|
||||
# Ready to handle events
|
||||
self.ready = True
|
||||
logger.info("Timer system ready to process events.")
|
||||
|
||||
# ----- Event Handlers -----
|
||||
@LionCog.listener('on_voice_state_update')
|
||||
@log_wrap(action='Timer Voice Events')
|
||||
async def timer_voice_events(self, member, before, after):
|
||||
if not self.ready:
|
||||
# Trust initialise to trigger update status
|
||||
return
|
||||
if member.bot:
|
||||
return
|
||||
|
||||
# If a member is leaving or joining a running timer, trigger a status update
|
||||
if before.channel != after.channel:
|
||||
leaving = self.get_channel_timer(before.channel.id) if before.channel else None
|
||||
joining = self.get_channel_timer(after.channel.id) if after.channel else None
|
||||
|
||||
tasks = []
|
||||
if leaving:
|
||||
tasks.append(leaving.update_status_card())
|
||||
if joining is not None:
|
||||
joining.last_seen[member.id] = utc_now()
|
||||
if not joining.running and joining.auto_restart:
|
||||
tasks.append(joining.start())
|
||||
else:
|
||||
tasks.append(joining.update_status_card())
|
||||
|
||||
if tasks:
|
||||
try:
|
||||
await asyncio.gather(*tasks)
|
||||
except Exception:
|
||||
logger.exception(
|
||||
"Exception occurred while handling timer voice event. "
|
||||
f"Leaving: {leaving!r} "
|
||||
f"Joining: {joining!r}"
|
||||
)
|
||||
|
||||
@LionCog.listener('on_guild_remove')
|
||||
@log_wrap(action='Unload Guild Timers')
|
||||
async def _unload_guild_timers(self, guild: discord.Guild):
|
||||
"""
|
||||
When we leave a guild, perform an unload for all timers in the Guild.
|
||||
"""
|
||||
if not self.ready:
|
||||
# Trust initialiser to ignore the guild
|
||||
return
|
||||
|
||||
timers = self.timers.pop(guild.id)
|
||||
tasks = []
|
||||
for timer in timers:
|
||||
tasks.append(asyncio.create_task(timer.unload()))
|
||||
if tasks:
|
||||
try:
|
||||
await asyncio.gather(*tasks)
|
||||
except Exception:
|
||||
logger.warning(
|
||||
"Exception occurred while unloading timers for removed guild.",
|
||||
exc_info=True
|
||||
)
|
||||
logger.info(
|
||||
f"Unloaded {len(timers)} from removed guild <gid: {guild.id}>."
|
||||
)
|
||||
|
||||
@LionCog.listener('on_guild_join')
|
||||
@log_wrap(action='Load Guild Timers')
|
||||
async def _load_guild_timers(self, guild: discord.Guild):
|
||||
"""
|
||||
When we join a guild, reload any saved timers for this guild.
|
||||
"""
|
||||
timer_data = await self.data.Timer.fetch_where(guildid=guild.id)
|
||||
if timer_data:
|
||||
await self._load_timers(timer_data)
|
||||
|
||||
@LionCog.listener('on_guild_channel_delete')
|
||||
@log_wrap(action='Destroy Channel Timer')
|
||||
async def _destroy_channel_timer(self, channel: discord.abc.GuildChannel):
|
||||
"""
|
||||
If a voice channel with a timer was deleted, destroy the timer.
|
||||
"""
|
||||
timer = self.get_channel_timer(channel.id)
|
||||
if timer is not None:
|
||||
await timer.destroy(reason="Voice Channel Deleted")
|
||||
|
||||
@LionCog.listener('on_guildset_pomodoro_channel')
|
||||
@log_wrap(action='Update Pomodoro Channels')
|
||||
async def _update_pomodoro_channels(self, guildid: int, setting: TimerSettings.PomodoroChannel):
|
||||
"""
|
||||
Request a send_status for all guild timers which need to move channel.
|
||||
"""
|
||||
timers = self.get_guild_timers(guildid).values()
|
||||
tasks = []
|
||||
for timer in timers:
|
||||
current_channel = timer.notification_channel
|
||||
current_hook = timer._hook
|
||||
if current_channel and (not current_hook or current_hook.channelid != current_channel.id):
|
||||
tasks.append(asyncio.create_task(timer.send_status()))
|
||||
|
||||
if tasks:
|
||||
try:
|
||||
await asyncio.gather(*tasks)
|
||||
except Exception:
|
||||
logger.warning(
|
||||
"Exception occurred which refreshing status for timers with new notification_channel.",
|
||||
exc_info=True
|
||||
)
|
||||
|
||||
# ----- Timer API -----
|
||||
def get_guild_timers(self, guildid: int) -> dict[int, Timer]:
|
||||
"""
|
||||
Get all timers in the given guild as a map channelid -> Timer.
|
||||
"""
|
||||
return self.timers[guildid]
|
||||
|
||||
def get_channel_timer(self, channelid: int) -> Optional[Timer]:
|
||||
"""
|
||||
Get the timer bound to the given channel, or None if it does not exist.
|
||||
"""
|
||||
channel = self.bot.get_channel(channelid)
|
||||
if channel:
|
||||
return self.timers[channel.guild.id].get(channelid, None)
|
||||
|
||||
async def create_timer(self, **kwargs):
|
||||
timer_data = await self.data.Timer.create(**kwargs)
|
||||
lguild = await self.bot.core.lions.fetch_guild(timer_data.guildid)
|
||||
timer = Timer(self.bot, timer_data, lguild)
|
||||
self.timers[timer_data.guildid][timer_data.channelid] = timer
|
||||
|
||||
return timer
|
||||
|
||||
async def destroy_timer(self, timer: Timer, **kwargs):
|
||||
"""
|
||||
Destroys the provided timer and removes it from the registry.
|
||||
"""
|
||||
self.timers[timer.data.guildid].pop(timer.data.channelid, None)
|
||||
await timer.destroy(**kwargs)
|
||||
|
||||
# ----- Timer Commands -----
|
||||
@cmds.hybrid_group(
|
||||
name=_p('cmd:pomodoro', "pomodoro"),
|
||||
desc=_p('cmd:pomodoro|desc', "Base group for all pomodoro timer commands.")
|
||||
)
|
||||
@cmds.guild_only()
|
||||
async def pomodoro_group(self, ctx: LionContext):
|
||||
...
|
||||
|
||||
# -- User Display Commands --
|
||||
@pomodoro_group.command(
|
||||
name=_p('cmd:pomodoro_status', "show"),
|
||||
description=_p('cmd:pomodoro_status|desc', "Display the status of a single pomodoro timer.")
|
||||
)
|
||||
@appcmds.rename(
|
||||
channel=_p('cmd:pomodoro_status|param:channel', "timer_channel")
|
||||
)
|
||||
@appcmds.describe(
|
||||
channel=_p(
|
||||
'cmd:pomodoro_status|param:channel|desc',
|
||||
"The channel for which you want to view the timer."
|
||||
)
|
||||
)
|
||||
async def cmd_pomodoro_status(self, ctx: LionContext, channel: discord.VoiceChannel):
|
||||
t = self.bot.translator.t
|
||||
|
||||
if not ctx.guild:
|
||||
return
|
||||
if not ctx.interaction:
|
||||
return
|
||||
|
||||
# Check if a timer exists in the given channel
|
||||
timer = self.get_channel_timer(channel.id)
|
||||
if timer is None:
|
||||
embed = discord.Embed(
|
||||
colour=discord.Colour.brand_red(),
|
||||
description=t(_p(
|
||||
'cmd:pomodoro_status|error:no_timer',
|
||||
"The channel {channel} does not have a timer set up!"
|
||||
)).format(channel=channel.mention)
|
||||
)
|
||||
await ctx.reply(embed=embed, ephemeral=True)
|
||||
else:
|
||||
# Display the timer status ephemerally
|
||||
status = await timer.current_status(with_notify=False, with_warnings=False)
|
||||
await ctx.reply(**status.send_args, ephemeral=True)
|
||||
|
||||
@pomodoro_group.command(
|
||||
name=_p('cmd:pomodoro_list', "list"),
|
||||
description=_p('cmd:pomodoro_list|desc', "List the available pomodoro timers.")
|
||||
)
|
||||
async def cmd_pomodoro_list(self, ctx: LionContext):
|
||||
t = self.bot.translator.t
|
||||
|
||||
if not ctx.guild:
|
||||
return
|
||||
if not ctx.interaction:
|
||||
return
|
||||
|
||||
timers = list(self.get_guild_timers(ctx.guild.id).values())
|
||||
visible_timers = [
|
||||
timer for timer in timers
|
||||
if timer.channel and timer.channel.permissions_for(ctx.author).view_channel
|
||||
]
|
||||
|
||||
if not timers:
|
||||
# No timers in this guild!
|
||||
embed = discord.Embed(
|
||||
colour=discord.Colour.brand_red(),
|
||||
description=t(_p(
|
||||
'cmd:pomodoro_list|error:no_timers',
|
||||
"No timers have been setup in this server!\n"
|
||||
"You can ask an admin to create one with {command}."
|
||||
)).format(command='`/pomodoro admin create`')
|
||||
)
|
||||
# TODO: Update command mention when we have command mentions
|
||||
await ctx.reply(embed=embed, ephemeral=True)
|
||||
elif not visible_timers:
|
||||
# Timers exist, but the member can't see any
|
||||
embed = discord.Embed(
|
||||
colour=discord.Colour.brand_red(),
|
||||
description=t(_p(
|
||||
'cmd:pomodoro_list|error:no_visible_timers',
|
||||
"There are no visible timers in this server!"
|
||||
))
|
||||
)
|
||||
await ctx.reply(embed=embed, ephemeral=True)
|
||||
else:
|
||||
# Timers exist and are visible!
|
||||
embed = discord.Embed(
|
||||
colour=discord.Colour.orange(),
|
||||
title=t(_p(
|
||||
'cmd:pomodoro_list|embed:timer_list|title',
|
||||
"Pomodoro Timers in **{guild}**"
|
||||
)).format(guild=ctx.guild.name),
|
||||
)
|
||||
for timer in visible_timers:
|
||||
stage = timer.current_stage
|
||||
if stage is None:
|
||||
if timer.auto_restart:
|
||||
lazy_status = _p(
|
||||
'cmd:pomodoro_list|status:stopped_auto',
|
||||
"`{pattern}` timer is stopped with no members!\nJoin {channel} to restart it."
|
||||
)
|
||||
else:
|
||||
lazy_status = _p(
|
||||
'cmd:pomodoro_list|status:stopped_manual',
|
||||
"`{pattern}` timer is stopped with `{members}` members!\n"
|
||||
"Join {channel} and press `Start` to start it!"
|
||||
)
|
||||
else:
|
||||
if stage.focused:
|
||||
lazy_status = _p(
|
||||
'cmd:pomodoro_list|status:running_focus',
|
||||
"`{pattern}` timer is running with `{members}` members!\n"
|
||||
"Currently **focusing**, with break starting {timestamp}"
|
||||
)
|
||||
else:
|
||||
lazy_status = _p(
|
||||
'cmd:pomodoro_list|status:running_break',
|
||||
"`{pattern}` timer is running with `{members}` members!\n"
|
||||
"Currently **resting**, with focus starting {timestamp}"
|
||||
)
|
||||
status = t(lazy_status).format(
|
||||
pattern=timer.pattern,
|
||||
channel=timer.channel.mention,
|
||||
members=len(timer.members),
|
||||
timestamp=f"<t:{int(stage.end.timestamp())}:R>" if stage else None
|
||||
)
|
||||
embed.add_field(name=timer.channel.mention, value=status, inline=False)
|
||||
await ctx.reply(embed=embed, ephemeral=False)
|
||||
|
||||
# -- Admin Commands --
|
||||
@pomodoro_group.group(
|
||||
name=_p('cmd:pomodoro_admin', "admin"),
|
||||
desc=_p('cmd:pomodoro_admin|desc', "Command group for pomodoro admin controls.")
|
||||
)
|
||||
async def pomodoro_admin_group(self, ctx: LionContext):
|
||||
...
|
||||
|
||||
@pomodoro_admin_group.command(
|
||||
name=_p('cmd:pomodoro_create', "create"),
|
||||
description=_p(
|
||||
'cmd:pomodoro_create|desc',
|
||||
"Create a new Pomodoro timer. Requires admin permissions."
|
||||
)
|
||||
)
|
||||
@appcmds.rename(
|
||||
channel=_p('cmd:pomodoro_create|param:channel', "timer_channel"),
|
||||
**{param: option._display_name for param, (option, _) in _param_options.items()}
|
||||
)
|
||||
@appcmds.describe(
|
||||
channel=_p(
|
||||
'cmd:pomodoro_create|param:channel|desc',
|
||||
"Voice channel to create the timer in. (Defaults to your current channel, or makes a new one.)"
|
||||
),
|
||||
**{param: option._desc for param, (option, _) in _param_options.items()}
|
||||
)
|
||||
async def cmd_pomodoro_create(self, ctx: LionContext,
|
||||
focus_length: appcmds.Range[int, 1, 24*60],
|
||||
break_length: appcmds.Range[int, 1, 24*60],
|
||||
channel: Optional[discord.VoiceChannel] = None,
|
||||
notification_channel: Optional[discord.TextChannel | discord.VoiceChannel] = None,
|
||||
inactivity_threshold: Optional[appcmds.Range[int, 0, 127]] = None,
|
||||
manager_role: Optional[discord.Role] = None,
|
||||
voice_alerts: Optional[bool] = None,
|
||||
name: Optional[appcmds.Range[str, 0, 100]] = None,
|
||||
channel_name: Optional[appcmds.Range[str, 0, 100]] = None,
|
||||
):
|
||||
t = self.bot.translator.t
|
||||
|
||||
# Type guards
|
||||
if not ctx.guild:
|
||||
return
|
||||
if not ctx.interaction:
|
||||
return
|
||||
|
||||
# Check permissions
|
||||
if not ctx.author.guild_permissions.administrator:
|
||||
embed = discord.Embed(
|
||||
colour=discord.Colour.brand_red(),
|
||||
description=t(_p(
|
||||
'cmd:pomodoro_create|error:insufficient_perms',
|
||||
"Only server administrators can create timers!"
|
||||
))
|
||||
)
|
||||
await ctx.reply(embed=embed, ephemeral=True)
|
||||
return
|
||||
|
||||
# If a voice channel was not given, attempt to resolve it or make one
|
||||
if channel is None:
|
||||
# Resolving order: command channel, author voice channel, new channel
|
||||
if ctx.channel.type is discord.ChannelType.voice:
|
||||
channel = ctx.channel
|
||||
elif ctx.author.voice and ctx.author.voice.channel:
|
||||
channel = ctx.author.voice.channel
|
||||
else:
|
||||
# Attempt to create new channel in current category
|
||||
if ctx.guild.me.guild_permissions.manage_channels:
|
||||
try:
|
||||
channel = await ctx.guild.create_voice_channel(
|
||||
name="Timer",
|
||||
reason="Creating Pomodoro Voice Channel",
|
||||
category=ctx.channel.category
|
||||
)
|
||||
except discord.HTTPException:
|
||||
embed = discord.Embed(
|
||||
colour=discord.Colour.brand_red(),
|
||||
title=t(_p(
|
||||
'cmd:pomodoro_create|error:channel_create_failed|title',
|
||||
"Could not create pomodoro voice channel!"
|
||||
)),
|
||||
description=t(_p(
|
||||
'cmd:pomodoro_create|error:channel_create|desc',
|
||||
"Failed to create a new pomodoro voice channel due to an unknown "
|
||||
"Discord communication error. "
|
||||
"Please try creating the channel manually and pass it to the "
|
||||
"`timer_channel` argument of this command."
|
||||
))
|
||||
)
|
||||
await ctx.reply(embed=embed, ephemeral=True)
|
||||
return
|
||||
else:
|
||||
# Error
|
||||
embed = discord.Embed(
|
||||
colour=discord.Colour.brand_red(),
|
||||
title=t(_p(
|
||||
'cmd:pomodoro_create|error:channel_create_permissions|title',
|
||||
"Could not create pomodoro voice channel!"
|
||||
)),
|
||||
description=t(_p(
|
||||
'cmd:pomodoro_create|error:channel_create_permissions|desc',
|
||||
"No `timer_channel` was provided, and I lack the `MANAGE_CHANNELS` permission "
|
||||
"needed to create a new voice channel."
|
||||
))
|
||||
)
|
||||
await ctx.reply(embed=embed, ephemeral=True)
|
||||
return
|
||||
|
||||
# At this point, we have a voice channel
|
||||
# Make sure a timer does not already exist in the channel
|
||||
if (self.get_channel_timer(channel.id)) is not None:
|
||||
embed = discord.Embed(
|
||||
colour=discord.Colour.brand_red(),
|
||||
description=t(_p(
|
||||
'cmd:pomodoro_create|error:timer_exists',
|
||||
"A timer already exists in {channel}! Use `/pomodoro admin edit` to modify it."
|
||||
)).format(channel=channel.mention)
|
||||
)
|
||||
await ctx.reply(embed=embed, ephemeral=True)
|
||||
return
|
||||
|
||||
# Build the creation arguments from the rest of the provided args
|
||||
provided = {
|
||||
'focus_length': focus_length * 60,
|
||||
'break_length': break_length * 60,
|
||||
'notification_channel': notification_channel,
|
||||
'inactivity_threshold': inactivity_threshold,
|
||||
'manager_role': manager_role,
|
||||
'voice_alerts': voice_alerts,
|
||||
'name': name or channel.name,
|
||||
'channel_name': channel_name or None,
|
||||
}
|
||||
|
||||
create_args = {'channelid': channel.id, 'guildid': channel.guild.id}
|
||||
for param, value in provided.items():
|
||||
if value is not None:
|
||||
setting, _ = _param_options[param]
|
||||
create_args[setting._column] = setting._data_from_value(channel.id, value)
|
||||
|
||||
# Permission checks and input checking done
|
||||
await ctx.interaction.response.defer(thinking=True)
|
||||
|
||||
# Create timer
|
||||
timer = await self.create_timer(**create_args)
|
||||
|
||||
# Start timer
|
||||
await timer.start()
|
||||
|
||||
# Ack with a config UI
|
||||
ui = TimerOptionsUI(self.bot, timer, TimerRole.ADMIN, callerid=ctx.author.id)
|
||||
await ui.run(
|
||||
ctx.interaction,
|
||||
content=t(_p(
|
||||
'cmd:pomodoro_create|response:success|content',
|
||||
"Timer created successfully! Use the panel below to reconfigure."
|
||||
))
|
||||
)
|
||||
await ui.wait()
|
||||
|
||||
@pomodoro_admin_group.command(
|
||||
name=_p('cmd:pomodoro_destroy', "destroy"),
|
||||
description=_p(
|
||||
'cmd:pomodoro_destroy|desc',
|
||||
"Delete a pomodoro timer from a voice channel. Requires admin permissions."
|
||||
)
|
||||
)
|
||||
@appcmds.rename(
|
||||
channel=_p('cmd:pomodoro_destroy|param:channel', "timer_channel"),
|
||||
)
|
||||
@appcmds.describe(
|
||||
channel=_p('cmd:pomodoro_destroy|param:channel', "Channel with the timer to delete."),
|
||||
)
|
||||
async def cmd_pomodoro_delete(self, ctx: LionContext, channel: discord.VoiceChannel):
|
||||
t = self.bot.translator.t
|
||||
|
||||
# Type guards
|
||||
if not ctx.guild:
|
||||
return
|
||||
if not ctx.interaction:
|
||||
return
|
||||
|
||||
# Check the timer actually exists
|
||||
timer = self.get_channel_timer(channel.id)
|
||||
if timer is None:
|
||||
embed = discord.Embed(
|
||||
colour=discord.Colour.brand_red(),
|
||||
description=t(_p(
|
||||
'cmd:pomodoro_destroy|error:no_timer',
|
||||
"This channel doesn't have an attached pomodoro timer!"
|
||||
))
|
||||
)
|
||||
await ctx.interaction.response.send_message(embed=embed, ephemeral=True)
|
||||
return
|
||||
|
||||
# Check the user has sufficient permissions to delete the timer
|
||||
# TODO: Should we drop the admin requirement down to manage channel?
|
||||
timer_role = timer.get_member_role(ctx.author)
|
||||
if timer.owned:
|
||||
if timer_role < TimerRole.OWNER:
|
||||
embed = discord.Embed(
|
||||
colour=discord.Colour.brand_red(),
|
||||
description=t(_p(
|
||||
'cmd:pomodoro_destroy|error:insufficient_perms|owned',
|
||||
"You need to be an administrator or own this channel to remove this timer!"
|
||||
))
|
||||
)
|
||||
await ctx.interaction.response.send_message(embed=embed, ephemeral=True)
|
||||
return
|
||||
elif timer_role is not TimerRole.ADMIN:
|
||||
embed = discord.Embed(
|
||||
colour=discord.Colour.brand_red(),
|
||||
description=t(_p(
|
||||
'cmd:pomodoro_destroy|error:insufficient_perms|notowned',
|
||||
"You need to be a server administrator to remove this timer!"
|
||||
))
|
||||
)
|
||||
await ctx.interaction.response.send_message(embed=embed, ephemeral=True)
|
||||
return
|
||||
|
||||
await ctx.interaction.response.defer(thinking=True)
|
||||
await self.destroy_timer(timer, reason="Deleted by command")
|
||||
embed = discord.Embed(
|
||||
colour=discord.Colour.brand_green(),
|
||||
description=t(_p(
|
||||
'cmd:pomdoro_destroy|response:success|description',
|
||||
"Timer successfully removed from {channel}."
|
||||
)).format(channel=channel.mention)
|
||||
)
|
||||
await ctx.interaction.edit_original_response(embed=embed)
|
||||
|
||||
@pomodoro_admin_group.command(
|
||||
name=_p('cmd:pomodoro_edit', "edit"),
|
||||
description=_p(
|
||||
'cmd:pomodoro_edit|desc',
|
||||
"Edit a Timer"
|
||||
)
|
||||
)
|
||||
@appcmds.rename(
|
||||
channel=_p('cmd:pomodoro_edit|param:channel', "timer_channel"),
|
||||
**{param: option._display_name for param, (option, _) in _param_options.items()}
|
||||
)
|
||||
@appcmds.describe(
|
||||
channel=_p(
|
||||
'cmd:pomodoro_edit|param:channel|desc',
|
||||
"Channel holding the timer to edit."
|
||||
),
|
||||
**{param: option._desc for param, (option, _) in _param_options.items()}
|
||||
)
|
||||
async def cmd_pomodoro_edit(self, ctx: LionContext,
|
||||
channel: discord.VoiceChannel,
|
||||
focus_length: Optional[appcmds.Range[int, 1, 24*60]] = None,
|
||||
break_length: Optional[appcmds.Range[int, 1, 24*60]] = None,
|
||||
notification_channel: Optional[discord.TextChannel | discord.VoiceChannel] = None,
|
||||
inactivity_threshold: Optional[appcmds.Range[int, 0, 127]] = None,
|
||||
manager_role: Optional[discord.Role] = None,
|
||||
voice_alerts: Optional[bool] = None,
|
||||
name: Optional[appcmds.Range[str, 0, 100]] = None,
|
||||
channel_name: Optional[appcmds.Range[str, 0, 100]] = None,
|
||||
):
|
||||
t = self.bot.translator.t
|
||||
provided = {
|
||||
'focus_length': focus_length * 60 if focus_length else None,
|
||||
'break_length': break_length * 60 if break_length else None,
|
||||
'notification_channel': notification_channel,
|
||||
'inactivity_threshold': inactivity_threshold,
|
||||
'manager_role': manager_role,
|
||||
'voice_alerts': voice_alerts,
|
||||
'name': name or None,
|
||||
'channel_name': channel_name or None,
|
||||
}
|
||||
modified = set(param for param, value in provided.items() if value is not None)
|
||||
|
||||
# Type guards
|
||||
if not ctx.guild:
|
||||
return
|
||||
if not ctx.interaction:
|
||||
return
|
||||
|
||||
# Check the timer actually exists
|
||||
timer = self.get_channel_timer(channel.id)
|
||||
if timer is None:
|
||||
embed = discord.Embed(
|
||||
colour=discord.Colour.brand_red(),
|
||||
description=t(_p(
|
||||
'cmd:pomodoro_edit|error:no_timer',
|
||||
"This channel doesn't have an attached pomodoro timer to edit!"
|
||||
))
|
||||
)
|
||||
await ctx.interaction.response.send_message(embed=embed, ephemeral=True)
|
||||
return
|
||||
|
||||
# Check that the author has sufficient permissions to update the timer at all
|
||||
timer_role = timer.get_member_role(ctx.author)
|
||||
if timer_role is TimerRole.OTHER:
|
||||
embed = discord.Embed(
|
||||
colour=discord.Colour.brand_red(),
|
||||
description=t(_p(
|
||||
'cmd:pomodoro_edit|error:insufficient_perms|role:other',
|
||||
"Insufficient permissions to modifiy this timer!\n"
|
||||
"You need to be a server administrator, own this channel, or have the timer manager role."
|
||||
))
|
||||
)
|
||||
await ctx.reply(embed=embed, ephemeral=True)
|
||||
return
|
||||
|
||||
# Check that the author has sufficient permissions to modify the requested items
|
||||
# And build the list of arguments to write
|
||||
update_args = {}
|
||||
for param in modified:
|
||||
setting, required = _param_options[param]
|
||||
if timer_role < required:
|
||||
if required is TimerRole.OWNER and not timer.owned:
|
||||
required = TimerRole.ADMIN
|
||||
elif required is TimerRole.MANAGER and timer.data.manager_roleid is None:
|
||||
required = TimerRole.ADMIN
|
||||
|
||||
if required is TimerRole.ADMIN:
|
||||
error = t(_p(
|
||||
'cmd:pomodoro_edit|error:insufficient_permissions|role_needed:admin',
|
||||
"You need to be a guild admin to modify this option!"
|
||||
))
|
||||
elif required is TimerRole.OWNER:
|
||||
error = t(_p(
|
||||
'cmd:pomodoro_edit|error:insufficient_permissions|role_needed:owner',
|
||||
"You need to be a channel owner or guild admin to modify this option!"
|
||||
))
|
||||
elif required is TimerRole.MANAGER:
|
||||
error = t(_p(
|
||||
'cmd:pomodoro_edit|error:insufficient_permissions|role_needed:manager',
|
||||
"You need to be a guild admin or have the manager role to modify this option!"
|
||||
))
|
||||
|
||||
embed = discord.Embed(
|
||||
colour=discord.Colour.brand_red(),
|
||||
description=error
|
||||
)
|
||||
await ctx.reply(embed=embed, ephemeral=True)
|
||||
return
|
||||
update_args[setting._column] = setting._data_from_value(channel.id, provided[param])
|
||||
|
||||
await ctx.interaction.response.defer(thinking=True)
|
||||
|
||||
if update_args:
|
||||
# Update the timer data
|
||||
await timer.data.update(**update_args)
|
||||
# Regenerate or refresh the timer
|
||||
if ('focus_length' in modified) or ('break_length' in modified):
|
||||
await timer.start()
|
||||
elif ('notification_channel' in modified):
|
||||
await timer.send_status()
|
||||
else:
|
||||
await timer.update_status_card()
|
||||
|
||||
# Show the config UI
|
||||
ui = TimerOptionsUI(self.bot, timer, timer_role)
|
||||
await ui.run(ctx.interaction)
|
||||
await ui.wait()
|
||||
|
||||
# ----- Guild Config Commands -----
|
||||
@LionCog.placeholder_group
|
||||
@cmds.hybrid_group('configure', with_app_command=False)
|
||||
async def configure_group(self, ctx: LionContext):
|
||||
...
|
||||
|
||||
@configure_group.command(
|
||||
name=_p('cmd:configure_pomodoro', "pomodoro"),
|
||||
description=_p('cmd:configure_pomodoro|desc', "Configure Pomodoro Timer System")
|
||||
)
|
||||
@appcmds.rename(
|
||||
pomodoro_channel=TimerSettings.PomodoroChannel._display_name
|
||||
)
|
||||
@appcmds.describe(
|
||||
pomodoro_channel=TimerSettings.PomodoroChannel._desc
|
||||
)
|
||||
@cmds.check(low_management)
|
||||
async def configure_pomodoro_command(self, ctx: LionContext,
|
||||
pomodoro_channel: Optional[discord.VoiceChannel | discord.TextChannel] = None):
|
||||
t = self.bot.translator.t
|
||||
|
||||
# Type checking guards
|
||||
if not ctx.guild:
|
||||
return
|
||||
if not ctx.interaction:
|
||||
return
|
||||
|
||||
await ctx.interaction.response.defer(thinking=True)
|
||||
|
||||
pomodoro_channel_setting = await self.settings.PomodoroChannel.get(ctx.guild.id)
|
||||
|
||||
if pomodoro_channel is not None:
|
||||
# VALIDATE PERMISSIONS!
|
||||
pomodoro_channel_setting.value = pomodoro_channel
|
||||
await pomodoro_channel_setting.write()
|
||||
modified = True
|
||||
else:
|
||||
modified = False
|
||||
|
||||
if modified:
|
||||
line = pomodoro_channel_setting.update_message
|
||||
embed = discord.Embed(
|
||||
colour=discord.Colour.brand_green(),
|
||||
description=f"{self.bot.config.emojis.tick} {line}"
|
||||
)
|
||||
await ctx.reply(embed=embed)
|
||||
|
||||
if ctx.channel.id not in TimerConfigUI._listening or not modified:
|
||||
ui = TimerConfigUI(self.bot, ctx.guild.id, ctx.channel.id)
|
||||
await ui.run(ctx.interaction)
|
||||
await ui.wait()
|
||||
Reference in New Issue
Block a user