Ajout système de tâches et Fix: suggerer et signaler_bug

This commit is contained in:
Mathis 2026-06-07 16:22:17 +02:00
parent a49aa6717f
commit f1c0921a58
7 changed files with 1434 additions and 5 deletions

259
TÂCHE_VICY.py Executable file
View file

@ -0,0 +1,259 @@
import discord
from discord import app_commands
from discord.ext import commands, tasks
from datetime import datetime
from zoneinfo import ZoneInfo
from utils.gestion_taches import (
ajouter_tache, lister_taches_utilisateur, taches_proches,
update_tache_status, charger_taches, supprimer_tache, sauvegarder_taches
)
PRIORITES = ["Haute", "Moyenne", "Basse", "Critique", "Faible"]
class RappelTacheView(discord.ui.View):
def __init__(self, tid: str):
super().__init__(timeout=None)
self.tid = tid
@discord.ui.button(label="✅ Marquer comme faite", style=discord.ButtonStyle.success, custom_id="btn_mark_done")
async def mark_done(self, interaction: discord.Interaction, button: discord.ui.Button):
success = await update_tache_status(self.tid, "Terminée")
if success:
await interaction.response.send_message(
f"✅ La tâche #{self.tid} a été marquée comme terminée !", ephemeral=True)
else:
await interaction.response.send_message(
"❌ Impossible de mettre à jour la tâche. Peut-être déjà terminée ?", ephemeral=True)
self.disable_all_items()
await interaction.message.edit(view=self)
class MesTachesPaginator(discord.ui.View):
def __init__(self, interaction: discord.Interaction, taches_list: list[tuple], per_page=5):
super().__init__(timeout=120)
self.interaction = interaction
self.taches = taches_list
self.per_page = per_page
self.page = 0
self.max_page = (len(taches_list) - 1) // per_page
def get_embed(self):
embed = discord.Embed(title="📝 Vos Tâches Assignées", color=discord.Color.green())
start = self.page * self.per_page
end = start + self.per_page
for tid, t in self.taches[start:end]:
embed.add_field(
name=f"#{tid} - {t['titre']}",
value=f"📅 Échéance : {t['échéance']}\n⚠️ Priorité : {t['priorité']}\n📌 Statut : {t['statut']}",
inline=False
)
embed.set_footer(text=f"Page {self.page + 1} / {self.max_page + 1}")
return embed
@discord.ui.button(label="⬅️", style=discord.ButtonStyle.secondary)
async def previous_page(self, interaction: discord.Interaction, button: discord.ui.Button):
if interaction.user.id != self.interaction.user.id:
await interaction.response.send_message("❌ Ce n'est pas votre pagination.", ephemeral=True)
return
if self.page > 0:
self.page -= 1
await interaction.response.edit_message(embed=self.get_embed(), view=self)
else:
await interaction.response.defer()
@discord.ui.button(label="➡️", style=discord.ButtonStyle.secondary)
async def next_page(self, interaction: discord.Interaction, button: discord.ui.Button):
if interaction.user.id != self.interaction.user.id:
await interaction.response.send_message("❌ Ce n'est pas votre pagination.", ephemeral=True)
return
if self.page < self.max_page:
self.page += 1
await interaction.response.edit_message(embed=self.get_embed(), view=self)
else:
await interaction.response.defer()
class JiraDiscord(commands.Cog):
def __init__(self, bot):
self.bot = bot
self.rappels.start()
def cog_unload(self):
self.rappels.cancel()
async def autocomplete_priorite(self, interaction: discord.Interaction, current: str):
return [
app_commands.Choice(name=prio, value=prio)
for prio in PRIORITES if current.lower() in prio.lower()
][:5]
@app_commands.command(name="nouvelle_tache", description="Créer une nouvelle tâche")
@app_commands.describe(
titre="Titre de la tâche",
description="Description de la tâche",
assigne="Utilisateur assigné",
priorite="Niveau de priorité (ex: Haute, Moyenne, Basse)",
echeance="Date d'échéance au format JJ/MM/AAAA"
)
async def nouvelle_tache(
self, interaction: discord.Interaction, titre: str, description: str,
assigne: discord.User, priorite: str, echeance: str
):
data = await charger_taches()
for tid, tache in data.get("taches", {}).items():
if tache["assigné"] == assigne.id and tache["titre"].lower() == titre.lower():
return await interaction.response.send_message(
f"{assigne.mention} a déjà une tâche avec ce titre.", ephemeral=True)
try:
date_obj = datetime.strptime(echeance, "%d/%m/%Y")
date_str = date_obj.strftime("%Y-%m-%d")
except ValueError:
return await interaction.response.send_message(
"❌ Format de date invalide (JJ/MM/AAAA).", ephemeral=True)
id_tache = await ajouter_tache(titre, description, assigne.id, priorite, date_str, interaction.user.id)
embed = discord.Embed(
title=f"📌 Nouvelle Tâche #{id_tache}",
description=description,
color=discord.Color.blurple()
)
embed.set_author(name=f"Assignée à {assigne.display_name}", icon_url=assigne.display_avatar.url)
embed.add_field(name="🗂 Titre", value=titre, inline=False)
embed.add_field(name="⚠️ Priorité", value=priorite, inline=True)
embed.add_field(name="📅 Échéance", value=echeance, inline=True)
embed.set_footer(text=f"Créée par {interaction.user.display_name}", icon_url=interaction.user.display_avatar.url)
await interaction.response.send_message(embed=embed)
try:
await assigne.send(
f"👋 Bonjour {assigne.display_name},\n\nUne nouvelle tâche t'a été assignée par {interaction.user.mention} :",
embed=embed
)
except Exception:
pass
@nouvelle_tache.autocomplete("priorite")
async def autocomplete_priorite_handler(self, interaction: discord.Interaction, current: str):
return await self.autocomplete_priorite(interaction, current)
@app_commands.command(name="mes_taches", description="Lister mes tâches assignées")
async def mes_taches(self, interaction: discord.Interaction):
taches = await lister_taches_utilisateur(interaction.user.id)
if not taches:
await interaction.response.send_message("🚫 Aucune tâche ne vous est assignée.", ephemeral=True)
return
taches_list = list(taches.items())
if len(taches_list) <= 5:
embed = discord.Embed(title="📝 Vos Tâches Assignées", color=discord.Color.green())
for tid, t in taches_list:
embed.add_field(
name=f"#{tid} - {t['titre']}",
value=f"📅 Échéance : {t['échéance']}\n⚠️ Priorité : {t['priorité']}\n📌 Statut : {t['statut']}",
inline=False
)
await interaction.response.send_message(embed=embed, ephemeral=True)
else:
view = MesTachesPaginator(interaction, taches_list)
await interaction.response.send_message(embed=view.get_embed(), view=view, ephemeral=True)
@app_commands.command(name="update_tache", description="Mettre à jour le statut et la priorité d'une tâche")
@app_commands.describe(
id_tache="ID de la tâche à mettre à jour",
statut="Nouveau statut de la tâche",
priorite="Nouvelle priorité (optionnel)"
)
async def update_tache(
self, interaction: discord.Interaction, id_tache: str, statut: str, priorite: str = None
):
data = await charger_taches()
if id_tache not in data.get("taches", {}):
return await interaction.response.send_message("❌ Tâche non trouvée.", ephemeral=True)
tache = data["taches"][id_tache]
if tache["assigné"] != interaction.user.id:
return await interaction.response.send_message("❌ Vous ne pouvez pas mettre à jour le statut de cette tâche.", ephemeral=True)
statut_update = await update_tache_status(id_tache, statut)
priorite_update = True
if priorite:
tache["priorité"] = priorite
await sauvegarder_taches(data)
if statut_update and priorite_update:
await interaction.response.send_message(
f"✅ Le statut de la tâche #{id_tache} a été mis à jour à **{statut}**." +
(f"\nLa priorité est maintenant **{priorite}**." if priorite else ""),
ephemeral=True
)
if tache["créé_par"] != interaction.user.id:
try:
creator = await self.bot.fetch_user(tache["créé_par"])
await creator.send(
f"🔔 La tâche #{id_tache} que vous avez créée a été mise à jour par {interaction.user.mention} :\n"
f"**Nouveau statut** : {statut}" +
(f"\n**Nouvelle priorité** : {priorite}" if priorite else "")
)
except Exception:
pass
else:
await interaction.response.send_message("❌ Erreur lors de la mise à jour de la tâche.", ephemeral=True)
@update_tache.autocomplete("priorite")
async def autocomplete_priorite_update(self, interaction: discord.Interaction, current: str):
return await self.autocomplete_priorite(interaction, current)
@app_commands.command(name="supprimer_tache", description="Supprimer une tâche")
@app_commands.describe(
id_tache="ID de la tâche à supprimer"
)
async def supprimer_tache_cmd(self, interaction: discord.Interaction, id_tache: str):
data = await charger_taches()
if id_tache not in data.get("taches", {}):
return await interaction.response.send_message("❌ Tâche non trouvée.", ephemeral=True)
tache = data["taches"][id_tache]
perms = interaction.user.guild_permissions
can_delete = (
tache["créé_par"] == interaction.user.id or
tache["assigné"] == interaction.user.id or
perms.administrator
)
if not can_delete:
return await interaction.response.send_message("❌ Vous ne pouvez pas supprimer cette tâche.", ephemeral=True)
success = await supprimer_tache(id_tache)
if success:
await interaction.response.send_message(f"🗑️ La tâche #{id_tache} a été supprimée.", ephemeral=True)
else:
await interaction.response.send_message("❌ Erreur lors de la suppression de la tâche.", ephemeral=True)
@tasks.loop(seconds=30)
async def rappels(self):
now = datetime.now(ZoneInfo("Europe/Paris"))
if now.hour == 9 and now.minute == 0:
taches = await taches_proches()
for tid, t in taches.items():
try:
user = await self.bot.fetch_user(t["assigné"])
embed = discord.Embed(
title=f"🔔 Rappel Tâche #{tid}",
description=t["description"],
color=discord.Color.orange()
)
embed.add_field(name="🗂 Titre", value=t["titre"], inline=False)
embed.add_field(name="⚠️ Priorité", value=t["priorité"], inline=True)
embed.add_field(name="📅 Échéance", value=t["échéance"], inline=True)
embed.set_footer(text="Clique sur le bouton si tu as fini !")
view = RappelTacheView(tid)
await user.send(embed=embed, view=view)
except Exception:
continue
@rappels.before_loop
async def avant_rappel(self):
await self.bot.wait_until_ready()
async def setup(bot):
await bot.add_cog(JiraDiscord(bot))

