516 lines
22 KiB
Python
Executable file
516 lines
22 KiB
Python
Executable file
import disnake
|
|
from disnake.ext import commands, tasks
|
|
import json
|
|
import asyncio
|
|
import os
|
|
from datetime import datetime, timezone
|
|
try:
|
|
from zoneinfo import ZoneInfo
|
|
except ImportError:
|
|
from backports.zoneinfo import ZoneInfo
|
|
from typing import Optional, Dict, Any
|
|
|
|
PARIS_TZ = ZoneInfo("Europe/Paris")
|
|
DATE_FORMAT = "%d/%m/%Y %H:%M"
|
|
|
|
ABSENCE_SETTINGS_FILE = "data/absence_settings.json"
|
|
ABSENCE_DATA_FILE = "data/absences.json"
|
|
file_lock = asyncio.Lock()
|
|
|
|
def load_settings(guild_id: int) -> dict:
|
|
"""Charge les paramètres d'absence pour un serveur"""
|
|
if not os.path.exists(ABSENCE_SETTINGS_FILE):
|
|
return {}
|
|
try:
|
|
with open(ABSENCE_SETTINGS_FILE, "r", encoding='utf-8') as f:
|
|
return json.load(f).get(str(guild_id), {})
|
|
except (json.JSONDecodeError, FileNotFoundError):
|
|
return {}
|
|
|
|
def save_settings(guild_id: int, settings: dict) -> None:
|
|
"""Sauvegarde les paramètres d'absence"""
|
|
data = {}
|
|
if os.path.exists(ABSENCE_SETTINGS_FILE):
|
|
try:
|
|
with open(ABSENCE_SETTINGS_FILE, "r", encoding='utf-8') as f:
|
|
data = json.load(f)
|
|
except (json.JSONDecodeError, FileNotFoundError):
|
|
pass
|
|
|
|
data[str(guild_id)] = settings
|
|
os.makedirs(os.path.dirname(ABSENCE_SETTINGS_FILE), exist_ok=True)
|
|
with open(ABSENCE_SETTINGS_FILE, "w", encoding='utf-8') as f:
|
|
json.dump(data, f, indent=4, ensure_ascii=False)
|
|
|
|
async def load_absences() -> dict:
|
|
"""Charge toutes les absences (tous serveurs confondus)"""
|
|
async with file_lock:
|
|
if not os.path.exists(ABSENCE_DATA_FILE):
|
|
return {}
|
|
try:
|
|
with open(ABSENCE_DATA_FILE, "r", encoding='utf-8') as f:
|
|
return json.load(f)
|
|
except (json.JSONDecodeError, FileNotFoundError):
|
|
return {}
|
|
|
|
async def save_absences(data: dict) -> None:
|
|
"""Sauvegarde toutes les absences"""
|
|
async with file_lock:
|
|
os.makedirs(os.path.dirname(ABSENCE_DATA_FILE), exist_ok=True)
|
|
temp_file = ABSENCE_DATA_FILE + ".tmp"
|
|
with open(temp_file, "w", encoding='utf-8') as f:
|
|
json.dump(data, f, indent=4, ensure_ascii=False)
|
|
os.replace(temp_file, ABSENCE_DATA_FILE)
|
|
|
|
def get_guild_absences(all_absences: dict, guild_id: int) -> dict:
|
|
"""Filtre les absences pour un serveur spécifique"""
|
|
return all_absences.get(str(guild_id), {})
|
|
|
|
def update_guild_absences(all_absences: dict, guild_id: int, guild_data: dict) -> dict:
|
|
"""Met à jour les absences pour un serveur spécifique"""
|
|
all_absences[str(guild_id)] = guild_data
|
|
return all_absences
|
|
|
|
def get_paris_now():
|
|
"""Récupère l'heure actuelle à Paris"""
|
|
return datetime.now(PARIS_TZ)
|
|
|
|
def parse_date(date_str: Optional[str]):
|
|
"""Parse une date au format JJ/MM/AAAA HH:MM en timezone Paris"""
|
|
if not date_str:
|
|
raise ValueError("Date vide")
|
|
return datetime.strptime(date_str, DATE_FORMAT).replace(tzinfo=PARIS_TZ)
|
|
|
|
# --- MODALS & VIEWS ---
|
|
|
|
class AbsenceModal(disnake.ui.Modal):
|
|
def __init__(self, superieur: Optional[disnake.Member] = None):
|
|
super().__init__(title="Déclarer une Absence Staff", components=[self.motif, self.date_debut, self.date_fin])
|
|
self.superieur = superieur
|
|
|
|
motif = disnake.ui.TextInput(
|
|
label="Motif de l'absence",
|
|
style=disnake.TextInputStyle.paragraph,
|
|
placeholder="Ex: Vacances, Raisons personnelles, Travail...",
|
|
required=True,
|
|
max_length=500
|
|
)
|
|
date_debut = disnake.ui.TextInput(
|
|
label="Début (JJ/MM/AAAA HH:MM)",
|
|
placeholder="Ex: 15/05/2024 08:00",
|
|
min_length=16,
|
|
max_length=16,
|
|
required=True
|
|
)
|
|
date_fin = disnake.ui.TextInput(
|
|
label="Fin (JJ/MM/AAAA HH:MM)",
|
|
placeholder="Ex: 20/05/2024 18:00",
|
|
min_length=16,
|
|
max_length=16,
|
|
required=True
|
|
)
|
|
|
|
async def callback(self, interaction: disnake.ModalInteraction):
|
|
try:
|
|
val_debut = interaction.text_values.get(self.date_debut.custom_id) or self.date_debut.value
|
|
val_fin = interaction.text_values.get(self.date_fin.custom_id) or self.date_fin.value
|
|
|
|
debut = parse_date(val_debut)
|
|
fin = parse_date(val_fin)
|
|
except (ValueError, TypeError) as e:
|
|
return await interaction.response.send_message(f"❌ Format de date invalide ou manquant. Utilise le format: `{DATE_FORMAT.replace('%d', 'JJ').replace('%m', 'MM').replace('%Y', 'AAAA').replace('%H', 'HH').replace('%M', 'MM')}`", ephemeral=True)
|
|
|
|
if debut > fin:
|
|
return await interaction.response.send_message("❌ La date de début doit précéder la date de fin.", ephemeral=True)
|
|
|
|
# Vérification si déjà absent
|
|
all_abs = await load_absences()
|
|
guild_abs = get_guild_absences(all_abs, interaction.guild.id)
|
|
|
|
if str(interaction.user.id) in guild_abs:
|
|
return await interaction.response.send_message("❌ Tu as déjà déclaré une absence sur ce serveur.", ephemeral=True)
|
|
|
|
settings = load_settings(interaction.guild.id)
|
|
channel_id = settings.get("absence_channel_id")
|
|
|
|
if not channel_id:
|
|
return await interaction.response.send_message("❌ Le salon d'absence n'est pas configuré sur ce serveur. Contacte un administrateur.", ephemeral=True)
|
|
|
|
channel = interaction.guild.get_channel(int(channel_id))
|
|
if not channel:
|
|
return await interaction.response.send_message("❌ Salon d'absence introuvable. Vérifie la configuration.", ephemeral=True)
|
|
|
|
duree_delta = fin - debut
|
|
duree_jours = duree_delta.days
|
|
duree_heures = duree_delta.seconds // 3600
|
|
|
|
duree_str = f"{duree_jours}j {duree_heures}h" if duree_jours > 0 else f"{duree_heures}h"
|
|
|
|
now = get_paris_now()
|
|
etat = "à venir" if debut > now else "en cours"
|
|
|
|
# Message dans le salon
|
|
absence_message = await channel.send(
|
|
f"# ABSENCE STAFF :\n"
|
|
f"- **Pseudo Discord** : {interaction.user.mention}\n"
|
|
f"- **Durée prévue** : {duree_str}\n"
|
|
f"- **Débute le** : {self.date_debut.value} (Paris)\n"
|
|
f"- **Termine le** : {self.date_fin.value} (Paris)\n"
|
|
f"- **Motif** : {self.motif.value}\n"
|
|
f"- **Mention du supérieur hiérarchique** : {self.superieur.mention if self.superieur else 'Aucun'}"
|
|
)
|
|
|
|
# Enregistrement
|
|
guild_abs[str(interaction.user.id)] = {
|
|
"pseudo": interaction.user.name,
|
|
"debut": self.date_debut.value,
|
|
"fin": self.date_fin.value,
|
|
"motif": self.motif.value,
|
|
"superieur": self.superieur.mention if self.superieur else "Aucun",
|
|
"message_id": absence_message.id,
|
|
"etat": etat
|
|
}
|
|
|
|
all_abs = update_guild_absences(all_abs, interaction.guild.id, guild_abs)
|
|
await save_absences(all_abs)
|
|
|
|
# Attribution du rôle si déjà en cours
|
|
if etat == "en cours":
|
|
role_id = settings.get("absence_role_id")
|
|
if role_id:
|
|
role = interaction.guild.get_role(int(role_id))
|
|
if role:
|
|
try:
|
|
await interaction.user.add_roles(role, reason="Début d'absence staff")
|
|
except: pass
|
|
|
|
await interaction.response.send_message("✅ Absence enregistrée avec succès !", ephemeral=True)
|
|
|
|
class AbsenceConfigView(disnake.ui.View):
|
|
def __init__(self, guild_id: int):
|
|
super().__init__(timeout=60)
|
|
self.guild_id = guild_id
|
|
|
|
@disnake.ui.button(label="Définir le Salon", style=disnake.ButtonStyle.primary, emoji="📁")
|
|
async def set_channel(self, button: disnake.ui.Button, interaction: disnake.Interaction):
|
|
await interaction.response.send_message("Mentionne le salon ici (ou envoie son ID) :", ephemeral=True)
|
|
|
|
def check(m):
|
|
return m.author == interaction.user and m.channel == interaction.channel
|
|
|
|
try:
|
|
msg = await interaction.bot.wait_for('message', check=check, timeout=30)
|
|
channel = None
|
|
if msg.channel_mentions:
|
|
channel = msg.channel_mentions[0]
|
|
elif msg.content.isdigit():
|
|
channel = interaction.guild.get_channel(int(msg.content))
|
|
|
|
if not channel:
|
|
return await interaction.followup.send("❌ Salon invalide.", ephemeral=True)
|
|
|
|
settings = load_settings(self.guild_id)
|
|
settings["absence_channel_id"] = channel.id
|
|
save_settings(self.guild_id, settings)
|
|
await interaction.followup.send(f"✅ Salon d'absence configuré sur {channel.mention}", ephemeral=True)
|
|
try: await msg.delete()
|
|
except: pass
|
|
except asyncio.TimeoutError:
|
|
await interaction.followup.send("❌ Temps écoulé.", ephemeral=True)
|
|
|
|
@disnake.ui.button(label="Définir le Rôle", style=disnake.ButtonStyle.primary, emoji="🎭")
|
|
async def set_role(self, button: disnake.ui.Button, interaction: disnake.Interaction):
|
|
await interaction.response.send_message("Mentionne le rôle ici (ou envoie son ID) :", ephemeral=True)
|
|
|
|
def check(m):
|
|
return m.author == interaction.user and m.channel == interaction.channel
|
|
|
|
try:
|
|
msg = await interaction.bot.wait_for('message', check=check, timeout=30)
|
|
role = None
|
|
if msg.role_mentions:
|
|
role = msg.role_mentions[0]
|
|
elif msg.content.isdigit():
|
|
role = interaction.guild.get_role(int(msg.content))
|
|
|
|
if not role:
|
|
return await interaction.followup.send("❌ Rôle invalide.", ephemeral=True)
|
|
|
|
settings = load_settings(self.guild_id)
|
|
settings["absence_role_id"] = role.id
|
|
save_settings(self.guild_id, settings)
|
|
await interaction.followup.send(f"✅ Rôle d'absence configuré sur **{role.name}**", ephemeral=True)
|
|
try: await msg.delete()
|
|
except: pass
|
|
except asyncio.TimeoutError:
|
|
await interaction.followup.send("❌ Temps écoulé.", ephemeral=True)
|
|
|
|
class Absence(commands.Cog):
|
|
def __init__(self, bot):
|
|
self.bot = bot
|
|
self.check_absences.start()
|
|
|
|
@commands.slash_command(name="absence", description="Déclarer une absence staff")
|
|
async def absence(self, interaction: disnake.ApplicationCommandInteraction,
|
|
superieur: Optional[disnake.Member] = commands.Param(None, description="Le supérieur hiérarchique à prévenir")):
|
|
"""Déclare une absence via un formulaire"""
|
|
await interaction.response.send_modal(AbsenceModal(superieur))
|
|
|
|
@commands.slash_command(name="absence_config", description="Configurer le système d'absence (Admin)")
|
|
@commands.has_permissions(administrator=True)
|
|
async def absence_config(self, interaction: disnake.ApplicationCommandInteraction):
|
|
"""Ouvre le panneau de configuration des absences"""
|
|
settings = load_settings(interaction.guild.id)
|
|
|
|
embed = disnake.Embed(
|
|
title="⚙️ Configuration des Absences Staff",
|
|
description="Utilisez les boutons ci-dessous pour configurer le système.",
|
|
color=disnake.Color.blue()
|
|
)
|
|
|
|
channel_id = settings.get("absence_channel_id")
|
|
role_id = settings.get("absence_role_id")
|
|
|
|
embed.add_field(
|
|
name="📁 Salon d'affichage",
|
|
value=f"<#{channel_id}>" if channel_id else "❌ Non configuré",
|
|
inline=True
|
|
)
|
|
embed.add_field(
|
|
name="🎭 Rôle d'absence",
|
|
value=f"<@&{role_id}>" if role_id else "❌ Non configuré",
|
|
inline=True
|
|
)
|
|
|
|
await interaction.response.send_message(embed=embed, view=AbsenceConfigView(interaction.guild.id), ephemeral=True)
|
|
|
|
async def strike_absence_message(self, guild: disnake.Guild, channel_id: int, message_id: int):
|
|
"""Modifie le message d'absence pour le barrer au lieu de le supprimer"""
|
|
channel = guild.get_channel(channel_id)
|
|
if not channel:
|
|
try:
|
|
channel = await guild.fetch_channel(channel_id)
|
|
except:
|
|
return
|
|
|
|
try:
|
|
msg = await channel.fetch_message(message_id)
|
|
content = msg.content
|
|
|
|
if "~~" in content: # Déjà barré
|
|
return
|
|
|
|
lines = content.split("\n")
|
|
new_lines = []
|
|
for line in lines:
|
|
if line.startswith("# "):
|
|
# Conserver le style header mais barrer le texte
|
|
new_lines.append(f"# ~~{line[2:].strip()}~~ [TERMINÉ]")
|
|
elif line.strip():
|
|
new_lines.append(f"~~{line}~~")
|
|
else:
|
|
new_lines.append(line)
|
|
|
|
await msg.edit(content="\n".join(new_lines))
|
|
except Exception as e:
|
|
print(f"Erreur lors du barrage du message d'absence: {e}")
|
|
|
|
@commands.slash_command(name="fin_absence", description="Terminer son absence staff")
|
|
async def fin_absence(self, interaction: disnake.ApplicationCommandInteraction):
|
|
"""Met fin à l'absence en cours"""
|
|
all_abs = await load_absences()
|
|
guild_abs = get_guild_absences(all_abs, interaction.guild.id)
|
|
user_id = str(interaction.user.id)
|
|
|
|
if user_id not in guild_abs:
|
|
return await interaction.response.send_message("❌ Tu n'as pas d'absence enregistrée sur ce serveur.", ephemeral=True)
|
|
|
|
info = guild_abs[user_id]
|
|
message_id = info.get("message_id")
|
|
|
|
settings = load_settings(interaction.guild.id)
|
|
channel_id = settings.get("absence_channel_id")
|
|
|
|
if channel_id and message_id:
|
|
await self.strike_absence_message(interaction.guild, int(channel_id), int(message_id))
|
|
|
|
# Retrait du rôle
|
|
role_id = settings.get("absence_role_id")
|
|
if role_id:
|
|
role = interaction.guild.get_role(int(role_id))
|
|
if role:
|
|
try: await interaction.user.remove_roles(role, reason="Fin d'absence staff")
|
|
except: pass
|
|
|
|
del guild_abs[user_id]
|
|
all_abs = update_guild_absences(all_abs, interaction.guild.id, guild_abs)
|
|
await save_absences(all_abs)
|
|
await interaction.response.send_message("✅ Ton absence a été retirée.", ephemeral=True)
|
|
|
|
@commands.slash_command(name="liste_absences", description="Afficher la liste des absences staff en cours")
|
|
async def liste_absences(self, interaction: disnake.ApplicationCommandInteraction):
|
|
"""Affiche les absences en cours sur le serveur"""
|
|
all_abs = await load_absences()
|
|
guild_abs = get_guild_absences(all_abs, interaction.guild.id)
|
|
|
|
absences = [
|
|
(user_id, info) for user_id, info in guild_abs.items()
|
|
if info.get("etat") == "en cours"
|
|
]
|
|
|
|
if not absences:
|
|
return await interaction.response.send_message("Aucune absence staff en cours sur ce serveur.", ephemeral=True)
|
|
|
|
per_page = 5
|
|
pages = [absences[i:i + per_page] for i in range(0, len(absences), per_page)]
|
|
|
|
def create_embed(page_index: int) -> disnake.Embed:
|
|
embed = disnake.Embed(
|
|
title=f"📋 Absences staff - {interaction.guild.name} ({page_index + 1}/{len(pages)})",
|
|
color=disnake.Color.orange()
|
|
)
|
|
for user_id, info in pages[page_index]:
|
|
embed.add_field(
|
|
name=info["pseudo"],
|
|
value=(
|
|
f"**Période** : du {info['debut']} au {info['fin']}\n"
|
|
f"**Motif** : {info['motif']}\n"
|
|
f"**Supérieur** : {info['superieur']}"
|
|
),
|
|
inline=False
|
|
)
|
|
embed.set_footer(text="Gestion des absences du staff")
|
|
return embed
|
|
|
|
class PaginatorView(disnake.ui.View):
|
|
def __init__(self):
|
|
super().__init__(timeout=60)
|
|
self.page = 0
|
|
|
|
@disnake.ui.button(label="⬅️ Précédent", style=disnake.ButtonStyle.secondary, disabled=True)
|
|
async def previous(self, interaction_btn: disnake.Interaction, button: disnake.ui.Button):
|
|
self.page -= 1
|
|
self.update_buttons()
|
|
await interaction_btn.response.edit_message(embed=create_embed(self.page), view=self)
|
|
|
|
@disnake.ui.button(label="➡️ Suivant", style=disnake.ButtonStyle.secondary, disabled=len(pages) == 1)
|
|
async def next(self, interaction_btn: disnake.Interaction, button: disnake.ui.Button):
|
|
self.page += 1
|
|
self.update_buttons()
|
|
await interaction_btn.response.edit_message(embed=create_embed(self.page), view=self)
|
|
|
|
def update_buttons(self):
|
|
self.children[0].disabled = self.page == 0
|
|
self.children[1].disabled = self.page >= len(pages) - 1
|
|
|
|
await interaction.response.send_message(embed=create_embed(0), view=PaginatorView(), ephemeral=True)
|
|
|
|
@tasks.loop(minutes=1)
|
|
async def check_absences(self):
|
|
all_abs = await load_absences()
|
|
now = get_paris_now()
|
|
changed = False
|
|
|
|
for guild_id_str, guild_data in list(all_abs.items()):
|
|
guild_id = int(guild_id_str)
|
|
guild = self.bot.get_guild(guild_id)
|
|
if not guild:
|
|
try:
|
|
guild = await self.bot.fetch_guild(guild_id)
|
|
except:
|
|
continue
|
|
|
|
settings = load_settings(guild_id)
|
|
role_id = settings.get("absence_role_id")
|
|
role = guild.get_role(int(role_id)) if role_id else None
|
|
|
|
to_remove = []
|
|
for user_id_str, info in guild_data.items():
|
|
try:
|
|
debut = parse_date(info["debut"])
|
|
fin = parse_date(info["fin"])
|
|
except Exception as e:
|
|
print(f"Erreur parsing date pour {user_id_str}: {e}")
|
|
continue
|
|
|
|
user_id = int(user_id_str)
|
|
member = guild.get_member(user_id)
|
|
if not member:
|
|
try:
|
|
member = await guild.fetch_member(user_id)
|
|
except:
|
|
pass
|
|
|
|
etat = info.get("etat", "en cours")
|
|
|
|
# Activation si l'heure est arrivée
|
|
if etat == "à venir" and debut <= now:
|
|
print(f"Activation absence pour {info['pseudo']} ({user_id_str})")
|
|
info["etat"] = "en cours"
|
|
changed = True
|
|
if member and role:
|
|
try:
|
|
await member.add_roles(role, reason="Début d'absence staff")
|
|
except Exception as e:
|
|
print(f"Erreur ajout rôle absence: {e}")
|
|
|
|
# Désactivation si l'heure est passée
|
|
if fin < now:
|
|
print(f"Désactivation absence pour {info['pseudo']} ({user_id_str})")
|
|
if member and role:
|
|
try:
|
|
await member.remove_roles(role, reason="Fin d'absence staff")
|
|
except Exception as e:
|
|
print(f"Erreur retrait rôle absence: {e}")
|
|
|
|
# On barre aussi le message en automatique
|
|
message_id = info.get("message_id")
|
|
channel_id = settings.get("absence_channel_id")
|
|
if channel_id and message_id:
|
|
await self.strike_absence_message(guild, int(channel_id), int(message_id))
|
|
|
|
to_remove.append(user_id_str)
|
|
changed = True
|
|
|
|
for uid in to_remove:
|
|
guild_data.pop(uid, None)
|
|
|
|
if changed:
|
|
await save_absences(all_abs)
|
|
|
|
@commands.Cog.listener()
|
|
async def on_message(self, message):
|
|
if message.author.bot or not message.guild:
|
|
return
|
|
|
|
all_abs = await load_absences()
|
|
guild_abs = get_guild_absences(all_abs, message.guild.id)
|
|
|
|
# 1. Rappel à l'auteur s'il est absent
|
|
if str(message.author.id) in guild_abs:
|
|
if guild_abs[str(message.author.id)].get("etat") == "en cours":
|
|
await message.channel.send(
|
|
f"{message.author.mention}, tu es actuellement déclaré absent sur ce serveur. Merci de prévenir si tu es de retour !",
|
|
delete_after=10
|
|
)
|
|
|
|
# 2. Notification des mentions de staff absents
|
|
for user_id_str, info in guild_abs.items():
|
|
if info.get("etat") != "en cours":
|
|
continue
|
|
|
|
if f"<@{user_id_str}>" in message.content or f"<@!{user_id_str}>" in message.content:
|
|
msg_id = info.get("message_id")
|
|
settings = load_settings(message.guild.id)
|
|
channel_id = settings.get("absence_channel_id")
|
|
|
|
absence_url = f"https://disnake.com/channels/{message.guild.id}/{channel_id}/{msg_id}" if channel_id and msg_id else ""
|
|
|
|
reply_text = f"⚠️ {message.author.mention}, cet utilisateur est actuellement absent."
|
|
if absence_url:
|
|
reply_text += f"\nPlus d'infos ici : [Voir la déclaration]({absence_url})"
|
|
|
|
await message.reply(reply_text, delete_after=30)
|
|
|
|
def setup(bot):
|
|
bot.add_cog(Absence(bot))
|