Correction complète des systèmes anti-spam et anti-liens + optimisations des performances et de la mémoire
This commit is contained in:
parent
59e738c0d4
commit
343a4f879c
4 changed files with 434 additions and 110 deletions
|
|
@ -2,154 +2,184 @@ import discord
|
|||
from discord.ext import commands
|
||||
import json
|
||||
import os
|
||||
import asyncio
|
||||
from datetime import datetime
|
||||
from collections import deque
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
# Chemins des fichiers
|
||||
SETTINGS_FILE = "data/security_settings.json"
|
||||
WHITELIST_FILE = "data/whitelist.json"
|
||||
|
||||
# Cache TTL en secondes (5 secondes pour avoir une réactivité acceptable)
|
||||
CACHE_TTL = 5
|
||||
|
||||
# Triggers pour détection de liens
|
||||
LINK_TRIGGERS = ("discord.gg/", "http://", "https://", "www.")
|
||||
|
||||
class AntiSpam(commands.Cog):
|
||||
def __init__(self, bot):
|
||||
self.bot = bot
|
||||
# Compteurs pour le Spam
|
||||
self.message_counts = {}
|
||||
self.spam_warnings = {}
|
||||
# Optimisation: deque avec maxlen pour O(1) au lieu de O(n) list filtering
|
||||
self.message_counts = {}
|
||||
self.spam_warnings = {}
|
||||
self.link_warnings = {}
|
||||
# Cache avec timestamps pour éviter la lecture fichier à chaque message
|
||||
self._settings_cache = {} # {guild_id: (timestamp, settings_dict)}
|
||||
self._whitelist_cache = {} # {guild_id: (timestamp, whitelist_set)}
|
||||
|
||||
def _get_cached_settings(self, guild_id):
|
||||
"""Récupère les paramètres avec cache (TTL de 5 secondes)"""
|
||||
now = discord.utils.utcnow().timestamp()
|
||||
cached = self._settings_cache.get(guild_id)
|
||||
|
||||
# Compteurs pour les Liens
|
||||
self.link_warnings = {}
|
||||
if cached:
|
||||
cached_time, settings = cached
|
||||
if now - cached_time < CACHE_TTL:
|
||||
return settings
|
||||
|
||||
# Cache expiré ou absent - lire le fichier
|
||||
try:
|
||||
if os.path.exists(SETTINGS_FILE):
|
||||
with open(SETTINGS_FILE, "r", encoding='utf-8') as f:
|
||||
all_settings = json.load(f)
|
||||
settings = all_settings.get(str(guild_id), {})
|
||||
else:
|
||||
settings = {}
|
||||
except:
|
||||
settings = {}
|
||||
|
||||
self._settings_cache[guild_id] = (now, settings)
|
||||
return settings
|
||||
|
||||
def _get_cached_whitelist(self, guild_id):
|
||||
"""Récupère la whitelist avec cache (TTL de 5 secondes)"""
|
||||
now = discord.utils.utcnow().timestamp()
|
||||
cached = self._whitelist_cache.get(guild_id)
|
||||
|
||||
if cached:
|
||||
cached_time, whitelist_set = cached
|
||||
if now - cached_time < CACHE_TTL:
|
||||
return whitelist_set
|
||||
|
||||
# Cache expiré ou absent - lire le fichier
|
||||
whitelist_set = set()
|
||||
try:
|
||||
if os.path.exists(WHITELIST_FILE):
|
||||
with open(WHITELIST_FILE, "r", encoding='utf-8') as f:
|
||||
data = json.load(f)
|
||||
server_id_str = str(guild_id)
|
||||
whitelist_ids = data.get(server_id_str, [])
|
||||
whitelist_set = set(str(uid) for uid in whitelist_ids)
|
||||
except:
|
||||
pass
|
||||
|
||||
self._whitelist_cache[guild_id] = (now, whitelist_set)
|
||||
return whitelist_set
|
||||
|
||||
def is_whitelisted(self, guild_id, user_id):
|
||||
"""Vérifie si l'utilisateur est protégé (Whitelist ou Owner)"""
|
||||
# 1. Le propriétaire est toujours whitelist
|
||||
"""Vérifie l'immunité (Propriétaire ou Whitelist JSON avec cache)"""
|
||||
guild = self.bot.get_guild(guild_id)
|
||||
if guild and user_id == guild.owner_id:
|
||||
return True
|
||||
|
||||
# 2. Vérification fichier Whitelist
|
||||
if not os.path.exists(WHITELIST_FILE): return False
|
||||
try:
|
||||
with open(WHITELIST_FILE, "r", encoding='utf-8') as f:
|
||||
data = json.load(f)
|
||||
return user_id in data.get(str(guild_id), [])
|
||||
except: return False
|
||||
# Utilisation du cache whitelist
|
||||
whitelist = self._get_cached_whitelist(guild_id)
|
||||
return str(user_id) in whitelist
|
||||
|
||||
@commands.Cog.listener()
|
||||
async def on_message(self, message):
|
||||
if message.author.bot or not message.guild: return
|
||||
# 1. Filtres de base (early return pour performance)
|
||||
if message.author.bot or not message.guild:
|
||||
return
|
||||
|
||||
guild_id = message.guild.id
|
||||
user_id = message.author.id
|
||||
|
||||
# --- 1. IMMUNITÉ TOTALE ---
|
||||
# Si Whitelisté, on ignore tout (Spam et Liens)
|
||||
if self.is_whitelisted(message.guild.id, message.author.id):
|
||||
# 2. VÉRIFICATION WHITELIST (Immunité totale)
|
||||
if self.is_whitelisted(guild_id, user_id):
|
||||
return
|
||||
|
||||
# --- 2. CHARGEMENT CONFIG ---
|
||||
if not os.path.exists(SETTINGS_FILE): return
|
||||
with open(SETTINGS_FILE, "r", encoding='utf-8') as f:
|
||||
settings = json.load(f).get(str(message.guild.id), {})
|
||||
# --- LECTURE CACHÉ (dynamique mais rapide) ---
|
||||
settings = self._get_cached_settings(guild_id)
|
||||
|
||||
# --- TEST ANTI-LIENS ---
|
||||
# Bloque tout le monde sauf whitelist ET manage_messages
|
||||
anti_link_enabled = settings.get("anti_link", False) if settings else False
|
||||
has_perm = message.author.guild_permissions.manage_messages
|
||||
if anti_link_enabled and not has_perm:
|
||||
content_lower = message.content.lower()
|
||||
if any(trigger in content_lower for trigger in LINK_TRIGGERS):
|
||||
await self.handle_violation(message, "link")
|
||||
return
|
||||
|
||||
# =================================================================
|
||||
# MODULE ANTI-LINK (3 Avertissements -> Mute)
|
||||
# =================================================================
|
||||
if settings.get("anti_link"):
|
||||
# Liste des déclencheurs interdits
|
||||
forbidden = ["discord.gg/", "http://", "https://", "www."]
|
||||
# Sortir si pas de settings
|
||||
if not settings:
|
||||
return
|
||||
|
||||
# --- TEST ANTI-SPAM (deque pour O(1)) ---
|
||||
if settings.get("anti_spam") is True:
|
||||
now = discord.utils.utcnow().timestamp()
|
||||
|
||||
# On vérifie si le message contient un lien interdit
|
||||
if any(trigger in message.content.lower() for trigger in forbidden):
|
||||
# Si l'auteur a la permission de gérer les messages, on laisse passer (optionnel, sinon retirer cette ligne)
|
||||
if not message.author.guild_permissions.manage_messages:
|
||||
await self.handle_link_violation(message)
|
||||
return # On arrête là, pas besoin de vérifier le spam si le message est supprimé
|
||||
|
||||
# =================================================================
|
||||
# MODULE ANTI-SPAM (3 Avertissements -> Mute)
|
||||
# =================================================================
|
||||
if settings.get("anti_spam"):
|
||||
user_id = message.author.id
|
||||
now = datetime.now().timestamp()
|
||||
|
||||
# Initialiser ou récupérer le deque pour cet utilisateur
|
||||
if user_id not in self.message_counts:
|
||||
self.message_counts[user_id] = []
|
||||
# maxlen=10 suffit pour la fenêtre de 5 secondes avec seuil typique
|
||||
self.message_counts[user_id] = deque(maxlen=10)
|
||||
|
||||
user_deque = self.message_counts[user_id]
|
||||
user_deque.append(now)
|
||||
|
||||
# Nettoyer les timestamps expirés (plus vieux que 5 secondes)
|
||||
# Le deque supprime automatiquement les anciens éléments quand il dépasse maxlen
|
||||
# Mais on doit aussi filtrer par temps pour être précis
|
||||
while user_deque and now - user_deque[0] >= 5:
|
||||
user_deque.popleft()
|
||||
|
||||
threshold = settings.get("spam_threshold", 5)
|
||||
if len(user_deque) > threshold:
|
||||
await self.handle_violation(message, "spam")
|
||||
|
||||
self.message_counts[user_id].append(now)
|
||||
# Fenêtre de temps : 5 secondes
|
||||
self.message_counts[user_id] = [t for t in self.message_counts[user_id] if now - t < 5]
|
||||
|
||||
# Seuil dynamique (défaut: 5)
|
||||
limit = settings.get("spam_threshold", 5)
|
||||
|
||||
if len(self.message_counts[user_id]) > limit:
|
||||
await self.handle_spam_violation(message)
|
||||
|
||||
|
||||
# --- GESTION DES VIOLATIONS LIENS ---
|
||||
async def handle_link_violation(self, message):
|
||||
try:
|
||||
await message.delete()
|
||||
except:
|
||||
pass # Message déjà supprimé ou manque de perms
|
||||
|
||||
user_id = message.author.id
|
||||
warnings = self.link_warnings.get(user_id, 0) + 1
|
||||
self.link_warnings[user_id] = warnings
|
||||
|
||||
if warnings <= 3:
|
||||
embed = discord.Embed(
|
||||
title="🔗 Liens Interdits",
|
||||
description=f"{message.author.mention}, les liens ne sont pas autorisés.\n**Avertissement : {warnings}/3**",
|
||||
color=0xFFA500
|
||||
)
|
||||
await message.channel.send(embed=embed, delete_after=5)
|
||||
else:
|
||||
await self.sanction_user(message, "Publication de liens interdits")
|
||||
self.link_warnings[user_id] = 0 # Reset après sanction
|
||||
|
||||
# --- GESTION DES VIOLATIONS SPAM ---
|
||||
async def handle_spam_violation(self, message):
|
||||
async def handle_violation(self, message, type_violation):
|
||||
user_id = message.author.id
|
||||
|
||||
# Nettoyage visuel du spam
|
||||
try:
|
||||
await message.channel.purge(limit=5, check=lambda m: m.author == message.author)
|
||||
except: pass
|
||||
if type_violation == "link":
|
||||
try: await message.delete()
|
||||
except: pass
|
||||
self.link_warnings[user_id] = self.link_warnings.get(user_id, 0) + 1
|
||||
current_warns = self.link_warnings[user_id]
|
||||
reason = "Envoi de liens interdits"
|
||||
else:
|
||||
try: await message.channel.purge(limit=5, check=lambda m: m.author == message.author)
|
||||
except: pass
|
||||
self.spam_warnings[user_id] = self.spam_warnings.get(user_id, 0) + 1
|
||||
current_warns = self.spam_warnings[user_id]
|
||||
reason = "Spam excessif"
|
||||
# Réinitialiser le deque après spam détecté
|
||||
self.message_counts[user_id] = deque(maxlen=10)
|
||||
|
||||
warnings = self.spam_warnings.get(user_id, 0) + 1
|
||||
self.spam_warnings[user_id] = warnings
|
||||
|
||||
# Reset le compteur de messages pour éviter le spam d'avertissements
|
||||
self.message_counts[user_id] = []
|
||||
|
||||
if warnings <= 3:
|
||||
if current_warns <= 3:
|
||||
embed = discord.Embed(
|
||||
title="🛡️ Anti-Spam",
|
||||
description=f"{message.author.mention}, veuillez cesser de spammer.\n**Avertissement : {warnings}/3**",
|
||||
color=0xFFA500
|
||||
title="🛡️ Sécurité Kuby",
|
||||
description=f"{message.author.mention}, {reason}.\n**Avertissement : {current_warns}/3**",
|
||||
color=0xFFA500
|
||||
)
|
||||
await message.channel.send(embed=embed, delete_after=5)
|
||||
else:
|
||||
await self.sanction_user(message, "Spam excessif")
|
||||
self.spam_warnings[user_id] = 0 # Reset après sanction
|
||||
await self.apply_mute(message, reason)
|
||||
if type_violation == "link": self.link_warnings[user_id] = 0
|
||||
else: self.spam_warnings[user_id] = 0
|
||||
|
||||
# --- SANCTION COMMUNE (TIMEOUT 5 MIN) ---
|
||||
async def sanction_user(self, message, reason):
|
||||
async def apply_mute(self, message, reason):
|
||||
try:
|
||||
# Durée : 5 minutes
|
||||
duration = discord.utils.utcnow() + asyncio.timedelta(minutes=5)
|
||||
|
||||
# Application du Timeout (Exclusion temporaire)
|
||||
await message.author.timeout(duration, reason=f"Kuby Security: {reason} (3 Warns)")
|
||||
until = discord.utils.utcnow() + timedelta(minutes=5)
|
||||
await message.author.timeout(until, reason=f"Kuby Security: {reason} (3 Warns)")
|
||||
|
||||
embed = discord.Embed(
|
||||
title="⛔ Exclusion Temporaire",
|
||||
description=f"**Utilisateur :** {message.author.mention}\n**Durée :** 5 minutes\n**Raison :** {reason} (Limite atteinte).",
|
||||
color=0xFF0000
|
||||
description=f"**Membre :** {message.author.mention}\n**Durée :** 5 minutes\n**Raison :** {reason} répétitif.",
|
||||
color=0xFF0000
|
||||
)
|
||||
await message.channel.send(embed=embed)
|
||||
|
||||
except discord.Forbidden:
|
||||
await message.channel.send("⚠️ Je n'ai pas la permission d'exclure ce membre (Vérifiez ma hiérarchie).")
|
||||
except Exception as e:
|
||||
print(f"Erreur Sanction: {e}")
|
||||
print(f"Erreur lors de la sanction : {e}")
|
||||
|
||||
async def setup(bot):
|
||||
await bot.add_cog(AntiSpam(bot))
|
||||
Loading…
Add table
Add a link
Reference in a new issue