3534363f98
- advanced_logger.py: parsing NDJSON plus robuste pour les fichiers JSON corrompus ou malformés lors de la vérification des rôles précédents. Évite l'erreur "Expecting value: line 2 column 1". - bot.py: simplification de la détection d'erreur 403 Missing Permissions (codes 50013 et 20023) lors de la restauration des rôles au join d'un membre. Évite le log d'erreur "Failed to restore safe roles" quand c'est juste un problème de permissions/bot role hierarchy. - bug_report.py: simplification de la détection du code 50007 (Cannot send messages to this user) lors de l'envoi de notifications de labels GitLab. Évite le log "Erreur inattendue dans la notification de label". Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
275 lines
No EOL
13 KiB
Python
Executable file
275 lines
No EOL
13 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
|
|
# 403 with code 50013 = Missing Permissions - don't retry
|
|
if isinstance(e, disnake.HTTPException) and e.status == 403:
|
|
error_code = getattr(e, 'code', None)
|
|
# 50013 = Missing Permissions, 20023 = Cannot modify a role higher than bot's highest role
|
|
if error_code in (50013, 20023):
|
|
kuby_logger.warning(f"Missing Manage Roles permission to restore safe roles for {member} in {member.guild.name} - please ensure the bot has the Manage Roles permission")
|
|
break
|
|
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) |