246 lines
9.7 KiB
Python
Executable File
246 lines
9.7 KiB
Python
Executable File
import discord
|
|
from discord.ext import commands
|
|
from utils.sql_commands import DatabaseManager
|
|
from utils.bank_functions import *
|
|
from random import randint
|
|
from math import log2
|
|
from datetime import datetime
|
|
import io
|
|
from PIL import Image, ImageDraw, ImageFont
|
|
import logging
|
|
import os
|
|
|
|
|
|
def calculate_xp_needed_for_next_level(level: int) -> int:
|
|
"""Calculates the XP needed for the next level."""
|
|
return int(35 * (level**2) + 35 * level)
|
|
|
|
|
|
class XP(commands.Cog):
|
|
|
|
def __init__(self, client):
|
|
self.client = client
|
|
self.db = DatabaseManager("")
|
|
|
|
@commands.Cog.listener()
|
|
async def on_message(self, ctx: discord.Message) -> None:
|
|
if ctx.author.bot:
|
|
return
|
|
|
|
try:
|
|
data = self.db.fetch_one(
|
|
"SELECT XP, LEVEL FROM users WHERE ID = %s", (ctx.author.id,)
|
|
)
|
|
if data is None:
|
|
# Insert a new row for this user
|
|
current_level = 0
|
|
current_xp = 0
|
|
self.db.execute_query(
|
|
"INSERT INTO users (ID, XP, LEVEL) VALUES (%s, %s, %s)",
|
|
(ctx.author.id, current_xp, current_level),
|
|
)
|
|
else:
|
|
current_level = data.get("LEVEL", 0)
|
|
current_xp = data.get("XP", 0)
|
|
|
|
logging.info(f"XP: {current_xp}, Level: {current_level}")
|
|
|
|
# Balanced XP calculation
|
|
base = randint(5, 10)
|
|
length_bonus = (
|
|
min(len(ctx.content), 100) // 10
|
|
) # +1 XP per 10 chars, max +10
|
|
extra_xp = base + length_bonus
|
|
extra_xp = min(extra_xp, 20) # cap at 20 XP per message
|
|
current_xp += extra_xp
|
|
|
|
if current_xp > calculate_xp_needed_for_next_level(current_level):
|
|
current_level += 1
|
|
player_balance = self.db.fetch_one(
|
|
"SELECT BANK FROM economy WHERE ID = %s", (ctx.author.id,)
|
|
)
|
|
player_balance = player_balance.get("BANK", 0) if player_balance else 0
|
|
levelup_bonus = randint(500, 2000)
|
|
new_balance = player_balance + levelup_bonus
|
|
await update_money(ctx.author, bank=new_balance)
|
|
|
|
await ctx.channel.send(
|
|
f"Congrats!! You achieved a new level.\nYou are now level {current_level}. You received {levelup_bonus}<:flooney:1194943899765051473>"
|
|
)
|
|
current_xp = 0
|
|
|
|
# Use INSERT ... ON DUPLICATE KEY UPDATE for XP/level
|
|
self.db.execute_query(
|
|
"INSERT INTO users (ID, XP, LEVEL) VALUES (%s, %s, %s) "
|
|
"ON DUPLICATE KEY UPDATE XP = %s, LEVEL = %s",
|
|
(ctx.author.id, current_xp, current_level, current_xp, current_level),
|
|
)
|
|
except Exception as e:
|
|
logging.error(f"Error in XP on_message: {e}")
|
|
|
|
@commands.command()
|
|
async def top(self, ctx: commands.Context) -> None:
|
|
embed = discord.Embed(
|
|
colour=discord.Colour.purple(),
|
|
timestamp=ctx.message.created_at,
|
|
title="Top Messengers",
|
|
)
|
|
users = self.db.fetch_all("SELECT ID, XP, LEVEL FROM users")
|
|
if not users:
|
|
return
|
|
|
|
sorted_users = sorted(users, key=lambda x: int(x["XP"]), reverse=True)[:10]
|
|
data = []
|
|
|
|
for index, member in enumerate(sorted_users, start=1):
|
|
try:
|
|
member_obj = await self.client.fetch_user(int(member["ID"]))
|
|
member_name = (
|
|
member_obj.display_name
|
|
if hasattr(member_obj, "display_name")
|
|
else str(member_obj)
|
|
)
|
|
except Exception:
|
|
member_name = f"User {member['ID']}"
|
|
member_xp, member_lvl = int(member["XP"]), int(member["LEVEL"])
|
|
medals = ["🥇", "🥈", "🥉"]
|
|
msg = f"**{medals[index - 1] if index <= 3 else index} `{member_name}` -- {member_xp:,}xp, {member_lvl:,}level**"
|
|
data.append(msg)
|
|
|
|
msg = "\n".join(data)
|
|
embed.description = f"It's Based on XP of Global Users\n\n{msg}"
|
|
guild_name = ctx.guild.name if ctx.guild is not None else "Direct Message"
|
|
embed.set_footer(text=f"GLOBAL - {guild_name}")
|
|
embed.color = discord.Color(0x00FF00)
|
|
await ctx.reply(embed=embed, mention_author=False)
|
|
|
|
@commands.command()
|
|
async def stats(self, ctx: commands.Context) -> None:
|
|
"""Shows the user's current XP and level."""
|
|
try:
|
|
# Use a fallback background if the random one doesn't exist
|
|
bg_path = f"images/image_{randint(1,100)}.jpg"
|
|
if not os.path.exists(bg_path):
|
|
bg_path = "images/default.jpg"
|
|
profile_image = await self.create_profile_card(ctx.author, bg_path)
|
|
|
|
await ctx.send(file=discord.File(profile_image, "profile.png"))
|
|
|
|
if profile_image is not None:
|
|
profile_image.close()
|
|
|
|
except ZeroDivisionError:
|
|
await ctx.send(f"{ctx.author.mention}, you haven't earned any XP yet.")
|
|
except Exception as e:
|
|
await ctx.send(f"Error creating profile card: {e}")
|
|
logging.error(f"Error creating profile card: {e}")
|
|
|
|
async def create_profile_card(
|
|
self, user: discord.Member | discord.User, background_image_path: str
|
|
) -> io.BytesIO:
|
|
"""Create a profile card image for a user with a custom background and text overlay."""
|
|
card_width, card_height = 400, 200
|
|
text_color = (255, 255, 255)
|
|
progress_bar_color = (0, 255, 0)
|
|
outline_color = (255, 255, 255)
|
|
tint_color = (50, 50, 50)
|
|
transparency = 25
|
|
opacity = int(255 * transparency / 100)
|
|
|
|
# Load and resize the background image
|
|
try:
|
|
background_image = Image.open(background_image_path).resize(
|
|
(card_width, card_height)
|
|
)
|
|
except Exception:
|
|
# Fallback to a plain background if image fails
|
|
# type: ignore[reportArgumentType] - Pillow expects a tuple for RGBA color, but Pylance incorrectly warns here
|
|
background_image = Image.new(
|
|
"RGBA", (card_width, card_height), (30, 30, 30, 255) # type: ignore[reportArgumentType] - Pillow expects a tuple for RGBA color, but Pylance incorrectly warns here
|
|
)
|
|
|
|
card_image = Image.new("RGBA", (card_width, card_height))
|
|
card_image.paste(background_image, (0, 0))
|
|
|
|
overlay = Image.new(
|
|
"RGBA",
|
|
card_image.size,
|
|
tint_color + (opacity,), # type: ignore[reportArgumentType] - Pillow expects a tuple for RGBA color, but Pylance incorrectly warns here
|
|
)
|
|
card_image = Image.alpha_composite(card_image, overlay)
|
|
|
|
draw = ImageDraw.Draw(card_image)
|
|
|
|
# Load fonts with fallback
|
|
font_path = "arial.ttf"
|
|
try:
|
|
font = ImageFont.truetype(font_path, 20)
|
|
except Exception:
|
|
font = ImageFont.load_default()
|
|
|
|
draw.text((10, 10), f"User: {user.display_name}", fill=text_color, font=font)
|
|
user_info = self.db.fetch_one(
|
|
"SELECT XP, LEVEL FROM users WHERE ID = %s", (user.id,)
|
|
)
|
|
draw.text(
|
|
(10, 40), f"Level: {user_info.get('LEVEL', 0)}", fill=text_color, font=font
|
|
)
|
|
draw.text(
|
|
(10, 70),
|
|
f"XP: {user_info.get('XP', 0)}/{calculate_xp_needed_for_next_level(user_info.get('LEVEL', 0))}",
|
|
fill=text_color,
|
|
font=font,
|
|
)
|
|
draw.text(
|
|
(10, 100),
|
|
f"Next level in: {calculate_xp_needed_for_next_level(user_info.get('LEVEL', 0))-user_info.get('XP', 0)}XP",
|
|
fill=text_color,
|
|
font=font,
|
|
)
|
|
|
|
max_xp = calculate_xp_needed_for_next_level(user_info.get("LEVEL", 0))
|
|
progress_ratio = user_info.get("XP", 0) / max_xp if max_xp > 0 else 0
|
|
progress_width = progress_ratio * 200
|
|
|
|
draw.rectangle([(10, 130), (210, 150)], outline=outline_color, width=2)
|
|
draw.rectangle([(10, 130), (10 + progress_width, 150)], fill=progress_bar_color)
|
|
|
|
try:
|
|
profile_size = 96
|
|
margin = 30
|
|
if user.avatar is not None:
|
|
profile_picture_data = await user.avatar.read()
|
|
profile_image = Image.open(io.BytesIO(profile_picture_data)).resize(
|
|
(profile_size, profile_size)
|
|
)
|
|
else:
|
|
# Use a default avatar image if user has no avatar
|
|
default_avatar_path = "images/default_avatar.png"
|
|
if os.path.exists(default_avatar_path):
|
|
profile_image = Image.open(default_avatar_path).resize(
|
|
(profile_size, profile_size)
|
|
)
|
|
else:
|
|
# Create a blank image if default not found
|
|
# type: ignore[reportArgumentType] - Pillow expects a tuple for RGBA color, but Pylance incorrectly warns here
|
|
profile_image = Image.new(
|
|
"RGBA", (profile_size, profile_size), (100, 100, 100, 255) # type: ignore[reportArgumentType] - Pillow expects a tuple for RGBA color, but Pylance incorrectly warns here
|
|
)
|
|
card_image.paste(
|
|
profile_image,
|
|
(card_width - profile_size - margin, margin),
|
|
profile_image.convert("RGBA"),
|
|
)
|
|
except Exception as error:
|
|
logging.error(f"Error fetching profile picture: {error}")
|
|
|
|
image_bytes = io.BytesIO()
|
|
card_image.save(image_bytes, format="PNG")
|
|
image_bytes.seek(0)
|
|
|
|
return image_bytes
|
|
|
|
|
|
async def setup(client):
|
|
await client.add_cog(XP(client))
|