############################################################################## # # Gestion scolarite IUT # # Copyright (c) 1999 - 2024 Emmanuel Viennet. All rights reserved. # # This program is free software; you can redistribute it and/or modify # it under the terms of the GNU General Public License as published by # the Free Software Foundation; either version 2 of the License, or # (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program; if not, write to the Free Software # Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA # # Emmanuel Viennet emmanuel.viennet@viennet.net # ############################################################################## """Lecture du fichier "maquette" Apogée Le fichier CSV, champs séparés par des tabulations, a la structure suivante: <pre> XX-APO_TITRES-XX apoC_annee 2007/2008 apoC_cod_dip VDTCJ apoC_Cod_Exp 1 apoC_cod_vdi 111 apoC_Fichier_Exp VDTCJ_V1CJ.txt apoC_lib_dip DUT CJ apoC_Titre1 Export Apogée du 13/06/2008 à 14:29 apoC_Titre2 XX-APO_TYP_RES-XX ...section optionnelle au contenu quelconque... XX-APO_COLONNES-XX apoL_a01_code Type Objet Code Version Année Session Admission/Admissibilité Type Rés. Etudiant Numéro apoL_a02_nom 1 Nom apoL_a03_prenom 1 Prénom apoL_a04_naissance Session Admissibilité Naissance APO_COL_VAL_DEB apoL_c0001 VET V1CJ 111 2007 0 1 N V1CJ - DUT CJ an1 0 1 Note apoL_c0002 VET V1CJ 111 2007 0 1 B 0 1 Barème apoL_c0003 VET V1CJ 111 2007 0 1 R 0 1 Résultat APO_COL_VAL_FIN apoL_c0030 APO_COL_VAL_FIN XX-APO_VALEURS-XX apoL_a01_code apoL_a02_nom apoL_a03_prenom apoL_a04_naissance apoL_c0001 apoL_c0002 apoL_c0003 apoL_c0004 apoL_c0005 apoL_c0006 apoL_c0007 apoL_c0008 apoL_c0009 apoL_c0010 apoL_c0011 apoL_c0012 apoL_c0013 apoL_c0014 apoL_c0015 apoL_c0016 apoL_c0017 apoL_c0018 apoL_c0019 apoL_c0020 apoL_c0021 apoL_c0022 apoL_c0023 apoL_c0024 apoL_c0025 apoL_c0026 apoL_c0027 apoL_c0028 apoL_c0029 10601232 AARIF MALIKA 22/09/1986 18 20 ADM 18 20 ADM 18 20 ADM 18 20 ADM 18 20 ADM 18 20 18 20 ADM 18 20 ADM 18 20 ADM 18 20 ADM </pre> On récupère nos éléments pédagogiques dans la section XX-APO-COLONNES-XX et notre liste d'étudiants dans la section XX-APO_VALEURS-XX. Les champs de la section XX-APO_VALEURS-XX sont décrits par les lignes successives de la section XX-APO_COLONNES-XX. Le fichier CSV correspond à une étape, qui est récupérée sur la ligne <pre> apoL_c0001 VET V1CJ ... </pre> """ from collections import namedtuple import io import pprint import re # Pour la détection auto de l'encodage des fichiers Apogée: from chardet import detect as chardet_detect from app import log from app.scodoc.sco_exceptions import ScoFormatError from app.scodoc import sco_preferences APO_PORTAL_ENCODING = ( "utf8" # encodage du fichier CSV Apogée (était 'ISO-8859-1' avant jul. 2016) ) APO_INPUT_ENCODING = "ISO-8859-1" # APO_OUTPUT_ENCODING = APO_INPUT_ENCODING # encodage des fichiers Apogee générés APO_DECIMAL_SEP = "," # separateur décimal: virgule APO_SEP = "\t" APO_NEWLINE = "\r\n" ApoEtudTuple = namedtuple("ApoEtudTuple", ("nip", "nom", "prenom", "naissance", "cols")) class DictCol(dict): "A dict, where we can add attributes" class StringIOWithLineNumber(io.StringIO): "simple wrapper to use a string as a file with line numbers" def __init__(self, data: str): super().__init__(data) self.lineno = 0 def readline(self): self.lineno += 1 return super().readline() class ApoCSVReadWrite: "Gestion lecture/écriture de fichiers csv Apogée" def __init__(self, data: str): if not data: raise ScoFormatError("Fichier Apogée vide !") self.data = data self._file = StringIOWithLineNumber(data) # pour traiter comme un fichier self.apo_elts: dict = None self.cols: dict[str, dict[str, str]] = None self.column_titles: str = None self.col_ids: list[str] = None self.csv_etuds: list[ApoEtudTuple] = [] # section_str: utilisé pour ré-écrire les headers sans aucune altération self.sections_str: dict[str, str] = {} "contenu initial de chaque section" # self.header: str = "" # "début du fichier Apogée jusqu'à XX-APO_TYP_RES-XX non inclu (sera ré-écrit non modifié)" self.header_apo_typ_res: str = "" "section XX-APO_TYP_RES-XX (qui peut en option ne pas être ré-écrite)" self.titles: dict[str, str] = {} "titres Apogée (section XX-APO_TITRES-XX)" self.read_sections() # Check that we have collected all requested infos: if not self.header_apo_typ_res: # on pourrait rendre XX-APO_TYP_RES-XX optionnelle mais mieux vaut vérifier: raise ScoFormatError( "format incorrect: pas de XX-APO_TYP_RES-XX", filename=self.get_filename(), ) if self.cols is None: raise ScoFormatError( "format incorrect: pas de XX-APO_COLONNES-XX", filename=self.get_filename(), ) if self.column_titles is None: raise ScoFormatError( "format incorrect: pas de XX-APO_VALEURS-XX", filename=self.get_filename(), ) def read_sections(self): """Lit une à une les sections du fichier Apogée""" # sanity check: we are at the begining of Apogee CSV start_pos = self._file.tell() section = self._file.readline().strip() if section != "XX-APO_TITRES-XX": raise ScoFormatError("format incorrect: pas de XX-APO_TITRES-XX") while True: self.read_section(section) line, end_pos = _apo_next_non_blank_line(self._file) self.sections_str[section] = self.data[start_pos:end_pos] if not line: break section = line start_pos = end_pos def read_section(self, section_name: str): """Read a section: _file is on the first line after section title""" if section_name == "XX-APO_TITRES-XX": # Titres: # on va y chercher apoC_Fichier_Exp qui donnera le nom du fichier # ainsi que l'année scolaire et le code diplôme. self.titles = self._apo_read_titres(self._file) elif section_name == "XX-APO_TYP_RES-XX": self.header_apo_typ_res = _apo_read_typ_res(self._file) elif section_name == "XX-APO_COLONNES-XX": self.cols = self.apo_read_cols() self.apo_elts = self.group_elt_cols(self.cols) elif section_name == "XX-APO_VALEURS-XX": # les étudiants self.apo_read_section_valeurs() else: raise ScoFormatError( f"format incorrect: section inconnue: {section_name}", filename=self.get_filename(), ) def apo_read_cols(self): """Lecture colonnes apo : Démarre après la balise XX-APO_COLONNES-XX et s'arrête après la ligne suivant la balise APO_COL_VAL_FIN Colonne Apogee: les champs sont données par la ligne apoL_a01_code de la section XX-APO_COLONNES-XX col_id est apoL_c0001, apoL_c0002, ... :return: { col_id : { title : value } } Example: { 'apoL_c0001' : { 'Type Objet' : 'VET', 'Code' : 'V1IN', ... }, ... } """ line = self._file.readline().strip(" " + APO_NEWLINE) fields = line.split(APO_SEP) if fields[0] != "apoL_a01_code": raise ScoFormatError( f"invalid line: {line} (expecting apoL_a01_code)", filename=self.get_filename(), ) col_keys = fields while True: # skip premiere partie (apoL_a02_nom, ...) line = self._file.readline().strip(" " + APO_NEWLINE) if line == "APO_COL_VAL_DEB": break # après APO_COL_VAL_DEB cols = {} i = 0 while True: line = self._file.readline().strip(" " + APO_NEWLINE) if line == "APO_COL_VAL_FIN": break i += 1 fields = line.split(APO_SEP) # sanity check col_id = fields[0] # apoL_c0001, ... if col_id in cols: raise ScoFormatError( f"duplicate column definition: {col_id}", filename=self.get_filename(), ) m = re.match(r"^apoL_c([0-9]{4})$", col_id) if not m: raise ScoFormatError( f"invalid column id: {line} (expecting apoL_c{col_id})", filename=self.get_filename(), ) if int(m.group(1)) != i: raise ScoFormatError( f"invalid column id: {col_id} for index {i}", filename=self.get_filename(), ) cols[col_id] = DictCol(list(zip(col_keys, fields))) cols[col_id].lineno = self._file.lineno # for debuging purpose self._file.readline() # skip next line return cols def group_elt_cols(self, cols) -> dict: """Return (ordered) dict of ApoElt from list of ApoCols. Clé: id apogée, eg 'V1RT', 'V1GE2201', ... Valeur: ApoElt, avec les attributs code, type_objet Si les id Apogée ne sont pas uniques (ce n'est pas garanti), garde le premier """ elts = {} for col_id in sorted(list(cols.keys()), reverse=True): col = cols[col_id] if col["Code"] in elts: elts[col["Code"]].append(col) else: elts[col["Code"]] = ApoElt([col]) return elts # { code apo : ApoElt } def apo_read_section_valeurs(self): "traitement de la section XX-APO_VALEURS-XX" self.column_titles = self._file.readline() self.col_ids = self.column_titles.strip().split() self.csv_etuds = self.apo_read_etuds() def apo_read_etuds(self) -> list[ApoEtudTuple]: """Lecture des étudiants (et résultats) du fichier CSV Apogée. Les lignes "étudiant" commencent toujours par `12345678 NOM PRENOM 15/05/2003` le premier code étant le NIP. """ etud_tuples = [] while True: line = self._file.readline() # cette section est impérativement la dernière du fichier # donc on arrête ici: if not line: break if not line.strip(): continue # silently ignore blank lines line = line.strip(APO_NEWLINE) fields = line.split(APO_SEP) if len(fields) < 4: raise ScoFormatError( """Ligne étudiant invalide (doit commencer par 'NIP NOM PRENOM dd/mm/yyyy')""", filename=self.get_filename(), ) cols = {} # { col_id : value } try: for i, field in enumerate(fields): cols[self.col_ids[i]] = field except IndexError as exc: raise raise ScoFormatError( f"Fichier Apogee incorrect (colonnes excédentaires ? (<tt>{i}/{field}</tt>))", filename=self.get_filename(), ) from exc etud_tuples.append( ApoEtudTuple( nip=fields[0], # id etudiant nom=fields[1], prenom=fields[2], naissance=fields[3], cols=cols, ) # XXX à remettre dans apogee_csv.py # export_res_etape=self.export_res_etape, # export_res_sem=self.export_res_sem, # export_res_ues=self.export_res_ues, # export_res_modules=self.export_res_modules, # export_res_sdj=self.export_res_sdj, # export_res_rat=self.export_res_rat, # ) ) return etud_tuples def _apo_read_titres(self, f) -> dict: "Lecture section TITRES du fichier Apogée, renvoie dict" d = {} while True: line = f.readline().strip( " " + APO_NEWLINE ) # ne retire pas le \t (pour les clés vides) if not line.strip(): # stoppe sur ligne pleines de \t break fields = line.split(APO_SEP) if len(fields) == 2: k, v = fields else: log(f"Error read CSV: \nline={line}\nfields={fields}") log(dir(f)) raise ScoFormatError( f"Fichier Apogee incorrect (section titres, {len(fields)} champs au lieu de 2)", filename=self.get_filename(), ) d[k] = v # if not d.get("apoC_Fichier_Exp", None): raise ScoFormatError( "Fichier Apogee incorrect: pas de titre apoC_Fichier_Exp", filename=self.get_filename(), ) # keep only basename: may be a windows or unix pathname s = d["apoC_Fichier_Exp"].split("/")[-1] s = s.split("\\")[-1] # for DOS paths, eg C:\TEMP\VL4RT_V3ASR.TXT d["apoC_Fichier_Exp"] = s return d def get_filename(self) -> str: """Le nom du fichier APogée, tel qu'indiqué dans le fichier ou vide.""" if self.titles: return self.titles.get("apoC_Fichier_Exp", "") return "" def write(self, apo_etuds: list["ApoEtud"]) -> bytes: """Renvoie le contenu actualisé du fichier Apogée""" f = io.StringIO() self._write_header(f) self._write_etuds(f, apo_etuds) return f.getvalue().encode(APO_OUTPUT_ENCODING) def _write_etuds(self, f, apo_etuds: list["ApoEtud"]): """write apo CSV etuds on f""" for apo_etud in apo_etuds: fields = [] # e['nip'], e['nom'], e['prenom'], e['naissance'] ] for col_id in self.col_ids: try: fields.append(str(apo_etud.new_cols[col_id])) except KeyError: log( f"""Error: {apo_etud["nip"]} {apo_etud["nom"]} missing column key {col_id} Details:\napo_etud = {pprint.pformat(apo_etud)} col_ids={pprint.pformat(self.col_ids)} étudiant ignoré. """ ) f.write(APO_SEP.join(fields) + APO_NEWLINE) def _write_header(self, f): """write apo CSV header on f (beginning of CSV until columns titles just after XX-APO_VALEURS-XX line) """ remove_typ_res = sco_preferences.get_preference("export_res_remove_typ_res") for section, data in self.sections_str.items(): # ne recopie pas la section résultats, et en option supprime APO_TYP_RES if (section != "XX-APO_VALEURS-XX") and ( section != "XX-APO_TYP_RES-XX" or not remove_typ_res ): f.write(data) f.write("XX-APO_VALEURS-XX" + APO_NEWLINE) f.write(self.column_titles) class ApoElt: """Définition d'un Element Apogée sur plusieurs colonnes du fichier CSV """ def __init__(self, cols): assert len(cols) > 0 assert len(set([c["Code"] for c in cols])) == 1 # colonnes de meme code assert len(set([c["Type Objet"] for c in cols])) == 1 # colonnes de meme type self.cols = cols self.code = cols[0]["Code"] self.version = cols[0]["Version"] self.type_objet = cols[0]["Type Objet"] def append(self, col): """ajoute une "colonne" à l'élément""" assert col["Code"] == self.code if col["Type Objet"] != self.type_objet: log( f"""Warning: ApoElt: duplicate id { self.code} ({self.type_objet} and {col["Type Objet"]})""" ) self.type_objet = col["Type Objet"] self.cols.append(col) def __repr__(self): return f"ApoElt(code='{self.code}', cols={pprint.pformat(self.cols)})" def guess_data_encoding(text: bytes, threshold=0.6): """Guess string encoding, using chardet heuristics. Returns encoding, or None if detection failed (confidence below threshold) """ r = chardet_detect(text) if r["confidence"] < threshold: return None else: return r["encoding"] def fix_data_encoding( text: bytes, default_source_encoding=APO_INPUT_ENCODING, dest_encoding=APO_INPUT_ENCODING, ) -> tuple[bytes, str]: """Try to ensure that text is using dest_encoding returns converted text, and a message describing the conversion. Raises UnicodeEncodeError en cas de problème, en général liée à une auto-détection errornée. """ message = "" detected_encoding = guess_data_encoding(text) if not detected_encoding: if default_source_encoding != dest_encoding: message = f"converting from {default_source_encoding} to {dest_encoding}" text = text.decode(default_source_encoding).encode(dest_encoding) else: if detected_encoding != dest_encoding: message = ( f"converting from detected {default_source_encoding} to {dest_encoding}" ) text = text.decode(detected_encoding).encode(dest_encoding) return text, message def _apo_read_typ_res(f) -> str: "Lit la section XX-APO_TYP_RES-XX" text = "XX-APO_TYP_RES-XX" + APO_NEWLINE while True: line = f.readline() stripped_line = line.strip() if not stripped_line: break text += line return text def _apo_next_non_blank_line(f: StringIOWithLineNumber) -> tuple[str, int]: "Ramène prochaine ligne non blanche, stripped, et l'indice de son début" while True: pos = f.tell() line = f.readline() if not line: return "", -1 stripped_line = line.strip() if stripped_line: return stripped_line, pos