Refactor Shop cog to enhance role refund process by implementing a more robust interactive confirmation system. Improve user feedback during role removal transactions and streamline error handling for a better user experience.

leaderboard/__init__.py

@@ -0,0 +1,7 @@
+from .leaderboard import Leaderboard
+
+async def setup(bot):
+    """Load the Leaderboard cog."""
+    cog = Leaderboard(bot)
+    await cog.initialize()
+    await bot.add_cog(cog) 

leaderboard/info.json

@@ -0,0 +1,28 @@
+{
+    "name": "Leaderboard",
+    "author": [
+        "Valerie"
+    ],
+    "description": "Global leaderboard system for Ruby, tracking user activity and points across all servers.",
+    "short": "Global activity leaderboard system",
+    "tags": [
+        "leaderboard",
+        "points",
+        "activity",
+        "ranking"
+    ],
+    "type": "COG",
+    "end_user_data_statement": "This cog stores Discord user IDs and usernames along with their activity points. Data is stored on Ruby's API server and can be cleared after 30 days of inactivity.",
+    "min_bot_version": "3.5.0",
+    "hidden": false,
+    "disabled": false,
+    "required_cogs": {},
+    "requirements": [
+        "aiohttp"
+    ],
+    "min_python_version": [
+        3,
+        8,
+        0
+    ]
+}

leaderboard/leaderboard.py

