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 = None # Initialize in setup 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): if self.session: asyncio.create_task(self.session.close()) async def initialize(self): """Load the admin secret from bot config and initialize session.""" 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. Please set it using [p]set api ruby_api admin_secret,") return False # Initialize aiohttp session if not self.session: self.session = aiohttp.ClientSession() return True async def _get_leaderboard(self) -> Optional[list]: """Fetch the global leaderboard from the API.""" if not self.admin_secret or not self.admin_secret.get("admin_secret"): log.error("Admin secret not configured") return None if not self.session: log.error("Session not initialized") 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() if not isinstance(data, dict) or "leaderboard" not in data: log.error("Invalid API response format") return None self._cache["leaderboard"] = data["leaderboard"] self._last_update["leaderboard"] = now return self._cache["leaderboard"] elif resp.status == 401: log.error("Unauthorized: Invalid admin secret") return None else: log.error(f"Failed to fetch leaderboard: Status {resp.status}") return None except aiohttp.ClientError as e: log.error(f"API connection error: {e}") return None except Exception as e: log.error(f"Unexpected 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 or not self.admin_secret.get("admin_secret"): log.error("Admin secret not configured") return False if not self.session: log.error("Session not initialized") 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: if resp.status == 200: # Clear cache to ensure fresh data on next fetch self._cache.pop("leaderboard", None) return True elif resp.status == 401: log.error("Unauthorized: Invalid admin secret") return False else: log.error(f"Failed to update points: Status {resp.status}") return False except aiohttp.ClientError as e: log.error(f"API connection error: {e}") return False except Exception as e: log.error(f"Unexpected error updating points: {e}") return False @commands.group(name="globalboard", aliases=["glb"]) async def globalboard(self, ctx: commands.Context): """Global leaderboard commands.""" if ctx.invoked_subcommand is None: # Don't send help here, just show the base command response await ctx.send("Use `!help globalboard` to see available commands.") @globalboard.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: if not self.admin_secret or not self.admin_secret.get("admin_secret"): return await ctx.send("Leaderboard is not configured. Please contact the bot administrator.") return await ctx.send("Failed to fetch leaderboard data. Please try again later.") if not leaderboard_data: # Empty leaderboard return await ctx.send("The leaderboard is currently empty!") 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 pages: # No data after pagination return await ctx.send("The leaderboard is currently empty!") 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.get("username", "Unknown User") points = entry.get("points", 0) lines.append(f"{i}. {username}: {points:,} points") if not lines: # No valid entries return await ctx.send("No valid leaderboard entries found.") header = f"🏆 Global Leaderboard (Page {page}/{len(pages)})" footer = f"Use {ctx.prefix}globalboard show to view other pages" content = box("\n".join([header, *lines, "", footer]), lang="md") await ctx.send(content) @globalboard.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: if not self.admin_secret or not self.admin_secret.get("admin_secret"): return await ctx.send("Leaderboard is not configured. Please contact the bot administrator.") return await ctx.send("Failed to fetch leaderboard data. Please try again later.") user_data = next( (entry for entry in leaderboard_data if entry.get("userId") == str(member.id)), None ) if user_data: rank = next( (i for i, entry in enumerate(leaderboard_data, 1) if entry.get("userId") == str(member.id)), None ) await ctx.send( f"🏆 **{member.display_name}** has **{user_data.get('points', 0):,}** 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.get("userId") == str(message.author.id)), None ) if user_data: current_points = user_data.get("points", 0) # Update points new_points = current_points + points_per_message await self._update_points( str(message.author.id), str(message.author), new_points )