316 lines
No EOL
12 KiB
Python
316 lines
No EOL
12 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)
|
|
self.session = None # Initialize in setup
|
|
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 = {
|
|
"points_per_message": 1,
|
|
"points_decay": 0.5, # Points lost per day of inactivity
|
|
"min_message_length": 5,
|
|
"cooldown": 60, # Seconds between point gains
|
|
}
|
|
|
|
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."""
|
|
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_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:
|
|
balance = await bank.get_balance(user)
|
|
total_points += balance
|
|
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:
|
|
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 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):
|
|
"""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 leaderboard."""
|
|
async with ctx.typing():
|
|
leaderboard_data = await self._get_leaderboard()
|
|
|
|
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
|
|
return await ctx.send("The leaderboard is currently empty!")
|
|
|
|
embeds = []
|
|
for page_num, entries in enumerate(chunks, 1):
|
|
embed = discord.Embed(
|
|
title="🏆 Global Leaderboard",
|
|
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")
|
|
points = entry.get("points", 0)
|
|
|
|
# Format each entry with position, name, points, and user ID
|
|
description.append(
|
|
f"`{i}.` <@{user_id}> • **{humanize_number(points)}** points"
|
|
)
|
|
|
|
embed.description = "\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="points")
|
|
async def check_points(self, ctx: commands.Context, member: commands.MemberConverter = None):
|
|
"""Check your points or another member's points."""
|
|
member = member or ctx.author
|
|
leaderboard_data = await self._get_leaderboard()
|
|
|
|
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)),
|
|
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 Points",
|
|
color=await ctx.embed_color()
|
|
)
|
|
embed.description = (
|
|
f"**User:** <@{member.id}>\n"
|
|
f"**Points:** {humanize_number(user_data.get('points', 0))}\n"
|
|
f"**Rank:** #{humanize_number(rank)}\n"
|
|
f"**ID:** {member.id}"
|
|
)
|
|
embed.set_thumbnail(url=member.display_avatar.url)
|
|
|
|
await ctx.send(embed=embed)
|
|
else:
|
|
embed = discord.Embed(
|
|
title="No Points Found",
|
|
description=f"<@{member.id}> has no points yet!",
|
|
color=await ctx.embed_color()
|
|
)
|
|
embed.set_footer(text=f"ID: {member.id}")
|
|
await ctx.send(embed=embed)
|
|
|
|
@commands.Cog.listener()
|
|
async def on_message(self, message):
|
|
"""Award points for activity."""
|
|
if message.author.bot or not message.guild:
|
|
return
|
|
|
|
# Get guild settings
|
|
guild_settings = await self.config.guild(message.guild).all()
|
|
points_per_message = guild_settings["points_per_message"]
|
|
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
|
|
|
|
# Get current points
|
|
leaderboard_data = await self._get_leaderboard()
|
|
current_points = 0
|
|
|
|
if leaderboard_data:
|
|
user_data = next(
|
|
(entry for entry in leaderboard_data if entry.get("userId") == str(message.author.id)),
|
|
None
|
|
)
|
|
if user_data:
|
|
current_points = user_data.get("points", 0)
|
|
|
|
# Update points
|
|
new_points = current_points + points_per_message
|
|
await self._update_points(
|
|
str(message.author.id),
|
|
str(message.author),
|
|
new_points
|
|
) |