# 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))