Croc/croccybot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'rewrite' into pillow

Browse Source
This commit is contained in:
Conatum
2023-09-24 10:58:59 +03:00
parent 81734f00ef 5675f72853
commit 5db32df3e2
178 changed files with 29691 additions and 2267 deletions
Download Patch File Download Diff File Expand all files Collapse all files

427
src/modules/pomodoro/cog.py
View File

@@ -10,6 +10,7 @@ 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 meta.monitor import ComponentMonitor, ComponentStatus, StatusLevel
from utils.lib import utc_now
from wards import low_management_ward
@@ -21,7 +22,6 @@ 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
@@ -43,12 +43,25 @@ class TimerCog(LionCog):
self.bot = bot
self.data = bot.db.load_registry(TimerData())
self.settings = TimerSettings()
self.monitor = ComponentMonitor('TimerCog', self._monitor)
self.timer_options = TimerOptions()
self.ready = False
self.timers = defaultdict(dict)
async def _monitor(self):
if not self.ready:
level = StatusLevel.STARTING
info = "(STARTING) Not ready. {timers} timers loaded."
else:
level = StatusLevel.OKAY
info = "(OK) {timers} timers loaded."
data = dict(timers=len(self.timers))
return ComponentStatus(level, info, info, data)
async def cog_load(self):
self.bot.system_monitor.add_component(self.monitor)
await self.data.init()
self.bot.core.guild_config.register_model_setting(self.settings.PomodoroChannel)
@@ -319,29 +332,24 @@ class TimerCog(LionCog):
await timer.destroy(**kwargs)
# ----- Timer Commands -----
@cmds.hybrid_group(
name=_p('cmd:pomodoro', "timers"),
description=_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.")
@cmds.hybrid_command(
name=_p('cmd:timer', "timer"),
description=_p('cmd:timer|desc', "Show your current (or selected) pomodoro timer.")
)
@appcmds.rename(
channel=_p('cmd:pomodoro_status|param:channel', "timer_channel")
channel=_p('cmd:timer|param:channel', "timer_channel")
)
@appcmds.describe(
channel=_p(
'cmd:pomodoro_status|param:channel|desc',
"The channel for which you want to view the timer."
'cmd:timer|param:channel|desc',
"Select a timer to display (by selecting the timer voice channel)"
)
)
async def cmd_pomodoro_status(self, ctx: LionContext, channel: discord.VoiceChannel):
@cmds.guild_only()
async def cmd_timer(self, ctx: LionContext,
channel: Optional[discord.VoiceChannel] = None):
t = self.bot.translator.t
if not ctx.guild:
@@ -349,27 +357,64 @@ class TimerCog(LionCog):
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(
timers: list[Timer] = list(self.get_guild_timers(ctx.guild.id).values())
error: Optional[discord.Embed] = None
if not timers:
# Guild has no timers
error = 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)
'cmd:timer|error:no_timers|desc',
"**This server has no timers set up!**\n"
"Ask an admin to set up and configure a timer with {create_cmd} first, "
"or rent a private room with {room_cmd} and create one yourself!"
)).format(create_cmd=self.bot.core.mention_cmd('pomodoro create'),
room_cmd=self.bot.core.mention_cmd('rooms rent'))
)
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)
elif channel is None:
if ctx.author.voice and ctx.author.voice.channel:
channel = ctx.author.voice.channel
else:
error = discord.Embed(
colour=discord.Colour.brand_red(),
description=t(_p(
'cmd:timer|error:no_channel|desc',
"**I don't know what timer to show you.**\n"
"No channel selected and you are not in a voice channel! "
"Use {timers_cmd} to list the available timers in this server."
)).format(timers_cmd=self.bot.core.mention_cmd('timers'))
)
@pomodoro_group.command(
name=_p('cmd:pomodoro_list', "list"),
description=_p('cmd:pomodoro_list|desc', "List the available pomodoro timers.")
if channel is not None:
timer = self.get_channel_timer(channel.id)
if timer is None:
error = discord.Embed(
colour=discord.Colour.brand_red(),
description=t(_p(
'cmd:timer|error:no_timer_in_channel',
"The channel {channel} is not a pomodoro timer room!\n"
"Use {timers_cmd} to list the available timers in this server."
)).format(
channel=channel.mention,
timers_cmd=self.bot.core.mention_cmd('timers')
)
)
else:
# Display the timer status ephemerally
await ctx.interaction.response.defer(thinking=True, ephemeral=True)
status = await timer.current_status(with_notify=False, with_warnings=False)
await ctx.interaction.edit_original_response(**status.edit_args)
if error is not None:
await ctx.reply(embed=error, ephemeral=True)
@cmds.hybrid_command(
name=_p('cmd:timers', "timers"),
description=_p('cmd:timers|desc', "List the available pomodoro timer rooms.")
)
async def cmd_pomodoro_list(self, ctx: LionContext):
@cmds.guild_only()
async def cmd_timers(self, ctx: LionContext):
t = self.bot.translator.t
if not ctx.guild:
@@ -378,6 +423,8 @@ class TimerCog(LionCog):
return
timers = list(self.get_guild_timers(ctx.guild.id).values())
# Extra filter here to exclude owned timers, but include ones the author is a member of
visible_timers = [
timer for timer in timers
if timer.channel and timer.channel.permissions_for(ctx.author).connect
@@ -385,26 +432,29 @@ class TimerCog(LionCog):
]
if not timers:
# No timers in this guild!
# No timers in the 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}, "
"or rent a private room and create one yourself!"
)).format(command='`/pomodoro admin create`')
'cmd:timer|error:no_timers|desc',
"**This server has no timers set up!**\n"
"Ask an admin to set up and configure a timer with {create_cmd} first, "
"or rent a private room with {room_cmd} and create one yourself!"
)).format(create_cmd=self.bot.core.mention_cmd('pomodoro create'),
room_cmd=self.bot.core.mention_cmd('rooms rent'))
)
# 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 timers you can join in this server!"
))
'cmd:timer|error:no_visible_timers|desc',
"**There are no available pomodoro timers!**\n"
"Ask an admin to set up a new timer with {create_cmd}, "
"or rent a private room with {room_cmd} and create one yourself!"
)).format(create_cmd=self.bot.core.mention_cmd('pomodoro create'),
room_cmd=self.bot.core.mention_cmd('rooms rent'))
)
await ctx.reply(embed=embed, ephemeral=True)
else:
@@ -412,8 +462,8 @@ class TimerCog(LionCog):
embed = discord.Embed(
colour=discord.Colour.orange(),
title=t(_p(
'cmd:pomodoro_list|embed:timer_list|title',
"Pomodoro Timers in **{guild}**"
'cmd:timers|embed:timer_list|title',
"Pomodoro Timer Rooms in **{guild}**"
)).format(guild=ctx.guild.name),
)
for timer in visible_timers:
@@ -421,25 +471,26 @@ class TimerCog(LionCog):
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."
'cmd:timers|status:stopped_auto',
"`{pattern}` timer is stopped with no members!\n"
"Join {channel} to restart it."
)
else:
lazy_status = _p(
'cmd:pomodoro_list|status:stopped_manual',
'cmd:timers|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',
'cmd:timers|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',
'cmd:timers|status:running_break',
"`{pattern}` timer is running with `{members}` members!\n"
"Currently **resting**, with focus starting {timestamp}"
)
@@ -453,18 +504,19 @@ class TimerCog(LionCog):
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.")
@cmds.hybrid_group(
name=_p('cmd:pomodoro', "pomodoro"),
description=_p('cmd:pomodoro|desc', "Create and configure pomodoro timer rooms.")
)
async def pomodoro_admin_group(self, ctx: LionContext):
@cmds.guild_only()
async def pomodoro_group(self, ctx: LionContext):
...
@pomodoro_admin_group.command(
@pomodoro_group.command(
name=_p('cmd:pomodoro_create', "create"),
description=_p(
'cmd:pomodoro_create|desc',
"Create a new Pomodoro timer. Requires admin permissions."
"Create a new Pomodoro timer. Requires manage channel permissions."
)
)
@appcmds.rename(
@@ -497,17 +549,15 @@ class TimerCog(LionCog):
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
# Get private room if applicable
room_cog = self.bot.get_cog('RoomCog')
if room_cog is None:
logger.warning("Running pomodoro create without private room cog loaded!")
private_room = None
else:
rooms = room_cog.get_rooms(ctx.guild.id, ctx.author.id)
cid = next((cid for cid, room in rooms.items() if room.data.ownerid == ctx.author.id), None)
private_room = ctx.guild.get_channel(cid) if cid is not None else None
# If a voice channel was not given, attempt to resolve it or make one
if channel is None:
@@ -516,112 +566,155 @@ class TimerCog(LionCog):
channel = ctx.channel
elif ctx.author.voice and ctx.author.voice.channel:
channel = ctx.author.voice.channel
elif not ctx.author.guild_permissions.manage_channels:
embed = discord.Embed(
colour=discord.Colour.brand_red(),
title=t(_p(
'cmd:pomodoro_create|new_channel|error:your_insufficient_perms|title',
"Could not create pomodoro voice channel!"
)),
description=t(_p(
'cmd:pomodoro_create|new_channel|error:your_insufficient_perms',
"No `timer_channel` was provided, and you lack the 'Manage Channels` permission "
"required to create a new timer room!"
))
)
await ctx.reply(embed=embed, ephemeral=True)
elif not ctx.guild.me.guild_permissions.manage_channels:
# Error
embed = discord.Embed(
colour=discord.Colour.brand_red(),
title=t(_p(
'cmd:pomodoro_create|new_channel|error:my_insufficient_perms|title',
"Could not create pomodoro voice channel!"
)),
description=t(_p(
'cmd:pomodoro_create|new_channel|error:my_insufficient_perms|desc',
"No `timer_channel` was provided, and I lack the 'Manage Channels' permission "
"required to create a new voice channel."
))
)
await ctx.reply(embed=embed, ephemeral=True)
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=name or "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
try:
channel = await ctx.guild.create_voice_channel(
name=name or t(_p(
'cmd:pomodoro_create|new_channel|default_name',
"Timer"
)),
reason=t(_p(
'cmd:pomodoro_create|new_channel|audit_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_permissions|title',
'cmd:pomodoro_create|new_channel|error:channel_create_failed|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."
'cmd:pomodoro_create|new_channel|error:channel_create_failed|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
# 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:
if not channel:
# Already handled the creation error
pass
elif (self.get_channel_timer(channel.id)) is not None:
# A timer already exists in the resolved channel
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)
'cmd:pomodoro_create|add_timer|error:timer_exists',
"A timer already exists in {channel}! "
"Reconfigure it with {edit_cmd}."
)).format(
channel=channel.mention,
edit_cmd=self.bot.core.mention_cmd('pomodoro edit')
)
)
await ctx.reply(embed=embed, ephemeral=True)
return
elif not channel.permissions_for(ctx.author).manage_channels:
# Note that this takes care of private room owners as well
embed = discord.Embed(
colour=discord.Colour.brand_red(),
description=t(_p(
'cmd:pomodoro_create|add_timer|error:your_insufficient_perms',
"You must have the 'Manage Channel' permission in {channel} "
"in order to add a timer there!"
))
)
await ctx.reply(embed=embed, ephemeral=True)
else:
# Finally, we are sure they can create a timer here
# Build the creation arguments from the rest of the provided args
provided = {
'focus_length': focus_length * 60,
'break_length': break_length * 60,
'inactivity_threshold': inactivity_threshold,
'voice_alerts': voice_alerts,
'name': name or channel.name,
'channel_name': channel_name or None,
}
create_args = {'channelid': channel.id, 'guildid': channel.guild.id}
# 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,
}
owned = (private_room and (channel == private_room))
if owned:
provided['manager_role'] = manager_role or ctx.guild.default_role
create_args['notification_channelid'] = channel.id
create_args['ownerid'] = ctx.author.id
else:
provided['notification_channel'] = notification_channel
provided['manager_role'] = manager_role
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)
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)
# Permission checks and input checking done
await ctx.interaction.response.defer(thinking=True)
# Create timer
timer = await self.create_timer(**create_args)
# Create timer
timer = await self.create_timer(**create_args)
# Start timer
await timer.start()
# 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()
# Ack with a config UI
ui = TimerOptionsUI(
self.bot, timer, TimerRole.ADMIN if not owned else TimerRole.OWNER, 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(
@pomodoro_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."
"Remove a pomodoro timer from a voice channel."
)
)
@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."),
channel=_p('cmd:pomodoro_destroy|param:channel', "Select a timer voice channel to remove the timer from."),
)
async def cmd_pomodoro_delete(self, ctx: LionContext, channel: discord.VoiceChannel):
t = self.bot.translator.t
@@ -646,46 +739,42 @@ class TimerCog(LionCog):
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:
if timer.owned and 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)
elif timer_role is not TimerRole.ADMIN and not channel.permissions_for(ctx.author).manage_channels:
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!"
))
"You need to have the `Manage Channels` permission in {channel} to remove this timer!"
)).format(channel=channel.mention)
)
await ctx.interaction.response.send_message(embed=embed, ephemeral=True)
return
else:
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)
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(
@pomodoro_group.command(
name=_p('cmd:pomodoro_edit', "edit"),
description=_p(
'cmd:pomodoro_edit|desc',
"Edit a Timer"
"Reconfigure a pomodoro timer."
)
)
@appcmds.rename(
@@ -695,7 +784,7 @@ class TimerCog(LionCog):
@appcmds.describe(
channel=_p(
'cmd:pomodoro_edit|param:channel|desc',
"Channel holding the timer to edit."
"Select a timer voice channel to reconfigure."
),
**{param: option._desc for param, (option, _) in _param_options.items()}
)
@@ -829,8 +918,6 @@ class TimerCog(LionCog):
@low_management_ward
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

62
src/modules/pomodoro/options.py
View File

@@ -4,6 +4,7 @@ import discord
from meta import LionBot
from meta.errors import UserInputError
from utils.lib import replace_multiple
from babel.translator import ctx_translator
from settings import ModelData
from settings.groups import SettingGroup, ModelConfig, SettingDotDict
@@ -37,6 +38,8 @@ class TimerOptions(SettingGroup):
_model = TimerData.Timer
_column = TimerData.Timer.channelid.name
_create_row = False
_allow_object = False
@TimerConfig.register_model_setting
class NotificationChannel(ModelData, ChannelSetting):
@@ -50,6 +53,8 @@ class TimerOptions(SettingGroup):
_model = TimerData.Timer
_column = TimerData.Timer.notification_channelid.name
_create_row = False
_allow_object = False
@classmethod
async def _check_value(cls, parent_id: int, value: Optional[discord.abc.GuildChannel], **kwargs):
@@ -86,6 +91,7 @@ class TimerOptions(SettingGroup):
)
_model = TimerData.Timer
_column = TimerData.Timer.inactivity_threshold.name
_create_row = False
_min = 0
_max = 64
@@ -94,6 +100,19 @@ class TimerOptions(SettingGroup):
def input_formatted(self):
return str(self._data) if self._data is not None else ''
@classmethod
async def _parse_string(cls, parent_id, string, **kwargs):
try:
return await super()._parse_string(parent_id, string, **kwargs)
except UserInputError:
t = ctx_translator.get().t
raise UserInputError(
t(_p(
'timerset:inactivity_length|desc',
"The inactivity threshold must be a positive whole number!"
))
)
@TimerConfig.register_model_setting
class ManagerRole(ModelData, RoleSetting):
setting_id = 'manager_role'
@@ -106,6 +125,8 @@ class TimerOptions(SettingGroup):
_model = TimerData.Timer
_column = TimerData.Timer.manager_roleid.name
_create_row = False
_allow_object = False
@classmethod
def _format_data(cls, parent_id, data, timer=None, **kwargs):
@@ -132,6 +153,7 @@ class TimerOptions(SettingGroup):
_model = TimerData.Timer
_column = TimerData.Timer.voice_alerts.name
_create_row = False
@TimerConfig.register_model_setting
class BaseName(ModelData, StringSetting):
@@ -153,6 +175,7 @@ class TimerOptions(SettingGroup):
_model = TimerData.Timer
_column = TimerData.Timer.pretty_name.name
_create_row = False
@TimerConfig.register_model_setting
class ChannelFormat(ModelData, StringSetting):
@@ -172,6 +195,43 @@ class TimerOptions(SettingGroup):
_model = TimerData.Timer
_column = TimerData.Timer.channel_name.name
_create_row = False
@classmethod
async def _parse_string(cls, parent_id, string, **kwargs):
# Enforce a length limit on a test-rendered string.
# TODO: Localised formatkey transformation
if string.lower() in ('', 'none', 'default'):
# Special cases for unsetting
return None
testmap = {
'{remaining}': "10m",
'{name}': "Longish name",
'{stage}': "FOCUS",
'{members}': "25",
'{pattern}': "50/10",
}
testmapped = replace_multiple(string, testmap)
if len(testmapped) > 100:
t = ctx_translator.get().t
raise UserInputError(
t(_p(
'timerset:channel_name_format|error:too_long',
"The provided name is too long! Channel names can be at most `100` characters."
))
)
else:
return string
@classmethod
def _format_data(cls, parent_id, data, **kwargs):
"""
Overriding format to truncate displayed string.
"""
if data is not None and len(data) > 100:
data = data[:97] + '...'
return super()._format_data(parent_id, data, **kwargs)
@TimerConfig.register_model_setting
class FocusLength(ModelData, DurationSetting):
@@ -191,6 +251,7 @@ class TimerOptions(SettingGroup):
_model = TimerData.Timer
_column = TimerData.Timer.focus_length.name
_create_row = False
_default_multiplier = 60
allow_zero = False
@@ -231,6 +292,7 @@ class TimerOptions(SettingGroup):
_model = TimerData.Timer
_column = TimerData.Timer.break_length.name
_create_row = False
_default_multiplier = 60
allow_zero = False

70
src/modules/pomodoro/timer.py
View File

@@ -12,6 +12,7 @@ from utils.lib import MessageArgs, utc_now, replace_multiple
from core.lion_guild import LionGuild
from core.data import CoreData
from babel.translator import ctx_locale
from gui.errors import RenderingException
from . import babel, logger
from .data import TimerData
@@ -45,6 +46,7 @@ class Timer:
'_voice_update_lock',
'_run_task',
'_loop_task',
'destroyed',
)
break_name = _p('timer|stage:break|name', "BREAK")
@@ -79,7 +81,10 @@ class Timer:
# Main loop task. Should not be cancelled.
self._loop_task = None
self.destroyed = False
def __repr__(self):
# TODO: Add lock status and current state and stage
return (
"<Timer "
f"channelid={self.data.channelid} "
@@ -403,7 +408,7 @@ class Timer:
"Remember to press {tick} to register your presence every stage.",
len(needs_kick)
), locale=self.locale.value).format(
channel=self.channel.mention,
channel=f"<#{self.data.channelid}>",
mentions=', '.join(member.mention for member in needs_kick),
tick=self.bot.config.emojis.tick
)
@@ -436,7 +441,7 @@ class Timer:
if not stage:
return
if not self.channel.permissions_for(self.guild.me).speak:
if not self.channel or not self.channel.permissions_for(self.guild.me).speak:
return
async with self.lguild.voice_lock:
@@ -498,7 +503,7 @@ class Timer:
"{channel} is now on **BREAK**! Take a rest, **FOCUS** starts {timestamp}"
)
stageline = t(lazy_stageline).format(
channel=self.channel.mention,
channel=f"<#{self.data.channelid}>",
timestamp=f"<t:{int(stage.end.timestamp())}:R>"
)
return stageline
@@ -555,29 +560,27 @@ class Timer:
content = t(_p(
'timer|status|stopped:auto',
"Timer stopped! Join {channel} to start the timer."
)).format(channel=self.channel.mention)
embed = None
)).format(channel=f"<#{self.data.channelid}>")
else:
content = t(_p(
'timer|status|stopped:manual',
"Timer stopped! Press `Start` to restart the timer."
)).format(channel=self.channel.mention)
embed = None
card = await get_timer_card(self.bot, self, stage)
await card.render()
)).format(channel=f"<#{self.data.channelid}>")
if (ui := self.status_view) is None:
ui = self.status_view = TimerStatusUI(self.bot, self, self.channel)
await ui.refresh()
return MessageArgs(
content=content,
embed=embed,
file=card.as_file(f"pomodoro_{self.data.channelid}.png"),
view=ui
)
card = await get_timer_card(self.bot, self, stage)
try:
await card.render()
file = card.as_file(f"pomodoro_{self.data.channelid}.png")
args = MessageArgs(content=content, file=file, view=ui)
except RenderingException:
args = MessageArgs(content=content, view=ui)
return args
@log_wrap(action='Send Timer Status')
async def send_status(self, delete_last=True, **kwargs):
@@ -676,6 +679,7 @@ class Timer:
if repost:
await self.send_status(delete_last=False, with_notify=False)
@log_wrap(action='Update Channel Name')
async def _update_channel_name(self):
"""
Submit a task to update the voice channel name.
@@ -683,15 +687,19 @@ class Timer:
Attempts to ensure that only one task is running at a time.
Attempts to wait until the next viable channel update slot (via ratelimit).
"""
if self._voice_update_task and not self._voice_update_task.done():
# Voice update request already submitted
if self._voice_update_lock.locked():
# Voice update is already running
# Note that if channel editing takes a long time,
# and the lock is waiting on that,
# we may actually miss a channel update in this period.
# Erring on the side of less ratelimits.
return
async with self._voice_update_lock:
if self._last_voice_update:
to_wait = ((self._last_voice_update + timedelta(minutes=5)) - utc_now()).total_seconds()
if to_wait > 0:
self._voice_update_task = asyncio.create_task(asyncio.sleep(to_wait))
self._voice_update_task = asyncio.create_task(asyncio.sleep(to_wait), name='timer-voice-wait')
try:
await self._voice_update_task
except asyncio.CancelledError:
@@ -706,8 +714,18 @@ class Timer:
if new_name == self.channel.name:
return
self._last_voice_update = utc_now()
await self.channel.edit(name=self.channel_name)
try:
logger.debug(f"Requesting channel name update for timer {self}")
await self.channel.edit(name=new_name)
except discord.HTTPException:
logger.warning(
f"Voice channel name update failed for timer {self}",
exc_info=True
)
finally:
# Whether we fail or not, update ratelimit marker
# (Repeatedly sending failing requests is even worse than normal ratelimits.)
self._last_voice_update = utc_now()
@log_wrap(action="Stop Timer")
async def stop(self, auto_restart=False):
@@ -736,7 +754,12 @@ class Timer:
if self._run_task and not self._run_task.done():
self._run_task.cancel()
channelid = self.data.channelid
if self.channel:
task = asyncio.create_task(
self.channel.edit(name=self.data.pretty_name, reason="Reverting timer channel name")
)
await self.data.delete()
self.destroyed = True
if self.last_status_message:
try:
await self.last_status_message.delete()
@@ -770,8 +793,8 @@ class Timer:
to_next_stage = (current.end - utc_now()).total_seconds()
# TODO: Consider request rate and load
if to_next_stage > 1 * 60 - drift:
time_to_sleep = 1 * 60
if to_next_stage > 5 * 60 - drift:
time_to_sleep = 5 * 60
else:
time_to_sleep = to_next_stage
@@ -795,6 +818,7 @@ class Timer:
if current.end < utc_now():
self._state = self.current_stage
task = asyncio.create_task(self.notify_change_stage(current, self._state))
background_tasks.add(task)
task.add_done_callback(background_tasks.discard)
current = self._state
elif self.members:

17
src/modules/pomodoro/ui/config.py
View File

@@ -37,6 +37,23 @@ class TimerOptionsUI(MessageUI):
self.timer = timer
self.role = role
async def interaction_check(self, interaction: discord.Interaction):
if self.timer.destroyed:
t = self.bot.translator.t
error = t(_p(
'ui:timer_options|error:timer_destroyed',
"This timer no longer exists! Closing option menu."
))
embed = discord.Embed(
colour=discord.Colour.brand_red(),
description=error
)
await interaction.response.send_message(embed=embed, ephemeral=True)
await self.quit()
return False
else:
return await super().interaction_check(interaction)
@button(label="EDIT_PLACEHOLDER", style=ButtonStyle.blurple)
async def edit_button(self, press: discord.Interaction, pressed: Button):
"""