ruby/Ruby-Cogs
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Refactor Leaderboard cog to enhance user credit retrieval and API integration. Simplify session management, remove redundant initialization logic, and improve error handling for credit updates. Introduce a new method for fetching all user balances and streamline leaderboard display logic.
Some checks are pending
Run pre-commit / Run pre-commit (push) Waiting to run
Details

Browse source
This commit is contained in:
Valerie 2025-05-26 04:56:55 -04:00
parent 5ffc18d442
commit 0114307e67
1 changed files with 64 additions and 158 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

222
leaderboard/leaderboard.py
View file

@ -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,<your_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()
asyncio.create_task(self.session.close())