mirror of
https://github.com/Lilac-Rose/Lacie.git
synced 2026-10-01 04:36:40 -05:00
Added and changed things so the website dashboard works :3
This commit is contained in:
@@ -2,7 +2,7 @@ import discord
|
||||
from discord import app_commands
|
||||
from discord.ext import commands
|
||||
from .database import get_db
|
||||
from .utils import xp_for_level, can_get_xp, get_multiplier, COOLDOWN
|
||||
from .utils import xp_for_level, can_get_xp, get_multiplier, load_config
|
||||
import time
|
||||
|
||||
class CalculateCommand(commands.Cog):
|
||||
@@ -29,6 +29,11 @@ class CalculateCommand(commands.Cog):
|
||||
if user is None:
|
||||
user = interaction.user
|
||||
|
||||
# Load config fresh
|
||||
config = load_config()
|
||||
COOLDOWN = config["COOLDOWN"]
|
||||
random_xp_config = config.get("RANDOM_XP", {"min": 50, "max": 100})
|
||||
|
||||
# Determine which database to use
|
||||
use_lifetime = True if (board_type is None or board_type.value == "lifetime") else False
|
||||
board_name = "Lifetime" if use_lifetime else "Annual"
|
||||
@@ -59,8 +64,8 @@ class CalculateCommand(commands.Cog):
|
||||
|
||||
multiplier = get_multiplier(user, apply_multiplier=True)
|
||||
|
||||
min_xp_per_msg = int(50 * multiplier)
|
||||
max_xp_per_msg = int(100 * multiplier)
|
||||
min_xp_per_msg = int(random_xp_config["min"] * multiplier)
|
||||
max_xp_per_msg = int(random_xp_config["max"] * multiplier)
|
||||
avg_xp_per_msg = (min_xp_per_msg + max_xp_per_msg) / 2
|
||||
|
||||
max_messages = int(remaining_xp / min_xp_per_msg)
|
||||
@@ -111,4 +116,4 @@ class CalculateCommand(commands.Cog):
|
||||
|
||||
|
||||
async def setup(bot):
|
||||
await bot.add_cog(CalculateCommand(bot))
|
||||
await bot.add_cog(CalculateCommand(bot))
|
||||
@@ -5,7 +5,7 @@ import traceback
|
||||
from discord.ext import commands
|
||||
from discord import app_commands
|
||||
from .database import get_db
|
||||
from .utils import xp_for_level, MULTIPLIERS, COOLDOWN, get_multiplier
|
||||
from .utils import xp_for_level, get_multiplier, load_config
|
||||
|
||||
class Rank(commands.Cog):
|
||||
def __init__(self, bot):
|
||||
@@ -31,6 +31,11 @@ class Rank(commands.Cog):
|
||||
board_type_value = board_type.value if board_type else "lifetime"
|
||||
lifetime = board_type_value == "lifetime"
|
||||
|
||||
# Load config fresh every time
|
||||
config = load_config()
|
||||
MULTIPLIERS = config["MULTIPLIERS"]
|
||||
COOLDOWN = config["COOLDOWN"]
|
||||
|
||||
conn, cur = get_db(lifetime)
|
||||
|
||||
# Fetch XP data for the requested user
|
||||
@@ -123,4 +128,4 @@ class Rank(commands.Cog):
|
||||
)
|
||||
|
||||
async def setup(bot):
|
||||
await bot.add_cog(Rank(bot))
|
||||
await bot.add_cog(Rank(bot))
|
||||
129
xp/sync.py
129
xp/sync.py
@@ -2,42 +2,47 @@ import discord
|
||||
from discord.ext import commands
|
||||
from discord import app_commands
|
||||
from moderation.loader import ModerationBase
|
||||
from xp.add_xp import get_db
|
||||
from xp.utils import ROLE_REWARDS
|
||||
from .database import get_db
|
||||
from .utils import load_config, xp_for_level
|
||||
from discord.utils import get
|
||||
import asyncio
|
||||
from concurrent.futures import ThreadPoolExecutor
|
||||
|
||||
class XPSync(commands.Cog):
|
||||
"""Sync XP role rewards for users."""
|
||||
|
||||
def __init__(self, bot: commands.Bot):
|
||||
self.bot = bot
|
||||
self.executor = ThreadPoolExecutor(max_workers=2)
|
||||
|
||||
async def sync_roles_for_user(self, member: discord.Member) -> tuple[int, list[str]]:
|
||||
"""
|
||||
Sync roles for a member based on their lifetime XP level.
|
||||
Returns tuple of (llevel, list of role names added).
|
||||
Returns tuple of (level, list of role names added).
|
||||
"""
|
||||
# Load config fresh
|
||||
config = load_config()
|
||||
ROLE_REWARDS = {int(k): int(v) for k, v in config["ROLE_REWARDS"].items()}
|
||||
|
||||
conn, cur = get_db(lifetime=True)
|
||||
cur.execute("SELECT level FROM xp WHERE user_id = ?", (str(member.id),))
|
||||
row = cur.fetchone()
|
||||
conn.close()
|
||||
|
||||
|
||||
if not row:
|
||||
return (0, [])
|
||||
|
||||
|
||||
level = row[0]
|
||||
roles_added = []
|
||||
|
||||
|
||||
for lvl, role_id in ROLE_REWARDS.items():
|
||||
if level >= lvl:
|
||||
role = get(member.guild.roles, id=role_id)
|
||||
if role and role not in member.roles:
|
||||
await member.add_roles(role)
|
||||
roles_added.append(role.name)
|
||||
|
||||
|
||||
return (level, roles_added)
|
||||
|
||||
|
||||
|
||||
@app_commands.command(name="sync", description="Sync your XP role rewards.")
|
||||
@app_commands.describe(user="[Admin only] The user to sync roles for.")
|
||||
async def sync(
|
||||
@@ -45,7 +50,7 @@ class XPSync(commands.Cog):
|
||||
interaction: discord.Interaction,
|
||||
user: discord.User | None = None,
|
||||
):
|
||||
#If user parameter is provided, check if requester is admin
|
||||
# If user parameter is provided, check if requester is admin
|
||||
if user is not None:
|
||||
if not await ModerationBase.is_admin().predicate(interaction):
|
||||
await interaction.response.send_message("You don't have permission to sync roles for other users.", ephemeral=True)
|
||||
@@ -55,13 +60,12 @@ class XPSync(commands.Cog):
|
||||
await interaction.response.send_message("User is not in this server.", ephemeral=True)
|
||||
return
|
||||
else:
|
||||
#Sync for the user who ran the command
|
||||
# Sync for the user who ran the command
|
||||
target_member = interaction.user
|
||||
|
||||
|
||||
await interaction.response.defer()
|
||||
|
||||
level, roles_added = await self.sync_roles_for_user(target_member)
|
||||
|
||||
|
||||
if level == 0:
|
||||
await interaction.followup.send(f"{target_member.mention} has no lifetime XP recorded.")
|
||||
elif roles_added:
|
||||
@@ -73,5 +77,100 @@ class XPSync(commands.Cog):
|
||||
else:
|
||||
await interaction.followup.send(f"{target_member.mention} (Level {level}) already has all eligible role rewards.")
|
||||
|
||||
def _recalc_worker(self, lifetime: bool, user_id: str = None):
|
||||
"""Worker function that runs in a separate thread"""
|
||||
conn, cur = get_db(lifetime)
|
||||
total_updated = 0
|
||||
|
||||
if user_id:
|
||||
# Single user
|
||||
cur.execute("SELECT xp FROM xp WHERE user_id = ?", (user_id,))
|
||||
row = cur.fetchone()
|
||||
if row:
|
||||
xp = row[0]
|
||||
new_level = 0
|
||||
while xp >= xp_for_level(new_level + 1):
|
||||
new_level += 1
|
||||
cur.execute("UPDATE xp SET level = ? WHERE user_id = ?", (new_level, user_id))
|
||||
total_updated += 1
|
||||
else:
|
||||
# All users - batch process
|
||||
cur.execute("SELECT user_id, xp FROM xp")
|
||||
rows = cur.fetchall()
|
||||
|
||||
updates = []
|
||||
for uid, xp in rows:
|
||||
new_level = 0
|
||||
while xp >= xp_for_level(new_level + 1):
|
||||
new_level += 1
|
||||
updates.append((new_level, uid))
|
||||
total_updated += 1
|
||||
|
||||
# Execute all updates at once
|
||||
cur.executemany("UPDATE xp SET level = ? WHERE user_id = ?", updates)
|
||||
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
||||
return total_updated
|
||||
|
||||
@app_commands.command(name="recalc", description="[Admin] Recalculate levels based on current XP curve")
|
||||
@app_commands.describe(
|
||||
user="Specific user to recalculate (leave empty for everyone)",
|
||||
board_type="Which board to recalculate"
|
||||
)
|
||||
@app_commands.choices(board_type=[
|
||||
app_commands.Choice(name="Lifetime", value="lifetime"),
|
||||
app_commands.Choice(name="Annual", value="annual"),
|
||||
app_commands.Choice(name="Both", value="both")
|
||||
])
|
||||
@ModerationBase.is_admin()
|
||||
async def recalc(
|
||||
self,
|
||||
interaction: discord.Interaction,
|
||||
user: discord.Member = None,
|
||||
board_type: app_commands.Choice[str] = None
|
||||
):
|
||||
# Respond IMMEDIATELY
|
||||
if user:
|
||||
await interaction.response.send_message(f"🔄 Recalculating levels for {user.mention}...", ephemeral=True)
|
||||
else:
|
||||
await interaction.response.send_message("🔄 Starting recalculation for all users... I'll update you when done!")
|
||||
|
||||
# NOW do the work in background
|
||||
board_value = board_type.value if board_type else "both"
|
||||
boards = [True, False] if board_value == "both" else [board_value == "lifetime"]
|
||||
user_id = str(user.id) if user else None
|
||||
|
||||
async def process():
|
||||
try:
|
||||
total_updated = 0
|
||||
for lifetime in boards:
|
||||
board_name = "Lifetime" if lifetime else "Annual"
|
||||
|
||||
if not user: # Only send progress for all users
|
||||
await interaction.followup.send(f"⏳ Processing {board_name} database...")
|
||||
|
||||
count = await asyncio.get_event_loop().run_in_executor(
|
||||
self.executor, self._recalc_worker, lifetime, user_id
|
||||
)
|
||||
total_updated += count
|
||||
|
||||
if not user:
|
||||
await interaction.followup.send(f"✅ {board_name} complete: {count} entries updated")
|
||||
|
||||
board_text = "Lifetime and Annual" if board_value == "both" else board_value.title()
|
||||
if user:
|
||||
await interaction.followup.send(f"✅ Recalculated {board_text} level for {user.mention}")
|
||||
else:
|
||||
await interaction.followup.send(
|
||||
f"🎉 All done! Recalculated {board_text} levels for {total_updated} total entries. {interaction.user.mention}"
|
||||
)
|
||||
except Exception as e:
|
||||
await interaction.followup.send(f"❌ Error during recalculation: {str(e)}")
|
||||
|
||||
# Start background task
|
||||
self.bot.loop.create_task(process())
|
||||
|
||||
async def setup(bot: commands.Bot):
|
||||
await bot.add_cog(XPSync(bot))
|
||||
74
xp/utils.py
74
xp/utils.py
@@ -1,80 +1,58 @@
|
||||
import math, random, time
|
||||
import math, random, time, json
|
||||
from discord.utils import get
|
||||
from pathlib import Path
|
||||
CONFIG_PATH = Path("/home/lilacrose/lilacrose.dev2.0/bots/lacie/xp_config.json")
|
||||
|
||||
# Role rewards for leveling up (only apply to lifetime XP)
|
||||
ROLE_REWARDS = {
|
||||
2: 963498974222885054,
|
||||
8: 1372088164377694208,
|
||||
100: 1296055376009101384,
|
||||
150: 1296056209543008287,
|
||||
200: 1296056657784340480,
|
||||
}
|
||||
|
||||
# XP multipliers by role (only apply to lifetime XP)
|
||||
MULTIPLIERS = {
|
||||
"1038402681376612413": 1.25,
|
||||
"1213171315259736155": 1.25,
|
||||
"881560923494547477": 1.5,
|
||||
"1008830505166323742": 1.5,
|
||||
"880055414434201600": 1.5,
|
||||
"931234149774270524": 2,
|
||||
"1238966782962958377": 2.5,
|
||||
"1113751318918602762": 3,
|
||||
}
|
||||
|
||||
# Cooldown in seconds between XP gains
|
||||
COOLDOWN = 60
|
||||
|
||||
def load_config():
|
||||
with CONFIG_PATH.open() as f:
|
||||
return json.load(f)
|
||||
|
||||
def get_multiplier(member, apply_multiplier=True):
|
||||
"""Return XP multiplier from roles, or 1 if multiplier shouldn't be applied."""
|
||||
if not apply_multiplier:
|
||||
return 1
|
||||
|
||||
config = load_config()
|
||||
multipliers = config["MULTIPLIERS"]
|
||||
highest = 1
|
||||
for role in member.roles:
|
||||
if str(role.id) in MULTIPLIERS:
|
||||
highest = max(highest, MULTIPLIERS[str(role.id)])
|
||||
if str(role.id) in multipliers:
|
||||
highest = max(highest, multipliers[str(role.id)])
|
||||
return highest
|
||||
|
||||
|
||||
def xp_for_level(level: int) -> int:
|
||||
"""Calculate total XP required for a given level."""
|
||||
xp = (level ** 3) + (50 * level ** 2) + (100 * level)
|
||||
return int(math.floor(xp / 100) * 100) # round down to nearest 100
|
||||
|
||||
config = load_config()
|
||||
curve = config.get("XP_CURVE", {"base": 1, "square": 50, "linear": 100, "divisor": 100})
|
||||
|
||||
xp = (level ** 3 * curve["base"]) + (level ** 2 * curve["square"]) + (level * curve["linear"])
|
||||
xp = xp / curve["divisor"]
|
||||
return int(math.floor(xp / 100) * 100)
|
||||
|
||||
def random_xp() -> int:
|
||||
"""Random XP awarded per message."""
|
||||
return random.randint(50, 100)
|
||||
|
||||
config = load_config()
|
||||
xp_range = config.get("RANDOM_XP", {"min": 50, "max": 100})
|
||||
return random.randint(xp_range["min"], xp_range["max"])
|
||||
|
||||
def can_get_xp(last_message_time: int) -> bool:
|
||||
"""Check if enough time has passed since last XP gain."""
|
||||
return (time.time() - last_message_time) >= COOLDOWN
|
||||
|
||||
config = load_config()
|
||||
cooldown = config["COOLDOWN"]
|
||||
return (time.time() - last_message_time) >= cooldown
|
||||
|
||||
async def check_level_up(member, cur, conn, lifetime=True):
|
||||
"""Check if user leveled up and give rewards (only for lifetime XP)."""
|
||||
config = load_config()
|
||||
role_rewards = {int(k): v for k,v in config["ROLE_REWARDS"].items()}
|
||||
cur.execute("SELECT xp, level FROM xp WHERE user_id = ?", (str(member.id),))
|
||||
row = cur.fetchone()
|
||||
if not row:
|
||||
return
|
||||
|
||||
xp, level = row
|
||||
new_level = level
|
||||
while xp >= xp_for_level(new_level + 1):
|
||||
new_level += 1
|
||||
|
||||
if new_level > level:
|
||||
cur.execute("UPDATE xp SET level = ? WHERE user_id = ?", (new_level, str(member.id)))
|
||||
conn.commit()
|
||||
|
||||
# Only assign role rewards for lifetime XP
|
||||
if lifetime:
|
||||
from discord.utils import get
|
||||
for lvl, role_id in ROLE_REWARDS.items():
|
||||
for lvl, role_id in role_rewards.items():
|
||||
if new_level >= lvl:
|
||||
role = get(member.guild.roles, id=role_id)
|
||||
if role:
|
||||
await member.add_roles(role)
|
||||
await member.add_roles(role)
|
||||
Reference in New Issue
Block a user