diff --git a/leaderboard/__init__.py b/leaderboard/__init__.py new file mode 100644 index 0000000..dd117d5 --- /dev/null +++ b/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) \ No newline at end of file diff --git a/leaderboard/info.json b/leaderboard/info.json new file mode 100644 index 0000000..968f884 --- /dev/null +++ b/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 + ] +} \ No newline at end of file diff --git a/leaderboard/leaderboard.py b/leaderboard/leaderboard.py new file mode 100644 index 0000000..5bbf339 --- /dev/null +++ b/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 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 + ) \ No newline at end of file