Kuby/commandes/backup.py

# commandes/backup.py
import os
import io
import json
import zipfile
import asyncio
from datetime import datetime
from typing import Dict, Any, List

import disnake
from disnake.ext import commands

BACKUPS_ROOT = "data/security_backups"  # dossier de sortie des sauvegardes


def ensure_dir(path: str) -> None:
    os.makedirs(path, exist_ok=True)


def human_dt() -> str:
    return datetime.now().strftime("%Y%m%d_%H%M%S")


def progress_bar(pct: float, width: int = 20) -> str:
    pct = max(0.0, min(1.0, pct))
    filled = int(round(width * pct))
    return f"[{'█'*filled}{'─'*(width-filled)}] {int(pct*100)}%"


def estimate_duration_seconds(
    nb_roles: int,
    nb_categories: int,
    nb_text_channels: int,
    nb_members: int,
    save_messages: bool,
    message_limit_per_channel: int,
) -> int:
    meta_cost = nb_roles * 0.02 + nb_categories * 0.03 + nb_text_channels * 0.05 + nb_members * 0.01
    if save_messages:
        pages_per_chan = max(1, (message_limit_per_channel + 99) // 100)
        msg_cost = nb_text_channels * pages_per_chan * 0.12
    else:
        msg_cost = 0.0
    base_overhead = 3.0
    return int(meta_cost + msg_cost + base_overhead)


def serialize_overwrites(channel: disnake.abc.GuildChannel) -> List[Dict[str, Any]]:
    serialized: List[Dict[str, Any]] = []
    for target, perms in channel.overwrites.items():
        target_type = "role" if isinstance(target, disnake.Role) else "member"
        serialized.append({
            "target_type": target_type,
            "target_id": target.id,
            "permissions": perms._values
        })
    return serialized


class Backup(commands.Cog):
    """Sauvegarde de serveur avec estimation & progression."""

    def __init__(self, bot: commands.Bot):
        self.bot = bot

    @commands.slash_command(name="backup", description="Outils de sauvegarde du serveur")
    async def backup_group(self, interaction: disnake.ApplicationCommandInteraction):
        pass

    @backup_group.sub_command(name="create", description="Créer une sauvegarde du serveur")
    async def backup_create(
        self,
        interaction: disnake.ApplicationCommandInteraction,
        save_messages: bool = commands.Param(True, description="Inclure l'export des messages"),
        privacy_notice: bool = commands.Param(True, description="Afficher un avertissement public si messages sauvegardés"),
        message_limit_per_channel: int = commands.Param(10000, description="Nombre max de messages à exporter par salon (0 = tous)")
    ):
        guild = interaction.guild
        assert guild is not None, "Cette commande doit être utilisée dans un serveur."

        # Vérification de la whitelist
        whitelist_cog = self.bot.get_cog("whitelist")
        if not whitelist_cog:
            await interaction.response.send_message("❌ Cog Whitelist introuvable.", ephemeral=True)
            return
        if not whitelist_cog.is_whitelisted(guild.id, interaction.user.id):
            await interaction.response.send_message(
                "❌ Vous n'êtes pas dans la Whitelist.", ephemeral=True
            )
            return

        # Deferre l'interaction
        await interaction.response.defer(ephemeral=True)

        # Avertissement de confidentialité
        if privacy_notice and save_messages:
            try:
                await interaction.channel.send(
                    embed=disnake.Embed(
                        title="📢 Information de confidentialité (transparence)",
                        description=(
                            "Une **sauvegarde du serveur** est en cours et **inclus les messages**.\n"
                            "Vous êtes dans l'obligation de prévenir les membres."
                        ),
                        color=disnake.Color.orange()
                    )
                )
            except Exception:
                pass

        # Préparation chemins
        ts = human_dt()
        base_dir = os.path.join(BACKUPS_ROOT, f"{guild.id}_{ts}")
        ensure_dir(base_dir)
        ensure_dir(os.path.join(base_dir, "messages"))

        # Estimation
        nb_roles = len(guild.roles)
        nb_categories = len(guild.categories)
        nb_text_channels = len(guild.text_channels)
        nb_members = guild.member_count or len(guild.members)

        if message_limit_per_channel < 0:
            message_limit_per_channel = 0

        estimation_s = estimate_duration_seconds(
            nb_roles, nb_categories, nb_text_channels, nb_members, save_messages,
            message_limit_per_channel if message_limit_per_channel else 100000
        )

        prog_embed = disnake.Embed(
            title="🗄️ Sauvegarde en cours",
            description=(
                f"**Serveur :** {guild.name} (`{guild.id}`)\n"
                f"**Estimation :** ~{estimation_s} secondes\n"
                f"**Options** : messages={'oui' if save_messages else 'non'} • limite/chan={message_limit_per_channel or 'tous'}"
            ),
            color=disnake.Color.blurple()
        )
        prog_embed.add_field(name="Progression", value=progress_bar(0.0), inline=False)
        progress_msg = await interaction.followup.send(embed=prog_embed, ephemeral=False)

        # ================= EXPORT =================
        weights = {"roles":0.08,"categories":0.12,"channels":0.20,"members":0.10,"messages":0.10 if save_messages else 0.0,"zip":0.10}
        done = 0.0
        async def bump(step_key:str, extra_pct:float=0.0,note:str=""):
            nonlocal done
            done += weights.get(step_key,0.0)+extra_pct
            done = min(done,0.999)
            prog_embed.set_field_at(0,name="Progression",value=progress_bar(done),inline=False)
            if note:
                prog_embed.set_footer(text=note)
            await progress_msg.edit(embed=prog_embed)

        # --- RÔLES ---
        roles_dump = [{"id": r.id,"name": r.name,"color": r.color.value,"hoist": r.hoist,"mentionable": r.mentionable,"permissions": r.permissions.value,"position": r.position} for r in guild.roles]
        with open(os.path.join(base_dir,"roles.json"),"w",encoding="utf-8") as f:
            json.dump(roles_dump,f,indent=2,ensure_ascii=False)
        await bump("roles", note="Rôles exportés")

        # --- CATÉGORIES ---
        cats_dump = [{"id": c.id,"name": c.name,"position": c.position,"nsfw": getattr(c,"nsfw",False),"overwrites": serialize_overwrites(c)} for c in guild.categories]
        with open(os.path.join(base_dir,"categories.json"),"w",encoding="utf-8") as f:
            json.dump(cats_dump,f,indent=2,ensure_ascii=False)
        await bump("categories", note="Catégories exportées")


        # --- SALONS ---
        chans_dump=[]
        for ch in guild.channels:
            base={"id":ch.id,"name":ch.name,"type":str(ch.type),"position":ch.position,"category_id":ch.category.id if getattr(ch,"category",None) else None,"overwrites":serialize_overwrites(ch)}
            if isinstance(ch,disnake.TextChannel):
                base.update({"topic":ch.topic,"nsfw":ch.nsfw,"slowmode_delay":ch.slowmode_delay,"default_auto_archive_duration":ch.default_auto_archive_duration})
            elif isinstance(ch,disnake.VoiceChannel):
                base.update({"bitrate":ch.bitrate,"user_limit":ch.user_limit,"rtc_region":str(ch.rtc_region) if ch.rtc_region else None})
            elif isinstance(ch,disnake.ForumChannel):
                base.update({"nsfw":ch.nsfw,"default_auto_archive_duration":ch.default_auto_archive_duration})
            chans_dump.append(base)
        with open(os.path.join(base_dir,"channels.json"),"w",encoding="utf-8") as f:
            json.dump(chans_dump,f,indent=2,ensure_ascii=False)
        await bump("channels", note="Salons exportés")


        # --- MEMBRES ---
        members_dump=[]
        for m in guild.members:
            members_dump.append({"id":m.id,"name":str(m),"nick":m.nick,"roles":[r.id for r in m.roles if r!=guild.default_role],"bot":m.bot,"joined_at":m.joined_at.isoformat() if m.joined_at else None})
        with open(os.path.join(base_dir,"members.json"),"w",encoding="utf-8") as f:
            json.dump(members_dump,f,indent=2,ensure_ascii=False)
        await bump("members", note="Membres exportés")

        # --- MESSAGES ---
        if save_messages:
            text_chans=list(guild.text_channels)
            total_tc=len(text_chans)
            per_chan_weight=weights["messages"]/max(1,total_tc)
            exported_total=0
            for idx,tc in enumerate(text_chans,start=1):
                file_path=os.path.join(base_dir,"messages",f"{tc.id}.jsonl")
                count=0
                with open(file_path,"w",encoding="utf-8") as f:
                    try:
                        async for msg in tc.history(limit=message_limit_per_channel if message_limit_per_channel>0 else None, oldest_first=True):
                            data={"id":msg.id,"channel_id":msg.channel.id,"author_id":msg.author.id,"author_bot":msg.author.bot,"created_at":msg.created_at.isoformat(),"content":msg.content,"attachments":[a.url for a in msg.attachments],"embeds":[e.to_dict() for e in msg.embeds],"reference":{"message_id":getattr(msg.reference,"message_id",None),"channel_id":getattr(msg.reference,"channel_id",None)} if msg.reference else None}
                            f.write(json.dumps(data,ensure_ascii=False)+"\n")
                            count+=1
                    except disnake.Forbidden: pass
                    except Exception: pass
                exported_total+=count
                done+=per_chan_weight
                prog_embed.set_field_at(0,name="Progression",value=progress_bar(min(done,0.999)),inline=False)
                prog_embed.set_footer(text=f"Messages exportés : {exported_total} | Salon {idx}/{total_tc}")
                await progress_msg.edit(embed=prog_embed)

        # --- ZIP FINAL ---
        zip_path=os.path.join(BACKUPS_ROOT,f"{guild.id}_{ts}.zip")
        try:
            with zipfile.ZipFile(zip_path,"w",compression=zipfile.ZIP_DEFLATED,compresslevel=6) as zf:
                for root,_,files in os.walk(base_dir):
                    for name in files:
                        full=os.path.join(root,name)
                        rel=os.path.relpath(full,base_dir)
                        zf.write(full,rel)
        except Exception as exc:
            prog_embed.color=disnake.Color.red()
            prog_embed.set_field_at(0,name="Progression",value=progress_bar(1.0),inline=False)
            prog_embed.set_footer(text=f"Erreur lors de la création du ZIP : {exc}")
            await progress_msg.edit(embed=prog_embed)
            await interaction.followup.send("❌ Erreur lors de la compression ZIP. Si l'erreur persiste veuillez ouvrir un ticket sur Omega Kube", ephemeral=True)
            return

        # Finalisation
        prog_embed.color=disnake.Color.green()
        prog_embed.set_field_at(0,name="Progression",value=progress_bar(1.0),inline=False)
        prog_embed.add_field(name="Archive",value=f"`{zip_path}`",inline=False)
        prog_embed.set_footer(text="Sauvegarde terminée ✅")
        await progress_msg.edit(embed=prog_embed)
        await interaction.followup.send("🎉 Sauvegarde terminée. Un message de progression avec le chemin de l’archive a été publié.", ephemeral=True)

    @backup_group.sub_command(name="list", description="Lister les archives de sauvegarde disponibles")
    async def backup_list(self, interaction: disnake.ApplicationCommandInteraction):
        guild = interaction.guild
        assert guild is not None, "Cette commande doit être utilisée dans un serveur."

        whitelist_cog = self.bot.get_cog("whitelist")
        if not whitelist_cog:
            await interaction.response.send_message("❌ Cog Whitelist introuvable.", ephemeral=True)
            return
        if not whitelist_cog.is_whitelisted(guild.id, interaction.user.id):
            await interaction.response.send_message(
                "❌ Vous n'êtes pas autorisé à utiliser cette commande.", ephemeral=True
            )
            return

        await interaction.response.defer(ephemeral=True)
        ensure_dir(BACKUPS_ROOT)
        files = [f for f in os.listdir(BACKUPS_ROOT) if f.endswith(".zip")]
        files.sort(reverse=True)

        if not files:
            await interaction.followup.send("Aucune archive trouvée dans `backups/`.", ephemeral=True)
            return

        desc="\n".join(f"• `{name}`" for name in files[:20])
        embed=disnake.Embed(title="📦 Archives disponibles",description=desc,color=disnake.Color.blurple())
        if len(files)>20:
            embed.set_footer(text=f"... et {len(files)-20} de plus")
        await interaction.followup.send(embed=embed,ephemeral=True)


def setup(bot):
    bot.add_cog(Backup(bot))