From ac81f9fd49aedb76aedd5ee7689c41567ddb22aa Mon Sep 17 00:00:00 2001 From: Lilac-Rose Date: Tue, 21 Oct 2025 00:31:18 +0200 Subject: [PATCH] Backup xp :3 --- .gitignore | 3 ++- commands/ping.py | 29 +++++++++++++++------- moderation/loader.py | 57 ++++++++++++++++++++++++++++++++++++++++---- xp/backup_xp.py | 52 ++++++++++++++++++++++++++++++++++++++++ xp/export.py | 4 +--- 5 files changed, 128 insertions(+), 17 deletions(-) create mode 100644 xp/backup_xp.py diff --git a/.gitignore b/.gitignore index fcd079f..30fe136 100644 --- a/.gitignore +++ b/.gitignore @@ -2,4 +2,5 @@ __pycache__ venv/ *.db -*.json \ No newline at end of file +*.json +*/backups/ \ No newline at end of file diff --git a/commands/ping.py b/commands/ping.py index a8dd544..88cbc10 100644 --- a/commands/ping.py +++ b/commands/ping.py @@ -2,31 +2,44 @@ import discord from discord.ext import commands from discord import app_commands import time +import aiosqlite +import os class Ping(commands.Cog): def __init__(self, bot: commands.Bot): self.bot = bot - + self.db_path = os.path.join(os.path.dirname(__file__), "suggestions.db") # or any db file you use + @app_commands.command(name="ping", description="Check the bot's latency") async def ping(self, interaction: discord.Interaction): - # Measure API latency (time to respond to the interaction) start_time = time.perf_counter() await interaction.response.send_message("Pinging...") end_time = time.perf_counter() api_latency = round((end_time - start_time) * 1000) - - # Get WebSocket latency + ws_latency = round(self.bot.latency * 1000) - - # Edit the message with the results + + # Measure database latency + db_start = time.perf_counter() + try: + async with aiosqlite.connect(self.db_path) as db: + await db.execute("SELECT 1") + await db.commit() + except Exception: + db_latency = "Error" + else: + db_end = time.perf_counter() + db_latency = round((db_end - db_start) * 1000) + embed = discord.Embed( title="Pong!", color=discord.Color.green() ) embed.add_field(name="WebSocket Latency", value=f"{ws_latency}ms", inline=True) embed.add_field(name="API Latency", value=f"{api_latency}ms", inline=True) - + embed.add_field(name="Database Latency", value=f"{db_latency}ms", inline=True) + await interaction.edit_original_response(content=None, embed=embed) async def setup(bot: commands.Bot): - await bot.add_cog(Ping(bot)) \ No newline at end of file + await bot.add_cog(Ping(bot)) diff --git a/moderation/loader.py b/moderation/loader.py index a9cf087..d6f9ec0 100644 --- a/moderation/loader.py +++ b/moderation/loader.py @@ -38,13 +38,60 @@ class ModerationBase(commands.Cog): @staticmethod def is_admin(): - """Decorator that checks if the command author has the admin role.""" - async def predicate(ctx: commands.Context): - if not any(role.id == ADMIN_ROLE_ID for role in ctx.author.roles): - await ctx.send("You do not have permission to use this command.") + """Decorator that works for both prefix and slash commands.""" + async def predicate(target): + # Handle both ctx (prefix) and interaction (slash) + user = getattr(target, "author", None) or getattr(target, "user", None) + + # Identify if it's a slash or prefix command + is_interaction = hasattr(target, "response") + + # Determine the correct send method + async def send_message(msg, ephemeral=False): + if is_interaction: + try: + if not target.response.is_done(): + await target.response.send_message(msg, ephemeral=ephemeral) + else: + await target.followup.send(msg, ephemeral=ephemeral) + except Exception: + pass + else: + try: + await target.send(msg) + except Exception: + pass + + # Role-based check + if not hasattr(user, "roles"): + await send_message("Unable to check permissions in this context.", ephemeral=is_interaction) return False + + has_admin_role = any(role.id == ADMIN_ROLE_ID for role in user.roles) + if not has_admin_role: + await send_message("You do not have permission to use this command.", ephemeral=is_interaction) + # ❗ Important: explicitly raise to stop execution + from discord.app_commands import CheckFailure + raise CheckFailure("User lacks admin permissions.") + return True - return commands.check(predicate) + + # Register for both command types + import inspect + from discord import app_commands + from discord.ext import commands + + # Return a hybrid decorator + def decorator(func): + # Add prefix check + func = commands.check(predicate)(func) + # Add slash check + func = app_commands.check(predicate)(func) + return func + + return decorator + + async def log_infraction(self, guild_id: int, user_id: int, mod_id: int, type_: str, reason: str | None): """Log an infraction to the database.""" diff --git a/xp/backup_xp.py b/xp/backup_xp.py new file mode 100644 index 0000000..c4dfcc7 --- /dev/null +++ b/xp/backup_xp.py @@ -0,0 +1,52 @@ +import discord +from discord import app_commands +from discord.ext import commands +import os +import shutil +from datetime import datetime +from moderation.loader import ModerationBase + +class BackupXP(commands.Cog): + def __init__(self, bot): + self.bot = bot + self.base_dir = os.path.dirname(os.path.abspath(__file__)) + self.backup_dir = os.path.join(self.base_dir, "backups") + os.makedirs(self.backup_dir, exist_ok=True) + + @app_commands.command(name="backup_xp", description="Backup both lifetime and annual XP databases") + @ModerationBase.is_admin() + async def backup_xp(self, interaction: discord.Interaction): + await interaction.response.defer(ephemeral=True) + + # Define source files + lifetime_db = os.path.join(self.base_dir, "lifetime.db") + annual_db = os.path.join(self.base_dir, "annual.db") + + # Make sure both databases exist + missing = [db for db in [lifetime_db, annual_db] if not os.path.exists(db)] + if missing: + return await interaction.followup.send( + f"❌ Missing database files: {', '.join(os.path.basename(m) for m in missing)}" + ) + + # Create timestamp + timestamp = datetime.now().strftime("%Y-%m-%d_%H-%M-%S") + + # Define backup file paths + lifetime_backup = os.path.join(self.backup_dir, f"lifetime_{timestamp}.db") + annual_backup = os.path.join(self.backup_dir, f"annual_{timestamp}.db") + + try: + shutil.copy2(lifetime_db, lifetime_backup) + shutil.copy2(annual_db, annual_backup) + except Exception as e: + return await interaction.followup.send(f"❌ Backup failed: `{e}`") + + await interaction.followup.send( + f"✅ Databases backed up successfully!\n" + f"**Lifetime:** `{os.path.basename(lifetime_backup)}`\n" + f"**Annual:** `{os.path.basename(annual_backup)}`" + ) + +async def setup(bot: commands.Bot): + await bot.add_cog(BackupXP(bot)) \ No newline at end of file diff --git a/xp/export.py b/xp/export.py index 1dc2c16..9fe2141 100644 --- a/xp/export.py +++ b/xp/export.py @@ -46,14 +46,12 @@ class ExportXP(commands.Cog): app_commands.Choice(name="Annual", value="annual") ] ) + @ModerationBase.is_admin() async def export_xp(self, interaction: discord.Interaction, xp_type: app_commands.Choice[str]): try: # Defer immediately await interaction.response.defer() - if not ModerationBase.is_admin(): - await interaction.followup.send("You do not have permission to run this command") - return print(f"Export started for {xp_type.value}")