5
bot.py
View file

@ -129,6 +129,7 @@ class MyBot(commands.Bot):
"commandes.absence_staff", "commandes.absence_staff",
"commandes.moderation", "commandes.moderation",
"commandes.invites", "commandes.invites",
"commandes.taches",
] ]
for extension in extensions: for extension in extensions:
@ -138,6 +139,10 @@ class MyBot(commands.Bot):
except Exception as e: except Exception as e:
kuby_logger.error(f"❌ Erreur lors du chargement de {extension} : {e}", exc_info=True) kuby_logger.error(f"❌ Erreur lors du chargement de {extension} : {e}", exc_info=True)
# Synchroniser les commandes slash
await self._sync_application_commands()
kuby_logger.info("✅ Commandes slash synchronisées avec Discord")
# --- LOGGING DES INTERACTIONS --- # --- LOGGING DES INTERACTIONS ---
@self.listen("on_interaction") @self.listen("on_interaction")
async def log_interaction(inter: disnake.Interaction): async def log_interaction(inter: disnake.Interaction):

View file

@ -6,6 +6,7 @@ import logging
import json import json
import os import os
import asyncio import asyncio
from typing import Dict, Any, Optional
kuby_logger = logging.getLogger("KubyBot") kuby_logger = logging.getLogger("KubyBot")
BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
@ -222,6 +223,13 @@ class BugReport(commands.Cog):
os.makedirs(REPORTS_DIR, exist_ok=True) os.makedirs(REPORTS_DIR, exist_ok=True)
await self._start_webhook_server() await self._start_webhook_server()
# ⚡ Enregistrement des vues persistantes (boutons panel)
# Sans ça, après un redémarrage les boutons "Remplir le formulaire"
# dans les anciens messages de panel ne fonctionnent plus.
self.bot.add_view(BugReportView(self))
self.bot.add_view(FeatureSuggestionView(self))
kuby_logger.info("✅ Vues persistantes BugReport enregistrées (open_bug_form, open_suggestion_form)")
async def _start_webhook_server(self): async def _start_webhook_server(self):
try: try:
app = web.Application() app = web.Application()

View file

