Valerie
49902e8719
Some checks are pending
Run pre-commit / Run pre-commit (push) Waiting to run
Implement user options to opt-out of the global leaderboard, ensuring their credits are not displayed. Update leaderboard display logic to only show users with 10,000 or more credits, enhancing user privacy and experience. Adjust error messages and embed descriptions accordingly for clarity.
340 lines
No EOL
13 KiB
Python
340 lines
No EOL
13 KiB
Python
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
|
|
from discord.ui import Button, View
|
|
import aiohttp
|
|
import asyncio
|
|
import logging
|
|
from datetime import datetime, timezone
|
|
from typing import Dict, Optional, List
|
|
|
|
log = logging.getLogger("red.ruby.leaderboard")
|
|
|
|
class LeaderboardView(View):
|
|
def __init__(self, cog, ctx, pages: List[discord.Embed], timeout: float = 180.0):
|
|
super().__init__(timeout=timeout)
|
|
self.cog = cog
|
|
self.ctx = ctx
|
|
self.pages = pages
|
|
self.current_page = 0
|
|
self.message = None
|
|
|
|
# Add buttons
|
|
self.first_page.disabled = True
|
|
self.prev_page.disabled = True
|
|
if len(self.pages) == 1:
|
|
self.next_page.disabled = True
|
|
self.last_page.disabled = True
|
|
|
|
@discord.ui.button(label="≪", style=discord.ButtonStyle.gray)
|
|
async def first_page(self, interaction: discord.Interaction, button: discord.ui.Button):
|
|
self.current_page = 0
|
|
await self.update_page(interaction)
|
|
|
|
@discord.ui.button(label="Previous", style=discord.ButtonStyle.blurple)
|
|
async def prev_page(self, interaction: discord.Interaction, button: discord.ui.Button):
|
|
self.current_page = max(0, self.current_page - 1)
|
|
await self.update_page(interaction)
|
|
|
|
@discord.ui.button(label="Next", style=discord.ButtonStyle.blurple)
|
|
async def next_page(self, interaction: discord.Interaction, button: discord.ui.Button):
|
|
self.current_page = min(len(self.pages) - 1, self.current_page + 1)
|
|
await self.update_page(interaction)
|
|
|
|
@discord.ui.button(label="≫", style=discord.ButtonStyle.gray)
|
|
async def last_page(self, interaction: discord.Interaction, button: discord.ui.Button):
|
|
self.current_page = len(self.pages) - 1
|
|
await self.update_page(interaction)
|
|
|
|
async def update_page(self, interaction: discord.Interaction):
|
|
# Update button states
|
|
self.first_page.disabled = self.current_page == 0
|
|
self.prev_page.disabled = self.current_page == 0
|
|
self.next_page.disabled = self.current_page == len(self.pages) - 1
|
|
self.last_page.disabled = self.current_page == len(self.pages) - 1
|
|
|
|
await interaction.response.edit_message(embed=self.pages[self.current_page], view=self)
|
|
|
|
async def interaction_check(self, interaction: discord.Interaction) -> bool:
|
|
if interaction.user.id == self.ctx.author.id:
|
|
return True
|
|
await interaction.response.send_message("This menu is not for you!", ephemeral=True)
|
|
return False
|
|
|
|
async def on_timeout(self):
|
|
try:
|
|
for item in self.children:
|
|
item.disabled = True
|
|
await self.message.edit(view=self)
|
|
except discord.HTTPException:
|
|
pass
|
|
|
|
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)
|
|
|
|
# Optional API integration
|
|
self.session = None
|
|
self.api_base_url = "https://ruby.valerie.lol/api"
|
|
self.admin_secret = None
|
|
|
|
default_guild = {
|
|
"min_message_length": 5,
|
|
"cooldown": 60,
|
|
}
|
|
|
|
default_user = {
|
|
"opted_out": False
|
|
}
|
|
|
|
self.config.register_guild(**default_guild)
|
|
self.config.register_user(**default_user)
|
|
|
|
async def initialize(self):
|
|
"""Initialize optional API connection."""
|
|
try:
|
|
self.admin_secret = await self.bot.get_shared_api_tokens("ruby_api")
|
|
if self.admin_secret.get("admin_secret"):
|
|
self.session = aiohttp.ClientSession()
|
|
except Exception as e:
|
|
log.error(f"Failed to initialize API connection: {e}")
|
|
# Don't return anything - initialization is optional
|
|
|
|
async def cog_load(self):
|
|
"""Initialize the cog when it's loaded."""
|
|
await self.initialize()
|
|
|
|
async def get_user_credits(self, user: discord.Member) -> int:
|
|
"""Get a user's total credits from Red's bank system."""
|
|
try:
|
|
return await bank.get_balance(user)
|
|
except Exception as e:
|
|
log.error(f"Error getting bank balance for {user}: {e}")
|
|
return 0
|
|
|
|
async def get_all_balances(self) -> List[dict]:
|
|
"""Get all users' credit balances across all servers."""
|
|
all_users = {}
|
|
min_credits = 10000 # Minimum credits to show on leaderboard
|
|
|
|
# Collect all unique members and sum their balances across all guilds
|
|
for guild in self.bot.guilds:
|
|
for member in guild.members:
|
|
if member.bot:
|
|
continue
|
|
|
|
# Skip opted-out users
|
|
if await self.config.user(member).opted_out():
|
|
continue
|
|
|
|
try:
|
|
# Get balance for this guild
|
|
credits = await bank.get_balance(member)
|
|
|
|
if member.id not in all_users:
|
|
all_users[member.id] = {
|
|
"userId": str(member.id),
|
|
"username": str(member),
|
|
"points": credits
|
|
}
|
|
else:
|
|
# Add this guild's balance to user's total
|
|
all_users[member.id]["points"] += credits
|
|
|
|
except Exception as e:
|
|
log.error(f"Error getting balance for {member} in {guild}: {e}")
|
|
continue
|
|
|
|
# Filter out users with less than minimum credits and sort by total
|
|
sorted_users = sorted(
|
|
[user for user in all_users.values() if user["points"] >= min_credits],
|
|
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={
|
|
"Authorization": self.admin_secret.get("admin_secret"),
|
|
"Content-Type": "application/json"
|
|
},
|
|
json={
|
|
"userId": user_id,
|
|
"username": username,
|
|
"points": credits
|
|
}
|
|
) as resp:
|
|
return resp.status == 200
|
|
except Exception as 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:
|
|
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_all_balances()
|
|
|
|
if not leaderboard_data:
|
|
return await ctx.send("No users have 10,000 or more credits!")
|
|
|
|
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:
|
|
return await ctx.send("No users have 10,000 or more credits!")
|
|
|
|
embeds = []
|
|
for page_num, entries in enumerate(chunks, 1):
|
|
embed = discord.Embed(
|
|
title="🏆 Global Credits Leaderboard",
|
|
description="*Only showing users with 10,000+ credits*",
|
|
color=await ctx.embed_color()
|
|
)
|
|
|
|
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)
|
|
|
|
description.append(
|
|
f"`{i}.` <@{user_id}> • **{humanize_number(credits)}** credits"
|
|
)
|
|
|
|
embed.description = embed.description + "\n\n" + "\n".join(description)
|
|
embed.set_footer(text=f"Page {page_num}/{len(chunks)} • Total Users: {len(leaderboard_data)}")
|
|
embeds.append(embed)
|
|
|
|
view = LeaderboardView(self, ctx, embeds)
|
|
view.message = await ctx.send(embed=embeds[0], view=view)
|
|
|
|
@globalboard.command(name="optout")
|
|
async def opt_out(self, ctx: commands.Context):
|
|
"""Opt out of the global leaderboard. Your credits will no longer be visible."""
|
|
user_settings = self.config.user(ctx.author)
|
|
current_status = await user_settings.opted_out()
|
|
|
|
if current_status:
|
|
await ctx.send("You are already opted out of the global leaderboard.")
|
|
return
|
|
|
|
await user_settings.opted_out.set(True)
|
|
await ctx.send("You have been opted out of the global leaderboard. Your credits will no longer be visible.")
|
|
|
|
@globalboard.command(name="optin")
|
|
async def opt_in(self, ctx: commands.Context):
|
|
"""Opt back into the global leaderboard."""
|
|
user_settings = self.config.user(ctx.author)
|
|
current_status = await user_settings.opted_out()
|
|
|
|
if not current_status:
|
|
await ctx.send("You are already opted into the global leaderboard.")
|
|
return
|
|
|
|
await user_settings.opted_out.set(False)
|
|
await ctx.send("You have been opted back into the global leaderboard. Your credits will now be visible.")
|
|
|
|
@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
|
|
|
|
# Check if the target user has opted out
|
|
if await self.config.user(member).opted_out():
|
|
if member == ctx.author:
|
|
await ctx.send("You have opted out of the global leaderboard. Use `[p]globalboard optin` to show your credits again.")
|
|
else:
|
|
await ctx.send(f"{member.display_name} has opted out of the global leaderboard.")
|
|
return
|
|
|
|
credits = await self.get_user_credits(member)
|
|
|
|
# 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
|
|
)
|
|
|
|
embed = discord.Embed(
|
|
title="🏆 Global Credits",
|
|
color=await ctx.embed_color()
|
|
)
|
|
|
|
if credits >= 10000:
|
|
embed.description = (
|
|
f"**User:** <@{member.id}>\n"
|
|
f"**Credits:** {humanize_number(credits)}\n"
|
|
f"**Rank:** #{humanize_number(rank) if rank else 'Unranked'}\n"
|
|
f"**ID:** {member.id}"
|
|
)
|
|
else:
|
|
embed.description = f"<@{member.id}> has less than 10,000 credits."
|
|
|
|
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 API (if configured)."""
|
|
async with ctx.typing():
|
|
success_count = 0
|
|
fail_count = 0
|
|
|
|
for guild in self.bot.guilds:
|
|
for member in guild.members:
|
|
if member.bot:
|
|
continue
|
|
|
|
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
|
|
|
|
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_bank_update(self, member: discord.Member, before: int, after: int):
|
|
"""Update API when a member's bank balance changes."""
|
|
if member.bot:
|
|
return
|
|
|
|
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()) |