@@ -0,0 +1,204 @@
+from redbot.core import commands, Config
+from redbot.core.bot import Red
+from redbot.core.utils.chat_formatting import box, pagify
+import aiohttp
+import asyncio
+import logging
+from datetime import datetime, timezone
+from typing import Dict, Optional
+
+log = logging.getLogger("red.ruby.leaderboard")
+
+class Leaderboard(commands.Cog):
+    """Global leaderboard system for Ruby."""
+
+    def __init__(self, bot: Red):
+        self.bot = bot
+        self.config = Config.get_conf(self, identifier=867530999, force_registration=True)
+        self.session = aiohttp.ClientSession()
+        self.api_base_url = "https://ruby.valerie.lol/api"
+        self.admin_secret = None
+        self.cache_time = 300  # 5 minutes cache
+        self._cache = {}
+        self._last_update = {}
+        
+        # Default settings
+        default_guild = {
+            "points_per_message": 1,
+            "points_decay": 0.5,  # Points lost per day of inactivity
+            "min_message_length": 5,
+            "cooldown": 60,  # Seconds between point gains
+        }
+        
+        self.config.register_guild(**default_guild)
+        
+    def cog_unload(self):
+        asyncio.create_task(self.session.close())
+
+    async def initialize(self):
+        """Load the admin secret from bot config."""
+        self.admin_secret = await self.bot.get_shared_api_tokens("ruby_api")
+        if not self.admin_secret.get("admin_secret"):
+            log.error("No admin secret found. Leaderboard functionality will be limited.")
+
+    async def _get_leaderboard(self) -> Optional[list]:
+        """Fetch the global leaderboard from the API."""
+        if not self.admin_secret:
+            return None
+            
+        try:
+            now = datetime.now(timezone.utc).timestamp()
+            
+            # Return cached data if available and fresh
+            if self._cache and now - self._last_update.get("leaderboard", 0) < self.cache_time:
+                return self._cache.get("leaderboard")
+                
+            async with self.session.get(
+                f"{self.api_base_url}/leaderboard",
+                headers={"Authorization": self.admin_secret.get("admin_secret")}
+            ) as resp:
+                if resp.status == 200:
+                    data = await resp.json()
+                    self._cache["leaderboard"] = data.get("leaderboard", [])
+                    self._last_update["leaderboard"] = now
+                    return self._cache["leaderboard"]
+                else:
+                    log.error(f"Failed to fetch leaderboard: {resp.status}")
+                    return None
+        except Exception as e:
+            log.error(f"Error fetching leaderboard: {e}")
+            return None
+
+    async def _update_points(self, user_id: str, username: str, points: int) -> bool:
+        """Update a user's points in the global leaderboard."""
+        if not self.admin_secret:
+            return False
+            
+        try:
+            async with self.session.post(
+                f"{self.api_base_url}/leaderboard",
+                headers={
+                    "Authorization": self.admin_secret.get("admin_secret"),
+                    "Content-Type": "application/json"
+                },
+                json={
+                    "userId": user_id,
+                    "username": username,
+                    "points": points
+                }
+            ) as resp:
+                return resp.status == 200
+        except Exception as e:
+            log.error(f"Error updating points: {e}")
+            return False
+
+    @commands.group(name="leaderboard", aliases=["lb"])
+    async def leaderboard(self, ctx: commands.Context):
+        """Leaderboard commands."""
+        if ctx.invoked_subcommand is None:
+            await ctx.send_help(ctx.command)
+
+    @leaderboard.command(name="show")
+    async def show_leaderboard(self, ctx: commands.Context, page: int = 1):
+        """Show the global leaderboard."""
+        async with ctx.typing():
+            leaderboard_data = await self._get_leaderboard()
+            
+            if not leaderboard_data:
+                return await ctx.send("Failed to fetch leaderboard data.")
+                
+            items_per_page = 10
+            pages = [leaderboard_data[i:i + items_per_page] 
+                    for i in range(0, len(leaderboard_data), items_per_page)]
+            
+            if not 1 <= page <= len(pages):
+                return await ctx.send(f"Invalid page number. Please choose between 1 and {len(pages)}.")
+            
+            entries = pages[page - 1]
+            
+            # Format leaderboard
+            lines = []
+            start_pos = (page - 1) * items_per_page
+            
+            for i, entry in enumerate(entries, start=start_pos + 1):
+                username = entry["username"]
+                points = entry["points"]
+                lines.append(f"{i}. {username}: {points:,} points")
+            
+            header = f"🏆 Global Leaderboard (Page {page}/{len(pages)})"
+            footer = f"Use {ctx.prefix}leaderboard show <page> to view other pages"
+            
+            content = box("\n".join([header, *lines, "", footer]), lang="md")
+            await ctx.send(content)
+
+    @leaderboard.command(name="points")
+    async def check_points(self, ctx: commands.Context, member: commands.MemberConverter = None):
+        """Check your points or another member's points."""
+        member = member or ctx.author
+        leaderboard_data = await self._get_leaderboard()
+        
+        if not leaderboard_data:
+            return await ctx.send("Failed to fetch leaderboard data.")
+            
+        user_data = next(
+            (entry for entry in leaderboard_data if entry["userId"] == str(member.id)),
+            None
+        )
+        
+        if user_data:
+            rank = next(
+                (i for i, entry in enumerate(leaderboard_data, 1)
+                 if entry["userId"] == str(member.id)),
+                None
+            )
+            await ctx.send(
+                f"🏆 **{member.display_name}** has **{user_data['points']:,}** points "
+                f"(Rank: #{rank:,})"
+            )
+        else:
+            await ctx.send(f"**{member.display_name}** has no points yet!")
+
+    @commands.Cog.listener()
+    async def on_message(self, message):
+        """Award points for activity."""
+        if message.author.bot or not message.guild:
+            return
+            
+        # Get guild settings
+        guild_settings = await self.config.guild(message.guild).all()
+        points_per_message = guild_settings["points_per_message"]
+        min_length = guild_settings["min_message_length"]
+        cooldown = guild_settings["cooldown"]
+        
+        # Check message length
+        if len(message.content) < min_length:
+            return
+            
+        # Check cooldown
+        now = datetime.now(timezone.utc).timestamp()
+        last_msg_time = self._last_update.get(f"msg_{message.author.id}", 0)
+        
+        if now - last_msg_time < cooldown:
+            return
+            
+        self._last_update[f"msg_{message.author.id}"] = now
+        
+        # Get current points
+        leaderboard_data = await self._get_leaderboard()
+        current_points = 0
+        
+        if leaderboard_data:
+            user_data = next(
+                (entry for entry in leaderboard_data if entry["userId"] == str(message.author.id)),
+                None
+            )
+            if user_data:
+                current_points = user_data["points"]
+        
+        # Update points
+        new_points = current_points + points_per_message
+        await self._update_points(
+            str(message.author.id),
+            str(message.author),
+            new_points
+        )