From da273be3c33ed0d5dc7a258ec7ed5d4f7b364e1b Mon Sep 17 00:00:00 2001 From: Valerie Date: Mon, 26 May 2025 04:19:01 -0400 Subject: [PATCH] Enhance leaderboard functionality in Leaderboard cog by integrating user points retrieval from both bank and LevelUp systems. Refactor leaderboard data fetching to combine results and improve error handling. Remove unnecessary point update logic as it is now derived from user data sources. --- leaderboard/leaderboard.py | 132 ++++++++++++++++--------------------- 1 file changed, 55 insertions(+), 77 deletions(-) diff --git a/leaderboard/leaderboard.py b/leaderboard/leaderboard.py index 926fa41..23694df 100644 --- a/leaderboard/leaderboard.py +++ b/leaderboard/leaderboard.py @@ -1,4 +1,4 @@ -from redbot.core import commands, Config +from redbot.core import commands, Config, bank from redbot.core.bot import Red from redbot.core.utils.chat_formatting import box, pagify, humanize_number import discord @@ -109,87 +109,65 @@ class Leaderboard(commands.Cog): 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 - + async def get_user_points(self, user: discord.Member) -> int: + """Get a user's total points combining bank balance and levelup XP.""" + total_points = 0 + + # Get bank balance 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 + balance = await bank.get_balance(user) + total_points += balance except Exception as e: - log.error(f"Unexpected error fetching leaderboard: {e}") + log.error(f"Error getting bank balance for {user}: {e}") + + # Get LevelUp XP if cog is loaded + levelup = self.bot.get_cog("LevelUp") + if levelup: + try: + conf = levelup.db.get_conf(user.guild) + profile = conf.get_profile(user) + total_points += int(profile.xp) + except Exception as e: + log.error(f"Error getting LevelUp XP for {user}: {e}") + + return total_points + + async def _get_leaderboard(self) -> Optional[list]: + """Fetch and combine leaderboard data from bank and LevelUp.""" + try: + all_users = {} + + # Get all guild members + for guild in self.bot.guilds: + for member in guild.members: + if member.bot: + continue + + points = await self.get_user_points(member) + + if points > 0: + all_users[str(member.id)] = { + "userId": str(member.id), + "username": str(member), + "points": points + } + + # Sort by points and convert to list + sorted_users = sorted( + all_users.values(), + key=lambda x: x["points"], + reverse=True + ) + + return sorted_users + + except Exception as e: + log.error(f"Error generating 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 + """Update points is no longer needed as points are pulled from bank and LevelUp.""" + return True @commands.group(name="globalboard", aliases=["glb"]) async def globalboard(self, ctx: commands.Context):