# commandes/backup_restore.py import os import json import zipfile import tempfile import shutil from typing import List import discord from discord import app_commands from discord.ext import commands BACKUPS_ROOT = "data/security_backups" def ensure_dir(path: str): os.makedirs(path, exist_ok=True) class BackupRestore(commands.Cog): """Restauration granulaire d'une sauvegarde.""" def __init__(self, bot: commands.Bot): self.bot = bot async def check_whitelisted(self, guild_id: int, user_id: int) -> bool: """Vérifie si un utilisateur est whitelisté.""" # Recherche dynamique d’un cog ayant une méthode is_whitelisted for cog in self.bot.cogs.values(): if hasattr(cog, "is_whitelisted"): return cog.is_whitelisted(guild_id, user_id) raise RuntimeError("Cog Whitelist non chargé ou mal nommé !") async def list_backup_elements(self, backup_path: str) -> List[str]: """Liste les éléments exportables dans une archive ou dossier de backup.""" elements = [] if os.path.isdir(backup_path): for f in os.listdir(backup_path): if f.endswith(".json") or os.path.isdir(os.path.join(backup_path, f)): elements.append(f.replace(".json", "")) elif os.path.isfile(backup_path) and backup_path.endswith(".zip"): with zipfile.ZipFile(backup_path, "r") as zf: for f in zf.namelist(): base = f.split("/")[0] if base and base not in elements: elements.append(base) return elements async def element_autocomplete(self, interaction: discord.Interaction, current: str): """Complétion automatique pour l'option 'element'.""" guild = interaction.guild if guild is None: return [] ensure_dir(BACKUPS_ROOT) files = [f for f in os.listdir(BACKUPS_ROOT) if f.startswith(str(guild.id))] files.sort(reverse=True) if not files: return [] last_backup = files[0] backup_path = os.path.join(BACKUPS_ROOT, last_backup) elements = await self.list_backup_elements(backup_path) return [app_commands.Choice(name=e, value=e) for e in elements if current.lower() in e.lower()][:25] async def restore_roles(self, guild: discord.Guild, backup_path: str): """Restaure les rôles à partir du backup.""" roles_file = os.path.join(backup_path, "roles.json") if os.path.isfile(roles_file): with open(roles_file, "r", encoding="utf-8") as f: roles_data = json.load(f) for r in roles_data: try: await guild.create_role( name=r["name"], permissions=discord.Permissions(r["permissions"]), color=discord.Color(r.get("color", 0)), hoist=r.get("hoist", False), mentionable=r.get("mentionable", False), reason="Restauration backup" ) except discord.Forbidden: continue async def restore_channels(self, guild: discord.Guild, backup_path: str): """Restaure les channels à partir du backup.""" channels_file = os.path.join(backup_path, "channels.json") if os.path.isfile(channels_file): with open(channels_file, "r", encoding="utf-8") as f: channels_data = json.load(f) for c in channels_data: try: if c["type"] == 0: # text await guild.create_text_channel( name=c["name"], category=guild.get_channel(c.get("category_id")), topic=c.get("topic", ""), nsfw=c.get("nsfw", False), reason="Restauration backup" ) elif c["type"] == 2: # voice await guild.create_voice_channel( name=c["name"], category=guild.get_channel(c.get("category_id")), bitrate=c.get("bitrate", 64000), user_limit=c.get("user_limit", 0), reason="Restauration backup" ) except discord.Forbidden: continue async def restore_messages(self, channel: discord.TextChannel, backup_path: str): """Restaure les messages à partir du backup (solution ZIP incluse).""" if backup_path.endswith(".zip"): temp_dir = tempfile.mkdtemp() try: with zipfile.ZipFile(backup_path, "r") as zip_ref: zip_ref.extractall(temp_dir) messages_file = os.path.join(temp_dir, "messages.json") if not os.path.exists(messages_file): return with open(messages_file, "r", encoding="utf-8") as f: messages = json.load(f) for msg in messages: await channel.send(msg["content"]) finally: shutil.rmtree(temp_dir) else: messages_file = os.path.join(backup_path, "messages.json") if not os.path.exists(messages_file): return with open(messages_file, "r", encoding="utf-8") as f: messages = json.load(f) for msg in messages: await channel.send(msg["content"]) @app_commands.command(name="granulaire", description="Restaure un élément précis d'une sauvegarde") @app_commands.describe( backup="Nom ou ID de la sauvegarde", element="Élément à restaurer (roles, channels, messages...)" ) @app_commands.autocomplete(element=element_autocomplete) async def restore_granulaire( self, interaction: discord.Interaction, backup: str, element: str ): guild = interaction.guild if guild is None: await interaction.response.send_message("Cette commande doit être utilisée dans un serveur.", ephemeral=True) return if not await self.check_whitelisted(guild.id, interaction.user.id): await interaction.response.send_message( "❌ Vous n'êtes pas autorisé à utiliser cette commande.", ephemeral=True ) return backup_path = os.path.join(BACKUPS_ROOT, backup) if not os.path.exists(backup_path): await interaction.response.send_message(f"❌ La sauvegarde `{backup}` est introuvable.", ephemeral=True) return await interaction.response.defer(ephemeral=True, thinking=True) available_elements = await self.list_backup_elements(backup_path) if element not in available_elements: await interaction.followup.send( f"❌ Élément `{element}` introuvable dans la sauvegarde.", ephemeral=True ) return # Détection si c’est un zip pour extraire temporairement temp_path = backup_path if backup_path.endswith(".zip"): temp_path = tempfile.mkdtemp() with zipfile.ZipFile(backup_path, "r") as zip_ref: zip_ref.extractall(temp_path) try: if element == "roles": await self.restore_roles(guild, temp_path) elif element == "channels": await self.restore_channels(guild, temp_path) elif element == "messages": # Restaure messages dans tous les channels textuels ? for ch in guild.text_channels: await self.restore_messages(ch, temp_path) else: await interaction.followup.send(f"⚠️ Élément `{element}` non géré par la restauration automatique.", ephemeral=True) return finally: if backup_path.endswith(".zip"): shutil.rmtree(temp_path) await interaction.followup.send( f"✅ Restauration granulaire de `{element}` depuis la sauvegarde `{backup}` terminée.", ephemeral=True ) async def setup(bot: commands.Bot): await bot.add_cog(BackupRestore(bot))