mirror of
https://github.com/Lilac-Rose/Lacie.git
synced 2026-07-24 12:00:23 -05:00
Backup xp :3
This commit is contained in:
parent
4fbd6cfa90
commit
ac81f9fd49
3
.gitignore
vendored
3
.gitignore
vendored
|
|
@ -2,4 +2,5 @@
|
|||
__pycache__
|
||||
venv/
|
||||
*.db
|
||||
*.json
|
||||
*.json
|
||||
*/backups/
|
||||
|
|
@ -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))
|
||||
await bot.add_cog(Ping(bot))
|
||||
|
|
|
|||
|
|
@ -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."""
|
||||
|
|
|
|||
52
xp/backup_xp.py
Normal file
52
xp/backup_xp.py
Normal file
|
|
@ -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))
|
||||
|
|
@ -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}")
|
||||
|
||||
|
|
|
|||
Loading…
Reference in New Issue
Block a user