268 lines
No EOL
12 KiB
Python
Executable file
268 lines
No EOL
12 KiB
Python
Executable file
import os
|
|
import sys
|
|
from dotenv import load_dotenv
|
|
|
|
# Charger les variables d'environnement AVANT tout le reste
|
|
load_dotenv()
|
|
|
|
import disnake
|
|
from commandes.ticket.utils.permissions import can_access_config
|
|
from commandes.ticket.data.storage import get_storage
|
|
from disnake.ext import commands
|
|
import asyncio
|
|
from src.logger import kuby_logger, log_performance
|
|
import logging
|
|
from src.advanced_logger import AdvancedLogger
|
|
|
|
SENSITIVE_PERMISSIONS = {
|
|
"administrator", "manage_guild", "manage_roles", "manage_channels",
|
|
"kick_members", "ban_members", "manage_messages", "manage_webhooks",
|
|
"moderate_members"
|
|
}
|
|
|
|
class RoleApprovalView(disnake.ui.View):
|
|
def __init__(self, member: disnake.Member, roles: list, owner: disnake.Member):
|
|
super().__init__(timeout=86400) # 24 hours
|
|
self.member = member
|
|
self.roles = roles
|
|
self.owner = owner
|
|
|
|
@disnake.ui.button(label="Accepter", style=disnake.ButtonStyle.green, emoji="✅")
|
|
async def accept(self, button: disnake.ui.Button, interaction: disnake.MessageInteraction):
|
|
guild = self.member.guild
|
|
current_member = guild.get_member(self.member.id)
|
|
if not current_member:
|
|
await interaction.response.send_message("❌ L'utilisateur n'est plus sur le serveur.", ephemeral=True)
|
|
self.stop()
|
|
return
|
|
|
|
try:
|
|
await current_member.add_roles(*self.roles, reason="Restauration sécurisée approuvée par le propriétaire")
|
|
await interaction.response.send_message(f"✅ Rôles sensibles restaurés pour {self.member.mention}.", ephemeral=True)
|
|
|
|
components = [
|
|
disnake.ui.Container(
|
|
disnake.ui.TextDisplay(f"✅ Vous avez approuvé la restauration des rôles pour {self.member.mention}."),
|
|
disnake.ui.TextDisplay(f"🛡️ Rôles : {', '.join([r.name for r in self.roles])}")
|
|
)
|
|
]
|
|
await interaction.edit_original_response(embed=None, components=components, view=None)
|
|
self.stop()
|
|
except Exception as e:
|
|
await interaction.response.send_message(f"❌ Erreur lors de l'attribution : {e}", ephemeral=True)
|
|
|
|
@disnake.ui.button(label="Refuser", style=disnake.ButtonStyle.red, emoji="✖️")
|
|
async def deny(self, button: disnake.ui.Button, interaction: disnake.MessageInteraction):
|
|
await interaction.response.send_message(f"❌ Vous avez refusé la restauration pour {self.member.mention}.", ephemeral=True)
|
|
|
|
components = [
|
|
disnake.ui.Container(
|
|
disnake.ui.TextDisplay(f"❌ Vous avez refusé la restauration des rôles pour {self.member.mention}."),
|
|
disnake.ui.TextDisplay(f"🛡️ Rôles : {', '.join([r.name for r in self.roles])}")
|
|
)
|
|
]
|
|
await interaction.edit_original_response(embed=None, components=components, view=None)
|
|
self.stop()
|
|
|
|
|
|
# Récupération et validation de l'ID d'application
|
|
application_id_raw = os.getenv("APPLICATION_ID")
|
|
kuby_logger.debug(f"Raw APPLICATION_ID: {application_id_raw}")
|
|
try:
|
|
application_id = int(application_id_raw) if application_id_raw else None
|
|
kuby_logger.info(f"✅ APPLICATION_ID loaded successfully: {application_id}")
|
|
except ValueError:
|
|
kuby_logger.error(f"❌ APPLICATION_ID invalide : {application_id_raw}")
|
|
application_id = None
|
|
|
|
# Configuration des intents Disnake
|
|
intents = disnake.Intents.default()
|
|
intents.message_content = True
|
|
intents.members = True
|
|
intents.voice_states = True
|
|
kuby_logger.debug(f"Disnake intents configured: {intents}")
|
|
|
|
|
|
class MyBot(commands.Bot):
|
|
def __init__(self):
|
|
kuby_logger.debug("Initializing MyBot class")
|
|
super().__init__(
|
|
command_prefix="!",
|
|
intents=intents,
|
|
application_id=application_id,
|
|
command_sync_flags=commands.CommandSyncFlags.all()
|
|
)
|
|
kuby_logger.info("MyBot instance created successfully")
|
|
|
|
# Initialize advanced logger
|
|
self.advanced_logger = AdvancedLogger(self)
|
|
kuby_logger.info("✅ Advanced logger initialized")
|
|
|
|
# Flask est géré par PM2 (kuby-flask), pas besoin de le lancer ici
|
|
|
|
async def do_async_setup(self):
|
|
kuby_logger.info("🚀 Configuration initiale du bot Kuby...")
|
|
|
|
extensions = [
|
|
"commandes.whitelist",
|
|
"commandes.whitelist_monitor",
|
|
"commandes.sendbotmessages",
|
|
"commandes.goodbye",
|
|
"commandes.welcome",
|
|
"commandes.security",
|
|
"commandes.snipe",
|
|
"commandes.scan",
|
|
"commandes.no_link",
|
|
"commandes.blacklist",
|
|
"commandes.ticket",
|
|
"commandes.modules_security.antispam",
|
|
"commandes.modules_security.antiraid",
|
|
"commandes.modules_security.logs_manager",
|
|
"commandes.modules_security.rules",
|
|
"commandes.ticket_whitelist",
|
|
"commandes.bug_report",
|
|
"commandes.convocation",
|
|
"commandes.absence_staff",
|
|
"commandes.moderation",
|
|
"commandes.invites",
|
|
]
|
|
|
|
for extension in extensions:
|
|
try:
|
|
self.load_extension(extension)
|
|
kuby_logger.info(f"✅ Extension chargée : {extension}")
|
|
except Exception as e:
|
|
kuby_logger.error(f"❌ Erreur lors du chargement de {extension} : {e}", exc_info=True)
|
|
|
|
# --- LOGGING DES INTERACTIONS ---
|
|
@self.listen("on_interaction")
|
|
async def log_interaction(inter: disnake.Interaction):
|
|
if inter.type == disnake.InteractionType.application_command:
|
|
kuby_logger.info(f"⚡ Interaction reçue : /{inter.data.name} par {inter.user}")
|
|
|
|
kuby_logger.info("✅ Configuration terminée")
|
|
|
|
async def on_ready(self):
|
|
kuby_logger.info(f"🤖 Bot connecté : {self.user} (ID : {self.user.id})")
|
|
kuby_logger.info(f"📊 Serveurs : {len(self.guilds)}")
|
|
kuby_logger.info(f"🔍 Intents Status - Members: {self.intents.members}, Presences: {self.intents.presences}, Core: {self.intents.message_content}")
|
|
if not self.intents.members:
|
|
kuby_logger.warning("⚠️ L'intent MEMBERS est désactivé !")
|
|
kuby_logger.info("✅ Bot prêt et opérationnel !")
|
|
|
|
async def on_guild_join(self, guild):
|
|
kuby_logger.info(f"🆕 Bot joined new guild: {guild.name} (ID: {guild.id})")
|
|
kuby_logger.audit(f"Bot added to guild: {guild.name}")
|
|
|
|
async def on_guild_remove(self, guild):
|
|
kuby_logger.info(f"🚪 Bot removed from guild: {guild.name}")
|
|
kuby_logger.audit(f"Bot removed from guild: {guild.name}")
|
|
|
|
async def on_member_join(self, member):
|
|
kuby_logger.info(f"👋 {member} joined {member.guild.name}")
|
|
if hasattr(self, 'advanced_logger'):
|
|
self.advanced_logger.log_member_join(member)
|
|
previous_roles = self.advanced_logger.check_previous_roles(member.id)
|
|
if previous_roles:
|
|
guild = member.guild
|
|
roles_to_add = [guild.get_role(rid) for rid in previous_roles.get("role_ids", [])]
|
|
roles_to_add = [r for r in roles_to_add if r and r < guild.me.top_role]
|
|
if roles_to_add:
|
|
safe_roles = []
|
|
sensitive_roles = []
|
|
|
|
for role in roles_to_add:
|
|
is_sensitive = any(getattr(role.permissions, perm, False) for perm in SENSITIVE_PERMISSIONS)
|
|
if is_sensitive:
|
|
sensitive_roles.append(role)
|
|
else:
|
|
safe_roles.append(role)
|
|
|
|
if safe_roles:
|
|
retries = 3
|
|
for i in range(retries):
|
|
try:
|
|
await member.add_roles(*safe_roles)
|
|
kuby_logger.info(f"Restored {len(safe_roles)} safe roles for {member}")
|
|
break
|
|
except (disnake.HTTPException, disnake.GatewayNotFound, asyncio.TimeoutError) as e:
|
|
is_transient = isinstance(e, disnake.HTTPException) and e.status in [500, 502, 503, 504]
|
|
if not isinstance(e, disnake.HTTPException):
|
|
is_transient = True
|
|
if is_transient and i < retries - 1:
|
|
await asyncio.sleep((i + 1) * 2)
|
|
continue
|
|
kuby_logger.error(f"Failed to restore safe roles for {member}: {e}")
|
|
break
|
|
except Exception as e:
|
|
kuby_logger.error(f"Error restoring safe roles for {member}: {e}")
|
|
break
|
|
|
|
if sensitive_roles:
|
|
try:
|
|
owner = guild.owner or await guild.fetch_member(guild.owner_id)
|
|
embed = disnake.Embed(
|
|
title="⚠️ Restauration de rôles sensibles",
|
|
description=(
|
|
f"L'utilisateur **{member}** ({member.id}) vient de rejoindre **{guild.name}**.\n\n"
|
|
"Certains de ses anciens rôles possèdent des permissions sensibles :\n"
|
|
f"🛡️ **Rôles en attente :** {', '.join([r.mention for r in sensitive_roles])}\n\n"
|
|
"Voulez-vous restaurer ces rôles ?"
|
|
),
|
|
color=disnake.Color.yellow()
|
|
)
|
|
embed.set_thumbnail(url=member.display_avatar.url)
|
|
view = RoleApprovalView(member, sensitive_roles, owner)
|
|
await owner.send(embed=embed, view=view)
|
|
kuby_logger.info(f"Sent role approval request for {member} to owner {owner}")
|
|
except Exception as e:
|
|
kuby_logger.error(f"Failed to send role approval request to owner: {e}")
|
|
|
|
async def on_member_remove(self, member):
|
|
kuby_logger.info(f"👋 {member} left {member.guild.name}")
|
|
if hasattr(self, 'advanced_logger'):
|
|
self.advanced_logger.log_member_leave(member)
|
|
try:
|
|
from commandes.security import load_settings
|
|
settings = load_settings(member.guild.id)
|
|
log_channel_id = settings.get("log_channel_id")
|
|
if log_channel_id:
|
|
log_channel = member.guild.get_channel(log_channel_id)
|
|
if log_channel:
|
|
await self.advanced_logger.send_leave_notification(member, log_channel)
|
|
except Exception as e:
|
|
kuby_logger.error(f"Error in leave notification: {e}")
|
|
|
|
async def on_message_delete(self, message):
|
|
if message.author.bot:
|
|
return
|
|
if hasattr(self, 'advanced_logger'):
|
|
self.advanced_logger.log_message_event(message, "deleted")
|
|
|
|
async def on_message_edit(self, before, after):
|
|
if before.author.bot or before.content == after.content:
|
|
return
|
|
if hasattr(self, 'advanced_logger'):
|
|
self.advanced_logger.log_message_event(before, "edited", {
|
|
"before_content": before.content,
|
|
"after_content": after.content
|
|
})
|
|
|
|
async def on_voice_state_update(self, member, before, after):
|
|
if member.bot:
|
|
return
|
|
if before.channel != after.channel and hasattr(self, 'advanced_logger'):
|
|
if before.channel is None:
|
|
self.advanced_logger.log_voice_event(member, after.channel, "joined")
|
|
elif after.channel is None:
|
|
self.advanced_logger.log_voice_event(member, before.channel, "left")
|
|
else:
|
|
self.advanced_logger.log_voice_event(member, before.channel, "moved_from")
|
|
self.advanced_logger.log_voice_event(member, after.channel, "moved_to")
|
|
|
|
async def on_command_error(self, ctx, error):
|
|
if isinstance(error, commands.CommandNotFound):
|
|
return
|
|
if isinstance(error, commands.CheckFailure):
|
|
return
|
|
kuby_logger.error(f"Command error in {ctx.command}: {str(error)}", exc_info=True) |