diff --git a/xp/backup_xp.py b/xp/backup_xp.py index 6bedfc0..fe4bddd 100644 --- a/xp/backup_xp.py +++ b/xp/backup_xp.py @@ -1,52 +1,104 @@ import discord from discord import app_commands -from discord.ext import commands +from discord.ext import commands, tasks import os import shutil -from datetime import datetime +from datetime import datetime, timedelta from moderation.loader import ModerationBase +BACKUP_CHANNEL_ID = 946421558778417172 +BACKUP_INTERVAL = timedelta(days=7) + 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") + self.last_backup_file = os.path.join(self.backup_dir, "last_backup.txt") os.makedirs(self.backup_dir, exist_ok=True) + # Start the daily check task + self.auto_backup_task.start() + + async def cog_load(self): + """Check on startup if a backup is due""" + await self.check_last_backup() + + def cog_unload(self): + self.auto_backup_task.cancel() + @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=False) + success, message = await self.create_backup() + await interaction.followup.send(message) - # Define source files + @tasks.loop(hours=24) + async def auto_backup_task(self): + """Runs daily and checks if a weekly backup is due""" + await self.check_last_backup() + + async def check_last_backup(self): + """Check if the last backup was over 7 days ago or if no backup exists""" + now = datetime.now() + + if not os.path.exists(self.last_backup_file): + # No previous backup — make initial backup + await self.create_backup(log_channel=True, reason="Startup initial backup") + return + + with open(self.last_backup_file, "r") as f: + try: + last_time = datetime.fromisoformat(f.read().strip()) + except Exception: + last_time = datetime.min + + if now - last_time >= BACKUP_INTERVAL: + await self.create_backup(log_channel=True, reason="Auto weekly backup") + + async def create_backup(self, log_channel=False, reason=None): + """Handles the actual backup logic""" 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)}" - ) + return False, 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)}`" - ) + # Record last backup time + with open(self.last_backup_file, "w") as f: + f.write(datetime.now().isoformat()) + + # File sizes + lifetime_size = os.path.getsize(lifetime_db) / (1024 * 1024) + annual_size = os.path.getsize(annual_db) / (1024 * 1024) + total_size = lifetime_size + annual_size + + msg = ( + f"✅ Databases backed up successfully! ({reason or 'Manual backup'})\n" + f"**Lifetime:** `{os.path.basename(lifetime_backup)}` ({lifetime_size:.2f} MB)\n" + f"**Annual:** `{os.path.basename(annual_backup)}` ({annual_size:.2f} MB)\n" + f"**Total size:** {total_size:.2f} MB" + ) + + if log_channel: + channel = self.bot.get_channel(BACKUP_CHANNEL_ID) + if channel: + await channel.send(msg) + + return True, msg + + except Exception as e: + return False, f"❌ Backup failed: `{e}`" async def setup(bot: commands.Bot): - await bot.add_cog(BackupXP(bot)) \ No newline at end of file + await bot.add_cog(BackupXP(bot)) diff --git a/xp/sync.py b/xp/sync.py new file mode 100644 index 0000000..b7708f0 --- /dev/null +++ b/xp/sync.py @@ -0,0 +1,77 @@ +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 discord.utils import get + +class XPSync(commands.Cog): + """Sync XP role rewards for users.""" + + def __init__(self, bot: commands.Bot): + self.bot = bot + + 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). + """ + 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( + self, + interaction: discord.Interaction, + user: discord.User | None = None, + ): + #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) + return + target_member = interaction.guild.get_member(user.id) + if not target_member: + await interaction.response.send_message("User is not in this server.", ephemeral=True) + return + else: + #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: + roles_list = ", ".join(roles_added) + await interaction.followup.send( + f"Synced roles for {target_member.mention} (Level {level})\n" + f"**Roles added:** {roles_list}" + ) + else: + await interaction.followup.send(f"{target_member.mention} (Level {level}) already has all eligible role rewards.") + +async def setup(bot: commands.Bot): + await bot.add_cog(XPSync(bot)) \ No newline at end of file