272 lines
13 KiB
Python
Executable file
272 lines
13 KiB
Python
Executable file
# 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))
|