forked from ScoDoc/ScoDoc
232 lines
8.0 KiB
Python
232 lines
8.0 KiB
Python
"""
|
|
Gestion de l'archivage des justificatifs
|
|
|
|
Ecrit par Matthias HARTMANN
|
|
"""
|
|
import os
|
|
from datetime import datetime
|
|
from shutil import rmtree
|
|
|
|
from app.models import Identite
|
|
from app.scodoc.sco_archives import BaseArchiver
|
|
from app.scodoc.sco_exceptions import ScoValueError
|
|
from app.scodoc.sco_utils import is_iso_formated
|
|
|
|
|
|
class Trace:
|
|
"""gestionnaire de la trace des fichiers justificatifs"""
|
|
|
|
def __init__(self, path: str) -> None:
|
|
self.path: str = path + "/_trace.csv"
|
|
self.content: dict[str, list[datetime, datetime, str]] = {}
|
|
self.import_from_file()
|
|
|
|
def import_from_file(self):
|
|
"""import trace from file"""
|
|
if os.path.isfile(self.path):
|
|
with open(self.path, "r", encoding="utf-8") as file:
|
|
for line in file.readlines():
|
|
csv = line.split(",")
|
|
if len(csv) < 4:
|
|
continue
|
|
fname: str = csv[0]
|
|
entry_date: datetime = is_iso_formated(csv[1], True)
|
|
delete_date: datetime = is_iso_formated(csv[2], True)
|
|
user_id = csv[3]
|
|
|
|
self.content[fname] = [entry_date, delete_date, user_id]
|
|
|
|
def set_trace(self, *fnames: str, mode: str = "entry", current_user: str = None):
|
|
"""Ajoute une trace du fichier donné
|
|
mode : entry / delete
|
|
"""
|
|
modes: list[str] = ["entry", "delete", "user_id"]
|
|
for fname in fnames:
|
|
if fname in modes:
|
|
continue
|
|
traced: list[datetime, datetime, str] = self.content.get(fname, False)
|
|
if not traced:
|
|
self.content[fname] = [None, None, None]
|
|
traced = self.content[fname]
|
|
|
|
traced[modes.index(mode)] = (
|
|
datetime.now() if mode != "user_id" else current_user
|
|
)
|
|
self.save_trace()
|
|
|
|
def save_trace(self):
|
|
"""Enregistre la trace dans le fichier _trace.csv"""
|
|
lines: list[str] = []
|
|
for fname, traced in self.content.items():
|
|
date_fin: datetime or None = traced[1].isoformat() if traced[1] else "None"
|
|
if traced[0] is not None:
|
|
lines.append(f"{fname},{traced[0].isoformat()},{date_fin}, {traced[2]}")
|
|
with open(self.path, "w", encoding="utf-8") as file:
|
|
file.write("\n".join(lines))
|
|
|
|
def get_trace(
|
|
self, fnames: list[str] = None
|
|
) -> dict[str, list[datetime, datetime, str]]:
|
|
"""Récupère la trace pour les noms de fichiers.
|
|
si aucun nom n'est donné, récupère tous les fichiers"""
|
|
|
|
if fnames is None:
|
|
return self.content
|
|
|
|
traced: dict = {}
|
|
for fname in fnames:
|
|
traced[fname] = self.content.get(fname, None)
|
|
|
|
return traced
|
|
|
|
|
|
class JustificatifArchiver(BaseArchiver):
|
|
"""
|
|
|
|
TOTALK:
|
|
- oid -> etudid
|
|
- archive_id -> date de création de l'archive (une archive par dépot de document)
|
|
|
|
justificatif
|
|
└── <dept_id>
|
|
└── <etudid/oid>
|
|
├── [_trace.csv]
|
|
└── <archive_id>
|
|
├── [_description.txt]
|
|
└── [<filename.ext>]
|
|
|
|
"""
|
|
|
|
def __init__(self):
|
|
BaseArchiver.__init__(self, archive_type="justificatifs")
|
|
|
|
def save_justificatif(
|
|
self,
|
|
etud: Identite,
|
|
filename: str,
|
|
data: bytes or str,
|
|
archive_name: str = None,
|
|
description: str = "",
|
|
user_id: str = None,
|
|
) -> str:
|
|
"""
|
|
Ajoute un fichier dans une archive "justificatif" pour l'etudid donné
|
|
Retourne l'archive_name utilisé
|
|
"""
|
|
if archive_name is None:
|
|
archive_id: str = self.create_obj_archive(
|
|
oid=etud.id, description=description, dept_id=etud.dept_id
|
|
)
|
|
else:
|
|
archive_id: str = self.get_id_from_name(
|
|
etud.id, archive_name, dept_id=etud.dept_id
|
|
)
|
|
|
|
fname: str = self.store(archive_id, filename, data, dept_id=etud.dept_id)
|
|
|
|
trace = Trace(self.get_obj_dir(etud.id, dept_id=etud.dept_id))
|
|
trace.set_trace(fname, mode="entry")
|
|
if user_id is not None:
|
|
trace.set_trace(fname, mode="user_id", current_user=user_id)
|
|
|
|
return self.get_archive_name(archive_id), fname
|
|
|
|
def delete_justificatif(
|
|
self,
|
|
etud: Identite,
|
|
archive_name: str,
|
|
filename: str = None,
|
|
has_trace: bool = True,
|
|
):
|
|
"""
|
|
Supprime une archive ou un fichier particulier de l'archivage de l'étudiant donné
|
|
|
|
Si trace == True : sauvegarde le nom du/des fichier(s) supprimé(s)
|
|
dans la trace de l'étudiant
|
|
"""
|
|
if str(etud.id) not in self.list_oids():
|
|
raise ValueError(f"Aucune archive pour etudid[{etud.id}]")
|
|
|
|
archive_id = self.get_id_from_name(etud.id, archive_name, dept_id=etud.dept_id)
|
|
|
|
if filename is not None:
|
|
if filename not in self.list_archive(archive_id, dept_id=etud.dept_id):
|
|
raise ValueError(
|
|
f"""filename {filename} inconnu dans l'archive archive_id[{
|
|
archive_id}] -> etudid[{etud.id}]"""
|
|
)
|
|
|
|
path: str = os.path.join(
|
|
self.get_obj_dir(etud.id, dept_id=etud.dept_id), archive_id, filename
|
|
)
|
|
|
|
if os.path.isfile(path):
|
|
if has_trace:
|
|
trace = Trace(self.get_obj_dir(etud.id, dept_id=etud.dept_id))
|
|
trace.set_trace(filename, mode="delete")
|
|
os.remove(path)
|
|
|
|
else:
|
|
if has_trace:
|
|
trace = Trace(self.get_obj_dir(etud.id, dept_id=etud.dept_id))
|
|
trace.set_trace(
|
|
*self.list_archive(archive_id, dept_id=etud.dept_id), mode="delete"
|
|
)
|
|
|
|
self.delete_archive(
|
|
os.path.join(
|
|
self.get_obj_dir(etud.id, dept_id=etud.dept_id),
|
|
archive_id,
|
|
)
|
|
)
|
|
|
|
def list_justificatifs(
|
|
self, archive_name: str, etud: Identite
|
|
) -> list[tuple[str, int]]:
|
|
"""
|
|
Retourne la liste des noms de fichiers dans l'archive donnée
|
|
"""
|
|
filenames: list[str] = []
|
|
archive_id = self.get_id_from_name(etud.id, archive_name, dept_id=etud.dept_id)
|
|
|
|
filenames = self.list_archive(archive_id, dept_id=etud.dept_id)
|
|
trace: Trace = Trace(self.get_obj_dir(etud.id, dept_id=etud.dept_id))
|
|
traced = trace.get_trace(filenames)
|
|
retour = [(key, value[2]) for key, value in traced.items()]
|
|
|
|
return retour
|
|
|
|
def get_justificatif_file(self, archive_name: str, etud: Identite, filename: str):
|
|
"""
|
|
Retourne une réponse de téléchargement de fichier si le fichier existe
|
|
"""
|
|
archive_id: str = self.get_id_from_name(
|
|
etud.id, archive_name, dept_id=etud.dept_id
|
|
)
|
|
if filename in self.list_archive(archive_id, dept_id=etud.dept_id):
|
|
return self.get_archived_file(
|
|
etud.id, archive_name, filename, dept_id=etud.dept_id
|
|
)
|
|
raise ScoValueError(
|
|
f"Fichier {filename} introuvable dans l'archive {archive_name}"
|
|
)
|
|
|
|
def remove_dept_archive(self, dept_id: int = None):
|
|
"""
|
|
Supprime toutes les archives d'un département (ou de tous les départements)
|
|
⚠ Supprime aussi les fichiers de trace ⚠
|
|
"""
|
|
# juste pour récupérer .root, dept_id n'a pas d'importance
|
|
self.initialize(dept_id=1)
|
|
if dept_id is None:
|
|
rmtree(self.root, ignore_errors=True)
|
|
else:
|
|
rmtree(os.path.join(self.root, str(dept_id)), ignore_errors=True)
|
|
|
|
def get_trace( # XXX inutilisée ?
|
|
self, etud: Identite, *fnames: str
|
|
) -> dict[str, list[datetime, datetime]]:
|
|
"""Récupère la trace des justificatifs de l'étudiant"""
|
|
trace = Trace(self.get_obj_dir(etud.id, etud.dept_id))
|
|
return trace.get_trace(fnames)
|