Refactor Leaderboard cog to replace points with credits, updating related methods and commands for consistency. Enhance user experience by renaming commands and improving error handling during credit updates. Introduce a resync command for global leaderboard updates.
Some checks are pending
Run pre-commit / Run pre-commit (push) Waiting to run

This commit is contained in:
Valerie 2025-05-26 04:48:58 -04:00
parent 57c7b1aaa3
commit 8622d7e223

View file

@ -85,17 +85,11 @@ class Leaderboard(commands.Cog):
# 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
"cooldown": 60, # Seconds between updates
}
self.config.register_guild(**default_guild)
def cog_unload(self):
if self.session:
asyncio.create_task(self.session.close())
async def initialize(self):
"""Load the admin secret from bot config and initialize session."""
@ -109,31 +103,17 @@ class Leaderboard(commands.Cog):
self.session = aiohttp.ClientSession()
return True
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 (credits)
async def get_user_credits(self, user: discord.Member) -> int:
"""Get a user's total credits from Red's bank system."""
try:
if await bank.is_global():
balance = await bank.get_balance(user)
credits = await bank.get_balance(user)
else:
balance = await bank.get_balance(user, _forced=True)
total_points += balance
credits = await bank.get_balance(user, _forced=True)
return credits
except Exception as 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:
profile = await levelup.get_user_profile(user)
if profile:
total_points += int(profile.total_xp)
except Exception as e:
log.error(f"Error getting LevelUp XP for {user}: {e}")
return total_points
return 0
async def _get_leaderboard(self) -> Optional[list]:
"""Fetch the global leaderboard from the API."""
@ -174,8 +154,8 @@ class Leaderboard(commands.Cog):
log.error(f"Error fetching leaderboard: {e}")
return None
async def _update_points(self, user_id: str, username: str, points: int = None) -> bool:
"""Update a user's points in the global leaderboard."""
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
@ -185,8 +165,8 @@ class Leaderboard(commands.Cog):
return False
try:
# If points not provided, calculate from Red's systems
if points is None:
# 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))
@ -196,7 +176,7 @@ class Leaderboard(commands.Cog):
if not member:
return False
points = await self.get_user_points(member)
credits = await self.get_user_credits(member)
async with self.session.post(
f"{self.api_base_url}/leaderboard",
@ -207,7 +187,7 @@ class Leaderboard(commands.Cog):
json={
"userId": user_id,
"username": username,
"points": points
"points": credits # API still uses "points" field
}
) as resp:
if resp.status == 200:
@ -218,10 +198,10 @@ class Leaderboard(commands.Cog):
log.error("Unauthorized: Invalid admin secret")
return False
else:
log.error(f"Failed to update points: Status {resp.status}")
log.error(f"Failed to update credits: Status {resp.status}")
return False
except Exception as e:
log.error(f"Error updating points: {e}")
log.error(f"Error updating credits: {e}")
return False
@commands.group(name="globalboard", aliases=["glb"])
@ -233,7 +213,7 @@ class Leaderboard(commands.Cog):
@globalboard.command(name="show")
async def show_leaderboard(self, ctx: commands.Context, page: int = 1):
"""Show the global leaderboard."""
"""Show the global credits leaderboard."""
async with ctx.typing():
leaderboard_data = await self._get_leaderboard()
@ -255,7 +235,7 @@ class Leaderboard(commands.Cog):
embeds = []
for page_num, entries in enumerate(chunks, 1):
embed = discord.Embed(
title="🏆 Global Leaderboard",
title="🏆 Global Credits Leaderboard",
color=await ctx.embed_color()
)
@ -266,11 +246,11 @@ class Leaderboard(commands.Cog):
for i, entry in enumerate(entries, start=start_pos + 1):
username = entry.get("username", "Unknown User")
user_id = entry.get("userId", "0")
points = entry.get("points", 0)
credits = entry.get("points", 0) # API returns as "points"
# Format each entry with position, name, points, and user ID
# Format each entry with position, name, credits, and user ID
description.append(
f"`{i}.` <@{user_id}> • **{humanize_number(points)}** points"
f"`{i}.` <@{user_id}> • **{humanize_number(credits)}** credits"
)
embed.description = "\n".join(description)
@ -280,9 +260,9 @@ class Leaderboard(commands.Cog):
view = LeaderboardView(self, ctx, embeds)
view.message = await ctx.send(embed=embeds[0], view=view)
@globalboard.command(name="points")
async def check_points(self, ctx: commands.Context, member: commands.MemberConverter = None):
"""Check your points or another member's points."""
@globalboard.command(name="credits")
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()
@ -304,12 +284,12 @@ class Leaderboard(commands.Cog):
)
embed = discord.Embed(
title="🏆 Global Points",
title="🏆 Global Credits",
color=await ctx.embed_color()
)
embed.description = (
f"**User:** <@{member.id}>\n"
f"**Points:** {humanize_number(user_data.get('points', 0))}\n"
f"**Credits:** {humanize_number(user_data.get('points', 0))}\n"
f"**Rank:** #{humanize_number(rank)}\n"
f"**ID:** {member.id}"
)
@ -318,45 +298,60 @@ class Leaderboard(commands.Cog):
await ctx.send(embed=embed)
else:
embed = discord.Embed(
title="No Points Found",
description=f"<@{member.id}> has no points yet!",
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)
@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."""
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)):
success_count += 1
else:
fail_count += 1
embed = discord.Embed(
title="🔄 Global Leaderboard Resync Complete",
description=(
f"Successfully updated: **{success_count}** users\n"
f"Failed to update: **{fail_count}** users"
),
color=await ctx.embed_color()
)
await ctx.send(embed=embed)
@commands.Cog.listener()
async def on_member_update(self, before: discord.Member, after: discord.Member):
"""Update points when member's data changes."""
"""Update credits when member's data changes."""
if before.bot:
return
# Update points in API when bank balance or XP changes
await self._update_points(str(after.id), str(after))
# Update credits in API
await self._update_credits(str(after.id), str(after))
@commands.Cog.listener()
async def on_message(self, message: discord.Message):
"""Update points periodically during user activity."""
if message.author.bot or not message.guild:
async def on_bank_update(self, member: discord.Member, before: int, after: int):
"""Update credits when a member's bank balance changes."""
if member.bot:
return
# Get guild settings
guild_settings = await self.config.guild(message.guild).all()
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
# Update points in API
await self._update_points(str(message.author.id), str(message.author))
# Update credits in API with the new balance
await self._update_credits(str(member.id), str(member), after)