144 lines
6.5 KiB
Python
144 lines
6.5 KiB
Python
import discord
|
|
import logging
|
|
from typing import Dict
|
|
|
|
logger = logging.getLogger('Superviseur')
|
|
|
|
class ActionRouter:
|
|
"""Routs generated AI semantic actions to the physical Discord moderative or administrative commands."""
|
|
|
|
def __init__(self, bot):
|
|
self.bot = bot
|
|
self._module_cache = {}
|
|
|
|
async def execute_action(self, action: str, params: Dict, message: discord.Message) -> (bool, str):
|
|
"""Execute an action from LLM response."""
|
|
|
|
# --- INSIGHT HOOK (LLM generated task for Staff) ---
|
|
if action == 'INSIGHT':
|
|
content = params.get('message') or params.get('content') or params.get('insight') or "Pas de contenu"
|
|
|
|
if 'conflit' in content.lower() or 'tension' in content.lower():
|
|
logger.debug("Insight (Tension) logged.")
|
|
|
|
context_data = {
|
|
"author_mention": message.author.mention,
|
|
"author_name": message.author.display_name,
|
|
"channel_mention": message.channel.mention if hasattr(message.channel, 'mention') else 'DM',
|
|
"content": message.content[:200]
|
|
}
|
|
|
|
await self.bot.dispatcher.dispatch_insight(
|
|
insight=content,
|
|
importance=params.get('importance', 2),
|
|
guild=message.guild,
|
|
context_data=context_data
|
|
)
|
|
return True, None
|
|
# -----------------------------
|
|
|
|
# --- ALERT HOOK (LLM generated alert for Admin) ---
|
|
if action == 'ALERT':
|
|
content = params.get('message') or params.get('content') or params.get('alert') or "Alerte vide"
|
|
|
|
embed = discord.Embed(title="🚨 ALERTE DIRECTE (IA)", description=content, color=discord.Color.red())
|
|
embed.set_footer(text=f"Source: {message.author.display_name} | Salon: {message.channel}")
|
|
|
|
for admin_id in self.bot.admin_ids:
|
|
u = self.bot.get_user(admin_id) or await self.bot.fetch_user(admin_id)
|
|
if u: await u.send(embed=embed)
|
|
|
|
return True, None
|
|
# -----------------------------
|
|
|
|
# --- VOICE ACTIONS ---
|
|
if action == 'JOIN_VOICE':
|
|
voice_state = getattr(message.author, 'voice', None)
|
|
|
|
if not voice_state and self.bot.guild_id:
|
|
guild = self.bot.get_guild(self.bot.guild_id)
|
|
if guild:
|
|
member = guild.get_member(message.author.id) or await guild.fetch_member(message.author.id)
|
|
if member:
|
|
voice_state = member.voice
|
|
|
|
if voice_state and voice_state.channel:
|
|
await self.bot.voice_manager.join_channel(voice_state.channel)
|
|
return True, f"Connecté à {voice_state.channel.name}"
|
|
else:
|
|
return False, "Vous n'êtes pas dans un salon vocal identifiable par le système."
|
|
|
|
if action == 'LEAVE_VOICE':
|
|
await self.bot.voice_manager.leave_current()
|
|
return True, None
|
|
|
|
# --- PRIVACY ACTIONS ---
|
|
if action == 'FORGET_USER':
|
|
from ia.memoire import delete_user_memory
|
|
delete_user_memory(message.author.id)
|
|
return True, None
|
|
|
|
# Check whitelist
|
|
if self.bot.whitelist_manager.whitelist:
|
|
required_level = self.bot.whitelist_manager.get_required_level(action)
|
|
if required_level and not self.bot.has_whitelist_permission(message.author.id, required_level):
|
|
return False, f"Niveau whitelist requis: {required_level}"
|
|
|
|
action_map = {
|
|
'KICK': 'commandes.security.kick',
|
|
'BAN': 'commandes.security.ban',
|
|
'UNBAN': 'commandes.security.unban',
|
|
'UNMUTE': 'commandes.security.unmute',
|
|
'MUTE': 'commandes.security.mute',
|
|
'TIMEOUT': 'commandes.security.timeout',
|
|
'PURGE': 'commandes.security.purge',
|
|
'WARN': 'commandes.security.warn',
|
|
'LIST_WARNINGS': 'commandes.security.list_warnings',
|
|
'CREATE_CHANNEL': 'commandes.salons.creer',
|
|
'DELETE_CHANNEL': 'commandes.salons.supprimer',
|
|
'MODIFY_CHANNEL': 'commandes.salons.modifier',
|
|
'RENAME_CHANNEL': 'commandes.salons.renommer',
|
|
'MOVE_CHANNEL': 'commandes.salons.deplacer',
|
|
'CREATE_ROLE': 'commandes.roles.creer',
|
|
'DELETE_ROLE': 'commandes.roles.supprimer',
|
|
'MODIFY_ROLE': 'commandes.roles.modifier',
|
|
'RENAME_ROLE': 'commandes.roles.renommer',
|
|
'MOVE_ROLE': 'commandes.roles.deplacer',
|
|
'ADD_ROLE_TO_USER': 'commandes.roles.add_role',
|
|
'REMOVE_ROLE_FROM_USER': 'commandes.roles.remove_role',
|
|
'CREATE_CATEGORY': 'commandes.categories.creer',
|
|
'DELETE_CATEGORY': 'commandes.categories.supprimer',
|
|
'MODIFY_CATEGORY': 'commandes.categories.modifier',
|
|
'RENAME_CATEGORY': 'commandes.categories.renommer',
|
|
'MOVE_CATEGORY': 'commandes.categories.deplacer',
|
|
'PING': 'commandes.autres.ping',
|
|
'SET_WELCOME_CHANNEL': 'commandes.autres.setwelcomechannel',
|
|
'SET_GOODBYE_CHANNEL': 'commandes.autres.setgoodbyechannel',
|
|
'SET_MUTE_ROLE': 'commandes.autres.setmuterole',
|
|
'SEND_MESSAGE': 'commandes.autres.send_message',
|
|
'HELP': 'commandes.autres.help',
|
|
'STATUS': 'commandes.autres.status',
|
|
}
|
|
|
|
module_name = action_map.get(action)
|
|
if module_name:
|
|
logger.debug(f"Importing module: {module_name}")
|
|
try:
|
|
module = self._module_cache.get(module_name)
|
|
if module is None:
|
|
module = __import__(module_name, fromlist=['execute'])
|
|
self._module_cache[module_name] = module
|
|
|
|
logger.debug(f"Module {module_name} imported successfully")
|
|
await module.execute(self.bot, params, message)
|
|
logger.info(f"Action '{action}' executed successfully")
|
|
return True, None
|
|
|
|
except Exception as e:
|
|
error_msg = str(e)
|
|
logger.error(f"Error during action execution '{action}': {error_msg}")
|
|
return False, error_msg
|
|
else:
|
|
error_msg = f"Unknown action: {action}"
|
|
logger.error(error_msg)
|
|
return False, error_msg
|