@ -251,6 +251,8 @@ class SecurityView(disnake.ui.View):
self.build_rules_view() self.build_rules_view()
elif self.category == "membres": elif self.category == "membres":
self.build_members_view() self.build_members_view()
elif self.category == "taches":
self.build_taches_view()
def build_spam_view(self): def build_spam_view(self):
"""Construit la vue Anti-Spam""" """Construit la vue Anti-Spam"""
@ -697,6 +699,27 @@ class SecurityView(disnake.ui.View):
btn_back.callback = self.back_to_main btn_back.callback = self.back_to_main
self.add_item(btn_back) self.add_item(btn_back)
def build_taches_view(self):
"""Construit la vue Gestion des Tâches"""
# Toggle Activer/Désactiver
taches_enabled = self.settings.get("taches_enabled", True)
btn_taches = ToggleButton(
label="Module Tâches",
emoji="📌",
is_enabled=taches_enabled,
custom_id="taches_enabled"
)
self.add_item(btn_taches)
# Bouton retour
btn_back = disnake.ui.Button(
label="Retour",
style=disnake.ButtonStyle.gray,
emoji="⬅️"
)
btn_back.callback = self.back_to_main
self.add_item(btn_back)
async def back_to_main(self, interaction: disnake.Interaction): async def back_to_main(self, interaction: disnake.Interaction):
"""Retour au menu principal""" """Retour au menu principal"""
self.category = None self.category = None
@ -844,6 +867,12 @@ class MainCategorySelect(disnake.ui.Select):
emoji="👤", emoji="👤",
value="membres", value="membres",
description="Paramètres liés aux membres" description="Paramètres liés aux membres"
),
disnake.SelectOption(
label="Tâches",
emoji="📌",
value="taches",
description="Gestion des tâches et rappels"
) )
] ]
super().__init__( super().__init__(
@ -902,6 +931,13 @@ class MainCategorySelect(disnake.ui.Select):
"Paramètres Membres", "Paramètres Membres",
"Paramètres liés à la gestion des membres.\n\n" "Paramètres liés à la gestion des membres.\n\n"
"• **Restauration des rôles** : Restaurer automatiquement les rôles des membres de retour" "• **Restauration des rôles** : Restaurer automatiquement les rôles des membres de retour"
),
"taches": view.create_category_embed(
"Gestion des Tâches",
"Configurez le système de gestion des tâches.\n\n"
"• **Activer/Désactiver** : Active ou désactive le module sur ce serveur\n"
"• **Création** : Seuls les administrateurs peuvent créer des tâches\n"
"• **Rappels** : Notifications automatiques 1h avant l'échéance"
) )
} }

717
commandes/taches.py Normal file
View file

