The begining

commandes/backup_restore.py

@@ -0,0 +1,204 @@
+# 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))