730d479e2d
- Slight bugfix to docsite execution counter due to global command tree silently resulting in duplicate triggers
114 lines
4.5 KiB
Python
114 lines
4.5 KiB
Python
# modules/usage/usage_stats.py
|
|
from __future__ import annotations
|
|
from discord.ext import commands
|
|
import discord
|
|
from collections import OrderedDict
|
|
|
|
COUNTER_KEY_PREFIX = "cmd::"
|
|
|
|
def _key_from_app(cmd: discord.app_commands.Command) -> str:
|
|
# app command qualified_name is "group sub" or "name"
|
|
name = getattr(cmd, "qualified_name", None) or getattr(cmd, "name", "unknown")
|
|
return f"{COUNTER_KEY_PREFIX}{name}"
|
|
|
|
def _key_from_ctx(ctx: commands.Context) -> str:
|
|
# prefix/hybrid qualified_name is "group sub" or "name"
|
|
c = getattr(ctx, "command", None)
|
|
name = getattr(c, "qualified_name", None) or getattr(c, "name", "unknown")
|
|
return f"{COUNTER_KEY_PREFIX}{name}"
|
|
|
|
class UsageStatsCog(commands.Cog):
|
|
"""Lightweight command run counters with de-dup + hybrid guards."""
|
|
|
|
def __init__(self, bot: commands.Bot):
|
|
self.bot = bot
|
|
# Simple LRU caches for seen events (avoid double/quad increments)
|
|
self._seen_interactions: OrderedDict[int, int] = OrderedDict() # slash
|
|
self._seen_messages: OrderedDict[int, int] = OrderedDict() # prefix/hybrid
|
|
print("[usage] UsageStatsCog init")
|
|
|
|
# ----- small LRU helpers -----
|
|
@staticmethod
|
|
def _lru_mark_seen(lru: OrderedDict, key, cap: int) -> bool:
|
|
"""Return True if already seen; otherwise record and trim to cap."""
|
|
if key is None:
|
|
return False
|
|
if key in lru:
|
|
# refresh order
|
|
lru.move_to_end(key, last=True)
|
|
return True
|
|
lru[key] = 1
|
|
# trim if needed
|
|
while len(lru) > cap:
|
|
lru.popitem(last=False)
|
|
return False
|
|
|
|
# ----- successful completions -----
|
|
|
|
@commands.Cog.listener()
|
|
async def on_app_command_completion(self, interaction: discord.Interaction, command: discord.app_commands.Command):
|
|
# De-dup by interaction id
|
|
iid = getattr(interaction, "id", None)
|
|
if self._lru_mark_seen(self._seen_interactions, iid, cap=2048):
|
|
print(f"[usage] app ~~ dup ignored (iid={iid})")
|
|
return
|
|
|
|
dm = getattr(self.bot, "data_manager", None)
|
|
if not dm:
|
|
print("[usage] app ~~ skip (no data_manager)")
|
|
return
|
|
try:
|
|
key = _key_from_app(command)
|
|
newv = dm.incr_counter(key, 1)
|
|
print(f"[usage] app ++ {key} -> {newv}")
|
|
except Exception as e:
|
|
print("[usage] app !! incr failed:", repr(e))
|
|
|
|
@commands.Cog.listener()
|
|
async def on_command_completion(self, ctx: commands.Context):
|
|
# If a HybridCommand was invoked via slash, some setups can still fire this.
|
|
# Guard: only count here when NOT an interaction-backed invoke.
|
|
if isinstance(getattr(ctx, "command", None), commands.HybridCommand) and getattr(ctx, "interaction", None):
|
|
print("[usage] px ~~ hybrid-as-slash; ignore here (app listener handled it)")
|
|
return
|
|
|
|
# De-dup by message id
|
|
msg = getattr(ctx, "message", None)
|
|
mid = getattr(msg, "id", None)
|
|
if self._lru_mark_seen(self._seen_messages, mid, cap=4096):
|
|
print(f"[usage] px ~~ dup ignored (mid={mid})")
|
|
return
|
|
|
|
dm = getattr(self.bot, "data_manager", None)
|
|
if not dm:
|
|
print("[usage] px ~~ skip (no data_manager)")
|
|
return
|
|
try:
|
|
key = _key_from_ctx(ctx)
|
|
newv = dm.incr_counter(key, 1)
|
|
print(f"[usage] px ++ {key} -> {newv}")
|
|
except Exception as e:
|
|
print("[usage] px !! incr failed:", repr(e))
|
|
|
|
# ----- attempts that error (optional, keep while validating) -----
|
|
|
|
@commands.Cog.listener()
|
|
async def on_app_command_error(self, interaction: discord.Interaction, error: Exception):
|
|
# We won't increment exec counters on error; this is only for debug visibility.
|
|
iid = getattr(interaction, "id", None)
|
|
print(f"[usage] app ** error ({type(error).__name__}) iid={iid}")
|
|
|
|
@commands.Cog.listener()
|
|
async def on_command_error(self, ctx: commands.Context, error: Exception):
|
|
mid = getattr(getattr(ctx, "message", None), "id", None)
|
|
print(f"[usage] px ** error ({type(error).__name__}) mid={mid}")
|
|
|
|
async def setup(bot: commands.Bot):
|
|
# Prevent duplicate registration if extensions are reloaded / auto-discovered twice
|
|
if getattr(bot, "_usage_stats_loaded", False):
|
|
print("[usage] UsageStatsCog already loaded; skipping duplicate add")
|
|
return
|
|
await bot.add_cog(UsageStatsCog(bot))
|
|
bot._usage_stats_loaded = True
|
|
print("[usage] UsageStatsCog loaded")
|