313 lines
15 KiB
Python
Executable file
313 lines
15 KiB
Python
Executable file
import disnake
|
|
from disnake.ext import commands
|
|
import json
|
|
import os
|
|
from typing import List, Optional, Dict
|
|
import datetime
|
|
import asyncio
|
|
import logging
|
|
from utils.premium import check_premium_tier
|
|
|
|
kuby_logger = logging.getLogger("KubyBot")
|
|
|
|
# Chemins ABSOLUS
|
|
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
|
|
DATA_DIR = os.path.join(BASE_DIR, "..", "data")
|
|
CONV_SETTINGS_FILE = os.path.join(DATA_DIR, "convocation_settings.json")
|
|
CONV_BACKUPS_FILE = os.path.join(DATA_DIR, "convocation_backups.json")
|
|
|
|
# Créer le dossier data/ s'il n'existe pas
|
|
os.makedirs(DATA_DIR, exist_ok=True)
|
|
|
|
def load_json(filepath: str) -> Dict:
|
|
if not os.path.exists(filepath):
|
|
return {}
|
|
try:
|
|
with open(filepath, "r", encoding='utf-8') as f:
|
|
return json.load(f)
|
|
except (json.JSONDecodeError, FileNotFoundError):
|
|
return {}
|
|
|
|
def save_json(filepath: str, data: Dict):
|
|
os.makedirs(os.path.dirname(filepath), exist_ok=True)
|
|
with open(filepath, "w", encoding='utf-8') as f:
|
|
json.dump(data, f, indent=4, ensure_ascii=False)
|
|
|
|
class Convocation(commands.Cog):
|
|
def __init__(self, bot):
|
|
self.bot = bot
|
|
self.settings = load_json(CONV_SETTINGS_FILE)
|
|
self.backups = load_json(CONV_BACKUPS_FILE)
|
|
|
|
def get_guild_settings(self, guild_id: int) -> Dict:
|
|
return self.settings.get(str(guild_id), {})
|
|
|
|
def save_guild_settings(self, guild_id: int, guild_settings: Dict):
|
|
self.settings[str(guild_id)] = guild_settings
|
|
save_json(CONV_SETTINGS_FILE, self.settings)
|
|
|
|
def get_backup(self, guild_id: int, user_id: int) -> Optional[Dict]:
|
|
key = f"{guild_id}_{user_id}"
|
|
return self.backups.get(key)
|
|
|
|
def save_backup(self, guild_id: int, user_id: int, backup_data: Dict):
|
|
key = f"{guild_id}_{user_id}"
|
|
self.backups[key] = backup_data
|
|
save_json(CONV_BACKUPS_FILE, self.backups)
|
|
|
|
def clear_backup(self, guild_id: int, user_id: int):
|
|
key = f"{guild_id}_{user_id}"
|
|
if key in self.backups:
|
|
del self.backups[key]
|
|
save_json(CONV_BACKUPS_FILE, self.backups)
|
|
|
|
@commands.slash_command(name="convocation", description="Gère les convocations de membres")
|
|
@check_premium_tier()
|
|
@commands.has_permissions(manage_roles=True)
|
|
async def convocation_group(self, interaction: disnake.ApplicationCommandInteraction):
|
|
pass
|
|
|
|
@convocation_group.sub_command(name="convoquer", description="Convoque un membre et verrouille ses accès")
|
|
async def convoquer(self, interaction: disnake.ApplicationCommandInteraction,
|
|
membre: disnake.Member = commands.Param(description="Le membre à convoquer"),
|
|
raison: str = commands.Param(description="La raison de la convocation"),
|
|
heure: str = commands.Param(description="L'heure de la convocation (ex: 14:30)")):
|
|
guild_id = interaction.guild.id
|
|
settings = self.get_guild_settings(guild_id)
|
|
role_conv_id = settings.get("role_id")
|
|
log_channel_id = settings.get("log_channel_id")
|
|
attente_id = settings.get("attente_id") # Salon d'attente à ignorer
|
|
admin_ids = settings.get("admin_ids", [])
|
|
|
|
if not role_conv_id:
|
|
await interaction.response.send_message("❌ Le rôle de convocation n'est pas configuré. Utilisez `/convocation config`.", ephemeral=True)
|
|
return
|
|
|
|
await interaction.response.defer(ephemeral=True)
|
|
|
|
try:
|
|
# --- BACKUP ---
|
|
bot_top_role = interaction.guild.me.top_role
|
|
backup_data = {
|
|
"roles": [r.id for r in membre.roles if not r.is_default() and r.id != role_conv_id and r.position < bot_top_role.position],
|
|
"overrides": {}
|
|
}
|
|
|
|
# 1. Backup and Remove current roles
|
|
current_roles = [interaction.guild.get_role(rid) for rid in backup_data["roles"] if interaction.guild.get_role(rid)]
|
|
if current_roles:
|
|
try:
|
|
await membre.remove_roles(*current_roles, reason=f"Convocation par {interaction.author}")
|
|
except disnake.Forbidden:
|
|
await interaction.followup.send("⚠️ Je n'ai pas pu retirer certains rôles (Permission refusée).", ephemeral=True)
|
|
|
|
# 2. Dynamic Lockdown: Loop through ALL channels
|
|
semaphore = asyncio.Semaphore(10)
|
|
|
|
async def lock_channel(channel):
|
|
async with semaphore:
|
|
# On ignore le salon d'attente et ses parents s'il y en a
|
|
if attente_id and (channel.id == attente_id or (hasattr(channel, 'category') and channel.category and channel.category.id == attente_id)):
|
|
return False
|
|
|
|
# Check existing override for the member
|
|
overwrites = channel.overwrites_for(membre)
|
|
if not overwrites.is_empty():
|
|
# Save existing override
|
|
backup_data["overrides"][str(channel.id)] = {k: v for k, v in dict(overwrites).items() if v is not None}
|
|
|
|
# Apply Lockdown override for the member
|
|
try:
|
|
await channel.set_permissions(membre, view_channel=False, reason="Convocation Lockdown")
|
|
return True
|
|
except disnake.Forbidden:
|
|
return False
|
|
except Exception as e:
|
|
kuby_logger.warning(f"⚠️ [Convocation] Impossible de verrouiller {channel.name}: {e}")
|
|
return False
|
|
|
|
tasks = [lock_channel(c) for c in interaction.guild.channels]
|
|
results = await asyncio.gather(*tasks)
|
|
lockdown_count = sum(1 for r in results if r is True)
|
|
|
|
self.save_backup(guild_id, membre.id, backup_data)
|
|
|
|
# 3. Add Convocation role
|
|
role_conv = interaction.guild.get_role(role_conv_id)
|
|
if role_conv:
|
|
if role_conv.position >= bot_top_role.position:
|
|
await interaction.followup.send(f"⚠️ Je ne peux pas donner le rôle {role_conv.mention} car il est plus haut que moi !", ephemeral=True)
|
|
else:
|
|
try:
|
|
await membre.add_roles(role_conv, reason="Début de convocation")
|
|
except disnake.Forbidden:
|
|
await interaction.followup.send(f"⚠️ Impossible de donner le rôle {role_conv.name} (Permission refusée).", ephemeral=True)
|
|
|
|
# 4. Embeds & Logs (Inchangés mais inclus pour la cohérence)
|
|
embed_conv = disnake.Embed(
|
|
title="CONVOCATION | 🔖",
|
|
description=f"> Bonjour {membre.mention}, vous êtes **convoqué** par {interaction.author.mention}. Voici les **détails** :",
|
|
color=0x0099ff,
|
|
timestamp=disnake.utils.utcnow()
|
|
)
|
|
embed_conv.add_field(name="**Raison de la convocation :**", value=f"➜ {raison}", inline=False)
|
|
embed_conv.add_field(name="**Heure de la convocation :**", value=f"➜ {heure}", inline=False)
|
|
embed_conv.add_field(name="\u200B", value=(
|
|
"> Merci de vous présenter dans le vocal attente **5 minutes** avant l'heure.\n\n"
|
|
"> La **levée** de suspension se fera uniquement à la **fin** de la convocation."
|
|
), inline=False)
|
|
embed_conv.set_footer(text=f"Gestion de {interaction.guild.name}")
|
|
|
|
embed_log = disnake.Embed(
|
|
title="📋 CONVOCATION ENVOYÉE",
|
|
color=0xff9900,
|
|
timestamp=disnake.utils.utcnow()
|
|
)
|
|
embed_log.add_field(name="👤 Membre convoqué", value=f"{membre.mention} ({membre.id})", inline=True)
|
|
embed_log.add_field(name="👮 Convoqué par", value=f"{interaction.author.mention}", inline=True)
|
|
embed_log.add_field(name="📄 Raison", value=raison, inline=False)
|
|
embed_log.add_field(name="🕒 Heure", value=heure, inline=False)
|
|
|
|
# DM
|
|
dm_sent = False
|
|
try:
|
|
await membre.send(embed=embed_conv)
|
|
dm_sent = True
|
|
except disnake.Forbidden:
|
|
dm_sent = False
|
|
|
|
# Admin & Channel Logs
|
|
for admin_id in admin_ids:
|
|
try:
|
|
admin = await self.bot.fetch_user(admin_id)
|
|
await admin.send(embed=embed_log)
|
|
except: pass
|
|
|
|
if log_channel_id:
|
|
chan = interaction.guild.get_channel(log_channel_id)
|
|
if chan: await chan.send(embed=embed_log)
|
|
|
|
await interaction.followup.send(f"✅ Convocation envoyée à {membre.mention}. ({lockdown_count} salons verrouillés) " +
|
|
("" if dm_sent else "(⚠️ DM impossible)"), ephemeral=True)
|
|
|
|
except Exception as e:
|
|
import traceback
|
|
traceback.print_exc()
|
|
await interaction.followup.send(f"❌ Une erreur critique est survenue : {e}", ephemeral=True)
|
|
|
|
@convocation_group.sub_command(name="lever", description="Lève la convocation d'un membre et restaure ses accès")
|
|
async def lever(self, interaction: disnake.ApplicationCommandInteraction,
|
|
membre: disnake.Member = commands.Param(description="Le membre à libérer")):
|
|
guild_id = interaction.guild.id
|
|
settings = self.get_guild_settings(guild_id)
|
|
role_conv_id = settings.get("role_id")
|
|
|
|
await interaction.response.defer(ephemeral=True)
|
|
|
|
backup_data = self.get_backup(guild_id, membre.id)
|
|
if not backup_data:
|
|
await interaction.followup.send("⚠️ Aucune donnée de convocation trouvée pour ce membre.", ephemeral=True)
|
|
return
|
|
|
|
# 1. Remove Convocation role & Restore original roles FIRST
|
|
try:
|
|
if role_conv_id:
|
|
role_conv = interaction.guild.get_role(role_conv_id)
|
|
if role_conv and role_conv in membre.roles:
|
|
await membre.remove_roles(role_conv, reason="Fin de convocation")
|
|
|
|
role_ids = backup_data.get("roles", [])
|
|
if role_ids:
|
|
restored_roles = []
|
|
for rid in role_ids:
|
|
role = interaction.guild.get_role(rid)
|
|
if role and role.position < interaction.guild.me.top_role.position:
|
|
restored_roles.append(role)
|
|
|
|
if restored_roles:
|
|
await membre.add_roles(*restored_roles, reason="Restauration fin de convocation")
|
|
except Exception as e:
|
|
kuby_logger.error(f"❌ [Convocation] Erreur lors de la restauration des rôles pour {membre}: {e}")
|
|
|
|
# 2. Restore channel overrides (Parallel)
|
|
semaphore = asyncio.Semaphore(10)
|
|
overrides = backup_data.get("overrides", {})
|
|
|
|
async def restore_channel(channel):
|
|
async with semaphore:
|
|
chan_id_str = str(channel.id)
|
|
try:
|
|
if chan_id_str in overrides:
|
|
# Restore previous override
|
|
ov_data = overrides[chan_id_str]
|
|
overwrite = disnake.PermissionOverwrite(**ov_data)
|
|
await channel.set_permissions(membre, overwrite=overwrite, reason="Restauration fin de convocation")
|
|
return "restored"
|
|
elif membre in channel.overwrites:
|
|
# Clear our lockdown override only if it exists
|
|
await channel.set_permissions(membre, overwrite=None, reason="Fin de convocation - Cleanup")
|
|
return "cleared"
|
|
except disnake.Forbidden:
|
|
return "forbidden"
|
|
except Exception as e:
|
|
kuby_logger.warning(f"⚠️ [Convocation] Erreur restauration {channel.name}: {e}")
|
|
return "error"
|
|
return "skipped"
|
|
|
|
tasks = [restore_channel(c) for c in interaction.guild.channels]
|
|
results = await asyncio.gather(*tasks)
|
|
|
|
restore_count = sum(1 for r in results if r == "restored")
|
|
clear_count = sum(1 for r in results if r == "cleared")
|
|
forbidden_count = sum(1 for r in results if r == "forbidden")
|
|
|
|
kuby_logger.info(f"🔓 [Convocation] Restauration terminée pour {membre}: {restore_count} restaurés, {clear_count} nettoyés, {forbidden_count} échecs permissions.")
|
|
|
|
self.clear_backup(guild_id, membre.id)
|
|
|
|
# 4. Notification
|
|
embed = disnake.Embed(
|
|
title="🔓 CONVOCATION TERMINÉE",
|
|
description=f"Votre convocation sur **{interaction.guild.name}** est terminée. Vos accès ont été restaurés.",
|
|
color=disnake.Color.green()
|
|
)
|
|
try:
|
|
await membre.send(embed=embed)
|
|
except: pass
|
|
|
|
await interaction.followup.send(f"✅ Convocation levée pour {membre.mention}. Access restaurés.", ephemeral=True)
|
|
|
|
@convocation_group.sub_command(name="config", description="Configure le système de convocation")
|
|
async def config(self, interaction: disnake.ApplicationCommandInteraction,
|
|
role: disnake.Role = commands.Param(description="Le rôle de convocation (Muet/Convocation)"),
|
|
channel: disnake.TextChannel = commands.Param(None, description="Le salon de logs"),
|
|
attente: disnake.abc.GuildChannel = commands.Param(None, description="Salon (ou catégorie) d'attente à laisser visible"),
|
|
admins: str = commands.Param(None, description="Liste d'IDs d'admins à notifier (séparés par des espaces)")):
|
|
guild_id = interaction.guild.id
|
|
settings = self.get_guild_settings(guild_id)
|
|
|
|
settings["role_id"] = role.id
|
|
if channel:
|
|
settings["log_channel_id"] = channel.id
|
|
if attente:
|
|
settings["attente_id"] = attente.id
|
|
|
|
if admins:
|
|
admin_list = []
|
|
for uid in admins.split():
|
|
if uid.isdigit():
|
|
admin_list.append(int(uid))
|
|
settings["admin_ids"] = admin_list
|
|
|
|
self.save_guild_settings(guild_id, settings)
|
|
|
|
embed = disnake.Embed(title="⚙️ Configuration Convocation Mise à jour", color=disnake.Color.blue())
|
|
embed.add_field(name="Rôle", value=role.mention)
|
|
embed.add_field(name="Salon Logs", value=channel.mention if channel else "Non défini")
|
|
embed.add_field(name="Salon/Cat d'attente", value=attente.mention if attente else "Non défini")
|
|
embed.add_field(name="Admins notifiés", value=str(len(settings.get("admin_ids", []))))
|
|
|
|
await interaction.response.send_message(embed=embed, ephemeral=True)
|
|
|
|
def setup(bot):
|
|
bot.add_cog(Convocation(bot))
|