404 lines
11 KiB
Python
404 lines
11 KiB
Python
"""
|
|
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)
|