89 lines
2.7 KiB
Python
89 lines
2.7 KiB
Python
import json
|
|
import os
|
|
import asyncio
|
|
from datetime import datetime
|
|
|
|
FICHIER_TACHES = "taches.json"
|
|
FICHIER_TACHES_TMP = "taches.json.tmp"
|
|
FICHIER_ID = "id_compteur.json"
|
|
file_lock = asyncio.Lock()
|
|
|
|
async def charger_taches():
|
|
async with file_lock:
|
|
if not os.path.exists(FICHIER_TACHES):
|
|
with open(FICHIER_TACHES, "w", encoding='utf-8') as f:
|
|
json.dump({"taches": {}}, f, indent=4)
|
|
with open(FICHIER_TACHES, "r", encoding='utf-8') as f:
|
|
return json.load(f)
|
|
|
|
async def sauvegarder_taches(donnees):
|
|
async with file_lock:
|
|
with open(FICHIER_TACHES_TMP, "w", encoding='utf-8') as f:
|
|
json.dump(donnees, f, indent=4, ensure_ascii=False)
|
|
os.replace(FICHIER_TACHES_TMP, FICHIER_TACHES)
|
|
|
|
async def generer_id_unique():
|
|
async with file_lock:
|
|
if not os.path.exists(FICHIER_ID):
|
|
with open(FICHIER_ID, "w", encoding='utf-8') as f:
|
|
json.dump({"dernier_id": 0}, f, indent=4)
|
|
|
|
with open(FICHIER_ID, "r+", encoding='utf-8') as f:
|
|
data = json.load(f)
|
|
data["dernier_id"] += 1
|
|
f.seek(0)
|
|
json.dump(data, f, indent=4)
|
|
f.truncate()
|
|
return str(data["dernier_id"])
|
|
|
|
async def ajouter_tache(titre, description, user_id, priorite, date_echeance, auteur_id):
|
|
data = await charger_taches()
|
|
taches = data["taches"]
|
|
id_tache = await generer_id_unique()
|
|
|
|
taches[id_tache] = {
|
|
"titre": titre,
|
|
"description": description,
|
|
"assigné": user_id,
|
|
"statut": "À faire",
|
|
"priorité": priorite,
|
|
"échéance": date_echeance,
|
|
"créé_par": auteur_id,
|
|
"commentaires": []
|
|
}
|
|
|
|
await sauvegarder_taches(data)
|
|
return id_tache
|
|
|
|
async def update_tache_status(id_tache, statut):
|
|
data = await charger_taches()
|
|
if id_tache in data["taches"]:
|
|
data["taches"][id_tache]["statut"] = statut
|
|
await sauvegarder_taches(data)
|
|
return True
|
|
return False
|
|
|
|
async def supprimer_tache(id_tache):
|
|
data = await charger_taches()
|
|
if id_tache in data["taches"]:
|
|
del data["taches"][id_tache]
|
|
await sauvegarder_taches(data)
|
|
return True
|
|
return False
|
|
|
|
async def lister_taches_utilisateur(user_id):
|
|
data = await charger_taches()
|
|
return {tid: t for tid, t in data["taches"].items() if t["assigné"] == user_id}
|
|
|
|
async def taches_proches():
|
|
data = await charger_taches()
|
|
proches = {}
|
|
now = datetime.now()
|
|
for tid, t in data["taches"].items():
|
|
try:
|
|
date_obj = datetime.strptime(t["échéance"], "%Y-%m-%d")
|
|
if 0 <= (date_obj - now).days <= 2:
|
|
proches[tid] = t
|
|
except ValueError:
|
|
continue
|
|
return proches
|