From b6b66fbf89ef0b53941f6c162d5ca7204c7ac3fe Mon Sep 17 00:00:00 2001 From: Jonathan Green Date: Mon, 9 Feb 2026 17:55:59 -0500 Subject: [PATCH 1/6] feat: phase 1 telemetry -- event capture and logging --- capy_discord/exts/core/__init__.py | 1 + capy_discord/exts/core/telemetry.py | 422 ++++++++++++++++++++++++++++ 2 files changed, 423 insertions(+) create mode 100644 capy_discord/exts/core/__init__.py create mode 100644 capy_discord/exts/core/telemetry.py diff --git a/capy_discord/exts/core/__init__.py b/capy_discord/exts/core/__init__.py new file mode 100644 index 0000000..94c1ea3 --- /dev/null +++ b/capy_discord/exts/core/__init__.py @@ -0,0 +1 @@ +"""Core bot functionality including telemetry and system monitoring.""" diff --git a/capy_discord/exts/core/telemetry.py b/capy_discord/exts/core/telemetry.py new file mode 100644 index 0000000..6a82a2c --- /dev/null +++ b/capy_discord/exts/core/telemetry.py @@ -0,0 +1,422 @@ +"""Telemetry extension for tracking Discord bot interactions. + +PHASE 1: Event Capture and Logging +This is a foundational implementation that captures Discord events and logs them to console. +No database, no queue, no background tasks - just pure event capture to prove the concept works. + +Key Design Decisions: +- We capture on_interaction (ALL interactions: commands, buttons, dropdowns, modals) +- We capture on_app_command (slash commands specifically with cleaner metadata) +- Data is extracted to simple dicts (not stored as Discord objects) +- All guild-specific fields handle None for DM scenarios +- Telemetry failures are caught and logged, never crashing the bot + +Future Phases: +- Phase 2: Add asyncio.Queue for async event buffering +- Phase 3: Add database storage (SQLite or PostgreSQL) +- Phase 4: Add web dashboard for analytics +""" + +import logging +from datetime import UTC, datetime +from typing import Any + +import discord +from discord import app_commands +from discord.ext import commands + +# Discord component type constants +COMPONENT_TYPE_BUTTON = 2 +COMPONENT_TYPE_SELECT = 3 + + +class Telemetry(commands.Cog): + """Telemetry Cog for capturing and logging Discord bot interactions. + + This cog listens to Discord events and extracts structured data for monitoring + bot usage patterns, user engagement, and command popularity. + + Captured Events: + - on_interaction: Captures ALL user interactions (commands, buttons, dropdowns, modals) + - on_app_command: Captures slash command completions with clean metadata + + Why both events? + - on_interaction fires BEFORE command execution (captures attempts, even failed ones) + - on_app_command fires AFTER successful command execution (cleaner data, only successful commands) + - Having both gives us a complete picture of user behavior + """ + + def __init__(self, bot: commands.Bot) -> None: + """Initialize the Telemetry cog. + + Args: + bot: The Discord bot instance + """ + self.bot = bot + self.log = logging.getLogger(__name__) + self.log.info("Telemetry cog initialized - Phase 1: Console logging only") + + # ======================================================================================== + # EVENT LISTENERS + # ======================================================================================== + + @commands.Cog.listener() + async def on_interaction(self, interaction: discord.Interaction) -> None: + """Capture ALL interactions (commands, buttons, dropdowns, modals, etc). + + This event fires for EVERY user interaction with the bot, including: + - Slash commands (/ping, /feedback, etc) + - Button clicks (Confirm, Cancel, etc) + - Dropdown selections (Select menus) + - Modal submissions (Forms) + + Why capture this? + - Gives us a complete picture of ALL user engagement + - Captures failed command attempts (before validation) + - Tracks non-command interactions (buttons, dropdowns) + + Args: + interaction: The Discord interaction object + """ + try: + # Extract structured event data + event_data = self._extract_interaction_data(interaction) + + # Log to console (Phase 1: console only, Phase 3 will add database) + self._log_event(event_data) + + except Exception: + # CRITICAL: Telemetry must never crash the bot + # Log the error but don't re-raise + self.log.exception("Failed to capture on_interaction event") + + @commands.Cog.listener() + async def on_app_command_completion( + self, + interaction: discord.Interaction, + command: app_commands.Command | app_commands.ContextMenu, + ) -> None: + """Capture successful slash command executions. + + This event fires AFTER a slash command successfully completes. + It provides cleaner metadata than on_interaction and only fires for actual commands. + + Why capture this separately from on_interaction? + - Cleaner command metadata (name, parameters) + - Only successful executions (on_interaction captures failed attempts too) + - Better for analytics on "what commands users actually complete" + + Args: + interaction: The Discord interaction object + command: The app command that was executed + """ + try: + # Extract structured event data + event_data = self._extract_app_command_data(interaction, command) + + # Log to console (Phase 1: console only) + self._log_event(event_data) + + except Exception: + # CRITICAL: Telemetry must never crash the bot + self.log.exception("Failed to capture on_app_command_completion event") + + # ======================================================================================== + # DATA EXTRACTION METHODS + # ======================================================================================== + + def _extract_interaction_data(self, interaction: discord.Interaction) -> dict[str, Any]: + """Extract structured data from a Discord interaction. + + This method converts a Discord interaction object into a simple dict + with only the data we care about. We don't store Discord objects directly + because they can't be serialized to JSON/database easily. + + Handles Edge Cases: + - DMs where guild_id is None + - Non-command interactions (buttons, dropdowns) where command name is missing + - Complex interaction types (modals, select menus) + + Args: + interaction: The Discord interaction object + + Returns: + Dict with structured event data ready for logging/storage + """ + # Determine interaction type (command, button, dropdown, modal, etc) + interaction_type = self._get_interaction_type(interaction) + + # Extract command name if this is a command interaction + # For buttons/dropdowns, this will be None or the custom_id + command_name = self._get_command_name(interaction) + + # Extract command options/parameters if available + # For slash commands: {"username": "john", "count": 5} + # For buttons: {"custom_id": "confirm_button"} + # For dropdowns: {"values": ["option1", "option2"]} + options = self._extract_interaction_options(interaction) + + return { + "event_type": "interaction", + "interaction_type": interaction_type, + "user_id": interaction.user.id, + "username": str(interaction.user), # "username#1234" or new format + "command_name": command_name, + "guild_id": interaction.guild_id, # None for DMs + "guild_name": interaction.guild.name if interaction.guild else None, + "channel_id": interaction.channel_id, + "timestamp": datetime.now(UTC), + "options": options, + } + + def _extract_app_command_data( + self, + interaction: discord.Interaction, + command: app_commands.Command | app_commands.ContextMenu, + ) -> dict[str, Any]: + """Extract structured data from a completed app command. + + This provides cleaner metadata than on_interaction since we have + the actual Command object with its name and parameters. + + Args: + interaction: The Discord interaction object + command: The app command that was executed + + Returns: + Dict with structured event data ready for logging/storage + """ + # Get command parameters from the interaction namespace + # For /ping: {} + # For /kick user:@john reason:"spam": {"user": "john", "reason": "spam"} + options = {} + if hasattr(interaction, "namespace"): + # Convert namespace to dict, filtering out private attributes + options = { + key: self._serialize_value(value) + for key, value in vars(interaction.namespace).items() + if not key.startswith("_") + } + + return { + "event_type": "app_command", + "command_name": command.name, + "command_type": "context_menu" if isinstance(command, app_commands.ContextMenu) else "slash_command", + "user_id": interaction.user.id, + "username": str(interaction.user), + "guild_id": interaction.guild_id, # None for DMs + "guild_name": interaction.guild.name if interaction.guild else None, + "channel_id": interaction.channel_id, + "timestamp": datetime.now(UTC), + "options": options, + } + + # ======================================================================================== + # HELPER METHODS + # ======================================================================================== + + def _get_interaction_type(self, interaction: discord.Interaction) -> str: + """Determine the type of interaction (command, button, dropdown, modal, etc). + + Discord has many interaction types. This method converts the enum to a readable string. + + Args: + interaction: The Discord interaction object + + Returns: + Human-readable interaction type string + """ + # Map Discord's InteractionType enum to readable strings + type_map = { + discord.InteractionType.application_command: "slash_command", + discord.InteractionType.component: "component", # Buttons, dropdowns + discord.InteractionType.modal_submit: "modal", + discord.InteractionType.autocomplete: "autocomplete", + } + + interaction_type = type_map.get(interaction.type, "unknown") + + # For component interactions, get more specific type + if interaction_type == "component" and interaction.data: + component_type = interaction.data.get("component_type") + if component_type == COMPONENT_TYPE_BUTTON: + interaction_type = "button" + elif component_type == COMPONENT_TYPE_SELECT: + interaction_type = "dropdown" + + return interaction_type + + def _get_command_name(self, interaction: discord.Interaction) -> str | None: + """Extract the command name from an interaction. + + For slash commands: Returns the command name (/ping -> "ping") + For buttons/dropdowns: Returns the custom_id or None + For modals: Returns the custom_id or None + + Args: + interaction: The Discord interaction object + + Returns: + Command name or custom_id, or None if not applicable + """ + # For slash commands, use the command attribute + if interaction.command: + return interaction.command.name + + # For components (buttons, dropdowns) or modals, use custom_id + if interaction.data: + return interaction.data.get("custom_id") + + return None + + def _extract_interaction_options(self, interaction: discord.Interaction) -> dict[str, Any]: + """Extract options/parameters from an interaction. + + Different interaction types have different data structures: + - Slash commands: Have "options" in data + - Buttons: Have "custom_id" in data + - Dropdowns: Have "values" in data + - Modals: Have "components" with field values in data + + Args: + interaction: The Discord interaction object + + Returns: + Dict of extracted options/data + """ + if not interaction.data: + return {} + + # Cast to dict to bypass TypedDict validation - Discord's interaction data + # structure is more flexible than the typed definitions suggest + data: dict[str, Any] = interaction.data # type: ignore[assignment] + options: dict[str, Any] = {} + + # Handle slash command options + if "options" in data: + for option in data["options"]: + options[option["name"]] = self._serialize_value(option.get("value")) + + # Handle button custom_id + if "custom_id" in data: + options["custom_id"] = data["custom_id"] + + # Handle dropdown values + if "values" in data: + options["values"] = data["values"] + + # Handle modal components (form fields) + if "components" in data: + for action_row in data["components"]: + for component in action_row.get("components", []): + field_id = component.get("custom_id") + field_value = component.get("value") + if field_id and field_value is not None: + options[field_id] = field_value + + return options + + def _serialize_value(self, value: Any) -> Any: # noqa: ANN401 + """Convert complex Discord objects to simple serializable types. + + Discord.py uses complex objects (Member, Channel, Role, etc) that can't be + easily logged or stored. This method converts them to simple types. + + Why we do this: + - Easier to log to console + - Easier to serialize to JSON + - Easier to store in database (Phase 3) + - Preserves only the data we actually need + + Args: + value: Any value from Discord interaction data + + Returns: + Serializable version of the value (int, str, list, dict) + """ + # Discord User/Member -> user ID + if isinstance(value, (discord.User, discord.Member)): + return value.id + + # Discord Channel -> channel ID + if isinstance(value, (discord.TextChannel, discord.VoiceChannel, discord.Thread)): + return value.id + + # Discord Role -> role ID + if isinstance(value, discord.Role): + return value.id + + # Lists (recursively serialize) + if isinstance(value, list): + return [self._serialize_value(v) for v in value] + + # Dicts (recursively serialize) + if isinstance(value, dict): + return {k: self._serialize_value(v) for k, v in value.items()} + + # Everything else (int, str, bool, None) passes through + return value + + def _log_event(self, event_data: dict[str, Any]) -> None: + """Log captured event data to console. + + Phase 1: Just console logging + Phase 2: Will add to asyncio.Queue + Phase 3: Will store in database + + Args: + event_data: Structured event data dict + """ + # Format timestamp for readability + timestamp = event_data["timestamp"].strftime("%Y-%m-%d %H:%M:%S UTC") + + # Build readable log message + event_type = event_data["event_type"] + user_id = event_data["user_id"] + username = event_data.get("username", "Unknown") + + if event_type == "interaction": + interaction_type = event_data["interaction_type"] + command_name = event_data.get("command_name", "N/A") + guild_name = event_data.get("guild_name") or "DM" + options = event_data.get("options", {}) + + self.log.info( + "[TELEMETRY] Interaction | Type=%s | Command=%s | User=%s(%s) | Guild=%s | Options=%s | Time=%s", + interaction_type, + command_name, + username, + user_id, + guild_name, + options, + timestamp, + ) + + elif event_type == "app_command": + command_name = event_data["command_name"] + command_type = event_data.get("command_type", "slash_command") + guild_name = event_data.get("guild_name") or "DM" + options = event_data.get("options", {}) + + self.log.info( + "[TELEMETRY] AppCommand | Type=%s | Command=%s | User=%s(%s) | Guild=%s | Options=%s | Time=%s", + command_type, + command_name, + username, + user_id, + guild_name, + options, + timestamp, + ) + + +async def setup(bot: commands.Bot) -> None: + """Set up the Telemetry cog. + + This function is called by Discord.py's extension loader. + It creates an instance of the Telemetry cog and adds it to the bot. + + Args: + bot: The Discord bot instance + """ + await bot.add_cog(Telemetry(bot)) From ecb0842c7606a82a13796da80473a00c3f0ce1be Mon Sep 17 00:00:00 2001 From: Jonathan Green Date: Mon, 9 Feb 2026 18:26:55 -0500 Subject: [PATCH 2/6] fix: minor fix in telemetry.py --- capy_discord/exts/core/telemetry.py | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/capy_discord/exts/core/telemetry.py b/capy_discord/exts/core/telemetry.py index 6a82a2c..ba41d7a 100644 --- a/capy_discord/exts/core/telemetry.py +++ b/capy_discord/exts/core/telemetry.py @@ -18,7 +18,6 @@ """ import logging -from datetime import UTC, datetime from typing import Any import discord @@ -165,7 +164,7 @@ def _extract_interaction_data(self, interaction: discord.Interaction) -> dict[st "guild_id": interaction.guild_id, # None for DMs "guild_name": interaction.guild.name if interaction.guild else None, "channel_id": interaction.channel_id, - "timestamp": datetime.now(UTC), + "timestamp": interaction.created_at, "options": options, } @@ -207,7 +206,7 @@ def _extract_app_command_data( "guild_id": interaction.guild_id, # None for DMs "guild_name": interaction.guild.name if interaction.guild else None, "channel_id": interaction.channel_id, - "timestamp": datetime.now(UTC), + "timestamp": interaction.created_at, "options": options, } From 9a4fbcc24094335c88c6604d35460e268124b8f8 Mon Sep 17 00:00:00 2001 From: Jonathan Green <150488726+GreenJonathan@users.noreply.github.com> Date: Mon, 9 Feb 2026 19:54:20 -0500 Subject: [PATCH 3/6] Apply suggestion from @sourcery-ai Co-authored-by: sourcery-ai[bot] <58596630+sourcery-ai[bot]@users.noreply.github.com> --- capy_discord/exts/core/telemetry.py | 21 ++++++++++++++++++--- 1 file changed, 18 insertions(+), 3 deletions(-) diff --git a/capy_discord/exts/core/telemetry.py b/capy_discord/exts/core/telemetry.py index ba41d7a..e9230d5 100644 --- a/capy_discord/exts/core/telemetry.py +++ b/capy_discord/exts/core/telemetry.py @@ -291,10 +291,25 @@ def _extract_interaction_options(self, interaction: discord.Interaction) -> dict data: dict[str, Any] = interaction.data # type: ignore[assignment] options: dict[str, Any] = {} - # Handle slash command options + # Handle slash command options (including nested subcommands/subcommand groups) if "options" in data: - for option in data["options"]: - options[option["name"]] = self._serialize_value(option.get("value")) + def _flatten_options(option_list: list[dict[str, Any]], prefix: str = "") -> None: + for opt in option_list: + # Build a stable, flattened key like "subcommand.param" + name = opt.get("name") + if not name: + continue + + full_name = f"{prefix}.{name}" if prefix else name + + # Subcommand or subcommand group with nested options + if "options" in opt and isinstance(opt["options"], list): + _flatten_options(opt["options"], full_name) + # Leaf option with a value + elif "value" in opt: + options[full_name] = self._serialize_value(opt.get("value")) + + _flatten_options(data["options"]) # Handle button custom_id if "custom_id" in data: From e6226fa32699572a4d621cf38f1ccd9f50e6d31a Mon Sep 17 00:00:00 2001 From: Jonathan Green Date: Mon, 9 Feb 2026 20:04:26 -0500 Subject: [PATCH 4/6] fix: fixed lint errors --- capy_discord/exts/core/telemetry.py | 64 ++++++++++++++++++----------- 1 file changed, 41 insertions(+), 23 deletions(-) diff --git a/capy_discord/exts/core/telemetry.py b/capy_discord/exts/core/telemetry.py index e9230d5..671f70e 100644 --- a/capy_discord/exts/core/telemetry.py +++ b/capy_discord/exts/core/telemetry.py @@ -293,23 +293,7 @@ def _extract_interaction_options(self, interaction: discord.Interaction) -> dict # Handle slash command options (including nested subcommands/subcommand groups) if "options" in data: - def _flatten_options(option_list: list[dict[str, Any]], prefix: str = "") -> None: - for opt in option_list: - # Build a stable, flattened key like "subcommand.param" - name = opt.get("name") - if not name: - continue - - full_name = f"{prefix}.{name}" if prefix else name - - # Subcommand or subcommand group with nested options - if "options" in opt and isinstance(opt["options"], list): - _flatten_options(opt["options"], full_name) - # Leaf option with a value - elif "value" in opt: - options[full_name] = self._serialize_value(opt.get("value")) - - _flatten_options(data["options"]) + self._extract_command_options(data["options"], options) # Handle button custom_id if "custom_id" in data: @@ -321,15 +305,49 @@ def _flatten_options(option_list: list[dict[str, Any]], prefix: str = "") -> Non # Handle modal components (form fields) if "components" in data: - for action_row in data["components"]: - for component in action_row.get("components", []): - field_id = component.get("custom_id") - field_value = component.get("value") - if field_id and field_value is not None: - options[field_id] = field_value + self._extract_modal_components(data["components"], options) return options + def _extract_command_options( + self, option_list: list[dict[str, Any]], options: dict[str, Any], prefix: str = "" + ) -> None: + """Recursively extract and flatten slash command options. + + Args: + option_list: List of command options from interaction data + options: Dictionary to populate with flattened options (modified in place) + prefix: Current prefix for nested options (e.g., "subcommand") + """ + for opt in option_list: + # Build a stable, flattened key like "subcommand.param" + name = opt.get("name") + if not name: + continue + + full_name = f"{prefix}.{name}" if prefix else name + + # Subcommand or subcommand group with nested options + if "options" in opt and isinstance(opt["options"], list): + self._extract_command_options(opt["options"], options, full_name) + # Leaf option with a value + elif "value" in opt: + options[full_name] = self._serialize_value(opt.get("value")) + + def _extract_modal_components(self, components: list[dict[str, Any]], options: dict[str, Any]) -> None: + """Extract form field values from modal components. + + Args: + components: List of modal components (action rows) + options: Dictionary to populate with field values (modified in place) + """ + for action_row in components: + for component in action_row.get("components", []): + field_id = component.get("custom_id") + field_value = component.get("value") + if field_id and field_value is not None: + options[field_id] = field_value + def _serialize_value(self, value: Any) -> Any: # noqa: ANN401 """Convert complex Discord objects to simple serializable types. From 81a7625dc42902ea860723adbd6b1a3a5bdfe0a4 Mon Sep 17 00:00:00 2001 From: Jonathan Green Date: Tue, 10 Feb 2026 18:23:35 -0500 Subject: [PATCH 5/6] feat: phase 1 complete --- capy_discord/bot.py | 26 +- capy_discord/exts/core/telemetry.py | 433 ++++++++++++++-------- capy_discord/logging.py | 17 +- tests/capy_discord/exts/test_telemetry.py | 185 +++++++++ 4 files changed, 479 insertions(+), 182 deletions(-) create mode 100644 tests/capy_discord/exts/test_telemetry.py diff --git a/capy_discord/bot.py b/capy_discord/bot.py index 06383b4..7696b1a 100644 --- a/capy_discord/bot.py +++ b/capy_discord/bot.py @@ -5,6 +5,7 @@ from discord.ext import commands from capy_discord.errors import UserFriendlyError +from capy_discord.exts.core.telemetry import Telemetry from capy_discord.ui.embeds import error_embed from capy_discord.utils import EXTENSIONS @@ -19,7 +20,7 @@ async def setup_hook(self) -> None: await self.load_extensions() def _get_logger_for_command( - self, command: app_commands.Command | app_commands.ContextMenu | commands.Command | None + self, command: app_commands.Command | app_commands.ContextMenu | None ) -> logging.Logger: if command and hasattr(command, "module") and command.module: return logging.getLogger(command.module) @@ -32,6 +33,11 @@ async def on_tree_error(self, interaction: discord.Interaction, error: app_comma if isinstance(error, app_commands.CommandInvokeError): actual_error = error.original + # Track all failures in telemetry (both user-friendly and unexpected) + telemetry = self.get_cog("Telemetry") + if isinstance(telemetry, Telemetry): + telemetry.log_command_failure(interaction, error) + if isinstance(actual_error, UserFriendlyError): embed = error_embed(description=actual_error.user_message) if interaction.response.is_done(): @@ -49,24 +55,6 @@ async def on_tree_error(self, interaction: discord.Interaction, error: app_comma else: await interaction.response.send_message(embed=embed, ephemeral=True) - async def on_command_error(self, ctx: commands.Context, error: commands.CommandError) -> None: - """Handle errors in prefix commands.""" - # Unpack CommandInvokeError - actual_error = error - if isinstance(error, commands.CommandInvokeError): - actual_error = error.original - - if isinstance(actual_error, UserFriendlyError): - embed = error_embed(description=actual_error.user_message) - await ctx.send(embed=embed) - return - - # Generic error handling - logger = self._get_logger_for_command(ctx.command) - logger.exception("Command error: %s", error) - embed = error_embed(description="An unexpected error occurred. Please try again later.") - await ctx.send(embed=embed) - async def load_extensions(self) -> None: """Load all enabled extensions.""" for extension in EXTENSIONS: diff --git a/capy_discord/exts/core/telemetry.py b/capy_discord/exts/core/telemetry.py index 671f70e..47b294a 100644 --- a/capy_discord/exts/core/telemetry.py +++ b/capy_discord/exts/core/telemetry.py @@ -1,8 +1,10 @@ """Telemetry extension for tracking Discord bot interactions. -PHASE 1: Event Capture and Logging -This is a foundational implementation that captures Discord events and logs them to console. -No database, no queue, no background tasks - just pure event capture to prove the concept works. +PHASE 2a: Queue Buffering and Error Categorization +Builds on Phase 1 event capture by adding: +- asyncio.Queue to decouple event listeners from I/O (fire-and-forget enqueue) +- Background consumer task that drains the queue and logs events +- Error categorization: "user_error" (UserFriendlyError) vs "internal_error" (real bugs) Key Design Decisions: - We capture on_interaction (ALL interactions: commands, buttons, dropdowns, modals) @@ -10,24 +12,46 @@ - Data is extracted to simple dicts (not stored as Discord objects) - All guild-specific fields handle None for DM scenarios - Telemetry failures are caught and logged, never crashing the bot +- Each interaction gets a UUID correlation_id linking interaction and completion logs +- Command failures are tracked via log_command_failure called from bot error handlers Future Phases: -- Phase 2: Add asyncio.Queue for async event buffering - Phase 3: Add database storage (SQLite or PostgreSQL) - Phase 4: Add web dashboard for analytics """ +import asyncio import logging +import time +import uuid +from dataclasses import dataclass from typing import Any import discord from discord import app_commands -from discord.ext import commands +from discord.ext import commands, tasks + +from capy_discord.errors import UserFriendlyError # Discord component type constants COMPONENT_TYPE_BUTTON = 2 COMPONENT_TYPE_SELECT = 3 +# Stale interaction entries older than this (seconds) are cleaned up +_STALE_THRESHOLD_SECONDS = 60 + +# Queue and consumer configuration +_QUEUE_MAX_SIZE = 1000 +_CONSUMER_INTERVAL_SECONDS = 1.0 + + +@dataclass(slots=True) +class TelemetryEvent: + """A telemetry event to be processed by the background consumer.""" + + event_type: str # "interaction" or "completion" + data: dict[str, Any] + class Telemetry(commands.Cog): """Telemetry Cog for capturing and logging Discord bot interactions. @@ -37,12 +61,11 @@ class Telemetry(commands.Cog): Captured Events: - on_interaction: Captures ALL user interactions (commands, buttons, dropdowns, modals) - - on_app_command: Captures slash command completions with clean metadata + - on_app_command_completion: Captures slash command completions with clean metadata + - log_command_failure: Called from bot error handler to capture failed commands - Why both events? - - on_interaction fires BEFORE command execution (captures attempts, even failed ones) - - on_app_command fires AFTER successful command execution (cleaner data, only successful commands) - - Having both gives us a complete picture of user behavior + Each interaction is assigned a UUID correlation_id that links the interaction log + to its corresponding completion or failure log. """ def __init__(self, bot: commands.Bot) -> None: @@ -53,7 +76,87 @@ def __init__(self, bot: commands.Bot) -> None: """ self.bot = bot self.log = logging.getLogger(__name__) - self.log.info("Telemetry cog initialized - Phase 1: Console logging only") + # Maps interaction.id -> (correlation_id, start_time_monotonic) + self._pending: dict[int, tuple[str, float]] = {} + self._queue: asyncio.Queue[TelemetryEvent] = asyncio.Queue(maxsize=_QUEUE_MAX_SIZE) + self.log.info("Telemetry cog initialized - Phase 2a: Queue buffering and error categorization") + + # ======================================================================================== + # LIFECYCLE + # ======================================================================================== + + async def cog_load(self) -> None: + """Start the background consumer task.""" + self._consumer_task.start() + + async def cog_unload(self) -> None: + """Stop the consumer and flush remaining events.""" + self._consumer_task.cancel() + self._drain_queue() + + # ======================================================================================== + # BACKGROUND CONSUMER + # ======================================================================================== + + @tasks.loop(seconds=_CONSUMER_INTERVAL_SECONDS) + async def _consumer_task(self) -> None: + """Periodically drain the queue and process pending telemetry events.""" + self._process_pending_events() + + @_consumer_task.before_loop + async def _before_consumer(self) -> None: + await self.bot.wait_until_ready() + + def _process_pending_events(self) -> None: + """Drain the queue and dispatch each event. Capped at _QUEUE_MAX_SIZE per tick.""" + processed = 0 + while processed < _QUEUE_MAX_SIZE: + try: + event = self._queue.get_nowait() + except asyncio.QueueEmpty: + break + self._dispatch_event(event) + processed += 1 + + def _drain_queue(self) -> None: + """Flush remaining events on unload. Warns if any events were pending.""" + count = 0 + while True: + try: + event = self._queue.get_nowait() + except asyncio.QueueEmpty: + break + self._dispatch_event(event) + count += 1 + if count: + self.log.warning("Drained %d telemetry event(s) during cog unload", count) + + def _dispatch_event(self, event: TelemetryEvent) -> None: + """Route an event to the appropriate logging method. + + Args: + event: The telemetry event to dispatch + """ + try: + if event.event_type == "interaction": + self._log_interaction(event.data) + elif event.event_type == "completion": + self._log_completion(**event.data) + else: + self.log.warning("Unknown telemetry event type: %s", event.event_type) + except Exception: + self.log.exception("Failed to dispatch telemetry event: %s", event.event_type) + + def _enqueue(self, event: TelemetryEvent) -> None: + """Enqueue a telemetry event. Drops the event if the queue is full. + + Args: + event: The telemetry event to enqueue + """ + try: + self._queue.put_nowait(event) + except asyncio.QueueFull: + self.log.warning("Telemetry queue full — dropping %s event", event.event_type) # ======================================================================================== # EVENT LISTENERS @@ -69,24 +172,26 @@ async def on_interaction(self, interaction: discord.Interaction) -> None: - Dropdown selections (Select menus) - Modal submissions (Forms) - Why capture this? - - Gives us a complete picture of ALL user engagement - - Captures failed command attempts (before validation) - - Tracks non-command interactions (buttons, dropdowns) - Args: interaction: The Discord interaction object """ try: + # Clean up stale entries that never got a completion/failure + self._cleanup_stale_entries() + + # Generate correlation ID and record start time + correlation_id = uuid.uuid4().hex[:12] + self._pending[interaction.id] = (correlation_id, time.monotonic()) + # Extract structured event data event_data = self._extract_interaction_data(interaction) + event_data["correlation_id"] = correlation_id - # Log to console (Phase 1: console only, Phase 3 will add database) - self._log_event(event_data) + # Enqueue for background processing + self._enqueue(TelemetryEvent("interaction", event_data)) except Exception: # CRITICAL: Telemetry must never crash the bot - # Log the error but don't re-raise self.log.exception("Failed to capture on_interaction event") @commands.Cog.listener() @@ -97,29 +202,78 @@ async def on_app_command_completion( ) -> None: """Capture successful slash command executions. - This event fires AFTER a slash command successfully completes. - It provides cleaner metadata than on_interaction and only fires for actual commands. - - Why capture this separately from on_interaction? - - Cleaner command metadata (name, parameters) - - Only successful executions (on_interaction captures failed attempts too) - - Better for analytics on "what commands users actually complete" + Logs a slim completion record with correlation_id, command name, + status, and execution time. Full metadata is in the interaction log. Args: interaction: The Discord interaction object command: The app command that was executed """ try: - # Extract structured event data - event_data = self._extract_app_command_data(interaction, command) - - # Log to console (Phase 1: console only) - self._log_event(event_data) + correlation_id, start_time = self._pop_pending(interaction.id) + duration_ms = round((time.monotonic() - start_time) * 1000, 1) + + self._enqueue( + TelemetryEvent( + "completion", + { + "correlation_id": correlation_id, + "command_name": command.name, + "status": "success", + "duration_ms": duration_ms, + }, + ) + ) except Exception: # CRITICAL: Telemetry must never crash the bot self.log.exception("Failed to capture on_app_command_completion event") + # ======================================================================================== + # FAILURE TRACKING (called from bot.py error handler) + # ======================================================================================== + + def log_command_failure( + self, + interaction: discord.Interaction, + error: app_commands.AppCommandError, + ) -> None: + """Log a command failure with correlation to the original interaction. + + Called from Bot.on_tree_error to track which commands fail and why. + Categorizes errors as "user_error" (UserFriendlyError) or "internal_error". + + Args: + interaction: The Discord interaction object + error: The error that occurred + """ + try: + correlation_id, start_time = self._pop_pending(interaction.id) + duration_ms = round((time.monotonic() - start_time) * 1000, 1) + + # Unwrap CommandInvokeError to get the actual cause + actual_error = error.original if isinstance(error, app_commands.CommandInvokeError) else error + + status = "user_error" if isinstance(actual_error, UserFriendlyError) else "internal_error" + + error_type = type(actual_error).__name__ + + self._enqueue( + TelemetryEvent( + "completion", + { + "correlation_id": correlation_id, + "command_name": interaction.command.name if interaction.command else "unknown", + "status": status, + "duration_ms": duration_ms, + "error_type": error_type, + }, + ) + ) + + except Exception: + self.log.exception("Failed to capture command failure event") + # ======================================================================================== # DATA EXTRACTION METHODS # ======================================================================================== @@ -131,104 +285,74 @@ def _extract_interaction_data(self, interaction: discord.Interaction) -> dict[st with only the data we care about. We don't store Discord objects directly because they can't be serialized to JSON/database easily. - Handles Edge Cases: - - DMs where guild_id is None - - Non-command interactions (buttons, dropdowns) where command name is missing - - Complex interaction types (modals, select menus) - Args: interaction: The Discord interaction object Returns: Dict with structured event data ready for logging/storage """ - # Determine interaction type (command, button, dropdown, modal, etc) interaction_type = self._get_interaction_type(interaction) - - # Extract command name if this is a command interaction - # For buttons/dropdowns, this will be None or the custom_id command_name = self._get_command_name(interaction) - - # Extract command options/parameters if available - # For slash commands: {"username": "john", "count": 5} - # For buttons: {"custom_id": "confirm_button"} - # For dropdowns: {"values": ["option1", "option2"]} options = self._extract_interaction_options(interaction) return { "event_type": "interaction", "interaction_type": interaction_type, "user_id": interaction.user.id, - "username": str(interaction.user), # "username#1234" or new format + "username": str(interaction.user), "command_name": command_name, - "guild_id": interaction.guild_id, # None for DMs + "guild_id": interaction.guild_id, "guild_name": interaction.guild.name if interaction.guild else None, "channel_id": interaction.channel_id, "timestamp": interaction.created_at, "options": options, } - def _extract_app_command_data( - self, - interaction: discord.Interaction, - command: app_commands.Command | app_commands.ContextMenu, - ) -> dict[str, Any]: - """Extract structured data from a completed app command. + # ======================================================================================== + # HELPER METHODS + # ======================================================================================== - This provides cleaner metadata than on_interaction since we have - the actual Command object with its name and parameters. + def _pop_pending(self, interaction_id: int) -> tuple[str, float]: + """Pop and return the pending entry for an interaction. + + If the entry doesn't exist (e.g. race condition or missed event), + returns a fallback with current time. Args: - interaction: The Discord interaction object - command: The app command that was executed + interaction_id: Discord interaction snowflake ID Returns: - Dict with structured event data ready for logging/storage + Tuple of (correlation_id, start_time) """ - # Get command parameters from the interaction namespace - # For /ping: {} - # For /kick user:@john reason:"spam": {"user": "john", "reason": "spam"} - options = {} - if hasattr(interaction, "namespace"): - # Convert namespace to dict, filtering out private attributes - options = { - key: self._serialize_value(value) - for key, value in vars(interaction.namespace).items() - if not key.startswith("_") - } + if interaction_id in self._pending: + return self._pending.pop(interaction_id) + return ("unknown", time.monotonic()) - return { - "event_type": "app_command", - "command_name": command.name, - "command_type": "context_menu" if isinstance(command, app_commands.ContextMenu) else "slash_command", - "user_id": interaction.user.id, - "username": str(interaction.user), - "guild_id": interaction.guild_id, # None for DMs - "guild_name": interaction.guild.name if interaction.guild else None, - "channel_id": interaction.channel_id, - "timestamp": interaction.created_at, - "options": options, - } + def _cleanup_stale_entries(self) -> None: + """Remove pending entries older than the stale threshold. - # ======================================================================================== - # HELPER METHODS - # ======================================================================================== + Prevents memory leaks from interactions that never get a + completion or failure callback. + """ + now = time.monotonic() + stale_ids = [ + iid for iid, (_, start_time) in self._pending.items() if now - start_time > _STALE_THRESHOLD_SECONDS + ] + for iid in stale_ids: + del self._pending[iid] def _get_interaction_type(self, interaction: discord.Interaction) -> str: """Determine the type of interaction (command, button, dropdown, modal, etc). - Discord has many interaction types. This method converts the enum to a readable string. - Args: interaction: The Discord interaction object Returns: Human-readable interaction type string """ - # Map Discord's InteractionType enum to readable strings type_map = { discord.InteractionType.application_command: "slash_command", - discord.InteractionType.component: "component", # Buttons, dropdowns + discord.InteractionType.component: "component", discord.InteractionType.modal_submit: "modal", discord.InteractionType.autocomplete: "autocomplete", } @@ -248,21 +372,15 @@ def _get_interaction_type(self, interaction: discord.Interaction) -> str: def _get_command_name(self, interaction: discord.Interaction) -> str | None: """Extract the command name from an interaction. - For slash commands: Returns the command name (/ping -> "ping") - For buttons/dropdowns: Returns the custom_id or None - For modals: Returns the custom_id or None - Args: interaction: The Discord interaction object Returns: Command name or custom_id, or None if not applicable """ - # For slash commands, use the command attribute if interaction.command: return interaction.command.name - # For components (buttons, dropdowns) or modals, use custom_id if interaction.data: return interaction.data.get("custom_id") @@ -271,12 +389,6 @@ def _get_command_name(self, interaction: discord.Interaction) -> str | None: def _extract_interaction_options(self, interaction: discord.Interaction) -> dict[str, Any]: """Extract options/parameters from an interaction. - Different interaction types have different data structures: - - Slash commands: Have "options" in data - - Buttons: Have "custom_id" in data - - Dropdowns: Have "values" in data - - Modals: Have "components" with field values in data - Args: interaction: The Discord interaction object @@ -286,24 +398,18 @@ def _extract_interaction_options(self, interaction: discord.Interaction) -> dict if not interaction.data: return {} - # Cast to dict to bypass TypedDict validation - Discord's interaction data - # structure is more flexible than the typed definitions suggest data: dict[str, Any] = interaction.data # type: ignore[assignment] options: dict[str, Any] = {} - # Handle slash command options (including nested subcommands/subcommand groups) if "options" in data: self._extract_command_options(data["options"], options) - # Handle button custom_id if "custom_id" in data: options["custom_id"] = data["custom_id"] - # Handle dropdown values if "values" in data: options["values"] = data["values"] - # Handle modal components (form fields) if "components" in data: self._extract_modal_components(data["components"], options) @@ -320,17 +426,14 @@ def _extract_command_options( prefix: Current prefix for nested options (e.g., "subcommand") """ for opt in option_list: - # Build a stable, flattened key like "subcommand.param" name = opt.get("name") if not name: continue full_name = f"{prefix}.{name}" if prefix else name - # Subcommand or subcommand group with nested options if "options" in opt and isinstance(opt["options"], list): self._extract_command_options(opt["options"], options, full_name) - # Leaf option with a value elif "value" in opt: options[full_name] = self._serialize_value(opt.get("value")) @@ -351,94 +454,100 @@ def _extract_modal_components(self, components: list[dict[str, Any]], options: d def _serialize_value(self, value: Any) -> Any: # noqa: ANN401 """Convert complex Discord objects to simple serializable types. - Discord.py uses complex objects (Member, Channel, Role, etc) that can't be - easily logged or stored. This method converts them to simple types. - - Why we do this: - - Easier to log to console - - Easier to serialize to JSON - - Easier to store in database (Phase 3) - - Preserves only the data we actually need - Args: value: Any value from Discord interaction data Returns: Serializable version of the value (int, str, list, dict) """ - # Discord User/Member -> user ID if isinstance(value, (discord.User, discord.Member)): return value.id - # Discord Channel -> channel ID if isinstance(value, (discord.TextChannel, discord.VoiceChannel, discord.Thread)): return value.id - # Discord Role -> role ID if isinstance(value, discord.Role): return value.id - # Lists (recursively serialize) if isinstance(value, list): return [self._serialize_value(v) for v in value] - # Dicts (recursively serialize) if isinstance(value, dict): return {k: self._serialize_value(v) for k, v in value.items()} - # Everything else (int, str, bool, None) passes through return value - def _log_event(self, event_data: dict[str, Any]) -> None: - """Log captured event data to console. + # ======================================================================================== + # LOGGING METHODS + # ======================================================================================== + + def _log_interaction(self, event_data: dict[str, Any]) -> None: + """Log the full interaction event at DEBUG level. - Phase 1: Just console logging - Phase 2: Will add to asyncio.Queue - Phase 3: Will store in database + Contains all metadata for the interaction. The completion/failure log + references this via correlation_id. Args: event_data: Structured event data dict """ - # Format timestamp for readability timestamp = event_data["timestamp"].strftime("%Y-%m-%d %H:%M:%S UTC") - - # Build readable log message - event_type = event_data["event_type"] - user_id = event_data["user_id"] + correlation_id = event_data["correlation_id"] + interaction_type = event_data["interaction_type"] + command_name = event_data.get("command_name", "N/A") username = event_data.get("username", "Unknown") + user_id = event_data["user_id"] + guild_name = event_data.get("guild_name") or "DM" + options = event_data.get("options", {}) + + self.log.debug( + "[TELEMETRY] Interaction | ID=%s | Type=%s | Command=%s | User=%s(%s) | Guild=%s | Options=%s | Time=%s", + correlation_id, + interaction_type, + command_name, + username, + user_id, + guild_name, + options, + timestamp, + ) + + def _log_completion( + self, + *, + correlation_id: str, + command_name: str, + status: str, + duration_ms: float, + error_type: str | None = None, + ) -> None: + """Log a slim completion/failure record at DEBUG level. - if event_type == "interaction": - interaction_type = event_data["interaction_type"] - command_name = event_data.get("command_name", "N/A") - guild_name = event_data.get("guild_name") or "DM" - options = event_data.get("options", {}) + Only contains correlation_id, command name, status, duration, and + optionally error type. Full metadata lives in the interaction log. - self.log.info( - "[TELEMETRY] Interaction | Type=%s | Command=%s | User=%s(%s) | Guild=%s | Options=%s | Time=%s", - interaction_type, + Args: + correlation_id: UUID linking to the interaction log + command_name: The command that completed/failed + status: "success", "user_error", or "internal_error" + duration_ms: Execution time in milliseconds + error_type: Error class name (only for failures) + """ + if error_type: + self.log.debug( + "[TELEMETRY] Completion | ID=%s | Command=%s | Status=%s | Error=%s | Duration=%sms", + correlation_id, command_name, - username, - user_id, - guild_name, - options, - timestamp, + status, + error_type, + duration_ms, ) - - elif event_type == "app_command": - command_name = event_data["command_name"] - command_type = event_data.get("command_type", "slash_command") - guild_name = event_data.get("guild_name") or "DM" - options = event_data.get("options", {}) - - self.log.info( - "[TELEMETRY] AppCommand | Type=%s | Command=%s | User=%s(%s) | Guild=%s | Options=%s | Time=%s", - command_type, + else: + self.log.debug( + "[TELEMETRY] Completion | ID=%s | Command=%s | Status=%s | Duration=%sms", + correlation_id, command_name, - username, - user_id, - guild_name, - options, - timestamp, + status, + duration_ms, ) diff --git a/capy_discord/logging.py b/capy_discord/logging.py index 4966f51..1075cfb 100644 --- a/capy_discord/logging.py +++ b/capy_discord/logging.py @@ -11,6 +11,9 @@ def setup_logging(level: int = logging.INFO) -> None: This configures the root logger to output to both the console (via discord.utils) and a unique timestamped log file in the 'logs/' directory. + + A separate telemetry log file captures all telemetry events at DEBUG level + regardless of the root log level, so telemetry data can be analyzed independently. """ # 1. Create logs directory if it doesn't exist log_dir = Path("logs") @@ -26,8 +29,20 @@ def setup_logging(level: int = logging.INFO) -> None: # 4. Setup Consolidated File Logging # We use mode="w" (or "a", but timestamp ensures uniqueness) - file_handler = logging.FileHandler(filename=log_file, encoding="utf-8", mode="w") dt_fmt = "%Y-%m-%d %H:%M:%S" formatter = logging.Formatter("[{asctime}] [{levelname:<8}] {name}: {message}", dt_fmt, style="{") + + file_handler = logging.FileHandler(filename=log_file, encoding="utf-8", mode="w") file_handler.setFormatter(formatter) logging.getLogger().addHandler(file_handler) + + # 5. Setup Dedicated Telemetry Log File + # Writes at DEBUG level so telemetry events are always captured even if root is INFO + telemetry_log_file = log_dir / f"telemetry_{timestamp}.log" + telemetry_handler = logging.FileHandler(filename=telemetry_log_file, encoding="utf-8", mode="w") + telemetry_handler.setLevel(logging.DEBUG) + telemetry_handler.setFormatter(formatter) + telemetry_logger = logging.getLogger("capy_discord.exts.core.telemetry") + telemetry_logger.addHandler(telemetry_handler) + telemetry_logger.setLevel(logging.DEBUG) + telemetry_logger.propagate = False diff --git a/tests/capy_discord/exts/test_telemetry.py b/tests/capy_discord/exts/test_telemetry.py new file mode 100644 index 0000000..b339f1e --- /dev/null +++ b/tests/capy_discord/exts/test_telemetry.py @@ -0,0 +1,185 @@ +import asyncio +from unittest.mock import MagicMock, patch + +import discord +import pytest +from discord import app_commands +from discord.ext import commands + +from capy_discord.errors import UserFriendlyError +from capy_discord.exts.core.telemetry import ( + Telemetry, + TelemetryEvent, + _QUEUE_MAX_SIZE, +) + + +@pytest.fixture +def bot(): + intents = discord.Intents.default() + b = MagicMock(spec=commands.Bot) + b.intents = intents + b.wait_until_ready = MagicMock(return_value=asyncio.Future()) + b.wait_until_ready.return_value.set_result(None) + return b + + +@pytest.fixture +def cog(bot): + with patch.object(Telemetry, "cog_load", return_value=None): + c = Telemetry(bot) + c.log = MagicMock() + return c + + +def _make_interaction(*, interaction_id=12345, command_name="test_cmd"): + interaction = MagicMock(spec=discord.Interaction) + interaction.id = interaction_id + interaction.type = discord.InteractionType.application_command + interaction.user = MagicMock() + interaction.user.id = 99 + interaction.user.__str__ = MagicMock(return_value="TestUser#0001") + interaction.guild_id = 1 + interaction.guild = MagicMock() + interaction.guild.name = "TestGuild" + interaction.channel_id = 2 + interaction.created_at = MagicMock() + interaction.created_at.strftime = MagicMock(return_value="2025-01-01 00:00:00 UTC") + interaction.command = MagicMock() + interaction.command.name = command_name + interaction.data = {"name": command_name} + return interaction + + +@pytest.mark.asyncio +async def test_interaction_event_enqueued(cog): + interaction = _make_interaction() + + await cog.on_interaction(interaction) + + assert cog._queue.qsize() == 1 + event = cog._queue.get_nowait() + assert event.event_type == "interaction" + assert event.data["command_name"] == "test_cmd" + assert "correlation_id" in event.data + + +@pytest.mark.asyncio +async def test_completion_event_enqueued(cog): + interaction = _make_interaction() + command = MagicMock(spec=app_commands.Command) + command.name = "ping" + + # Seed _pending so completion can find it + cog._pending[interaction.id] = ("abc123", 0.0) + + await cog.on_app_command_completion(interaction, command) + + assert cog._queue.qsize() == 1 + event = cog._queue.get_nowait() + assert event.event_type == "completion" + assert event.data["status"] == "success" + assert event.data["command_name"] == "ping" + + +@pytest.mark.asyncio +async def test_failure_user_error_categorized(cog): + interaction = _make_interaction() + cog._pending[interaction.id] = ("abc123", 0.0) + + user_err = UserFriendlyError("internal msg", "user msg") + wrapped = app_commands.CommandInvokeError(MagicMock(), user_err) + + cog.log_command_failure(interaction, wrapped) + + event = cog._queue.get_nowait() + assert event.data["status"] == "user_error" + assert event.data["error_type"] == "UserFriendlyError" + + +@pytest.mark.asyncio +async def test_failure_internal_error_categorized(cog): + interaction = _make_interaction() + cog._pending[interaction.id] = ("abc123", 0.0) + + internal_err = RuntimeError("something broke") + wrapped = app_commands.CommandInvokeError(MagicMock(), internal_err) + + cog.log_command_failure(interaction, wrapped) + + event = cog._queue.get_nowait() + assert event.data["status"] == "internal_error" + assert event.data["error_type"] == "RuntimeError" + + +def test_queue_full_drops_event(cog): + # Fill the queue to capacity + for i in range(_QUEUE_MAX_SIZE): + cog._queue.put_nowait(TelemetryEvent("interaction", {"i": i})) + + assert cog._queue.full() + + # This should not raise — it logs a warning and drops the event + cog._enqueue(TelemetryEvent("interaction", {"dropped": True})) + + cog.log.warning.assert_called_once() + assert "queue full" in cog.log.warning.call_args[0][0].lower() + + +def test_consumer_processes_events(cog): + events_to_process = 2 + cog._queue.put_nowait( + TelemetryEvent( + "completion", + { + "correlation_id": "abc", + "command_name": "ping", + "status": "success", + "duration_ms": 5.0, + }, + ) + ) + cog._queue.put_nowait( + TelemetryEvent( + "completion", + { + "correlation_id": "def", + "command_name": "help", + "status": "success", + "duration_ms": 3.0, + }, + ) + ) + + cog._process_pending_events() + + assert cog._queue.qsize() == 0 + assert cog.log.debug.call_count == events_to_process + + +def test_drain_on_unload(cog): + cog._queue.put_nowait( + TelemetryEvent( + "completion", + { + "correlation_id": "abc", + "command_name": "ping", + "status": "success", + "duration_ms": 1.0, + }, + ) + ) + + cog._drain_queue() + + assert cog._queue.qsize() == 0 + # Should have logged the completion + a warning about draining + cog.log.warning.assert_called_once() + assert "Drained" in cog.log.warning.call_args[0][0] + + +def test_dispatch_unknown_event_type(cog): + cog._dispatch_event(TelemetryEvent("bogus_type", {})) + + cog.log.warning.assert_called_once() + assert "Unknown telemetry event type" in cog.log.warning.call_args[0][0] From ed2c42ab657d06907162c276f2da00df5e2db6b0 Mon Sep 17 00:00:00 2001 From: Jonathan Green Date: Tue, 10 Feb 2026 22:49:20 -0500 Subject: [PATCH 6/6] fix: phase 1 fixed with tests --- capy_discord/bot.py | 19 ++++++++++++++++++- 1 file changed, 18 insertions(+), 1 deletion(-) diff --git a/capy_discord/bot.py b/capy_discord/bot.py index 7696b1a..1c9b8ea 100644 --- a/capy_discord/bot.py +++ b/capy_discord/bot.py @@ -20,7 +20,7 @@ async def setup_hook(self) -> None: await self.load_extensions() def _get_logger_for_command( - self, command: app_commands.Command | app_commands.ContextMenu | None + self, command: app_commands.Command | app_commands.ContextMenu | commands.Command | None ) -> logging.Logger: if command and hasattr(command, "module") and command.module: return logging.getLogger(command.module) @@ -55,6 +55,23 @@ async def on_tree_error(self, interaction: discord.Interaction, error: app_comma else: await interaction.response.send_message(embed=embed, ephemeral=True) + async def on_command_error(self, ctx: commands.Context, error: commands.CommandError) -> None: + """Handle errors in prefix commands.""" + actual_error = error + if isinstance(error, commands.CommandInvokeError): + actual_error = error.original + + if isinstance(actual_error, UserFriendlyError): + embed = error_embed(description=actual_error.user_message) + await ctx.send(embed=embed) + return + + # Generic error handling + logger = self._get_logger_for_command(ctx.command) + logger.exception("Prefix command error: %s", error) + embed = error_embed(description="An unexpected error occurred. Please try again later.") + await ctx.send(embed=embed) + async def load_extensions(self) -> None: """Load all enabled extensions.""" for extension in EXTENSIONS: