""" 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]] = {} 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(",") fname: str = csv[0] entry_date: datetime = is_iso_formated(csv[1], True) delete_date: datetime = is_iso_formated(csv[2], True) self.content[fname] = [entry_date, delete_date] def set_trace(self, *fnames: str, mode: str = "entry"): """Ajoute une trace du fichier donné mode : entry / delete """ modes: list[str] = ["entry", "delete"] for fname in fnames: if fname in modes: continue traced: list[datetime, datetime] = self.content.get(fname, False) if not traced: self.content[fname] = [None, None] traced = self.content[fname] traced[modes.index(mode)] = datetime.now() 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" lines.append(f"{fname},{traced[0].isoformat()},{date_fin}") with open(self.path, "w", encoding="utf-8") as file: file.write("\n".join(lines)) def get_trace(self, fnames: list[str] = ()) -> dict[str, list[datetime, datetime]]: """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 or len(fnames) == 0: 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 └── └── ├── [_trace.csv] └── ├── [_description.txt] └── [] """ def __init__(self): BaseArchiver.__init__(self, archive_type="justificatifs") def save_justificatif( self, etudid: int, filename: str, data: bytes or str, archive_name: str = None, description: str = "", ) -> str: """ Ajoute un fichier dans une archive "justificatif" pour l'etudid donné Retourne l'archive_name utilisé """ self._set_dept(etudid) if archive_name is None: archive_id: str = self.create_obj_archive( oid=etudid, description=description ) else: archive_id: str = self.get_id_from_name(etudid, archive_name) fname: str = self.store(archive_id, filename, data) trace = Trace(self.get_obj_dir(etudid)) trace.set_trace(fname, "entry") return self.get_archive_name(archive_id), fname def delete_justificatif( self, etudid: int, 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 """ self._set_dept(etudid) if str(etudid) not in self.list_oids(): raise ValueError(f"Aucune archive pour etudid[{etudid}]") archive_id = self.get_id_from_name(etudid, archive_name) if filename is not None: if filename not in self.list_archive(archive_id): raise ValueError( f"filename {filename} inconnu dans l'archive archive_id[{archive_id}] -> etudid[{etudid}]" ) path: str = os.path.join(self.get_obj_dir(etudid), archive_id, filename) if os.path.isfile(path): if has_trace: trace = Trace(self.get_obj_dir(etudid)) trace.set_trace(filename, "delete") os.remove(path) else: if has_trace: trace = Trace(self.get_obj_dir(etudid)) trace.set_trace(*self.list_archive(archive_id), mode="delete") self.delete_archive( os.path.join( self.get_obj_dir(etudid), archive_id, ) ) def list_justificatifs(self, archive_name: str, etudid: int) -> list[str]: """ Retourne la liste des noms de fichiers dans l'archive donnée """ self._set_dept(etudid) filenames: list[str] = [] archive_id = self.get_id_from_name(etudid, archive_name) filenames = self.list_archive(archive_id) return filenames def get_justificatif_file(self, archive_name: str, etudid: int, filename: str): """ Retourne une réponse de téléchargement de fichier si le fichier existe """ self._set_dept(etudid) archive_id: str = self.get_id_from_name(etudid, archive_name) if filename in self.list_archive(archive_id): return self.get_archived_file(etudid, archive_name, filename) raise ScoValueError( f"Fichier {filename} introuvable dans l'archive {archive_name}" ) def _set_dept(self, etudid: int): """ Mets à jour le dept_id de l'archiver en fonction du département de l'étudiant """ etud: Identite = Identite.query.filter_by(id=etudid).first() self.set_dept_id(etud.dept_id) 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 ⚠ """ self.set_dept_id(1) self.initialize() 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( self, etudid: int, *fnames: str ) -> dict[str, list[datetime, datetime]]: """Récupère la trace des justificatifs de l'étudiant""" trace = Trace(self.get_obj_dir(etudid)) return trace.get_trace(fnames)