541 lines
No EOL
22 KiB
Python
541 lines
No EOL
22 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
|
|
|
|
# Add task for auto-sync
|
|
self.auto_sync_task = 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 and start auto-sync task."""
|
|
try:
|
|
self.admin_secret = await self.bot.get_shared_api_tokens("ruby_api")
|
|
if self.admin_secret.get("admin_secret"):
|
|
self.session = aiohttp.ClientSession()
|
|
# Test API connection
|
|
try:
|
|
async with self.session.get(
|
|
f"{self.api_base_url}/leaderboard",
|
|
headers={"Authorization": self.admin_secret["admin_secret"]}
|
|
) as resp:
|
|
if resp.status == 200:
|
|
# Start auto-sync task if API connection successful
|
|
self.auto_sync_task = self.bot.loop.create_task(self._auto_sync_task())
|
|
log.info("Started automatic leaderboard sync task")
|
|
elif resp.status == 401:
|
|
log.error("Failed to initialize API: Invalid admin secret")
|
|
else:
|
|
log.error(f"Failed to initialize API: Status {resp.status}")
|
|
try:
|
|
data = await resp.json()
|
|
if data.get("error"):
|
|
log.error(f"Error message: {data['error']}")
|
|
except aiohttp.ContentTypeError:
|
|
log.error("API returned invalid JSON response during initialization")
|
|
except aiohttp.ClientError as e:
|
|
log.error(f"Failed to connect to API: {str(e)}")
|
|
except Exception as e:
|
|
log.error(f"Failed to initialize API connection: {str(e)}")
|
|
|
|
async def _auto_sync_task(self):
|
|
"""Task to automatically sync leaderboard every 6 hours."""
|
|
await self.bot.wait_until_ready()
|
|
while True:
|
|
try:
|
|
await self._perform_full_sync()
|
|
log.info("Completed automatic leaderboard sync")
|
|
await asyncio.sleep(21600) # 6 hours in seconds
|
|
except asyncio.CancelledError:
|
|
break
|
|
except Exception as e:
|
|
log.error(f"Error in automatic sync task: {e}")
|
|
await asyncio.sleep(300) # Wait 5 minutes before retrying on error
|
|
|
|
async def _perform_full_sync(self) -> tuple[int, int]:
|
|
"""Perform a full sync of all users to the API.
|
|
Returns tuple of (success_count, fail_count)"""
|
|
success_count = 0
|
|
fail_count = 0
|
|
processed_users = set()
|
|
|
|
for guild in self.bot.guilds:
|
|
for member in guild.members:
|
|
if member.bot or member.id in processed_users:
|
|
continue
|
|
|
|
try:
|
|
# Get total credits across all guilds
|
|
credits = await self.get_user_credits(member)
|
|
|
|
# Update API
|
|
if await self._try_api_update(str(member.id), str(member), credits):
|
|
success_count += 1
|
|
else:
|
|
fail_count += 1
|
|
|
|
processed_users.add(member.id)
|
|
except Exception as e:
|
|
log.error(f"Error syncing user {member} ({member.id}): {e}")
|
|
fail_count += 1
|
|
|
|
return success_count, fail_count
|
|
|
|
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 all servers."""
|
|
total_credits = 0
|
|
|
|
try:
|
|
# Get credits from all servers the user is in
|
|
for guild in self.bot.guilds:
|
|
try:
|
|
member = guild.get_member(user.id)
|
|
if member: # If user is in this guild
|
|
guild_credits = await bank.get_balance(member)
|
|
total_credits += guild_credits
|
|
except Exception as e:
|
|
log.error(f"Error getting bank balance for {user} in {guild}: {e}")
|
|
continue
|
|
|
|
except Exception as e:
|
|
log.error(f"Error getting total credits for {user}: {e}")
|
|
|
|
return total_credits
|
|
|
|
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 - matches API's MIN_CREDITS
|
|
|
|
# First get all unique members across all guilds
|
|
processed_users = set()
|
|
|
|
# Process each guild
|
|
for guild in self.bot.guilds:
|
|
for member in guild.members:
|
|
if member.bot or member.id in processed_users:
|
|
continue
|
|
|
|
# Skip opted-out users
|
|
try:
|
|
if await self.config.user(member).opted_out():
|
|
continue
|
|
except Exception as e:
|
|
log.error(f"Error checking opt-out status for {member}: {e}")
|
|
continue
|
|
|
|
try:
|
|
total_credits = await self.get_user_credits(member)
|
|
|
|
if total_credits >= min_credits:
|
|
all_users[member.id] = {
|
|
"userId": str(member.id),
|
|
"username": str(member),
|
|
"points": total_credits
|
|
}
|
|
|
|
processed_users.add(member.id)
|
|
except Exception as e:
|
|
log.error(f"Error getting balance for {member}: {e}")
|
|
continue
|
|
|
|
# Sort by total credits
|
|
sorted_users = sorted(
|
|
all_users.values(),
|
|
key=lambda x: x["points"],
|
|
reverse=True
|
|
)
|
|
|
|
# Debug logging
|
|
log.info(f"Found {len(sorted_users)} users with {min_credits}+ credits")
|
|
for user in sorted_users[:10]: # Log top 10 for debugging
|
|
log.info(f"User {user['username']} has {user['points']} credits")
|
|
|
|
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:
|
|
if credits < 0:
|
|
credits = 0 # API doesn't accept negative values
|
|
|
|
# Get opt-out status
|
|
opted_out = await self.config.user_from_id(int(user_id)).opted_out()
|
|
|
|
async with self.session.post(
|
|
f"{self.api_base_url}/leaderboard",
|
|
headers={
|
|
"Authorization": self.admin_secret["admin_secret"],
|
|
"Content-Type": "application/json"
|
|
},
|
|
json={
|
|
"userId": str(user_id),
|
|
"username": str(username),
|
|
"points": credits,
|
|
"optedOut": opted_out
|
|
}
|
|
) as resp:
|
|
if resp.status == 200:
|
|
try:
|
|
data = await resp.json()
|
|
if data.get("success"):
|
|
log.info(f"Updated leaderboard for {username}: {credits} credits (opted_out: {opted_out})")
|
|
return True
|
|
else:
|
|
log.error(f"API update failed: {data.get('error', 'Unknown error')}")
|
|
except aiohttp.ContentTypeError:
|
|
log.error("API returned invalid JSON response")
|
|
elif resp.status == 401:
|
|
log.error("Unauthorized: Invalid admin secret")
|
|
else:
|
|
log.error(f"API update failed: Status {resp.status}")
|
|
try:
|
|
data = await resp.json()
|
|
if data.get("error"):
|
|
log.error(f"Error message: {data['error']}")
|
|
except aiohttp.ContentTypeError:
|
|
pass
|
|
return False
|
|
except aiohttp.ClientError as e:
|
|
log.error(f"API request failed: {str(e)}")
|
|
return False
|
|
except Exception as e:
|
|
log.error(f"Unexpected error during API update: {str(e)}")
|
|
return False
|
|
|
|
async def _get_api_leaderboard(self) -> Optional[list]:
|
|
"""Fetch the leaderboard from the API."""
|
|
if not self.session or not self.admin_secret or not self.admin_secret.get("admin_secret"):
|
|
return None
|
|
|
|
try:
|
|
async with self.session.get(
|
|
f"{self.api_base_url}/leaderboard",
|
|
headers={"Authorization": self.admin_secret["admin_secret"]}
|
|
) as resp:
|
|
if resp.status == 200:
|
|
try:
|
|
data = await resp.json()
|
|
if "leaderboard" in data:
|
|
return data["leaderboard"]
|
|
log.error("Invalid API response format")
|
|
except aiohttp.ContentTypeError:
|
|
log.error("API returned invalid JSON response")
|
|
elif resp.status == 401:
|
|
log.error("Unauthorized: Invalid admin secret")
|
|
else:
|
|
log.error(f"Failed to fetch leaderboard: Status {resp.status}")
|
|
try:
|
|
data = await resp.json()
|
|
if data.get("error"):
|
|
log.error(f"Error message: {data['error']}")
|
|
except aiohttp.ContentTypeError:
|
|
pass
|
|
return None
|
|
except aiohttp.ClientError as e:
|
|
log.error(f"Error fetching leaderboard: {str(e)}")
|
|
return None
|
|
except Exception as e:
|
|
log.error(f"Unexpected error fetching leaderboard: {str(e)}")
|
|
return None
|
|
|
|
@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!")
|
|
|
|
# Send warning message first
|
|
await ctx.send("⚠️ **Note:** The global leaderboard may be slightly inaccurate due to synchronization delays.")
|
|
|
|
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)
|
|
|
|
# Update API with new opt-out status
|
|
credits = await self.get_user_credits(ctx.author)
|
|
await self._try_api_update(str(ctx.author.id), str(ctx.author), credits)
|
|
|
|
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)
|
|
|
|
# Update API with new opt-in status
|
|
credits = await self.get_user_credits(ctx.author)
|
|
await self._try_api_update(str(ctx.author.id), str(ctx.author), credits)
|
|
|
|
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)
|
|
|
|
# Debug logging
|
|
log.info(f"Credits check for {member}: {credits} credits")
|
|
|
|
# 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.mention}\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.mention} has less than 10,000 credits ({humanize_number(credits)} total)"
|
|
|
|
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)."""
|
|
if not self.session or not self.admin_secret or not self.admin_secret.get("admin_secret"):
|
|
await ctx.send("API is not configured.")
|
|
return
|
|
|
|
async with ctx.typing():
|
|
message = await ctx.send("🔄 Starting global leaderboard resync...")
|
|
|
|
try:
|
|
success_count, fail_count = await self._perform_full_sync()
|
|
|
|
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(),
|
|
timestamp=datetime.now(timezone.utc)
|
|
)
|
|
await message.edit(content=None, embed=embed)
|
|
except Exception as e:
|
|
await message.edit(content=f"❌ Error during resync: {str(e)}")
|
|
|
|
@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
|
|
|
|
total_credits = await self.get_user_credits(member)
|
|
await self._try_api_update(str(member.id), str(member), total_credits)
|
|
|
|
@globalboard.command(name="cleanup")
|
|
@commands.is_owner()
|
|
async def cleanup_leaderboard(self, ctx: commands.Context):
|
|
"""Clean up old leaderboard entries (older than 30 days)."""
|
|
if not self.session or not self.admin_secret or not self.admin_secret.get("admin_secret"):
|
|
await ctx.send("API is not configured.")
|
|
return
|
|
|
|
try:
|
|
async with self.session.delete(
|
|
f"{self.api_base_url}/leaderboard",
|
|
headers={"Authorization": self.admin_secret["admin_secret"]}
|
|
) as resp:
|
|
if resp.status == 200:
|
|
data = await resp.json()
|
|
if data.get("success"):
|
|
await ctx.send(
|
|
f"Cleaned up {data['removedCount']} old entries. "
|
|
f"{data['remainingCount']} entries remaining."
|
|
)
|
|
else:
|
|
await ctx.send("Failed to clean up leaderboard.")
|
|
elif resp.status == 401:
|
|
await ctx.send("Unauthorized: Invalid admin secret")
|
|
else:
|
|
await ctx.send(f"Failed to clean up leaderboard: Status {resp.status}")
|
|
except Exception as e:
|
|
await ctx.send(f"Error cleaning up leaderboard: {e}")
|
|
|
|
def cog_unload(self):
|
|
"""Clean up when cog is unloaded."""
|
|
if self.session:
|
|
asyncio.create_task(self.session.close())
|
|
if self.auto_sync_task:
|
|
self.auto_sync_task.cancel() |