Feature/capr 23 scaffold event cog from deprecated repo - #88
Merged
shamikkarkhanis merged 7 commits intoFeb 13, 2026
Merged
Conversation
…po"" This reverts commit 47da7de.
…o feature/capr-23-scaffold-event_cog-from-deprecated-repo
… are fixed instead of a new message being posted.
Contributor
Reviewer's GuideIntroduces a new Event cog with slash-command driven event management (create/edit/show/delete/list/announce/myevents) based on a Pydantic EventSchema, wires it into the project, and adds configuration for an announcement channel while temporarily disabling admin-only protection for the sync slash command. Sequence diagram for event creation via slash commandsequenceDiagram
actor User
participant DiscordClient
participant Bot
participant Event as EventCog
participant ModelModal
User->>DiscordClient: Invoke /event action=create
DiscordClient->>Bot: InteractionCreate
Bot->>Event: event(interaction, action=create)
Event->>Event: handle_create_action(interaction)
Event->>ModelModal: create ModelModal(EventSchema, _handle_event_submit)
Event->>DiscordClient: interaction.response.send_modal(ModelModal)
User->>DiscordClient: Submit modal with event fields
DiscordClient->>Bot: ModalSubmit interaction
Bot->>Event: ModelModal callback -> _handle_event_submit(interaction, event)
Event->>Event: validate and store event in events[guild_id]
Event->>DiscordClient: interaction.edit_original_response(embed=Event_Created)
DiscordClient->>User: Ephemeral confirmation with event details
Sequence diagram for announcing an event and RSVP trackingsequenceDiagram
actor User
participant DiscordClient
participant Bot
participant Event as EventCog
participant EventDropdownView
participant AnnouncementChannel as AnnouncementChannel
User->>DiscordClient: Invoke /event action=announce
DiscordClient->>Bot: InteractionCreate
Bot->>Event: event(interaction, action=announce)
Event->>Event: handle_announce_action(interaction)
Event->>Event: _get_events_for_dropdown(interaction, announce, _on_announce_select)
Event->>DiscordClient: interaction.response.defer(ephemeral)
Event->>EventDropdownView: create view with events
Event->>DiscordClient: followup.send(view=EventDropdownView)
User->>DiscordClient: Select event from dropdown
DiscordClient->>Bot: Component interaction
Bot->>EventDropdownView: on_select(interaction, selected_event)
EventDropdownView->>Event: _on_announce_select(interaction, selected_event)
Event->>Event: _get_announcement_channel(guild)
Event->>AnnouncementChannel: send(embed=announcement)
AnnouncementChannel-->>Event: message
Event->>AnnouncementChannel: add_reaction("✅")
Event->>AnnouncementChannel: add_reaction("❌")
Event->>Event: store message.id in event_announcements[guild_id][event_name]
Event->>DiscordClient: interaction.response.send_message(ephemeral success)
loop Later RSVP check
participant Member
Member->>DiscordClient: React ✅ on announcement
Note over Event,AnnouncementChannel: Reaction is stored by Discord
end
Note over Event,AnnouncementChannel: On /event action=myevents, Event
Note over Event,AnnouncementChannel: fetches stored message ids and
Note over Event,AnnouncementChannel: scans reactions to see registrations
Class diagram for new Event cog and related componentsclassDiagram
class EventSchema {
+str event_name
+date event_date
+time event_time
+str location
+str description
+_parse_event_date(value)
+_parse_event_time(value)
}
class Event {
+commands.Bot bot
+Logger log
+dict~int, list~EventSchema~~ events
+dict~int, dict~str, int~~ event_announcements
+event(interaction, action)
+handle_create_action(interaction)
+handle_edit_action(interaction)
+handle_show_action(interaction)
+handle_delete_action(interaction)
+handle_list_action(interaction)
+handle_announce_action(interaction)
+handle_myevents_action(interaction)
+_get_events_for_dropdown(interaction, action_name, callback)
+_event_datetime(event)
+_format_event_time_est(event)
+_format_when_where(event)
+_apply_event_fields(embed, event)
+_get_announcement_channel(guild)
+_is_user_registered(event, guild, user)
+_on_edit_select(interaction, selected_event)
+_on_announce_select(interaction, selected_event)
+_handle_event_submit(interaction, event)
+_create_event_embed(title, description, event)
+_handle_event_update(interaction, updated_event, original_event)
+_on_show_select(interaction, selected_event)
+_on_delete_select(interaction, selected_event)
}
class EventDropdownView {
+list~EventSchema~ event_list
+Event cog
+Callable on_select
+EventDropdownView(events, cog, placeholder, on_select_callback)
}
class EventDropdownSelect {
+EventDropdownView view_ref
+EventDropdownSelect(options, view, placeholder)
+callback(interaction)
}
class ConfirmDeleteView {
+bool value
+confirm(interaction, button)
+cancel(interaction, button)
}
class ModelModal {
+type model_cls
+Callable callback
+str title
+dict initial_data
}
class BaseView {
+int timeout
+disable_all_items()
+stop()
}
class Settings {
+int ticket_feedback_channel_id
+str announcement_channel_name
}
EventSchema <|-- Event
BaseView <|-- EventDropdownView
BaseView <|-- ConfirmDeleteView
ui.Select <|-- EventDropdownSelect
Event o--> EventSchema
EventDropdownView o--> EventSchema
EventDropdownView o--> Event
EventDropdownSelect o--> EventDropdownView
Event o--> EventDropdownView
Event o--> ConfirmDeleteView
Event o--> ModelModal
Event o--> Settings
File-Level Changes
Tips and commandsInteracting with Sourcery
Customizing Your ExperienceAccess your dashboard to:
Getting Help
|
simtiaz5
requested review from
Copilot and
shamikkarkhanis
and removed request for
Copilot
February 13, 2026 02:42
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Hey - I've found 3 issues, and left some high level feedback:
- The
@app_commands.checks.has_permissions(administrator=True)decorator on the/syncslash command is commented out, which effectively removes the admin restriction; if this is not purely for local debugging, consider restoring or replacing it with an appropriate permission check to avoid exposing sync to all users. - In
event_announcements, announcements are keyed byevent.event_name, so two events with the same name in a guild will collide and overwrite each other; consider using a more stable unique key (e.g., an event ID or a composite of name and datetime) to track announcement message IDs.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- The `@app_commands.checks.has_permissions(administrator=True)` decorator on the `/sync` slash command is commented out, which effectively removes the admin restriction; if this is not purely for local debugging, consider restoring or replacing it with an appropriate permission check to avoid exposing sync to all users.
- In `event_announcements`, announcements are keyed by `event.event_name`, so two events with the same name in a guild will collide and overwrite each other; consider using a more stable unique key (e.g., an event ID or a composite of name and datetime) to track announcement message IDs.
## Individual Comments
### Comment 1
<location> `capy_discord/exts/tools/sync.py:83` </location>
<code_context>
@app_commands.command(name="sync", description="Sync application commands")
- @app_commands.checks.has_permissions(administrator=True)
+ # @app_commands.checks.has_permissions(administrator=True)
async def sync_slash(self, interaction: discord.Interaction) -> None:
"""Sync commands via slash command."""
</code_context>
<issue_to_address>
**🚨 issue (security):** Commenting out the admin permission check on the sync slash command weakens access control.
This makes the sync command callable by any user with access to it, which can be disruptive, especially at scale. If this is for local debugging, please protect it with a narrower mechanism (e.g., specific user/guild IDs or an environment flag) instead of removing the admin permission check entirely.
</issue_to_address>
### Comment 2
<location> `capy_discord/exts/event/event.py:105-106` </location>
<code_context>
+ self.log.info("Event cog initialized")
+ # In-memory storage for demonstration.
+ self.events: dict[int, list[EventSchema]] = {}
+ # Track announcement messages: guild_id -> {event_name: message_id}
+ self.event_announcements: dict[int, dict[str, int]] = {}
+
+ @app_commands.command(name="event", description="Manage events")
</code_context>
<issue_to_address>
**suggestion (bug_risk):** Using `event_name` as the key for announcement messages can cause collisions between distinct events.
Because `event_announcements` is keyed by `event_name`, two events with the same name in a guild will overwrite each other’s announcement message ID, breaking RSVP tracking. Consider using a stable unique event identifier (e.g., DB ID) as the key, or supporting multiple announcements per name.
Suggested implementation:
```python
# In-memory storage for demonstration.
self.events: dict[int, list[EventSchema]] = {}
# Track announcement messages: guild_id -> {event_id: message_id}
self.event_announcements: dict[int, dict[int, int]] = {}
```
To fully implement the suggestion, you will also need to:
1. Update all references to `self.event_announcements` elsewhere in this cog:
- Replace any lookups like `self.event_announcements[guild_id].get(event_name)` or assignments to `self.event_announcements[guild_id][event_name]` with versions that use a unique event ID, e.g. `event.id`.
- Ensure that when you announce an event, you store the announcement as `self.event_announcements[guild_id][event.id] = message_id`.
- When handling RSVP interactions or updates, resolve the event by ID (e.g. from a custom_id or metadata) and then fetch the message ID via `self.event_announcements[guild_id].get(event.id)`.
2. If you currently only have the event name available where you need to access `event_announcements`, add logic to:
- Find the correct `EventSchema` instance from `self.events[guild_id]` by name, and
- Use that instance’s unique ID as the key into `self.event_announcements[guild_id]`.
3. If `EventSchema` does not yet expose a unique, stable identifier (e.g. `id: int` from the DB), you should add such a field and populate it when events are created so that it can be safely used as the key.
</issue_to_address>
### Comment 3
<location> `capy_discord/exts/event/event.py:162` </location>
<code_context>
+ """Handle event deletion."""
+ await self._get_events_for_dropdown(interaction, "delete", self._on_delete_select)
+
+ async def handle_list_action(self, interaction: discord.Interaction) -> None:
+ """Handle listing all events."""
+ guild_id = interaction.guild_id
</code_context>
<issue_to_address>
**issue (complexity):** Consider extracting shared event-listing and RSVP helper functions to centralize filtering/sorting logic and simplify `_is_user_registered` into smaller composable pieces.
You can reduce a fair bit of complexity and duplication without changing behavior by extracting a couple of helpers and simplifying some types.
### 1. Factor out shared “list events” logic
`handle_list_action` and `handle_myevents_action` both do:
- fetch events
- partition/filter by time
- sort
- build an embed using `_format_when_where`
You can centralize that so each handler just focuses on its specific filter.
```python
# inside Event
def _get_guild_events(self, interaction: discord.Interaction) -> list[EventSchema] | None:
guild_id = interaction.guild_id
if not guild_id:
embed = error_embed("No Server", "Events must be used in a server.")
# caller is responsible for choosing response vs followup
if not interaction.response.is_done():
await interaction.response.send_message(embed=embed, ephemeral=True)
else:
await interaction.followup.send(embed=embed, ephemeral=True)
return None
events = self.events.get(guild_id, [])
if not events:
embed = error_embed("No Events", "No events found in this server.")
# same response rule as above
if not interaction.response.is_done():
await interaction.response.send_message(embed=embed, ephemeral=True)
else:
await interaction.followup.send(embed=embed, ephemeral=True)
return None
return events
def _build_events_list_embed(
self,
title: str,
description: str,
events: list[EventSchema],
prefix_old_for_past: bool = False,
) -> discord.Embed:
now = datetime.now(ZoneInfo("UTC"))
upcoming: list[EventSchema] = []
past: list[EventSchema] = []
for event in events:
event_time = self._event_datetime(event)
(upcoming if event_time >= now else past).append(event)
upcoming.sort(key=self._event_datetime)
past.sort(key=self._event_datetime, reverse=True)
total_count = len(upcoming) + len(past)
embed = success_embed(
title,
f"{description}\nFound {total_count} events (Upcoming: {len(upcoming)}, Past: {len(past)})",
)
for event in upcoming:
embed.add_field(
name=event.event_name,
value=self._format_when_where(event),
inline=False,
)
if prefix_old_for_past:
for event in past:
embed.add_field(
name=f"[OLD] {event.event_name}",
value=self._format_when_where(event),
inline=False,
)
else:
for event in past:
embed.add_field(
name=event.event_name,
value=self._format_when_where(event),
inline=False,
)
return embed
```
Then your handlers become much shorter:
```python
async def handle_list_action(self, interaction: discord.Interaction) -> None:
events = self.events.get(interaction.guild_id or 0, [])
if not events:
embed = error_embed("No Events", "No events found in this server.")
await interaction.response.send_message(embed=embed, ephemeral=True)
return
self.log.info("Listing events for guild %s", interaction.guild_id)
await interaction.response.defer(ephemeral=True)
embed = self._build_events_list_embed(
"Events",
"All events in this server.",
events,
prefix_old_for_past=True,
)
await interaction.followup.send(embed=embed, ephemeral=True)
async def handle_myevents_action(self, interaction: discord.Interaction) -> None:
guild = interaction.guild
if not guild:
embed = error_embed("No Server", "Events must be viewed in a server.")
await interaction.response.send_message(embed=embed, ephemeral=True)
return
events = self.events.get(guild.id, [])
if not events:
embed = error_embed("No Events", "No events found in this server.")
await interaction.response.send_message(embed=embed, ephemeral=True)
return
self.log.info("Listing registered events for user %s", interaction.user)
await interaction.response.defer(ephemeral=True)
now = datetime.now(ZoneInfo("UTC"))
upcoming = [
event
for event in events
if self._event_datetime(event) >= now
and await self._is_user_registered(event, guild, interaction.user)
]
upcoming.sort(key=self._event_datetime)
embed = success_embed(
"Your Registered Events",
"Events you have registered for by reacting with ✅",
)
if not upcoming:
embed.description = (
"You haven't registered for any upcoming events.\n"
"React to event announcements with ✅ to register!"
)
else:
for event in upcoming:
embed.add_field(
name=event.event_name,
value=self._format_when_where(event),
inline=False,
)
await interaction.followup.send(embed=embed, ephemeral=True)
```
This keeps behavior identical but removes the duplicated filtering/sorting/field‑building sequences.
### 2. Split `_is_user_registered` into smaller helpers
The RSVP check is doing several different things. Splitting it into small, composable helpers makes the control flow much easier to follow:
```python
# inside Event
async def _get_announcement_message_for_event(
self,
event: EventSchema,
guild: discord.Guild,
) -> discord.Message | None:
guild_announcements = self.event_announcements.get(guild.id, {})
message_id = guild_announcements.get(event.event_name)
if not message_id:
return None
channel = self._get_announcement_channel(guild)
if not channel:
return None
try:
return await channel.fetch_message(message_id)
except (discord.NotFound, discord.Forbidden, discord.HTTPException):
self.log.warning("Could not fetch announcement message %s", message_id)
return None
async def _users_with_reaction(
self,
message: discord.Message,
emoji: str,
) -> list[discord.abc.User]:
for reaction in message.reactions:
if str(reaction.emoji) == emoji:
return [user async for user in reaction.users()]
return []
async def _is_user_registered(
self, event: EventSchema, guild: discord.Guild, user: discord.User | discord.Member
) -> bool:
message = await self._get_announcement_message_for_event(event, guild)
if not message:
return False
users = await self._users_with_reaction(message, "✅")
return user in users
```
The public semantics of `_is_user_registered` stay unchanged, but the responsibilities (lookup, fetch, reaction scan) are now individually understandable and reusable.
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
|
|
||
| @app_commands.command(name="sync", description="Sync application commands") | ||
| @app_commands.checks.has_permissions(administrator=True) | ||
| # @app_commands.checks.has_permissions(administrator=True) |
Contributor
There was a problem hiding this comment.
🚨 issue (security): Commenting out the admin permission check on the sync slash command weakens access control.
This makes the sync command callable by any user with access to it, which can be disruptive, especially at scale. If this is for local debugging, please protect it with a narrower mechanism (e.g., specific user/guild IDs or an environment flag) instead of removing the admin permission check entirely.
Comment on lines
+105
to
+106
| # Track announcement messages: guild_id -> {event_name: message_id} | ||
| self.event_announcements: dict[int, dict[str, int]] = {} |
Contributor
There was a problem hiding this comment.
suggestion (bug_risk): Using event_name as the key for announcement messages can cause collisions between distinct events.
Because event_announcements is keyed by event_name, two events with the same name in a guild will overwrite each other’s announcement message ID, breaking RSVP tracking. Consider using a stable unique event identifier (e.g., DB ID) as the key, or supporting multiple announcements per name.
Suggested implementation:
# In-memory storage for demonstration.
self.events: dict[int, list[EventSchema]] = {}
# Track announcement messages: guild_id -> {event_id: message_id}
self.event_announcements: dict[int, dict[int, int]] = {}To fully implement the suggestion, you will also need to:
- Update all references to
self.event_announcementselsewhere in this cog:- Replace any lookups like
self.event_announcements[guild_id].get(event_name)or assignments toself.event_announcements[guild_id][event_name]with versions that use a unique event ID, e.g.event.id. - Ensure that when you announce an event, you store the announcement as
self.event_announcements[guild_id][event.id] = message_id. - When handling RSVP interactions or updates, resolve the event by ID (e.g. from a custom_id or metadata) and then fetch the message ID via
self.event_announcements[guild_id].get(event.id).
- Replace any lookups like
- If you currently only have the event name available where you need to access
event_announcements, add logic to:- Find the correct
EventSchemainstance fromself.events[guild_id]by name, and - Use that instance’s unique ID as the key into
self.event_announcements[guild_id].
- Find the correct
- If
EventSchemadoes not yet expose a unique, stable identifier (e.g.id: intfrom the DB), you should add such a field and populate it when events are created so that it can be safely used as the key.
| """Handle event deletion.""" | ||
| await self._get_events_for_dropdown(interaction, "delete", self._on_delete_select) | ||
|
|
||
| async def handle_list_action(self, interaction: discord.Interaction) -> None: |
Contributor
There was a problem hiding this comment.
issue (complexity): Consider extracting shared event-listing and RSVP helper functions to centralize filtering/sorting logic and simplify _is_user_registered into smaller composable pieces.
You can reduce a fair bit of complexity and duplication without changing behavior by extracting a couple of helpers and simplifying some types.
1. Factor out shared “list events” logic
handle_list_action and handle_myevents_action both do:
- fetch events
- partition/filter by time
- sort
- build an embed using
_format_when_where
You can centralize that so each handler just focuses on its specific filter.
# inside Event
def _get_guild_events(self, interaction: discord.Interaction) -> list[EventSchema] | None:
guild_id = interaction.guild_id
if not guild_id:
embed = error_embed("No Server", "Events must be used in a server.")
# caller is responsible for choosing response vs followup
if not interaction.response.is_done():
await interaction.response.send_message(embed=embed, ephemeral=True)
else:
await interaction.followup.send(embed=embed, ephemeral=True)
return None
events = self.events.get(guild_id, [])
if not events:
embed = error_embed("No Events", "No events found in this server.")
# same response rule as above
if not interaction.response.is_done():
await interaction.response.send_message(embed=embed, ephemeral=True)
else:
await interaction.followup.send(embed=embed, ephemeral=True)
return None
return events
def _build_events_list_embed(
self,
title: str,
description: str,
events: list[EventSchema],
prefix_old_for_past: bool = False,
) -> discord.Embed:
now = datetime.now(ZoneInfo("UTC"))
upcoming: list[EventSchema] = []
past: list[EventSchema] = []
for event in events:
event_time = self._event_datetime(event)
(upcoming if event_time >= now else past).append(event)
upcoming.sort(key=self._event_datetime)
past.sort(key=self._event_datetime, reverse=True)
total_count = len(upcoming) + len(past)
embed = success_embed(
title,
f"{description}\nFound {total_count} events (Upcoming: {len(upcoming)}, Past: {len(past)})",
)
for event in upcoming:
embed.add_field(
name=event.event_name,
value=self._format_when_where(event),
inline=False,
)
if prefix_old_for_past:
for event in past:
embed.add_field(
name=f"[OLD] {event.event_name}",
value=self._format_when_where(event),
inline=False,
)
else:
for event in past:
embed.add_field(
name=event.event_name,
value=self._format_when_where(event),
inline=False,
)
return embedThen your handlers become much shorter:
async def handle_list_action(self, interaction: discord.Interaction) -> None:
events = self.events.get(interaction.guild_id or 0, [])
if not events:
embed = error_embed("No Events", "No events found in this server.")
await interaction.response.send_message(embed=embed, ephemeral=True)
return
self.log.info("Listing events for guild %s", interaction.guild_id)
await interaction.response.defer(ephemeral=True)
embed = self._build_events_list_embed(
"Events",
"All events in this server.",
events,
prefix_old_for_past=True,
)
await interaction.followup.send(embed=embed, ephemeral=True)
async def handle_myevents_action(self, interaction: discord.Interaction) -> None:
guild = interaction.guild
if not guild:
embed = error_embed("No Server", "Events must be viewed in a server.")
await interaction.response.send_message(embed=embed, ephemeral=True)
return
events = self.events.get(guild.id, [])
if not events:
embed = error_embed("No Events", "No events found in this server.")
await interaction.response.send_message(embed=embed, ephemeral=True)
return
self.log.info("Listing registered events for user %s", interaction.user)
await interaction.response.defer(ephemeral=True)
now = datetime.now(ZoneInfo("UTC"))
upcoming = [
event
for event in events
if self._event_datetime(event) >= now
and await self._is_user_registered(event, guild, interaction.user)
]
upcoming.sort(key=self._event_datetime)
embed = success_embed(
"Your Registered Events",
"Events you have registered for by reacting with ✅",
)
if not upcoming:
embed.description = (
"You haven't registered for any upcoming events.\n"
"React to event announcements with ✅ to register!"
)
else:
for event in upcoming:
embed.add_field(
name=event.event_name,
value=self._format_when_where(event),
inline=False,
)
await interaction.followup.send(embed=embed, ephemeral=True)This keeps behavior identical but removes the duplicated filtering/sorting/field‑building sequences.
2. Split _is_user_registered into smaller helpers
The RSVP check is doing several different things. Splitting it into small, composable helpers makes the control flow much easier to follow:
# inside Event
async def _get_announcement_message_for_event(
self,
event: EventSchema,
guild: discord.Guild,
) -> discord.Message | None:
guild_announcements = self.event_announcements.get(guild.id, {})
message_id = guild_announcements.get(event.event_name)
if not message_id:
return None
channel = self._get_announcement_channel(guild)
if not channel:
return None
try:
return await channel.fetch_message(message_id)
except (discord.NotFound, discord.Forbidden, discord.HTTPException):
self.log.warning("Could not fetch announcement message %s", message_id)
return None
async def _users_with_reaction(
self,
message: discord.Message,
emoji: str,
) -> list[discord.abc.User]:
for reaction in message.reactions:
if str(reaction.emoji) == emoji:
return [user async for user in reaction.users()]
return []
async def _is_user_registered(
self, event: EventSchema, guild: discord.Guild, user: discord.User | discord.Member
) -> bool:
message = await self._get_announcement_message_for_event(event, guild)
if not message:
return False
users = await self._users_with_reaction(message, "✅")
return user in usersThe public semantics of _is_user_registered stay unchanged, but the responsibilities (lookup, fetch, reaction scan) are now individually understandable and reusable.
There was a problem hiding this comment.
Pull request overview
This PR scaffolds a new event management cog from a deprecated repository, enabling Discord servers to create, edit, list, delete, and announce events with RSVP tracking. It also includes a security-related change to the sync command that removes administrator permission requirements.
Changes:
- New event management system with CRUD operations, announcement functionality, and RSVP tracking via reactions
- Removed administrator permission check from the sync slash command (security concern)
- Added announcement channel configuration setting
Reviewed changes
Copilot reviewed 4 out of 4 changed files in this pull request and generated 17 comments.
Show a summary per file
| File | Description |
|---|---|
| capy_discord/exts/tools/sync.py | Commented out administrator permission check for the sync slash command |
| capy_discord/exts/event/event.py | New event management cog with views, dropdowns, and handlers for event CRUD operations, announcements, and RSVP tracking |
| capy_discord/exts/event/_schemas.py | Pydantic schema for event validation with custom date/time parsing |
| capy_discord/exts/event/init.py | Module initialization with docstring |
| capy_discord/config.py | Added announcement_channel_name configuration setting |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| def __init__(self) -> None: | ||
| """Initialize the ConfirmDeleteView.""" | ||
| super().__init__(timeout=60) | ||
| self.value: bool | None = None |
There was a problem hiding this comment.
The ConfirmDeleteView has inconsistent initialization of the value attribute compared to the pattern in profile.py. In profile.py (line 22), it's initialized as self.value = None without a type annotation, whereas here it's self.value: bool | None = None with a type annotation. While the type annotation is actually better practice, consider applying this pattern consistently across the codebase for maintainability.
Suggested change
| self.value: bool | None = None | |
| self.value = None |
| self.log = logging.getLogger(__name__) | ||
| self.log.info("Event cog initialized") | ||
| # In-memory storage for demonstration. | ||
| self.events: dict[int, list[EventSchema]] = {} |
There was a problem hiding this comment.
The in-memory event storage uses mutable lists in a dictionary without synchronization. In concurrent scenarios (multiple users editing/deleting events simultaneously), this could lead to race conditions. For example, if two users delete the same event concurrently, the second operation might fail or cause unexpected behavior. Consider using thread-safe data structures or adding locking mechanisms, especially since this will eventually be replaced with a database that will need proper transaction handling.
Comment on lines
+408
to
+410
| users = [user async for user in reaction.users()] | ||
| if user in users: | ||
| return True |
There was a problem hiding this comment.
The RSVP checking logic fetches all users who reacted with ✅ and checks if the querying user is in that list. For events with many attendees, this could be inefficient as it fetches all reacting users into memory. Discord.py's reaction.users() returns an async iterator that could be used more efficiently by checking each user as they're fetched, allowing early exit when a match is found. Consider using: async for u in reaction.users(): if u.id == user.id: return True
Suggested change
| users = [user async for user in reaction.users()] | |
| if user in users: | |
| return True | |
| async for reacting_user in reaction.users(): | |
| if reacting_user.id == user.id: | |
| return True |
Comment on lines
+95
to
+629
| class Event(commands.Cog): | ||
| """Cog for event-related commands.""" | ||
|
|
||
| def __init__(self, bot: commands.Bot) -> None: | ||
| """Initialize the Event cog.""" | ||
| self.bot = bot | ||
| self.log = logging.getLogger(__name__) | ||
| self.log.info("Event cog initialized") | ||
| # In-memory storage for demonstration. | ||
| self.events: dict[int, list[EventSchema]] = {} | ||
| # Track announcement messages: guild_id -> {event_name: message_id} | ||
| self.event_announcements: dict[int, dict[str, int]] = {} | ||
|
|
||
| @app_commands.command(name="event", description="Manage events") | ||
| @app_commands.describe(action="The action to perform with events") | ||
| @app_commands.choices( | ||
| action=[ | ||
| app_commands.Choice(name="create", value="create"), | ||
| app_commands.Choice(name="edit", value="edit"), | ||
| app_commands.Choice(name="show", value="show"), | ||
| app_commands.Choice(name="delete", value="delete"), | ||
| app_commands.Choice(name="list", value="list"), | ||
| app_commands.Choice(name="announce", value="announce"), | ||
| app_commands.Choice(name="myevents", value="myevents"), | ||
| ] | ||
| ) | ||
| async def event(self, interaction: discord.Interaction, action: app_commands.Choice[str]) -> None: | ||
| """Manage events based on the action specified.""" | ||
| match action.value: | ||
| case "create": | ||
| await self.handle_create_action(interaction) | ||
| case "edit": | ||
| await self.handle_edit_action(interaction) | ||
| case "show": | ||
| await self.handle_show_action(interaction) | ||
| case "delete": | ||
| await self.handle_delete_action(interaction) | ||
| case "list": | ||
| await self.handle_list_action(interaction) | ||
| case "announce": | ||
| await self.handle_announce_action(interaction) | ||
| case "myevents": | ||
| await self.handle_myevents_action(interaction) | ||
|
|
||
| async def handle_create_action(self, interaction: discord.Interaction) -> None: | ||
| """Handle event creation.""" | ||
| self.log.info("Opening event creation modal for %s", interaction.user) | ||
|
|
||
| modal = ModelModal( | ||
| model_cls=EventSchema, | ||
| callback=self._handle_event_submit, | ||
| title="Create Event", | ||
| ) | ||
| await interaction.response.send_modal(modal) | ||
|
|
||
| async def handle_edit_action(self, interaction: discord.Interaction) -> None: | ||
| """Handle event editing.""" | ||
| await self._get_events_for_dropdown(interaction, "edit", self._on_edit_select) | ||
|
|
||
| async def handle_show_action(self, interaction: discord.Interaction) -> None: | ||
| """Handle showing event details.""" | ||
| await self._get_events_for_dropdown(interaction, "view", self._on_show_select) | ||
|
|
||
| async def handle_delete_action(self, interaction: discord.Interaction) -> None: | ||
| """Handle event deletion.""" | ||
| await self._get_events_for_dropdown(interaction, "delete", self._on_delete_select) | ||
|
|
||
| async def handle_list_action(self, interaction: discord.Interaction) -> None: | ||
| """Handle listing all events.""" | ||
| guild_id = interaction.guild_id | ||
| if not guild_id: | ||
| embed = error_embed("No Server", "Events must be listed in a server.") | ||
| await interaction.response.send_message(embed=embed, ephemeral=True) | ||
| return | ||
|
|
||
| # [DB CALL]: Fetch guild events | ||
| events = self.events.get(guild_id, []) | ||
|
|
||
| if not events: | ||
| embed = error_embed("No Events", "No events found in this server.") | ||
| await interaction.response.send_message(embed=embed, ephemeral=True) | ||
| return | ||
|
|
||
| self.log.info("Listing events for guild %s", guild_id) | ||
|
|
||
| await interaction.response.defer(ephemeral=True) | ||
|
|
||
| # Separate into upcoming and past events | ||
| now = datetime.now(ZoneInfo("UTC")) | ||
| upcoming_events: list[EventSchema] = [] | ||
| past_events: list[EventSchema] = [] | ||
|
|
||
| for event in events: | ||
| event_time = self._event_datetime(event) | ||
|
|
||
| if event_time >= now: | ||
| upcoming_events.append(event) | ||
| else: | ||
| past_events.append(event) | ||
|
|
||
| # Sort events | ||
| upcoming_events.sort(key=self._event_datetime) | ||
| past_events.sort(key=self._event_datetime, reverse=True) | ||
|
|
||
| # Build embed | ||
| total_count = len(upcoming_events) + len(past_events) | ||
| embed = success_embed( | ||
| "Events", | ||
| f"Found {total_count} events (Upcoming: {len(upcoming_events)}, Past: {len(past_events)})", | ||
| ) | ||
|
|
||
| # Add upcoming events | ||
| for event in upcoming_events: | ||
| embed.add_field( | ||
| name=event.event_name, | ||
| value=self._format_when_where(event), | ||
| inline=False, | ||
| ) | ||
|
|
||
| # Add past events with [OLD] prefix | ||
| for event in past_events: | ||
| embed.add_field( | ||
| name=f"[OLD] {event.event_name}", | ||
| value=self._format_when_where(event), | ||
| inline=False, | ||
| ) | ||
|
|
||
| await interaction.followup.send(embed=embed, ephemeral=True) | ||
|
|
||
| async def handle_announce_action(self, interaction: discord.Interaction) -> None: | ||
| """Handle announcing an event and user registrations.""" | ||
| await self._get_events_for_dropdown(interaction, "announce", self._on_announce_select) | ||
|
|
||
| async def handle_myevents_action(self, interaction: discord.Interaction) -> None: | ||
| """Handle showing events the user has registered for via RSVP.""" | ||
| guild_id = interaction.guild_id | ||
| guild = interaction.guild | ||
| if not guild_id or not guild: | ||
| embed = error_embed("No Server", "Events must be viewed in a server.") | ||
| await interaction.response.send_message(embed=embed, ephemeral=True) | ||
| return | ||
|
|
||
| # [DB CALL]: Fetch guild events | ||
| events = self.events.get(guild_id, []) | ||
|
|
||
| if not events: | ||
| embed = error_embed("No Events", "No events found in this server.") | ||
| await interaction.response.send_message(embed=embed, ephemeral=True) | ||
| return | ||
|
|
||
| self.log.info("Listing registered events for user %s", interaction.user) | ||
|
|
||
| await interaction.response.defer(ephemeral=True) | ||
|
|
||
| # Get upcoming events the user has registered for | ||
| now = datetime.now(ZoneInfo("UTC")) | ||
| registered_events: list[EventSchema] = [] | ||
|
|
||
| for event in events: | ||
| event_time = self._event_datetime(event) | ||
|
|
||
| # Only include upcoming events | ||
| if event_time < now: | ||
| continue | ||
|
|
||
| # Check if user has registered for this event | ||
| if await self._is_user_registered(event, guild, interaction.user): | ||
| registered_events.append(event) | ||
|
|
||
| registered_events.sort(key=self._event_datetime) | ||
|
|
||
| # Build embed | ||
| embed = success_embed( | ||
| "Your Registered Events", | ||
| "Events you have registered for by reacting with ✅", | ||
| ) | ||
|
|
||
| if not registered_events: | ||
| embed.description = ( | ||
| "You haven't registered for any upcoming events.\nReact to event announcements with ✅ to register!" | ||
| ) | ||
| await interaction.followup.send(embed=embed, ephemeral=True) | ||
| return | ||
|
|
||
| # Add registered events | ||
| for event in registered_events: | ||
| embed.add_field( | ||
| name=event.event_name, | ||
| value=self._format_when_where(event), | ||
| inline=False, | ||
| ) | ||
|
|
||
| await interaction.followup.send(embed=embed, ephemeral=True) | ||
|
|
||
| async def _get_events_for_dropdown( | ||
| self, | ||
| interaction: discord.Interaction, | ||
| action_name: str, | ||
| callback: Callable[[discord.Interaction, EventSchema], Coroutine[Any, Any, None]], | ||
| ) -> None: | ||
| """Generic handler to get events and show dropdown for selection. | ||
|
|
||
| Args: | ||
| interaction: The Discord interaction. | ||
| action_name: Name of the action (e.g., "edit", "view", "delete"). | ||
| callback: Async callback to handle the selected event. | ||
| """ | ||
| guild_id = interaction.guild_id | ||
| if not guild_id: | ||
| embed = error_embed("No Server", f"Events must be {action_name}ed in a server.") | ||
| await interaction.response.send_message(embed=embed, ephemeral=True) | ||
| return | ||
|
|
||
| # [DB CALL]: Fetch guild events | ||
| events = self.events.get(guild_id, []) | ||
|
|
||
| if not events: | ||
| embed = error_embed("No Events", f"No events found in this server to {action_name}.") | ||
| await interaction.response.send_message(embed=embed, ephemeral=True) | ||
| return | ||
|
|
||
| self.log.info("Opening event selection for %s in guild %s", action_name, guild_id) | ||
|
|
||
| await interaction.response.defer(ephemeral=True) | ||
|
|
||
| view = EventDropdownView(events, self, f"Select an event to {action_name}", callback) | ||
| await interaction.followup.send(content=f"Select an event to {action_name}:", view=view, ephemeral=True) | ||
|
|
||
| await view.wait() | ||
|
|
||
| @staticmethod | ||
| def _event_datetime(event: EventSchema) -> datetime: | ||
| """Convert event date and time to a timezone-aware datetime in UTC. | ||
|
|
||
| User input is treated as EST, then converted to UTC for storage. | ||
|
|
||
| Args: | ||
| event: The event containing date and time information. | ||
|
|
||
| Returns: | ||
| A UTC timezone-aware datetime object. | ||
| """ | ||
| est = ZoneInfo("America/New_York") | ||
| event_time = datetime.combine(event.event_date, event.event_time) | ||
| # Treat user input as EST | ||
| if event_time.tzinfo is None: | ||
| event_time = event_time.replace(tzinfo=est) | ||
| # Convert to UTC for storage | ||
| return event_time.astimezone(ZoneInfo("UTC")) | ||
|
|
||
| def _format_event_time_est(self, event: EventSchema) -> str: | ||
| """Format an event's date/time in EST for user-facing display.""" | ||
| event_dt_est = self._event_datetime(event).astimezone(ZoneInfo("America/New_York")) | ||
| return event_dt_est.strftime("%B %d, %Y at %I:%M %p EST") | ||
|
|
||
| def _format_when_where(self, event: EventSchema) -> str: | ||
| """Format the when/where field for embeds.""" | ||
| time_str = self._format_event_time_est(event) | ||
| return f"**When:** {time_str}\n**Where:** {event.location or 'TBD'}" | ||
|
|
||
| def _apply_event_fields(self, embed: discord.Embed, event: EventSchema) -> None: | ||
| """Append event detail fields to an embed.""" | ||
| embed.add_field(name="Event", value=event.event_name, inline=False) | ||
| embed.add_field(name="Date/Time", value=self._format_event_time_est(event), inline=True) | ||
| embed.add_field(name="Location", value=event.location or "TBD", inline=True) | ||
| if event.description: | ||
| embed.add_field(name="Description", value=event.description, inline=False) | ||
|
|
||
| def _get_announcement_channel(self, guild: discord.Guild) -> discord.TextChannel | None: | ||
| """Get the announcement channel from config name. | ||
|
|
||
| Args: | ||
| guild: The guild to search for the announcement channel. | ||
|
|
||
| Returns: | ||
| The announcement channel if found, None otherwise. | ||
| """ | ||
| for channel in guild.text_channels: | ||
| if channel.name.lower() == settings.announcement_channel_name.lower(): | ||
| return channel | ||
| return None | ||
|
|
||
| async def _is_user_registered( | ||
| self, event: EventSchema, guild: discord.Guild, user: discord.User | discord.Member | ||
| ) -> bool: | ||
| """Check if a user has registered for an event via RSVP reaction. | ||
|
|
||
| Args: | ||
| event: The event to check registration for. | ||
| guild: The guild where the event was announced. | ||
| user: The user to check registration for. | ||
|
|
||
| Returns: | ||
| True if the user has reacted with ✅ to the event announcement, False otherwise. | ||
| """ | ||
| # Get announcement messages for this guild | ||
| guild_announcements = self.event_announcements.get(guild.id, {}) | ||
| message_id = guild_announcements.get(event.event_name) | ||
|
|
||
| if not message_id: | ||
| return False | ||
|
|
||
| # Try to find the announcement message and check reactions | ||
| announcement_channel = self._get_announcement_channel(guild) | ||
|
|
||
| if not announcement_channel: | ||
| return False | ||
|
|
||
| try: | ||
| message = await announcement_channel.fetch_message(message_id) | ||
| # Check if user reacted with ✅ | ||
| for reaction in message.reactions: | ||
| if str(reaction.emoji) == "✅": | ||
| users = [user async for user in reaction.users()] | ||
| if user in users: | ||
| return True | ||
| except (discord.NotFound, discord.Forbidden, discord.HTTPException): | ||
| # Message not found or no permission - skip this event | ||
| self.log.warning("Could not fetch announcement message %s", message_id) | ||
| return False | ||
|
|
||
| return False | ||
|
|
||
| async def _on_edit_select(self, interaction: discord.Interaction, selected_event: EventSchema) -> None: | ||
| """Handle event selection for editing.""" | ||
| initial_data = { | ||
| "event_name": selected_event.event_name, | ||
| "event_date": selected_event.event_date.strftime("%m-%d-%Y"), | ||
| "event_time": selected_event.event_time.strftime("%H:%M"), | ||
| "location": selected_event.location, | ||
| "description": selected_event.description, | ||
| } | ||
|
|
||
| self.log.info("Opening edit modal for event '%s'", selected_event.event_name) | ||
|
|
||
| modal = ModelModal( | ||
| model_cls=EventSchema, | ||
| callback=lambda modal_interaction, event: self._handle_event_update( | ||
| modal_interaction, event, selected_event | ||
| ), | ||
| title="Edit Event", | ||
| initial_data=initial_data, | ||
| ) | ||
| await interaction.response.send_modal(modal) | ||
|
|
||
| async def _on_announce_select(self, interaction: discord.Interaction, selected_event: EventSchema) -> None: | ||
| """Handle event selection for announcement.""" | ||
| guild = interaction.guild | ||
| if not guild: | ||
| embed = error_embed("No Server", "Cannot determine server.") | ||
| await interaction.response.send_message(embed=embed, ephemeral=True) | ||
| return | ||
|
|
||
| # Get the announcement channel | ||
| announcement_channel = self._get_announcement_channel(guild) | ||
|
|
||
| if not announcement_channel: | ||
| embed = error_embed( | ||
| "No Announcement Channel", | ||
| f"Could not find a channel named '{settings.announcement_channel_name}'. " | ||
| "Please rename or create an announcement channel.", | ||
| ) | ||
| await interaction.response.send_message(embed=embed, ephemeral=True) | ||
| return | ||
|
|
||
| # Check if bot has permission to post in the channel | ||
| if not announcement_channel.permissions_for(guild.me).send_messages: | ||
| embed = error_embed( | ||
| "No Permission", | ||
| "I don't have permission to send messages in the announcement channel.", | ||
| ) | ||
| await interaction.response.send_message(embed=embed, ephemeral=True) | ||
| return | ||
|
|
||
| try: | ||
| # Create announcement embed | ||
| announce_embed = self._create_announcement_embed(selected_event) | ||
|
|
||
| # Post to announcement channel | ||
| message = await announcement_channel.send(embed=announce_embed) | ||
|
|
||
| # Add RSVP reactions | ||
| await message.add_reaction("✅") # Attending | ||
| await message.add_reaction("❌") # Not attending | ||
|
|
||
| # [DB CALL]: Store announcement message ID for RSVP tracking | ||
| if guild.id not in self.event_announcements: | ||
| self.event_announcements[guild.id] = {} | ||
| self.event_announcements[guild.id][selected_event.event_name] = message.id | ||
|
|
||
| self.log.info( | ||
| "Announced event '%s' to guild %s in channel %s", | ||
| selected_event.event_name, | ||
| guild.id, | ||
| announcement_channel.name, | ||
| ) | ||
|
|
||
| success = success_embed( | ||
| "Event Announced", | ||
| f"Event announced successfully in {announcement_channel.mention}!\n" | ||
| "Users can react with ✅ to attend or ❌ to decline.", | ||
| ) | ||
| self._apply_event_fields(success, selected_event) | ||
| await interaction.response.send_message(embed=success, ephemeral=True) | ||
|
|
||
| except discord.Forbidden: | ||
| embed = error_embed("Permission Denied", "I don't have permission to send messages in that channel.") | ||
| await interaction.response.send_message(embed=embed, ephemeral=True) | ||
| except discord.HTTPException: | ||
| self.log.exception("Failed to announce event") | ||
| embed = error_embed("Announcement Failed", "Failed to announce the event. Please try again.") | ||
| await interaction.response.send_message(embed=embed, ephemeral=True) | ||
|
|
||
| def _create_announcement_embed(self, event: EventSchema) -> discord.Embed: | ||
| """Create an announcement embed for an event.""" | ||
| embed = discord.Embed( | ||
| title=f"📅 {event.event_name}", | ||
| description=event.description or "No description provided.", | ||
| color=discord.Color.gold(), | ||
| ) | ||
|
|
||
| embed.add_field(name="🕐 When", value=self._format_event_time_est(event), inline=False) | ||
| embed.add_field(name="📍 Where", value=event.location or "TBD", inline=False) | ||
|
|
||
| embed.add_field( | ||
| name="📋 RSVP", | ||
| value="React with ✅ to attend or ❌ to decline.", | ||
| inline=False, | ||
| ) | ||
|
|
||
| now = datetime.now(ZoneInfo("UTC")).strftime("%Y-%m-%d %H:%M") | ||
| embed.set_footer(text=f"Announced: {now}") | ||
| return embed | ||
|
|
||
| async def _handle_event_submit(self, interaction: discord.Interaction, event: EventSchema) -> None: | ||
| """Process the valid event submission.""" | ||
| guild_id = interaction.guild_id | ||
|
|
||
| # Defer if not already done (ModelModal may have sent error) | ||
| if not interaction.response.is_done(): | ||
| await interaction.response.defer(ephemeral=True) | ||
|
|
||
| if not guild_id: | ||
| embed = error_embed("No Server", "Events must be created in a server.") | ||
| await interaction.edit_original_response(content="", embeds=[embed]) | ||
| return | ||
|
|
||
| # [DB CALL]: Save event | ||
| self.events.setdefault(guild_id, []).append(event) | ||
|
|
||
| self.log.info("Created event '%s' for guild %s", event.event_name, guild_id) | ||
|
|
||
| embed = success_embed("Event Created", "Your event has been created successfully!") | ||
| self._apply_event_fields(embed, event) | ||
| now = datetime.now(ZoneInfo("UTC")).strftime("%Y-%m-%d %H:%M") | ||
| embed.set_footer(text=f"Created: {now}") | ||
|
|
||
| await interaction.edit_original_response(content="", embeds=[embed], view=ui.View()) | ||
|
|
||
| def _create_event_embed(self, title: str, description: str, event: EventSchema) -> discord.Embed: | ||
| """Helper to build a success-styled event display embed.""" | ||
| embed = success_embed(title, description) | ||
| self._apply_event_fields(embed, event) | ||
| return embed | ||
|
|
||
| async def _handle_event_update( | ||
| self, interaction: discord.Interaction, updated_event: EventSchema, original_event: EventSchema | ||
| ) -> None: | ||
| """Process the event update submission.""" | ||
| guild_id = interaction.guild_id | ||
|
|
||
| # Defer if not already done (ModelModal may have sent error) | ||
| if not interaction.response.is_done(): | ||
| await interaction.response.defer(ephemeral=True) | ||
|
|
||
| if not guild_id: | ||
| embed = error_embed("No Server", "Events must be updated in a server.") | ||
| await interaction.edit_original_response(content="", embeds=[embed]) | ||
| return | ||
|
|
||
| # [DB CALL]: Update event | ||
| guild_events = self.events.setdefault(guild_id, []) | ||
| if original_event in guild_events: | ||
| idx = guild_events.index(original_event) | ||
| guild_events[idx] = updated_event | ||
|
|
||
| self.log.info("Updated event '%s' for guild %s", updated_event.event_name, guild_id) | ||
|
|
||
| embed = self._create_event_embed( | ||
| "Event Updated", | ||
| "Your event has been updated successfully!", | ||
| updated_event, | ||
| ) | ||
| now = datetime.now(ZoneInfo("UTC")).strftime("%Y-%m-%d %H:%M") | ||
| embed.set_footer(text=f"Updated: {now}") | ||
|
|
||
| await interaction.edit_original_response(content="", embeds=[embed], view=ui.View()) | ||
|
|
||
| async def _on_show_select(self, interaction: discord.Interaction, selected_event: EventSchema) -> None: | ||
| """Handle event selection for showing details.""" | ||
| embed = self._create_event_embed( | ||
| "Event Details", | ||
| "Here are the details for this event.", | ||
| selected_event, | ||
| ) | ||
| await interaction.response.send_message(embed=embed, ephemeral=True) | ||
|
|
||
| async def _on_delete_select(self, interaction: discord.Interaction, selected_event: EventSchema) -> None: | ||
| """Handle event selection for deletion.""" | ||
| view = ConfirmDeleteView() | ||
| embed = discord.Embed( | ||
| title="Confirm Deletion", | ||
| description=f"Are you sure you want to delete **{selected_event.event_name}**?", | ||
| color=discord.Color.red(), | ||
| ) | ||
| await interaction.response.send_message(embed=embed, view=view, ephemeral=True) | ||
|
|
||
| await view.wait() | ||
|
|
||
| if view.value is True: | ||
| # [DB CALL]: Delete event from guild | ||
| guild_id = interaction.guild_id | ||
| if guild_id: | ||
| guild_events = self.events.setdefault(guild_id, []) | ||
| if selected_event in guild_events: | ||
| guild_events.remove(selected_event) | ||
| self.log.info("Deleted event '%s' from guild %s", selected_event.event_name, guild_id) | ||
|
|
||
| success = success_embed("Event Deleted", "The event has been deleted successfully!") | ||
| await interaction.followup.send(embed=success, ephemeral=True) | ||
|
|
||
|
|
||
| async def setup(bot: commands.Bot) -> None: | ||
| """Set up the Event cog.""" | ||
| await bot.add_cog(Event(bot)) |
There was a problem hiding this comment.
The event cog lacks test coverage. The repository has tests for several cogs (ping, sync, telemetry) but the new event functionality is not tested. Key areas that should have test coverage include: event CRUD operations, timezone conversions, announcement message creation, RSVP tracking logic, and validation of event schemas (especially the date/time parsing logic). Consider adding comprehensive tests to ensure the event management functionality works correctly.
| # In-memory storage for demonstration. | ||
| self.events: dict[int, list[EventSchema]] = {} | ||
| # Track announcement messages: guild_id -> {event_name: message_id} | ||
| self.event_announcements: dict[int, dict[str, int]] = {} |
There was a problem hiding this comment.
The announcement tracking dictionary has a similar concurrency issue. When multiple announcements are made concurrently for different events in the same guild, there's potential for race conditions when checking and updating the nested dictionary structure. This could lead to lost announcement message IDs or inconsistent state.
| value = value.strip() | ||
| if " " in value: | ||
| # Handle 00:XX AM/PM by converting to 12:XX AM/PM | ||
| if value.lower().startswith("00:"): |
There was a problem hiding this comment.
The time parsing logic has a potential issue with "00:XX AM/PM" format. Converting "00:" to "12:" is correct for midnight (00:00 AM = 12:00 AM), but "00:00 PM" doesn't exist in standard 12-hour format. Consider adding validation to reject invalid combinations like "00:XX PM" to prevent user confusion.
Suggested change
| if value.lower().startswith("00:"): | |
| lower_value = value.lower() | |
| if lower_value.startswith("00:"): | |
| # 00:XX PM is not a valid 12-hour time; reject to avoid confusion | |
| if "pm" in lower_value: | |
| raise ValueError("Invalid time format: '00:XX PM' is not a valid 12-hour time.") |
| guild_events = self.events.setdefault(guild_id, []) | ||
| if original_event in guild_events: | ||
| idx = guild_events.index(original_event) | ||
| guild_events[idx] = updated_event |
There was a problem hiding this comment.
The event update logic uses object equality (if original_event in guild_events) to find and replace events. This relies on EventSchema's default Pydantic equality behavior, which compares all fields. However, this means if the event was modified elsewhere (e.g., by another user or if the object reference changed), the update will fail silently without notifying the user. Consider using a unique identifier for events or adding error handling to detect when an event can't be found for update.
Suggested change
| guild_events[idx] = updated_event | |
| guild_events[idx] = updated_event | |
| else: | |
| # Event not found; log and inform the user instead of silently succeeding. | |
| self.log.warning( | |
| "Attempted to update event '%s' for guild %s, but original event was not found", | |
| updated_event.event_name, | |
| guild_id, | |
| ) | |
| embed = error_embed( | |
| "Event Not Found", | |
| "The event you tried to update could not be found. It may have been changed or deleted.", | |
| ) | |
| await interaction.edit_original_response(content="", embeds=[embed]) | |
| return |
| if selected_event in guild_events: | ||
| guild_events.remove(selected_event) | ||
| self.log.info("Deleted event '%s' from guild %s", selected_event.event_name, guild_id) | ||
|
|
There was a problem hiding this comment.
When an event is deleted, the announcement message ID in self.event_announcements is not cleaned up. This causes a memory leak in the in-memory storage and will leave orphaned entries that could accumulate over time. When transitioning to a database, this could lead to stale references. Consider removing the announcement tracking entry when an event is deleted.
Suggested change
| # Clean up any associated announcement tracking for this event | |
| if hasattr(self, "event_announcements"): | |
| guild_announcements = self.event_announcements.get(guild_id) | |
| if guild_announcements: | |
| event_key = ( | |
| getattr(selected_event, "id", None) | |
| or getattr(selected_event, "event_id", None) | |
| or getattr(selected_event, "event_name", None) | |
| ) | |
| if event_key is not None: | |
| guild_announcements.pop(event_key, None) |
Comment on lines
+121
to
+123
| async def event(self, interaction: discord.Interaction, action: app_commands.Choice[str]) -> None: | ||
| """Manage events based on the action specified.""" | ||
| match action.value: |
There was a problem hiding this comment.
The action parameter is typed as app_commands.Choice[str] here, but in the profile cog (profile.py line 62), the equivalent parameter is typed as just str. Both work, but the profile cog pattern is simpler and more commonly used in discord.py. When using @app_commands.choices, discord.py automatically handles the Choice object and passes the value as a string. Consider using action: str for consistency with the rest of the codebase, then access it directly as action instead of action.value.
Suggested change
| async def event(self, interaction: discord.Interaction, action: app_commands.Choice[str]) -> None: | |
| """Manage events based on the action specified.""" | |
| match action.value: | |
| async def event(self, interaction: discord.Interaction, action: str) -> None: | |
| """Manage events based on the action specified.""" | |
| match action: |
| self.on_select = on_select_callback | ||
|
|
||
| if not events: | ||
| return |
There was a problem hiding this comment.
When no events exist, the EventDropdownView returns early (line 64) without adding any select component. However, the view is still instantiated and returned to the caller. This means the view has no interactive components, which could be confusing. Consider raising an error or returning None instead, or handle this case before instantiating the view at all. The calling code should check for this condition before creating the view.
Suggested change
| return | |
| raise ValueError("EventDropdownView requires at least one event.") |
shamikkarkhanis
approved these changes
Feb 13, 2026
shamikkarkhanis
deleted the
feature/capr-23-scaffold-event_cog-from-deprecated-repo
branch
February 13, 2026 03:22
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary by Sourcery
Add a new Discord cog for interactive event management, wired into existing UI patterns and configuration.
New Features:
Enhancements: