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="globalboard", aliases=["glb"]) async def globalboard(self, ctx: commands.Context): """Global leaderboard commands.""" if ctx.invoked_subcommand is None: await ctx.send_help(ctx.command) @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: 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}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: 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 )