diff --git a/leaderboard/leaderboard.py b/leaderboard/leaderboard.py index 1dc3164..96ad090 100644 --- a/leaderboard/leaderboard.py +++ b/leaderboard/leaderboard.py @@ -76,33 +76,19 @@ class Leaderboard(commands.Cog): 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 + + # Optional API integration + self.session = None 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 = { "min_message_length": 5, - "cooldown": 60, # Seconds between updates + "cooldown": 60, } self.config.register_guild(**default_guild) - 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_user_credits(self, user: discord.Member) -> int: """Get a user's total credits from Red's bank system.""" try: @@ -112,71 +98,39 @@ class Leaderboard(commands.Cog): log.error(f"Error getting bank balance for {user}: {e}") return 0 - 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") - await self.initialize() # Try to initialize if not done - if not self.session: - 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 Exception as e: - log.error(f"Error fetching leaderboard: {e}") - return None - - async def _update_credits(self, user_id: str, username: str, credits: int = None) -> bool: - """Update a user's credits 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: - # If credits not provided, get from Red's bank - if credits is None: - member = None - for guild in self.bot.guilds: - member = guild.get_member(int(user_id)) - if member: - break - - if not member: - return False + async def get_all_balances(self) -> List[dict]: + """Get all users' credit balances.""" + all_users = {} + + # Collect all unique members across guilds + for guild in self.bot.guilds: + for member in guild.members: + if member.bot or member.id in all_users: + continue credits = await self.get_user_credits(member) + if credits > 0: + all_users[member.id] = { + "userId": str(member.id), + "username": str(member), + "points": credits + } + + # Sort by credits and convert to list + sorted_users = sorted( + all_users.values(), + key=lambda x: x["points"], + reverse=True + ) + + return sorted_users + async def _try_api_update(self, user_id: str, username: str, credits: int) -> bool: + """Try to update the API if configured.""" + if not self.session or not self.admin_secret or not self.admin_secret.get("admin_secret"): + return False + + try: async with self.session.post( f"{self.api_base_url}/leaderboard", headers={ @@ -186,49 +140,34 @@ class Leaderboard(commands.Cog): json={ "userId": user_id, "username": username, - "points": credits # API still uses "points" field + "points": credits } ) 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 credits: Status {resp.status}") - return False + return resp.status == 200 except Exception as e: - log.error(f"Error updating credits: {e}") + log.error(f"API update failed: {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: - # Let the default help command handle it return @globalboard.command(name="show") async def show_leaderboard(self, ctx: commands.Context, page: int = 1): """Show the global credits leaderboard.""" async with ctx.typing(): - leaderboard_data = await self._get_leaderboard() + leaderboard_data = await self.get_all_balances() 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 chunks = [leaderboard_data[i:i + items_per_page] for i in range(0, len(leaderboard_data), items_per_page)] - if not chunks: # No data after pagination + if not chunks: return await ctx.send("The leaderboard is currently empty!") embeds = [] @@ -238,16 +177,14 @@ class Leaderboard(commands.Cog): color=await ctx.embed_color() ) - # Format leaderboard entries description = [] start_pos = (page_num - 1) * items_per_page for i, entry in enumerate(entries, start=start_pos + 1): username = entry.get("username", "Unknown User") user_id = entry.get("userId", "0") - credits = entry.get("points", 0) # API returns as "points" + credits = entry.get("points", 0) - # Format each entry with position, name, credits, and user ID description.append( f"`{i}.` <@{user_id}> • **{humanize_number(credits)}** credits" ) @@ -263,66 +200,49 @@ class Leaderboard(commands.Cog): async def check_credits(self, ctx: commands.Context, member: commands.MemberConverter = None): """Check your credits or another member's credits.""" member = member or ctx.author - leaderboard_data = await self._get_leaderboard() + credits = await self.get_user_credits(member) - 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)), + # Get user's rank + leaderboard_data = await self.get_all_balances() + rank = next( + (i for i, entry in enumerate(leaderboard_data, 1) + 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 - ) - - embed = discord.Embed( - title="🏆 Global Credits", - color=await ctx.embed_color() - ) + embed = discord.Embed( + title="🏆 Global Credits", + color=await ctx.embed_color() + ) + + if credits > 0: embed.description = ( f"**User:** <@{member.id}>\n" - f"**Credits:** {humanize_number(user_data.get('points', 0))}\n" - f"**Rank:** #{humanize_number(rank)}\n" + f"**Credits:** {humanize_number(credits)}\n" + f"**Rank:** #{humanize_number(rank) if rank else 'Unranked'}\n" f"**ID:** {member.id}" ) - embed.set_thumbnail(url=member.display_avatar.url) - - await ctx.send(embed=embed) else: - embed = discord.Embed( - title="No Credits Found", - description=f"<@{member.id}> has no credits yet!", - color=await ctx.embed_color() - ) - embed.set_footer(text=f"ID: {member.id}") - await ctx.send(embed=embed) + embed.description = f"<@{member.id}> has no credits yet!" + + embed.set_thumbnail(url=member.display_avatar.url) + await ctx.send(embed=embed) @globalboard.command(name="resync") @commands.is_owner() async def resync_leaderboard(self, ctx: commands.Context): - """Force a resync of all users' credits with the global leaderboard.""" + """Force a resync of all users' credits with the API (if configured).""" async with ctx.typing(): success_count = 0 fail_count = 0 - # Clear cache - self._cache.clear() - self._last_update.clear() - - # Update all members across all guilds for guild in self.bot.guilds: for member in guild.members: if member.bot: continue - if await self._update_credits(str(member.id), str(member)): + credits = await self.get_user_credits(member) + if await self._try_api_update(str(member.id), str(member), credits): success_count += 1 else: fail_count += 1 @@ -337,29 +257,15 @@ class Leaderboard(commands.Cog): ) await ctx.send(embed=embed) - @commands.Cog.listener() - async def on_member_update(self, before: discord.Member, after: discord.Member): - """Update credits when member's data changes.""" - if before.bot: - return - - # Update credits in API - await self._update_credits(str(after.id), str(after)) - @commands.Cog.listener() async def on_bank_update(self, member: discord.Member, before: int, after: int): - """Update credits when a member's bank balance changes.""" + """Update API when a member's bank balance changes.""" if member.bot: return - # Update credits in API with the new balance - await self._update_credits(str(member.id), str(member), after) + await self._try_api_update(str(member.id), str(member), after) def cog_unload(self): """Clean up when cog is unloaded.""" if self.session: - asyncio.create_task(self.session.close()) - - async def cog_load(self): - """Initialize the cog when it's loaded.""" - await self.initialize() \ No newline at end of file + asyncio.create_task(self.session.close()) \ No newline at end of file