@ -0,0 +1,717 @@
"""
Cog Tâches - Système de gestion des tâches modulaire pour Kuby
Utilise disnake >= 2.11 avec Components V2
"""
import disnake
import re
from disnake.ext import commands, tasks
from disnake import ui
from datetime import datetime
from zoneinfo import ZoneInfo
from typing import Optional, Dict, List, Any
from utils.gestion_taches import (
ajouter_tache, lister_taches_utilisateur, lister_toutes_taches, get_tache,
update_tache_status, update_tache_priorite, supprimer_tache,
taches_proches, get_taches_settings, set_taches_settings,
PRIORITES, STATUTS
)
from src.logger import kuby_logger
# ============================================================================
# MODALS
# ============================================================================
class NouvelleTacheModal(ui.Modal):
"""Modal pour créer une nouvelle tâche."""
def __init__(self, guild_id: int):
self.guild_id = guild_id
# Create components list - Modals can only contain TextInput components
components = [
ui.TextInput(
label="Titre",
custom_id="tache_titre",
placeholder="Ex: Corriger le bug du login",
max_length=100,
required=True
),
ui.TextInput(
label="Description",
custom_id="tache_description",
style=disnake.TextInputStyle.paragraph,
placeholder="Décrivez la tâche en détail...",
max_length=1000,
required=True
),
ui.TextInput(
label="ID de l'utilisateur assigné (ou mention)",
custom_id="tache_assigne",
placeholder="Ex: 123456789 ou @utilisateur",
max_length=100,
required=True
),
ui.TextInput(
label="Date d'échéance (JJ/MM/AAAA)",
custom_id="tache_echeance",
placeholder="31/12/2026",
required=True
),
ui.TextInput(
label="Priorité",
custom_id="tache_priorite",
placeholder=f"Une de: {', '.join(PRIORITES)}",
max_length=20,
required=False
)
]
super().__init__(title="📌 Nouvelle Tâche", custom_id="nouvelle_tache_modal", components=components)
async def callback(self, inter: disnake.ModalInteraction) -> None:
"""Callback quand le modal est soumis."""
try:
titre = inter.text_values["tache_titre"]
description = inter.text_values["tache_description"]
assigne_input = inter.text_values["tache_assigne"]
echeance_str = inter.text_values["tache_echeance"]
priorite = inter.text_values.get("tache_priorite", "Moyenne")
# Parse assigne_id from input (could be ID or mention)
mention_match = re.match(r'<@(\d+)>', assigne_input)
if mention_match:
assigne_id = int(mention_match.group(1))
else:
try:
assigne_id = int(assigne_input)
except ValueError:
await inter.response.send_message(
"❌ ID utilisateur invalide. Veuillez entrer un ID numérique ou une mention valide.",
ephemeral=True
)
return
# Validate priority
if priorite and priorite not in PRIORITES:
priorite = "Moyenne"
try:
echeance_date = datetime.strptime(echeance_str, "%d/%m/%Y")
echeance = echeance_date.strftime("%Y-%m-%d")
except ValueError:
return await inter.response.send_message(
"❌ Format de date invalide. Utilisez **JJ/MM/AAAA**.",
ephemeral=True
)
tache_id = await ajouter_tache(
titre=titre,
description=description,
assigne_id=assigne_id,
priorite=priorite,
echeance=echeance,
cree_par_id=inter.author.id,
serveur_id=self.guild_id
)
await inter.response.send_message(
f"✅ **Tâche #{tache_id} créée avec succès !**\n"
f"→ Assignée à : <@{assigne_id}>\n"
f"→ Échéance : {echeance_str}",
ephemeral=True
)
try:
assigne = inter.bot.get_user(assigne_id) or await inter.bot.fetch_user(assigne_id)
if assigne:
await assigne.send(
f"📌 **Nouvelle tâche assignée** par {inter.author.mention} :\n\n"
f"**#{tache_id} - {titre}**\n"
f"📅 Échéance : {echeance_str}\n"
f"⚠️ Priorité : {priorite}"
)
except Exception:
pass
except Exception as e:
kuby_logger.error(f"Erreur création tâche: {e}")
await inter.response.send_message(
"❌ Une erreur est survenue. Veuillez réessayer.",
ephemeral=True
)
class ConfirmDeleteModal(ui.Modal):
"""Modal de confirmation pour la suppression."""
def __init__(self, tache_id: str, tache_titre: str):
self.tache_id = tache_id
self.tache_titre = tache_titre
components = [
ui.TextInput(
label=f'Pour confirmer, tapez : "{tache_titre}"',
custom_id="confirm_text",
placeholder=f"Tapez exactement : {tache_titre}",
required=True
)
]
super().__init__(title=f"⚠️ Supprimer la tâche #{tache_id}", custom_id=f"confirm_delete_{tache_id}", components=components)
async def callback(self, inter: disnake.ModalInteraction) -> None:
confirmation = inter.text_values["confirm_text"]
if confirmation != self.tache_titre:
return await inter.response.send_message(
"❌ Confirmation incorrecte. La tâche n'a pas été supprimée.",
ephemeral=True
)
success = await supprimer_tache(self.tache_id)
if success:
await inter.response.send_message(
f"🗑️ **Tâche #{self.tache_id} supprimée avec succès.**",
ephemeral=True
)
else:
await inter.response.send_message(
"❌ Impossible de supprimer la tâche.",
ephemeral=True
)
# ============================================================================
# COMPONENTS V2 HELPER FUNCTIONS
# ============================================================================
def _get_statut_emoji(statut: str) -> str:
return {"À faire": "📌", "En cours": "", "Terminée": "", "Annulée": ""}.get(statut, "📌")
def _get_priorite_emoji(priorite: str) -> str:
return {"Critique": "🔴", "Haute": "🟠", "Moyenne": "🟡", "Basse": "🟢", "Faible": ""}.get(priorite, "🟡")
def get_main_menu_components(guild_id: int) -> list:
"""Menu principal des tâches"""
return [
ui.Container(
ui.TextDisplay("📝 **Gestion des Tâches**\nUtilisez les boutons ci-dessous pour interagir."),
ui.Section(" Nouvelle Tâche", accessory=ui.Button(label="Créer", style=disnake.ButtonStyle.green, custom_id=f"taches_nouvelle:{guild_id}")),
ui.Section("📋 Mes Tâches", accessory=ui.Button(label="Mes tâches", style=disnake.ButtonStyle.blurple, custom_id=f"taches_mes:{guild_id}")),
ui.Section("🌐 Toutes les Tâches", accessory=ui.Button(label="Toutes", style=disnake.ButtonStyle.blurple, custom_id=f"taches_toutes:{guild_id}")),
ui.Section("🔄 Rafraîchir", accessory=ui.Button(label="Rafraîchir", style=disnake.ButtonStyle.gray, custom_id=f"taches_rafraichir:{guild_id}")),
)
]
def get_no_permission_components(guild_id: int) -> list:
"""Message de permission refusée"""
return [
ui.Container(
ui.TextDisplay("❌ **Permission refusée**\nSeuls les administrateurs peuvent utiliser cette fonctionnalité."),
ui.Section("⬅️ Retour", accessory=ui.Button(label="Retour", style=disnake.ButtonStyle.gray, custom_id=f"taches_back:{guild_id}")),
)
]
def get_module_disabled_components(guild_id: int) -> list:
"""Message de module désactivé"""
return [
ui.Container(
ui.TextDisplay("❌ **Module désactivé**\nLe système de tâches est désactivé sur ce serveur."),
ui.Section("⬅️ Retour", accessory=ui.Button(label="Retour", style=disnake.ButtonStyle.gray, custom_id=f"taches_back:{guild_id}")),
)
]
def get_no_taches_components(user_id: int, guild_id: int, is_all: bool = False) -> list:
"""Message lorsqu'il n'y a pas de tâches"""
message = "📭 **Aucune tâche assignée**\nVous n'avez aucune tâche en cours sur ce serveur."
if is_all:
message = "📭 **Aucune tâche**\nAucune tâche trouvée sur ce serveur."
return [
ui.Container(
ui.TextDisplay(message),
ui.Section("⬅️ Retour au menu", accessory=ui.Button(label="Retour", style=disnake.ButtonStyle.gray, custom_id=f"taches_back_to_menu:{user_id}:{guild_id}")),
)
]
def get_taches_pagination_components(user_id: int, guild_id: int, taches_list: List[tuple], page: int, is_all: bool) -> list:
"""Pagination des tâches - divisé en plusieurs Containers pour respecter la limite de 5 éléments"""
start = page * 4
end = start + 4
page_taches = taches_list[start:end]
total_pages = max(1, (len(taches_list) - 1) // 4 + 1)
containers = []
# Premier Container: Titre + tâches (max 4)
container1_sections = [
ui.TextDisplay(f"📝 **{'Toutes les tâches du serveur' if is_all else 'Vos tâches'}** :")
]
if page_taches:
for tid, tache in page_taches:
container1_sections.append(ui.Section(
f"#{tid} - {tache['titre'][:30]}",
accessory=ui.Button(
label="Voir",
style=disnake.ButtonStyle.blurple,
custom_id=f"tache_detail:{tid}:{user_id}:{guild_id}:{page}:{1 if is_all else 0}"
)
))
else:
container1_sections.append(ui.TextDisplay("Aucune tâche à afficher."))
containers.append(ui.Container(*container1_sections))
# Deuxième Container: Navigation
prev_disabled = page == 0
next_disabled = end >= len(taches_list)
container2_sections = []
if prev_disabled and next_disabled:
container2_sections.append(ui.Section(
"⬅️ Retour au menu",
accessory=ui.Button(
label="Retour",
style=disnake.ButtonStyle.gray,
custom_id=f"taches_back_to_menu:{user_id}:{guild_id}"
)
))
else:
if not prev_disabled:
container2_sections.append(ui.Section(
"⬅️ Précédent",
accessory=ui.Button(
label="Précédent",
style=disnake.ButtonStyle.secondary,
custom_id=f"taches_prev:{user_id}:{guild_id}:{page}:{1 if is_all else 0}",
disabled=prev_disabled
)
))
container2_sections.append(ui.TextDisplay(f"Page {page + 1}/{total_pages}"))
if not next_disabled:
container2_sections.append(ui.Section(
"Suivant ➡️",
accessory=ui.Button(
label="Suivant",
style=disnake.ButtonStyle.secondary,
custom_id=f"taches_next:{user_id}:{guild_id}:{page}:{1 if is_all else 0}",
disabled=next_disabled
)
))
if len(container2_sections) >= 5:
containers.append(ui.Container(*container2_sections))
container2_sections = []
container2_sections.append(ui.Section(
"⬅️ Retour au menu",
accessory=ui.Button(
label="Retour",
style=disnake.ButtonStyle.gray,
custom_id=f"taches_back_to_menu:{user_id}:{guild_id}"
)
))
containers.append(ui.Container(*container2_sections))
return containers
def get_tache_detail_components(tache: Dict[str, Any], tache_id: str, user_id: int, guild_id: int, can_edit: bool) -> list:
"""Détails d'une tâche"""
statut_emoji = _get_statut_emoji(tache.get("statut", "À faire"))
priorite_emoji = _get_priorite_emoji(tache.get("priorite", "Moyenne"))
tache_text = (
f"{statut_emoji} **Tâche #{tache_id}**\n\n"
f"**Titre** : {tache.get('titre', 'N/A')}\n"
f"**Description** : {tache.get('description', 'N/A')}\n"
f"**Assigné à** : <@{tache.get('assigne', '0')}>\n"
f"**Créé par** : <@{tache.get('cree_par', '0')}>\n"
f"**Priorité** : {priorite_emoji} {tache.get('priorite', 'Moyenne')}\n"
f"**Statut** : {tache.get('statut', 'À faire')}\n"
f"**Échéance** : {tache.get('echeance', 'N/A')}\n"
f"**Créée le** : {tache.get('date_creation', 'N/A')}"
)
sections = [ui.TextDisplay(tache_text)]
if can_edit:
sections.append(ui.Section("✅ Marquer terminée", accessory=ui.Button(label="Terminée", style=disnake.ButtonStyle.green, custom_id=f"tache_done:{tache_id}:{guild_id}")))
sections.append(ui.Section("⚡ Changer priorité", accessory=ui.Button(label="Priorité", style=disnake.ButtonStyle.blurple, custom_id=f"tache_prio:{tache_id}:{guild_id}")))
sections.append(ui.Section("🗑️ Supprimer", accessory=ui.Button(label="Supprimer", style=disnake.ButtonStyle.red, custom_id=f"tache_delete:{tache_id}:{guild_id}")))
sections.append(ui.Section("⬅️ Retour", accessory=ui.Button(label="Retour", style=disnake.ButtonStyle.gray, custom_id=f"tache_back:{guild_id}")))
return [ui.Container(*sections)]
def get_tache_done_components(tache: Dict[str, Any], user_id: int, guild_id: int) -> list:
"""Confirmation que la tâche est terminée (avec titre barré)"""
priorite_emoji = _get_priorite_emoji(tache.get("priorite", "Moyenne"))
text = (
f"✅ **Tâche terminée !**\n\n"
f"📌 **Titre** : ~~{tache.get('titre', 'Inconnu')}~~\n"
f"📅 Échéance : {tache.get('echeance', 'N/A')}\n"
f"⚠️ Priorité : {tache.get('priorite', 'Moyenne')}\n"
f"📝 Statut : **Terminée**"
)
return [
ui.Container(
ui.TextDisplay(text),
# user_id est passé dans le custom_id pour que le handler puisse vérifier l'autorisation
ui.Section("⬅️ Retour", accessory=ui.Button(label="Retour", style=disnake.ButtonStyle.gray, custom_id=f"taches_back_to_menu:{user_id}:{guild_id}")),
)
]
def get_change_priority_components(tache_id: str, current_prio: str, user_id: int, guild_id: int) -> list:
"""Sélection de la priorité - divisé en plusieurs Containers pour respecter la limite de 5 éléments"""
containers = []
container1_sections = [
ui.TextDisplay("📝 Sélectionnez la nouvelle priorité :")
]
for prio in PRIORITES:
style = disnake.ButtonStyle.green if prio == current_prio else disnake.ButtonStyle.secondary
emoji = _get_priorite_emoji(prio)
container1_sections.append(ui.Section(
f"{emoji} {prio}",
accessory=ui.Button(label=prio, style=style, custom_id=f"prio_select:{tache_id}:{prio}:{user_id}:{guild_id}")
))
if len(container1_sections) >= 5:
containers.append(ui.Container(*container1_sections))
container1_sections = []
if container1_sections:
containers.append(ui.Container(*container1_sections))
containers.append(ui.Container(
ui.Section(
"❌ Annuler",
accessory=ui.Button(label="Annuler", style=disnake.ButtonStyle.red, custom_id=f"prio_cancel:{tache_id}:{user_id}:{guild_id}")
)
))
return containers
# ============================================================================
# COG PRINCIPAL
# ============================================================================
class TachesCog(commands.Cog):
"""Cog principal pour la gestion des tâches."""
def __init__(self, bot):
self.bot = bot
async def cog_load(self):
"""Démarre la tâche de rappel une fois le cog chargé."""
self.rappels_loop.start()
def cog_unload(self):
self.rappels_loop.cancel()
# ==========================================================================
# HANDLER MANUEL DES INTERACTIONS DE BOUTONS
# ==========================================================================
@commands.Cog.listener("on_message_interaction")
async def on_taches_interaction(self, interaction: disnake.MessageInteraction):
"""Gère toutes les interactions de boutons pour le système de tâches."""
custom_id = interaction.data.custom_id
if not custom_id or not (custom_id.startswith("taches_") or custom_id.startswith("tache_") or custom_id.startswith("prio_")):
return
kuby_logger.debug(f"Interaction tâches reçue : {custom_id} par {interaction.user}")
try:
parts = custom_id.split(":")
action = parts[0]
# Menu principal
if action == "taches_nouvelle":
guild_id = int(parts[1])
if not interaction.author.guild_permissions.administrator:
await interaction.response.edit_message(
components=get_no_permission_components(guild_id))
return
modal = NouvelleTacheModal(guild_id)
await interaction.response.send_modal(modal)
elif action == "taches_mes":
guild_id = int(parts[1])
user_id = interaction.author.id
settings = await get_taches_settings(guild_id)
if not settings.get("taches_enabled", True):
await interaction.response.edit_message(
components=get_module_disabled_components(guild_id))
return
taches = await lister_taches_utilisateur(user_id, guild_id)
if not taches:
await interaction.response.edit_message(
components=get_no_taches_components(user_id, guild_id, is_all=False))
return
taches_list = list(taches.items())
await interaction.response.edit_message(
components=get_taches_pagination_components(user_id, guild_id, taches_list, 0, False))
elif action == "taches_toutes":
guild_id = int(parts[1])
user_id = interaction.author.id
if not interaction.author.guild_permissions.administrator:
await interaction.response.edit_message(
components=get_no_permission_components(guild_id))
return
settings = await get_taches_settings(guild_id)
if not settings.get("taches_enabled", True):
await interaction.response.edit_message(
components=get_module_disabled_components(guild_id))
return
taches = await lister_toutes_taches(guild_id)
if not taches:
await interaction.response.edit_message(
components=get_no_taches_components(user_id, guild_id, is_all=True))
return
taches_list = list(taches.items())
await interaction.response.edit_message(
components=get_taches_pagination_components(user_id, guild_id, taches_list, 0, True))
elif action == "taches_rafraichir":
guild_id = int(parts[1])
await interaction.response.edit_message(
components=get_main_menu_components(guild_id))
# Navigation
elif action == "taches_prev":
user_id = int(parts[1])
guild_id = int(parts[2])
current_page = int(parts[3])
is_all = bool(int(parts[4]))
if interaction.author.id != user_id:
return
if current_page > 0:
new_page = current_page - 1
taches = await lister_toutes_taches(guild_id) if is_all else await lister_taches_utilisateur(user_id, guild_id)
taches_list = list(taches.items()) if taches else []
await interaction.response.edit_message(
components=get_taches_pagination_components(user_id, guild_id, taches_list, new_page, is_all))
elif action == "taches_next":
user_id = int(parts[1])
guild_id = int(parts[2])
current_page = int(parts[3])
is_all = bool(int(parts[4]))
if interaction.author.id != user_id:
return
taches = await lister_toutes_taches(guild_id) if is_all else await lister_taches_utilisateur(user_id, guild_id)
taches_list = list(taches.items()) if taches else []
if (current_page + 1) * 4 < len(taches_list):
new_page = current_page + 1
await interaction.response.edit_message(
components=get_taches_pagination_components(user_id, guild_id, taches_list, new_page, is_all))
elif action == "taches_back_to_menu":
if interaction.author.id != int(parts[1]):
return
guild_id = int(parts[2])
await interaction.response.edit_message(
components=get_main_menu_components(guild_id))
# Détails tâche
elif action == "tache_detail":
tache_id = parts[1]
user_id = int(parts[2])
guild_id = int(parts[3])
page = int(parts[4])
is_all = bool(int(parts[5]))
if interaction.author.id != user_id:
await interaction.response.send_message("❌ Non autorisé.", ephemeral=True)
return
tache = await get_tache(tache_id)
if not tache:
await interaction.response.edit_message(
components=get_taches_pagination_components(user_id, guild_id, [], page, is_all))
return
if isinstance(tache, dict) and len(tache) == 1:
tache = next(iter(tache.values()))
can_edit = tache.get("assigne") == user_id or tache.get("cree_par") == user_id
await interaction.response.edit_message(
components=get_tache_detail_components(tache, tache_id, user_id, guild_id, can_edit))
# Actions tâche
elif action == "tache_done":
tache_id = parts[1]
guild_id = int(parts[2])
tache = await get_tache(tache_id)
if not tache:
await interaction.response.send_message("❌ Tâche introuvable.", ephemeral=True)
return
if isinstance(tache, dict) and len(tache) == 1:
tache = next(iter(tache.values()))
if interaction.author.id != tache.get("assigne") and interaction.author.id != tache.get("cree_par"):
await interaction.response.send_message("❌ Non autorisé.", ephemeral=True)
return
success = await update_tache_status(tache_id, "Terminée")
if success:
tache["statut"] = "Terminée"
await interaction.response.edit_message(
components=get_tache_done_components(tache, interaction.author.id, guild_id))
else:
await interaction.response.send_message("❌ Erreur.", ephemeral=True)
elif action == "tache_prio":
tache_id = parts[1]
guild_id = int(parts[2])
tache = await get_tache(tache_id)
if not tache:
await interaction.response.send_message("❌ Tâche introuvable.", ephemeral=True)
return
if isinstance(tache, dict) and len(tache) == 1:
tache = next(iter(tache.values()))
if interaction.author.id != tache.get("assigne") and interaction.author.id != tache.get("cree_par"):
await interaction.response.send_message("❌ Non autorisé.", ephemeral=True)
return
current_prio = tache.get("priorite", "Moyenne")
await interaction.response.edit_message(
components=get_change_priority_components(tache_id, current_prio, interaction.author.id, guild_id))
elif action == "tache_delete":
tache_id = parts[1]
guild_id = int(parts[2])
tache = await get_tache(tache_id)
if not tache:
await interaction.response.send_message("❌ Tâche introuvable.", ephemeral=True)
return
if isinstance(tache, dict) and len(tache) == 1:
tache = next(iter(tache.values()))
if interaction.author.id != tache.get("assigne") and interaction.author.id != tache.get("cree_par"):
await interaction.response.send_message("❌ Non autorisé.", ephemeral=True)
return
tache_titre = tache.get("titre", "Tâche")
modal = ConfirmDeleteModal(tache_id, tache_titre)
await interaction.response.send_modal(modal)
# Retour
elif action == "tache_back":
guild_id = int(parts[1])
await interaction.response.edit_message(
components=get_main_menu_components(guild_id))
elif action == "taches_back":
guild_id = int(parts[1])
await interaction.response.edit_message(
components=get_main_menu_components(guild_id))
# Changement de priorité
elif action == "prio_select":
tache_id = parts[1]
new_prio = parts[2]
user_id = int(parts[3])
guild_id = int(parts[4])
if interaction.author.id != user_id:
await interaction.response.send_message("❌ Non autorisé.", ephemeral=True)
return
success = await update_tache_priorite(tache_id, new_prio)
if success:
await interaction.response.edit_message(
components=get_main_menu_components(guild_id))
await interaction.followup.send(f"✅ Priorité mise à jour : **{new_prio}**", ephemeral=True)
else:
await interaction.response.send_message("❌ Erreur.", ephemeral=True)
elif action == "prio_cancel":
guild_id = int(parts[3])
await interaction.response.edit_message(
components=get_main_menu_components(guild_id))
except Exception as e:
kuby_logger.error(f"Erreur dans on_taches_interaction: {e}", exc_info=True)
try:
await interaction.response.send_message("❌ Une erreur est survenue.", ephemeral=True)
except:
pass
# ==========================================================================
# COMMANDE SLASH
# ==========================================================================
@commands.slash_command(name="taches", description="Ouvrir le menu de gestion des tâches")
async def taches_command(self, inter: disnake.ApplicationCommandInteraction):
"""Commande principale pour accéder au menu des tâches."""
settings = await get_taches_settings(inter.guild.id)
if not settings.get("taches_enabled", True):
return await inter.response.send_message(
"❌ Le système de tâches est désactivé sur ce serveur.",
ephemeral=True
)
await inter.response.send_message(
components=get_main_menu_components(inter.guild.id)
)
# ==========================================================================
# BOUCLE DE RAPPELS
# ==========================================================================
@tasks.loop(minutes=15)
async def rappels_loop(self):
"""Vérifie les tâches proches de leur échéance tous les 15 minutes."""
for guild in self.bot.guilds:
try:
settings = await get_taches_settings(guild.id)
if not settings.get("taches_enabled", True):
continue
taches = await taches_proches(guild.id, heures_avant=1)
for tid, tache in taches.items():
try:
user = await self.bot.fetch_user(tache["assigne"])
if user:
statut_emoji = _get_statut_emoji(tache.get("statut", "À faire"))
priorite_emoji = _get_priorite_emoji(tache.get("priorite", "Moyenne"))
await user.send(
content=(
f"⏰ **Rappel : Tâche #{tid}**\n\n"
f"{statut_emoji} **{tache['titre']}** {priorite_emoji}\n"
f"📅 **Échéance dans 1 heure**\n"
f"👤 Assigné par : <@{tache['cree_par']}>"
)
)
except Exception as e:
kuby_logger.error(f"Erreur rappel tâche {tid}: {e}")
except Exception as e:
kuby_logger.error(f"Erreur dans rappels_loop pour {guild.id}: {e}")
@rappels_loop.before_loop
async def avant_rappels(self):
await self.bot.wait_until_ready()
# ============================================================================
# SETUP
# ============================================================================
def setup(bot):
"""Setup du cog Tâches."""
bot.add_cog(TachesCog(bot))

View file

@ -77,16 +77,16 @@ def run_mep_logic(author, commit_msg, branch="main"):
except subprocess.CalledProcessError as e: except subprocess.CalledProcessError as e:
send_discord_log("ERREUR MIGRATION", f"❌ Migration GitLab échouée: {e.stderr}", 15158332, author, commit_msg) send_discord_log("ERREUR MIGRATION", f"❌ Migration GitLab échouée: {e.stderr}", 15158332, author, commit_msg)
duration = (datetime.now() - start_time).total_seconds() # 3. PM2 (Appelé avant le message de succès)
msg = f"✅ **Déploiement réussi en {duration:.2f}s**\nLe bot est à jour et redémarré."
send_discord_log("VALIDÉ", msg, 3066993, author, commit_msg)
# 3. PM2 (Appelé après le message pour éviter que le kill du serv ne bloque l'envoi)
subprocess.run( subprocess.run(
f"source /home/discord/.nvm/nvm.sh && pm2 restart {PROCESS_NAME}", f"source /home/discord/.nvm/nvm.sh && pm2 restart {PROCESS_NAME}",
shell=True, check=True, capture_output=True, executable="/bin/bash" shell=True, check=True, capture_output=True, executable="/bin/bash"
) )
duration = (datetime.now() - start_time).total_seconds()
msg = f"✅ **Déploiement réussi en {duration:.2f}s**\nLe bot est à jour et redémarré."
send_discord_log("VALIDÉ", msg, 3066993, author, commit_msg)
except Exception as e: except Exception as e:
error_msg = f"❌ **ERREUR de déploiement**\n```python\n{str(e)}\n```" error_msg = f"❌ **ERREUR de déploiement**\n```python\n{str(e)}\n```"
send_discord_log("ÉCHEC", error_msg, 15158332, author, commit_msg) send_discord_log("ÉCHEC", error_msg, 15158332, author, commit_msg)

404
utils/gestion_taches.py Normal file
View file

@ -0,0 +1,404 @@
"""
Gestion des tâches - Module modulaire pour Kuby
Utilise disnake et un stockage JSON par serveur.
"""
import json
import os
from datetime import datetime, timedelta
from zoneinfo import ZoneInfo
from typing import Optional, Dict, List, Tuple, Any
# Chemin du fichier de stockage
DATA_DIR = os.path.join(os.path.dirname(os.path.dirname(__file__)), "data")
TACHES_FILE = os.path.join(DATA_DIR, "taches.json")
# Priorités disponibles
PRIORITES = ["Haute", "Moyenne", "Basse", "Critique", "Faible"]
STATUTS = ["À faire", "En cours", "Terminée", "Annulée"]
def _assurer_dossier():
"""Crée le dossier data/ si inexistant."""
os.makedirs(DATA_DIR, exist_ok=True)
def _charger_fichier() -> Dict[str, Any]:
"""Charge le fichier taches.json ou retourne une structure vide."""
_assurer_dossier()
if not os.path.exists(TACHES_FILE):
return {"taches": {}, "archive": {}, "settings": {}}
try:
with open(TACHES_FILE, "r", encoding="utf-8") as f:
return json.load(f)
except (json.JSONDecodeError, FileNotFoundError):
return {"taches": {}, "archive": {}, "settings": {}}
def _sauvegarder_fichier(data: Dict[str, Any]) -> None:
"""Sauvegarde les données dans taches.json."""
_assurer_dossier()
with open(TACHES_FILE, "w", encoding="utf-8") as f:
json.dump(data, f, indent=4, ensure_ascii=False)
# ============================================================================
# FONCTIONS DE BASE
# ============================================================================
async def charger_taches() -> Dict[str, Any]:
"""
Charge toutes les tâches depuis le fichier.
Returns:
Dict avec clés: taches, archive, settings
"""
return _charger_fichier()
async def sauvegarder_taches(data: Optional[Dict[str, Any]] = None) -> None:
"""
Sauvegarde les tâches dans le fichier.
Si data=None, recharge depuis le fichier et sauvegarde (pour rafraîchir).
"""
if data is None:
data = _charger_fichier()
_sauvegarder_fichier(data)
# ============================================================================
# GESTION DES TÂCHES
# ============================================================================
async def ajouter_tache(
titre: str,
description: str,
assigne_id: int,
priorite: str,
echeance: str, # Format: YYYY-MM-DD
cree_par_id: int,
serveur_id: int
) -> str:
"""
Ajoute une nouvelle tâche.
Args:
titre: Titre de la tâche
description: Description détaillée
assigne_id: ID Discord de l'utilisateur assigné
priorite: Priorité (doit être dans PRIORITES)
echeance: Date au format YYYY-MM-DD
cree_par_id: ID Discord du créateur
serveur_id: ID du serveur (guild)
Returns:
str: L'ID unique de la tâche créée
"""
data = _charger_fichier()
# Valider la priorité
if priorite not in PRIORITES:
priorite = "Moyenne"
# Générer un ID unique
tache_id = str(len(data["taches"]) + 1)
# Date actuelle
now = datetime.now(ZoneInfo("Europe/Paris"))
data["taches"][tache_id] = {
"titre": titre,
"description": description,
"assigne": assigne_id,
"cree_par": cree_par_id,
"priorite": priorite,
"echeance": echeance,
"statut": "À faire",
"date_creation": now.strftime("%Y-%m-%d %H:%M:%S"),
"date_fin": None,
"serveur_id": serveur_id,
"archive": False
}
_sauvegarder_fichier(data)
return tache_id
async def lister_taches_utilisateur(user_id: int, serveur_id: int) -> Dict[str, Dict[str, Any]]:
"""
Liste les tâches assignées à un utilisateur sur un serveur.
Args:
user_id: ID Discord de l'utilisateur
serveur_id: ID du serveur
Returns:
Dict des tâches (non archivées) de l'utilisateur
"""
data = _charger_fichier()
return {
tid: tache for tid, tache in data["taches"].items()
if tache.get("assigne") == user_id
and tache.get("serveur_id") == serveur_id
and not tache.get("archive", False)
}
async def lister_toutes_taches(serveur_id: int) -> Dict[str, Dict[str, Any]]:
"""
Liste toutes les tâches (non archivées) d'un serveur.
Args:
serveur_id: ID du serveur
Returns:
Dict de toutes les tâches du serveur
"""
data = _charger_fichier()
return {
tid: tache for tid, tache in data["taches"].items()
if tache.get("serveur_id") == serveur_id
and not tache.get("archive", False)
}
async def get_tache(tache_id: str) -> Optional[Dict[str, Any]]:
"""
Récupère une tâche par son ID.
Args:
tache_id: ID de la tâche
Returns:
Dict de la tâche ou None si non trouvée
"""
data = _charger_fichier()
return data["taches"].get(tache_id)
async def update_tache_status(tache_id: str, statut: str) -> bool:
"""
Met à jour le statut d'une tâche.
Args:
tache_id: ID de la tâche
statut: Nouveau statut
Returns:
bool: True si la mise à jour a réussi
"""
data = _charger_fichier()
if tache_id not in data["taches"]:
return False
if statut not in STATUTS:
statut = "À faire"
# Mettre à jour le statut
data["taches"][tache_id]["statut"] = statut
# Si terminée, enregistrer la date de fin
if statut == "Terminée":
now = datetime.now(ZoneInfo("Europe/Paris"))
data["taches"][tache_id]["date_fin"] = now.strftime("%Y-%m-%d %H:%M:%S")
else:
data["taches"][tache_id]["date_fin"] = None
_sauvegarder_fichier(data)
return True
async def update_tache_priorite(tache_id: str, priorite: str) -> bool:
"""
Met à jour la priorité d'une tâche.
Args:
tache_id: ID de la tâche
priorite: Nouvelle priorité
Returns:
bool: True si la mise à jour a réussi
"""
data = _charger_fichier()
if tache_id not in data["taches"]:
return False
if priorite not in PRIORITES:
return False
data["taches"][tache_id]["priorite"] = priorite
_sauvegarder_fichier(data)
return True
async def supprimer_tache(tache_id: str) -> bool:
"""
Supprime une tâche (la déplace dans l'archive).
Args:
tache_id: ID de la tâche
Returns:
bool: True si la suppression a réussi
"""
data = _charger_fichier()
if tache_id not in data["taches"]:
return False
# Archiver la tâche au lieu de la supprimer
tache = data["taches"].pop(tache_id)
tache["archive"] = True
tache["date_archivage"] = datetime.now(ZoneInfo("Europe/Paris")).strftime("%Y-%m-%d %H:%M:%S")
data["archive"][tache_id] = tache
_sauvegarder_fichier(data)
return True
# ============================================================================
# FONCTIONS DE RAPPEL
# ============================================================================
async def taches_proches(serveur_id: int, heures_avant: int = 1) -> Dict[str, Dict[str, Any]]:
"""
Récupère les tâches dont l'échéance est dans X heures.
Args:
serveur_id: ID du serveur
heures_avant: Nombre d'heures avant l'échéance (défaut: 1)
Returns:
Dict des tâches proches de leur échéance
"""
data = _charger_fichier()
now = datetime.now(ZoneInfo("Europe/Paris"))
result = {}
for tid, tache in data["taches"].items():
if tache.get("serveur_id") != serveur_id or tache.get("archive", False):
continue
if tache["statut"] == "Terminée":
continue
try:
echeance_date = datetime.strptime(tache["echeance"], "%Y-%m-%d")
echeance_date = echeance_date.replace(
hour=9, minute=0, second=0, microsecond=0,
tzinfo=ZoneInfo("Europe/Paris")
)
# Calculer la différence
delta = echeance_date - now
# Si l'échéance est dans X heures ou moins, et pas déjà passée
if timedelta(0) < delta <= timedelta(hours=heures_avant):
result[tid] = tache
except (ValueError, TypeError):
continue
return result
async def taches_aujourdhui(serveur_id: int) -> Dict[str, Dict[str, Any]]:
"""
Récupère les tâches dont l'échéance est aujourd'hui.
Args:
serveur_id: ID du serveur
Returns:
Dict des tâches d'aujourd'hui
"""
return await taches_proches(serveur_id, heures_avant=24)
# ============================================================================
# ARCHIVAGE AUTOMATIQUE
# ============================================================================
async def archiver_taches_anciennes(serveur_id: int, jours: int = 30) -> int:
"""
Archive les tâches terminées depuis plus de X jours.
Args:
serveur_id: ID du serveur
jours: Nombre de jours après lequel archiver (défaut: 30)
Returns:
int: Nombre de tâches archivées
"""
data = _charger_fichier()
now = datetime.now(ZoneInfo("Europe/Paris"))
count = 0
to_archive = []
for tid, tache in data["taches"].items():
if tache.get("serveur_id") != serveur_id:
continue
if tache.get("statut") != "Terminée":
continue
if tache.get("archive", False):
continue
try:
fin_date = datetime.strptime(tache["date_fin"], "%Y-%m-%d %H:%M:%S")
fin_date = fin_date.replace(tzinfo=ZoneInfo("Europe/Paris"))
if (now - fin_date) > timedelta(days=jours):
to_archive.append(tid)
except (ValueError, TypeError):
continue
for tid in to_archive:
tache = data["taches"].pop(tid)
tache["archive"] = True
tache["date_archivage"] = now.strftime("%Y-%m-%d %H:%M:%S")
data["archive"][tid] = tache
count += 1
if count > 0:
_sauvegarder_fichier(data)
return count
# ============================================================================
# CONFIGURATION PAR SERVEUR
# ============================================================================
async def get_taches_settings(serveur_id: int) -> Dict[str, Any]:
"""
Récupère les paramètres des tâches pour un serveur.
Args:
serveur_id: ID du serveur
Returns:
Dict des paramètres (taches_enabled, auto_archive_days, etc.)
"""
data = _charger_fichier()
return data["settings"].get(str(serveur_id), {
"taches_enabled": True,
"auto_archive_days": 30,
"notify_channel_id": None
})
async def set_taches_settings(serveur_id: int, settings: Dict[str, Any]) -> None:
"""
Met à jour les paramètres des tâches pour un serveur.
Args:
serveur_id: ID du serveur
settings: Dict des paramètres à mettre à jour
"""
data = _charger_fichier()
if str(serveur_id) not in data["settings"]:
data["settings"][str(serveur_id)] = {}
data["settings"][str(serveur_id)].update(settings)
_sauvegarder_fichier(data)