ScoDoc/app/scodoc/sco_apogee_csv.py

##############################################################################
#
# 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
#
##############################################################################

"""Exportation des résultats des étudiants vers Apogée.

Ce code a été au départ inspiré par les travaux de Damien Mascré, scodoc2apogee (en Java).

A utiliser en fin de semestre, après les jury.
On communique avec Apogée via des fichiers CSV.

XXX A vérifier: AJAC car 1 sem. validé et pas de NAR

"""

import datetime
from functools import reduce
import functools
import io
import os
import pprint
import re
import time
from zipfile import ZipFile

from flask import send_file
import numpy as np


from app import log
from app.comp import res_sem
from app.comp.res_compat import NotesTableCompat
from app.comp.res_but import ResultatsSemestreBUT
from app.models import (
    ApcValidationAnnee,
    ApcValidationRCUE,
    FormSemestre,
    Identite,
    ScolarFormSemestreValidation,
)

from app.models.config import ScoDocSiteConfig
from app.scodoc.sco_apogee_reader import (
    APO_DECIMAL_SEP,
    ApoCSVReadWrite,
    ApoEtudTuple,
)
import app.scodoc.sco_utils as scu
from app.scodoc.sco_exceptions import ScoValueError, ScoFormatError
from app.scodoc.gen_tables import GenTable
from app.scodoc.sco_vdi import ApoEtapeVDI
from app.scodoc.codes_cursus import code_semestre_validant
from app.scodoc.codes_cursus import (
    ADSUP,
    DEF,
    DEM,
    NAR,
    RAT,
)
from app.scodoc import sco_cursus
from app.scodoc import sco_formsemestre
from app.scodoc import sco_etud


def _apo_fmt_note(note, fmt="%3.2f"):
    "Formatte une note pour Apogée (séparateur décimal: ',')"
    # if not note and isinstance(note, float): changé le 31/1/2022, étrange ?
    #    return ""
    try:
        val = float(note)
    except ValueError:
        return ""
    if np.isnan(val):
        return ""
    return (fmt % val).replace(".", APO_DECIMAL_SEP)


class EtuCol:
    """Valeurs colonnes d'un element pour un etudiant"""

    def __init__(self, nip, apo_elt, init_vals):
        pass  # XXX


ETUD_OK = "ok"
ETUD_ORPHELIN = "orphelin"
ETUD_NON_INSCRIT = "non_inscrit"

VOID_APO_RES = dict(N="", B="", J="", R="", M="")


class ApoEtud(dict):
    """Étudiant Apogee:"""

    def __init__(
        self,
        apo_etud_tuple: ApoEtudTuple,
        export_res_etape=True,
        export_res_sem=True,
        export_res_ues=True,
        export_res_modules=True,
        export_res_sdj=True,
        export_res_rat=True,
    ):
        self["nip"] = apo_etud_tuple.nip
        self["nom"] = apo_etud_tuple.nom
        self["prenom"] = apo_etud_tuple.prenom
        self["naissance"] = apo_etud_tuple.naissance
        self.cols = apo_etud_tuple.cols
        "{ col_id : value }  colid = 'apoL_c0001'"
        self.is_apc = None
        "Vrai si BUT"
        self.col_elts = {}
        "{'V1RT': {'R': 'ADM', 'J': '', 'B': 20, 'N': '12.14'}}"
        self.etud: Identite = None
        "etudiant ScoDoc associé"
        self.etat = None  # ETUD_OK, ...
        self.is_nar = False
        "True si NARé dans un semestre"
        self.log = []
        self.has_logged_no_decision = False
        self.export_res_etape = export_res_etape  # VET, ...
        self.export_res_sem = export_res_sem  # elt_sem_apo
        self.export_res_ues = export_res_ues
        self.export_res_modules = export_res_modules
        self.export_res_sdj = export_res_sdj
        "export meme si pas de decision de jury"
        self.export_res_rat = export_res_rat
        self.fmt_note = functools.partial(
            _apo_fmt_note, fmt=ScoDocSiteConfig.get_code_apo("NOTES_FMT") or "3.2f"
        )
        # Initialisés par associate_sco:
        self.autre_sem: dict = None
        self.autre_res: NotesTableCompat = None
        self.cur_sem: dict = None
        self.cur_res: NotesTableCompat = None
        self.new_cols = {}
        "{ col_id : value to record in csv }"

        # Pour le BUT:
        self.validation_annee_but: ApcValidationAnnee = None
        "validation de jury annuelle BUT, ou None"

    def __repr__(self):
        return f"""ApoEtud( nom='{self["nom"]}', nip='{self["nip"]}' )"""

    def lookup_scodoc(self, etape_formsemestre_ids):
        """Cherche l'étudiant ScoDoc associé à cet étudiant Apogée.
        S'il n'est pas trouvé (état "orphelin", dans Apo mais pas chez nous),
        met .etud à None.
        Sinon, cherche le semestre, et met l'état à ETUD_OK ou ETUD_NON_INSCRIT.
        """

        # futur: #WIP
        # etud: Identite = Identite.query.filter_by(code_nip=self["nip"], dept_id=g.scodoc_dept_id).first()
        # self.etud = etud
        etuds = sco_etud.get_etud_info(code_nip=self["nip"], filled=True)
        if not etuds:
            # pas dans ScoDoc
            self.etud = None
            self.log.append("non inscrit dans ScoDoc")
            self.etat = ETUD_ORPHELIN
        else:
            # futur: #WIP
            # formsemestre_ids = {
            #    ins.formsemestre_id for ins in etud.formsemestre_inscriptions
            # }
            # in_formsemestre_ids = formsemestre_ids.intersection(etape_formsemestre_ids)
            self.etud = etuds[0]
            # cherche le semestre ScoDoc correspondant à l'un de ceux de l'etape:
            formsemestre_ids = {s["formsemestre_id"] for s in self.etud["sems"]}
            in_formsemestre_ids = formsemestre_ids.intersection(etape_formsemestre_ids)
            if not in_formsemestre_ids:
                self.log.append(
                    "connu dans ScoDoc, mais pas inscrit dans un semestre de cette étape"
                )
                self.etat = ETUD_NON_INSCRIT
            else:
                self.etat = ETUD_OK

    def associate_sco(self, apo_data: "ApoData"):
        """Recherche les valeurs des éléments Apogée pour cet étudiant
        Set .new_cols
        """
        self.col_elts = {}  # {'V1RT': {'R': 'ADM', 'J': '', 'B': 20, 'N': '12.14'}}
        if self.etat is None:
            self.lookup_scodoc(apo_data.etape_formsemestre_ids)
        if self.etat != ETUD_OK:
            self.new_cols = (
                self.cols
            )  # etudiant inconnu, recopie les valeurs existantes dans Apo
        else:
            sco_elts = {}  # valeurs trouvées dans ScoDoc   code : { N, B, J, R }
            for col_id in apo_data.apo_csv.col_ids[4:]:
                code = apo_data.apo_csv.cols[col_id]["Code"]  # 'V1RT'
                elt = sco_elts.get(code, None)
                # elt est {'R': ADM, 'J': '', 'B': 20, 'N': '12.14'}
                if elt is None:  # pas déjà trouvé
                    self.etud_set_semestres_de_etape(apo_data)
                    for sem in apo_data.sems_etape:
                        elt = self.search_elt_in_sem(code, sem)
                        if elt is not None:
                            sco_elts[code] = elt
                            break
                self.col_elts[code] = elt
                if elt is None:
                    try:
                        self.new_cols[col_id] = self.cols[col_id]
                    except KeyError as exc:
                        raise ScoFormatError(
                            f"""Fichier Apogee invalide : ligne mal formatée ? <br>colonne <tt>{col_id}</tt> non déclarée ?"""
                        ) from exc
                else:
                    try:
                        self.new_cols[col_id] = sco_elts[code][
                            apo_data.apo_csv.cols[col_id]["Type Rés."]
                        ]
                    except KeyError as exc:
                        log(
                            f"""associate_sco: missing key, etud={self}\ncode='{
                                code}'\netape='{apo_data.etape_apogee}'"""
                        )
                        raise ScoValueError(
                            f"""L'élément {code} n'a pas de résultat: peut-être une erreur
                            dans les codes sur le programme pédagogique
                            (vérifier qu'il est bien associé à une UE ou semestre)?"""
                        ) from exc
            # recopie les 4 premieres colonnes (nom, ..., naissance):
            for col_id in apo_data.apo_csv.col_ids[:4]:
                self.new_cols[col_id] = self.cols[col_id]

    # def unassociated_codes(self, apo_data):
    #     "list of apo elements for this student without a value in ScoDoc"
    #     codes = set([apo_data.apo_csv.cols[col_id].code for col_id in apo_data.apo_csv.col_ids])
    #     return codes - set(sco_elts)

    def search_elt_in_sem(self, code, sem) -> dict:
        """
        VET code jury etape (en BUT, le code annuel)
        ELP élément pédagogique: UE, module
        Autres éléments: résultats du semestre ou de l'année scolaire:
        => VRTW1: code additionnel au semestre ("code élement semestre", elt_sem_apo)
        => VRT1A: le même que le VET: ("code élement annuel", elt_annee_apo)
        Attention, si le semestre couvre plusieurs étapes, indiquer les codes des éléments,
        séparés par des virgules.

        Args:
           code (str): code apo de l'element cherché
           sem (dict): semestre dans lequel on cherche l'élément
           cur_sem (dict): semestre "courant" pour résultats annuels (VET)
           autre_sem (dict): autre semestre utilisé pour calculer les résultats annuels (VET)

        Returns:
           dict: with N, B, J, R keys, ou None si elt non trouvé
        """
        etudid = self.etud["etudid"]
        if not self.cur_res:
            log("search_elt_in_sem: no cur_res !")
            return None
        if sem["formsemestre_id"] == self.cur_res.formsemestre.id:
            res = self.cur_res
        elif (
            self.autre_res and sem["formsemestre_id"] == self.autre_res.formsemestre.id
        ):
            res = self.autre_res
        else:
            formsemestre = FormSemestre.query.get_or_404(sem["formsemestre_id"])
            res: NotesTableCompat = res_sem.load_formsemestre_results(formsemestre)

        if etudid not in res.identdict:
            return None  # etudiant non inscrit dans ce semestre

        if not self.export_res_sdj and not res.etud_has_decision(etudid):
            # pas de decision de jury, on n'enregistre rien
            # (meme si démissionnaire)
            if not self.has_logged_no_decision:
                self.log.append("Pas de décision (export désactivé)")
                self.has_logged_no_decision = True
            return VOID_APO_RES

        if res.is_apc:  # export BUT
            self._but_load_validation_annuelle()
        else:
            decision = res.get_etud_decision_sem(etudid)
            if decision and decision["code"] == NAR:
                self.is_nar = True
            # Element semestre: (non BUT donc)
            if code in {x.strip() for x in sem["elt_sem_apo"].split(",")}:
                if self.export_res_sem:
                    return self.comp_elt_semestre(res, decision, etudid)
                else:
                    return VOID_APO_RES

        # Element etape (annuel ou non):
        if sco_formsemestre.sem_has_etape(sem, code) or (
            code in {x.strip() for x in sem["elt_annee_apo"].split(",")}
        ):
            export_res_etape = self.export_res_etape
            if (not export_res_etape) and self.cur_sem:
                # exporte toujours le résultat de l'étape si l'étudiant est diplômé
                Se = sco_cursus.get_situation_etud_cursus(
                    self.etud, self.cur_sem["formsemestre_id"]
                )
                export_res_etape = Se.all_other_validated()

            if export_res_etape:
                return self.comp_elt_annuel(etudid)
            else:
                self.log.append("export étape désactivé")
                return VOID_APO_RES

        # Elements UE
        decisions_ue = res.get_etud_decisions_ue(etudid)
        for ue in res.get_ues_stat_dict():
            if ue["code_apogee"] and code in {
                x.strip() for x in ue["code_apogee"].split(",")
            }:
                if self.export_res_ues:
                    if (
                        decisions_ue and ue["ue_id"] in decisions_ue
                    ) or self.export_res_sdj:
                        ue_status = res.get_etud_ue_status(etudid, ue["ue_id"])
                        if decisions_ue and ue["ue_id"] in decisions_ue:
                            code_decision_ue = decisions_ue[ue["ue_id"]]["code"]
                            code_decision_ue_apo = ScoDocSiteConfig.get_code_apo(
                                code_decision_ue
                            )
                        else:
                            code_decision_ue_apo = ""
                        return dict(
                            N=self.fmt_note(ue_status["moy"] if ue_status else ""),
                            B=20,
                            J="",
                            R=code_decision_ue_apo,
                            M="",
                        )
                    else:
                        return VOID_APO_RES
                else:
                    return VOID_APO_RES

        # Elements Modules
        modimpls = res.get_modimpls_dict()
        module_code_found = False
        for modimpl in modimpls:
            module = modimpl["module"]
            if (
                res.modimpl_inscr_df[modimpl["moduleimpl_id"]][etudid]
                and module["code_apogee"]
                and code in {x.strip() for x in module["code_apogee"].split(",")}
            ):
                n = res.get_etud_mod_moy(modimpl["moduleimpl_id"], etudid)
                if n != "NI" and self.export_res_modules:
                    return dict(N=self.fmt_note(n), B=20, J="", R="")
                else:
                    module_code_found = True

        if module_code_found:
            return VOID_APO_RES
        #
        return None  # element Apogee non trouvé dans ce semestre

    def comp_elt_semestre(self, nt: NotesTableCompat, decision: dict, etudid: int):
        """Calcul résultat apo semestre.
        Toujours vide pour en BUT/APC.
        """
        if self.is_apc:  # garde fou: pas de code semestre en APC !
            return dict(N="", B=20, J="", R="", M="")
        if decision is None:
            etud = Identite.get_etud(etudid)
            nomprenom = etud.nomprenom if etud else "(inconnu)"
            raise ScoValueError(
                f"decision absente pour l'étudiant {nomprenom} ({etudid})"
            )
        # resultat du semestre
        decision_apo = ScoDocSiteConfig.get_code_apo(decision["code"])
        note = nt.get_etud_moy_gen(etudid)
        if decision_apo == "DEF" or decision["code"] == DEM or decision["code"] == DEF:
            note_str = "0,01"  # note non nulle pour les démissionnaires
        else:
            note_str = self.fmt_note(note)
        return dict(N=note_str, B=20, J="", R=decision_apo, M="")

    def comp_elt_annuel(self, etudid):
        """Calcul resultat annuel (VET) à partir du semestre courant
        et de l'autre (le suivant ou le précédent complétant l'année scolaire)
        En BUT, c'est la décision de jury annuelle (ApcValidationAnnee).
        """
        # Code annuel:
        #  - Note: moyenne des moyennes générales des deux semestres
        #        (pas vraiment de sens, mais faute de mieux)
        #    on pourrait aussi bien prendre seulement la note du dernier semestre (S2 ou S4).
        #    XXX APOBUT: à modifier pour prendre moyenne indicative annuelle ? non
        #
        #  - Résultat jury:
        #      si l'autre est validé, code du semestre courant (ex: S1 (ADM), S2 (AJ) => année AJ)
        #      si l'autre n'est pas validé ou est DEF ou DEM, code de l'autre
        #
        #    XXX cette règle est discutable, à valider

        # log('comp_elt_annuel cur_sem=%s autre_sem=%s' % (cur_sem['formsemestre_id'], autre_sem['formsemestre_id']))
        if not self.cur_sem:
            # l'étudiant n'a pas de semestre courant ?!
            self.log.append("pas de semestre courant")
            log(f"comp_elt_annuel: etudid {etudid} has no cur_sem")
            return VOID_APO_RES

        if self.is_apc:
            cur_decision = {}  # comp_elt_semestre sera vide.
        else:
            # Non BUT
            cur_decision = self.cur_res.get_etud_decision_sem(etudid)
            if not cur_decision:
                # pas de decision => pas de résultat annuel
                return VOID_APO_RES

            if (cur_decision["code"] == RAT) and not self.export_res_rat:
                # ne touche pas aux RATs
                return VOID_APO_RES

        if not self.autre_sem:
            # formations monosemestre, ou code VET semestriel,
            # ou jury intermediaire et etudiant non redoublant...
            return self.comp_elt_semestre(self.cur_res, cur_decision, etudid)

        # --- Traite le BUT à part:
        if self.is_apc:
            return self.comp_elt_annuel_apc()
        # --- Formations classiques
        decision_apo = ScoDocSiteConfig.get_code_apo(cur_decision["code"])
        autre_decision = self.autre_res.get_etud_decision_sem(etudid)
        if not autre_decision:
            # pas de decision dans l'autre => pas de résultat annuel
            return VOID_APO_RES
        autre_decision_apo = ScoDocSiteConfig.get_code_apo(autre_decision["code"])
        if (
            autre_decision_apo == "DEF"
            or autre_decision["code"] == DEM
            or autre_decision["code"] == DEF
        ) or (
            decision_apo == "DEF"
            or cur_decision["code"] == DEM
            or cur_decision["code"] == DEF
        ):
            note_str = "0,01"  # note non nulle pour les démissionnaires
        else:
            note = self.cur_res.get_etud_moy_gen(etudid)
            autre_note = self.autre_res.get_etud_moy_gen(etudid)
            # print 'note=%s autre_note=%s' % (note, autre_note)
            try:
                moy_annuelle = (note + autre_note) / 2
            except TypeError:
                moy_annuelle = ""
            note_str = self.fmt_note(moy_annuelle)

        if code_semestre_validant(autre_decision["code"]):
            decision_apo_annuelle = decision_apo
        else:
            decision_apo_annuelle = autre_decision_apo

        return dict(N=note_str, B=20, J="", R=decision_apo_annuelle, M="")

    def comp_elt_annuel_apc(self):
        """L'élément Apo pour un résultat annuel BUT.
        self.cur_res == résultats du semestre sur lequel a été appelé l'export.
        """
        if not self.validation_annee_but:
            # pas de décision ou pas de sem. impair
            return VOID_APO_RES

        return dict(
            N="",  # n'exporte pas de moyenne indicative annuelle, car pas de définition officielle
            B=20,
            J="",
            R=ScoDocSiteConfig.get_code_apo(self.validation_annee_but.code),
            M="",
        )

    def _but_load_validation_annuelle(self):
        """charge la validation de jury BUT annuelle.
        Ici impose qu'elle soit issue d'un semestre de l'année en cours
        (pas forcément nécessaire, voir selon les retours des équipes ?)
        """
        # le semestre impair de l'année scolaire
        if self.cur_res.formsemestre.semestre_id % 2:
            formsemestre = self.cur_res.formsemestre
        elif (
            self.autre_res
            and self.autre_res.formsemestre.annee_scolaire()
            == self.cur_res.formsemestre.annee_scolaire()
        ):
            formsemestre = self.autre_res.formsemestre
            assert formsemestre.semestre_id % 2
        else:
            # ne trouve pas de semestre impair
            self.validation_annee_but = None
            return
        self.validation_annee_but: ApcValidationAnnee = (
            ApcValidationAnnee.query.filter_by(
                formsemestre_id=formsemestre.id,
                etudid=self.etud["etudid"],
                referentiel_competence_id=self.cur_res.formsemestre.formation.referentiel_competence_id,
            ).first()
        )
        self.is_nar = (
            self.validation_annee_but and self.validation_annee_but.code == NAR
        )

    def etud_set_semestres_de_etape(self, apo_data: "ApoData"):
        """Set .cur_sem and .autre_sem et charge les résultats.
        Lorsqu'on a une formation semestrialisée mais avec un code étape annuel,
        il faut considérer les deux semestres ((S1,S2) ou (S3,S4)) pour calculer
        le code annuel (VET ou VRT1A (voir elt_annee_apo)).

        Pour les jurys intermediaires (janvier, S1 ou S3):  (S2 ou S4) de la même
        étape lors d'une année précédente ?

        Set cur_sem: le semestre "courant" et autre_sem, ou None s'il n'y en a pas.
        """
        # Cherche le semestre "courant":
        cur_sems = [
            sem
            for sem in self.etud["sems"]
            if (
                (sem["semestre_id"] == apo_data.cur_semestre_id)
                and (apo_data.etape in sem["etapes"])
                and (
                    sco_formsemestre.sem_in_semestre_scolaire(
                        sem,
                        apo_data.annee_scolaire,
                        0,  # annee complete
                    )
                )
            )
        ]
        if not cur_sems:
            cur_sem = None
        else:
            # prend le plus recent avec decision
            cur_sem = None
            for sem in cur_sems:
                formsemestre = FormSemestre.query.get_or_404(sem["formsemestre_id"])
                res: NotesTableCompat = res_sem.load_formsemestre_results(formsemestre)
                has_decision = res.etud_has_decision(self.etud["etudid"])
                if has_decision:
                    cur_sem = sem
                    self.cur_res = res
                    break
            if cur_sem is None:
                cur_sem = cur_sems[0]  # aucun avec décision, prend le plus recent
                if res.formsemestre.id == cur_sem["formsemestre_id"]:
                    self.cur_res = res
                else:
                    formsemestre = FormSemestre.query.get_or_404(
                        cur_sem["formsemestre_id"]
                    )
                    self.cur_res = res_sem.load_formsemestre_results(formsemestre)

        self.cur_sem = cur_sem

        if apo_data.cur_semestre_id <= 0:
            # "autre_sem" non pertinent pour sessions sans semestres:
            self.autre_sem = None
            self.autre_res = None
            return

        if apo_data.jury_intermediaire:  # jury de janvier
            # Le semestre suivant: exemple 2 si on est en jury de S1
            autre_semestre_id = apo_data.cur_semestre_id + 1
        else:
            # Le précédent (S1 si on est en S2)
            autre_semestre_id = apo_data.cur_semestre_id - 1

        # L'autre semestre DOIT être antérieur au courant indiqué par apo_data
        if apo_data.periode is not None:
            if apo_data.periode == 1:
                courant_annee_debut = apo_data.annee_scolaire
                courant_mois_debut = 9  # periode = 1 (sept-jan)
            elif apo_data.periode == 2:
                courant_annee_debut = apo_data.annee_scolaire + 1
                courant_mois_debut = 1  # ou 2 (fev-jul)
            else:
                raise ValueError("invalid periode value !")  # bug ?
            courant_date_debut = "%d-%02d-01" % (
                courant_annee_debut,
                courant_mois_debut,
            )
        else:
            courant_date_debut = "9999-99-99"

        # etud['sems'] est la liste des semestres de l'étudiant, triés par date,
        # le plus récemment effectué en tête.
        # Cherche les semestres (antérieurs) de l'indice autre de la même étape apogée
        # s'il y en a plusieurs, choisit le plus récent ayant une décision

        autres_sems = []
        for sem in self.etud["sems"]:
            if (
                sem["semestre_id"] == autre_semestre_id
                and apo_data.etape_apogee in sem["etapes"]
            ):
                if (
                    sem["date_debut_iso"] < courant_date_debut
                ):  # on demande juste qu'il ait démarré avant
                    autres_sems.append(sem)
        if not autres_sems:
            autre_sem = None
        elif len(autres_sems) == 1:
            autre_sem = autres_sems[0]
        else:
            autre_sem = None
            for sem in autres_sems:
                formsemestre = FormSemestre.query.get_or_404(sem["formsemestre_id"])
                res: NotesTableCompat = res_sem.load_formsemestre_results(formsemestre)
                if res.is_apc:
                    has_decision = res.etud_has_decision(self.etud["etudid"])
                else:
                    has_decision = res.get_etud_decision_sem(self.etud["etudid"])
                if has_decision:
                    autre_sem = sem
                    break
            if autre_sem is None:
                autre_sem = autres_sems[0]  # aucun avec decision, prend le plus recent

        self.autre_sem = autre_sem
        # Charge les résultats:
        if autre_sem:
            formsemestre = FormSemestre.query.get_or_404(autre_sem["formsemestre_id"])
            self.autre_res = res_sem.load_formsemestre_results(formsemestre)
        else:
            self.autre_res = None


class ApoData:
    def __init__(
        self,
        data: str,
        periode=None,
        export_res_etape=True,
        export_res_sem=True,
        export_res_ues=True,
        export_res_modules=True,
        export_res_sdj=True,
        export_res_rat=True,
        orig_filename=None,
    ):
        """Lecture du fichier CSV Apogée
        Regroupe les élements importants d'un fichier CSV Apogée
        periode = 1 (sept-jan) ou 2 (fev-jul), mais cette info n'est pas
         (toujours) présente dans les CSV Apogée et doit être indiquée par l'utilisateur
        Laisser periode à None si etape en 1 semestre (LP, décalés, ...)
        """
        self.export_res_etape = export_res_etape  # VET, ...
        self.export_res_sem = export_res_sem  # elt_sem_apo
        self.export_res_ues = export_res_ues
        self.export_res_modules = export_res_modules
        self.export_res_sdj = export_res_sdj
        self.export_res_rat = export_res_rat
        self.orig_filename = orig_filename
        self.periode = periode  #
        "1 sem. sept-jan, 2 sem. fev-jul. 0 si étape en 1 seul semestre."
        self.is_apc = None
        "Vrai si BUT"
        try:
            self.apo_csv = ApoCSVReadWrite(data)
        except ScoFormatError as e:
            # enrichit le message d'erreur
            filename = self.orig_filename or e.filename
            raise ScoFormatError(
                f"""<h3>Erreur lecture du fichier Apogée <tt>{filename}</tt></h3>
                <p>{e.args[0]}</p>"""
            ) from e
        self.etape_apogee = self.get_etape_apogee()  #  'V1RT'
        self.vdi_apogee = self.get_vdi_apogee()  # '111'
        self.etape = ApoEtapeVDI(etape=self.etape_apogee, vdi=self.vdi_apogee)
        self.cod_dip_apogee = self.get_cod_dip_apogee()
        self.annee_scolaire = self.get_annee_scolaire()
        self.jury_intermediaire = (
            False  # True si jury à mi-étape, eg jury de S1 dans l'étape (S1, S2)
        )
        # Crée les étudiants
        self.etuds = [
            ApoEtud(
                apo_etud_tuple,
                export_res_etape=export_res_etape,
                export_res_sem=export_res_sem,
                export_res_ues=export_res_ues,
                export_res_modules=export_res_modules,
                export_res_sdj=export_res_sdj,
                export_res_rat=export_res_rat,
            )
            for apo_etud_tuple in self.apo_csv.csv_etuds
        ]
        self.etud_by_nip = {apo_etud["nip"]: apo_etud for apo_etud in self.etuds}
        log(f"ApoData( periode={self.periode}, annee_scolaire={self.annee_scolaire} )")

    def setup(self):
        """Recherche semestres ScoDoc concernés"""
        self.sems_etape = comp_apo_sems(self.etape_apogee, self.annee_scolaire)
        if not self.sems_etape:
            raise ScoValueError("aucun semestre trouvé !")
        self.formsemestres_etape = [
            FormSemestre.query.get_or_404(s["formsemestre_id"]) for s in self.sems_etape
        ]
        apcs = {
            formsemestre.formation.is_apc() for formsemestre in self.formsemestres_etape
        }
        if len(apcs) != 1:
            raise ScoValueError(
                "l'ensemble mixe des semestres BUT (APC) et des semestres classiques !"
            )
        self.is_apc = apcs.pop()
        self.etape_formsemestre_ids = {s["formsemestre_id"] for s in self.sems_etape}
        if self.periode is not None:
            self.sems_periode = [
                s
                for s in self.sems_etape
                if (s["periode"] == self.periode) or s["semestre_id"] < 0
            ]
            if not self.sems_periode:
                log("** Warning: ApoData.setup: sems_periode is empty")
                log(
                    "**  (periode=%s, sems_etape [periode]=%s)"
                    % (self.periode, [s["periode"] for s in self.sems_etape])
                )
                self.sems_periode = None
                self.cur_semestre_id = -1  # ?
            else:
                self.cur_semestre_id = self.sems_periode[0]["semestre_id"]
                # Les semestres de la période ont le même indice, n'est-ce pas ?
                if not all(
                    self.cur_semestre_id == s["semestre_id"] for s in self.sems_periode
                ):
                    # debugging information
                    log(
                        f"""*** ApoData.set() error !
                        ApoData( periode={self.periode}, annee_scolaire={self.annee_scolaire
                        }, cur_semestre_id={self.cur_semestre_id} )
                        {len(self.sems_periode)} semestres dans la periode:
                        """
                    )
                    for s in self.sems_periode:
                        log(pprint.pformat(s))

                    raise ScoValueError(
                        f"""Incohérence détectée !

                        Les semestres de la période n'ont pas tous le même indice.

                        Période: {self.periode}. Indice courant: {self.cur_semestre_id}

                        (au besoin, contacter l'assistance sur {scu.SCO_DISCORD_ASSISTANCE})
                        """
                    )
            # Cette condition sera inadaptée si semestres décalés
            # (mais ils n'ont pas d'étape annuelle, espérons!)
            if self.cur_semestre_id >= 0:  # non pertinent pour sessions sans semestres
                self.jury_intermediaire = (self.cur_semestre_id % 2) != 0
        else:
            self.sems_periode = None

    def get_etape_apogee(self) -> str:
        """Le code etape: 'V1RT', donné par le code de l'élément VET"""
        for elt in self.apo_csv.apo_elts.values():
            if elt.type_objet == "VET":
                return elt.code
        raise ScoValueError("Pas de code etape Apogee (manque élément VET)")

    def get_vdi_apogee(self) -> str:
        """le VDI (version de diplôme), stocké dans l'élément VET
        (note: on pourrait peut-être aussi bien le récupérer dans
        l'en-tête XX-APO_TITRES-XX apoC_cod_vdi)
        """
        for elt in self.apo_csv.apo_elts.values():
            if elt.type_objet == "VET":
                return elt.version
        raise ScoValueError("Pas de VDI Apogee (manque élément VET)")

    def get_cod_dip_apogee(self) -> str:
        """Le code diplôme, indiqué dans l'en-tête de la maquette
        exemple: VDTRT
        Retourne '' si absent.
        """
        return self.apo_csv.titles.get("apoC_cod_dip", "")

    def get_annee_scolaire(self) -> int:
        """Annee scolaire du fichier Apogee: un integer
        = annee du mois de septembre de début
        """
        m = re.match("[12][0-9]{3}", self.apo_csv.titles["apoC_annee"])
        if not m:
            raise ScoFormatError(
                f"""Annee scolaire (apoC_annee) invalide: "{self.apo_csv.titles["apoC_annee"]}" """
            )
        return int(m.group(0))

    def list_unknown_elements(self) -> list[str]:
        """Liste des codes des elements Apogee non trouvés dans ScoDoc
        (après traitement de tous les étudiants)
        """
        codes = set()
        for apo_etud in self.etuds:
            codes.update(
                {code for code in apo_etud.col_elts if apo_etud.col_elts[code] is None}
            )
        codes_list = list(codes)
        codes_list.sort()
        return codes_list

    def list_elements(self) -> tuple[set[str], set[str]]:
        """Liste les codes des elements Apogée de la maquette
        et ceux des semestres ScoDoc associés
        Retourne deux ensembles
        """
        try:
            maq_elems = {
                self.apo_csv.cols[col_id]["Code"] for col_id in self.apo_csv.col_ids[4:]
            }
        except KeyError as exc:
            # une colonne déclarée dans l'en-tête n'est pas présente
            declared = self.apo_csv.col_ids[4:]  # id des colones dans l'en-tête
            present = sorted(self.apo_csv.cols.keys())  # colonnes présentes
            log("Fichier Apogee invalide:")
            log(f"Colonnes declarees: {declared}")
            log(f"Colonnes presentes: {present}")
            raise ScoFormatError(
                f"""Fichier Apogee invalide<br>Colonnes declarees: <tt>{declared}</tt>
            <br>Colonnes presentes: <tt>{present}</tt>"""
            ) from exc
        # l'ensemble de tous les codes des elements apo des semestres:
        sem_elems = reduce(set.union, list(self.get_codes_by_sem().values()), set())

        return maq_elems, sem_elems

    def get_codes_by_sem(self) -> dict[int, set[str]]:
        """Pour chaque semestre associé, donne l'ensemble des codes de cette maquette Apogée
        qui s'y trouvent (dans le semestre, les UE ou les modules).
        Return: { formsemestre_id : { 'code1', 'code2', ... }}
        """
        codes_by_sem = {}
        for sem in self.sems_etape:
            formsemestre: FormSemestre = FormSemestre.query.get_or_404(
                sem["formsemestre_id"]
            )
            # L'ensemble des codes apo associés aux éléments:
            codes_semestre = formsemestre.get_codes_apogee()
            codes_modules = set().union(
                *[
                    modimpl.module.get_codes_apogee()
                    for modimpl in formsemestre.modimpls
                ]
            )
            codes_ues = set().union(
                *[ue.get_codes_apogee() for ue in formsemestre.get_ues(with_sport=True)]
            )
            s = set()
            codes_by_sem[sem["formsemestre_id"]] = s
            for col_id in self.apo_csv.col_ids[4:]:
                code = self.apo_csv.cols[col_id]["Code"]  # 'V1RT'
                # associé à l'étape, l'année ou le semestre:
                if code in codes_semestre:
                    s.add(code)
                    continue
                # associé à une UE:
                if code in codes_ues:
                    s.add(code)
                    continue
                # associé à un module:
                if code in codes_modules:
                    s.add(code)
        # log('codes_by_sem=%s' % pprint.pformat(codes_by_sem))
        return codes_by_sem

    def build_cr_table(self):
        """Table compte rendu des décisions"""
        rows = []  # tableau compte rendu des decisions
        for apo_etud in self.etuds:
            cr = {
                "NIP": apo_etud["nip"],
                "nom": apo_etud["nom"],
                "prenom": apo_etud["prenom"],
                "est_NAR": apo_etud.is_nar,
                "commentaire": "; ".join(apo_etud.log),
            }
            if apo_etud.col_elts and apo_etud.col_elts[self.etape_apogee] is not None:
                cr["etape"] = apo_etud.col_elts[self.etape_apogee].get("R", "")
                cr["etape_note"] = apo_etud.col_elts[self.etape_apogee].get("N", "")
            else:
                cr["etape"] = ""
                cr["etape_note"] = ""
            rows.append(cr)

        columns_ids = ["NIP", "nom", "prenom"]
        columns_ids.extend(("etape", "etape_note", "est_NAR", "commentaire"))

        T = GenTable(
            columns_ids=columns_ids,
            titles=dict(zip(columns_ids, columns_ids)),
            rows=rows,
            xls_sheet_name="Decisions ScoDoc",
        )
        return T

    def build_adsup_table(self):
        """Construit une table listant les ADSUP émis depuis les formsemestres
        NIP nom prenom nom_formsemestre etape UE
        """
        validations_ues, validations_rcue = self.list_adsup()
        rows = [
            {
                "code_nip": v.etud.code_nip,
                "nom": v.etud.nom,
                "prenom": v.etud.prenom,
                "formsemestre": v.formsemestre.titre_formation(with_sem_idx=1),
                "etape": v.formsemestre.etapes_apo_str(),
                "ue": v.ue.acronyme,
            }
            for v in validations_ues
        ]
        rows += [
            {
                "code_nip": v.etud.code_nip,
                "nom": v.etud.nom,
                "prenom": v.etud.prenom,
                "formsemestre": v.formsemestre.titre_formation(with_sem_idx=1),
                "etape": "",  # on ne sait pas à quel étape rattacher le RCUE
                "rcue": f"{v.ue1.acronyme}/{v.ue2.acronyme}",
            }
            for v in validations_rcue
        ]

        return GenTable(
            columns_ids=(
                "code_nip",
                "nom",
                "prenom",
                "formsemestre",
                "etape",
                "ue",
                "rcue",
            ),
            titles={
                "code_nip": "NIP",
                "nom": "Nom",
                "prenom": "Prénom",
                "formsemestre": "Semestre",
                "etape": "Etape",
                "ue": "UE",
                "rcue": "RCUE",
            },
            rows=rows,
            xls_sheet_name="ADSUPs",
        )

    def list_adsup(
        self,
    ) -> tuple[list[ScolarFormSemestreValidation], list[ApcValidationRCUE]]:
        """Liste les validations ADSUP émises par des formsemestres de cet ensemble"""
        validations_ues = (
            ScolarFormSemestreValidation.query.filter_by(code=ADSUP)
            .filter(ScolarFormSemestreValidation.ue_id != None)
            .filter(
                ScolarFormSemestreValidation.formsemestre_id.in_(
                    self.etape_formsemestre_ids
                )
            )
        )
        validations_rcue = ApcValidationRCUE.query.filter_by(code=ADSUP).filter(
            ApcValidationRCUE.formsemestre_id.in_(self.etape_formsemestre_ids)
        )
        return validations_ues, validations_rcue


def comp_apo_sems(etape_apogee, annee_scolaire: int) -> list[dict]:
    """
    :param etape_apogee: etape (string or ApoEtapeVDI)
    :param annee_scolaire: annee (int)
    :return: list of sems for etape_apogee in annee_scolaire
    """
    return sco_formsemestre.list_formsemestre_by_etape(
        etape_apo=str(etape_apogee), annee_scolaire=annee_scolaire
    )


def nar_etuds_table(apo_data, nar_etuds):
    """Liste les NAR -> excel table"""
    code_etape = apo_data.etape_apogee
    today = datetime.datetime.today().strftime(scu.DATE_FMT)
    rows = []
    nar_etuds.sort(key=lambda k: k["nom"])
    for e in nar_etuds:
        rows.append(
            {
                "nom": e["nom"],
                "prenom": e["prenom"],
                "c0": "",
                "c1": "AD",
                "etape": code_etape,
                "c3": "",
                "c4": "",
                "c5": "",
                "c6": "N",
                "c7": "",
                "c8": "",
                "NIP": e["nip"],
                "c10": "",
                "c11": "",
                "c12": "",
                "c13": "NAR - Jury",
                "date": today,
            }
        )

    columns_ids = (
        "NIP",
        "nom",
        "prenom",
        "etape",
        "c0",
        "c1",
        "c3",
        "c4",
        "c5",
        "c6",
        "c7",
        "c8",
        "c10",
        "c11",
        "c12",
        "c13",
        "date",
    )
    table = GenTable(
        columns_ids=columns_ids,
        titles=dict(zip(columns_ids, columns_ids)),
        rows=rows,
        xls_sheet_name="NAR ScoDoc",
    )
    return table.excel()


def export_csv_to_apogee(
    apo_csv_data: str,
    periode=None,
    dest_zip=None,
    export_res_etape=True,
    export_res_sem=True,
    export_res_ues=True,
    export_res_modules=True,
    export_res_sdj=True,
    export_res_rat=True,
):
    """Genere un fichier CSV Apogée
    à partir d'un fichier CSV Apogée vide (ou partiellement rempli)
    et des résultats ScoDoc.
    Si dest_zip, ajoute les fichiers générés à ce zip
    sinon crée un zip et le publie
    """
    apo_data = ApoData(
        apo_csv_data,
        periode=periode,
        export_res_etape=export_res_etape,
        export_res_sem=export_res_sem,
        export_res_ues=export_res_ues,
        export_res_modules=export_res_modules,
        export_res_sdj=export_res_sdj,
        export_res_rat=export_res_rat,
    )
    apo_data.setup()  # -> .sems_etape
    apo_csv = apo_data.apo_csv
    for apo_etud in apo_data.etuds:
        apo_etud.is_apc = apo_data.is_apc
        apo_etud.lookup_scodoc(apo_data.etape_formsemestre_ids)
        apo_etud.associate_sco(apo_data)

    # Ré-écrit le fichier Apogée
    csv_data = apo_csv.write(apo_data.etuds)

    # Table des NAR:
    nar_etuds = [apo_etud for apo_etud in apo_data.etuds if apo_etud.is_nar]
    if nar_etuds:
        nar_xls = nar_etuds_table(apo_data, nar_etuds)
    else:
        nar_xls = None

    # Journaux & Comptes-rendus
    # Orphelins: etudiants dans fichier Apogée mais pas dans ScoDoc
    apo_non_scodoc = [
        apo_etud for apo_etud in apo_data.etuds if apo_etud.etat == ETUD_ORPHELIN
    ]
    # Non inscrits: connus de ScoDoc mais pas inscrit dans l'étape cette année
    apo_non_scodoc_inscrits = [
        apo_etud for apo_etud in apo_data.etuds if apo_etud.etat == ETUD_NON_INSCRIT
    ]
    # CR table
    cr_table = apo_data.build_cr_table()
    cr_xls = cr_table.excel()

    # ADSUPs
    adsup_table = apo_data.build_adsup_table()
    adsup_xls = adsup_table.excel() if len(adsup_table) else None

    # Create ZIP
    if not dest_zip:
        data = io.BytesIO()
        dest_zip = ZipFile(data, "w")
        my_zip = True
    else:
        my_zip = False
    # Ensure unique filenames
    filename = apo_csv.get_filename()
    basename, ext = os.path.splitext(filename)
    csv_filename = filename

    if csv_filename in dest_zip.namelist():
        basename = filename + "-" + apo_data.vdi_apogee
        csv_filename = basename + ext
    num_file = 1
    tmplname = basename
    while csv_filename in dest_zip.namelist():
        basename = f"{tmplname}-{num_file}"
        csv_filename = basename + ext
        num_file += 1

    log_filename = "scodoc-" + basename + ".log.txt"
    nar_filename = basename + "-nar" + scu.XLSX_SUFFIX
    cr_filename = basename + "-decisions" + scu.XLSX_SUFFIX
    adsup_filename = f"{basename}-adsups{scu.XLSX_SUFFIX}"

    logf = io.StringIO()
    logf.write(f"export_to_apogee du {time.ctime()}\n\n")
    logf.write("Semestres ScoDoc sources:\n")
    for sem in apo_data.sems_etape:
        logf.write("\t%(titremois)s\n" % sem)

    def vrai(val):
        return "vrai" if int(val) else "faux"

    logf.write(f"Période: {periode}\n")
    logf.write(f"exporte résultat à l'étape: {vrai(export_res_etape)}\n")
    logf.write(f"exporte résultat à l'année: {vrai(export_res_sem)}\n")
    logf.write(f"exporte résultats des UEs: {vrai(export_res_ues)}\n")
    logf.write(f"exporte résultats des modules: {vrai(export_res_modules)}\n")
    logf.write(f"exporte résultats sans décision de jury: {vrai(export_res_sdj)}\n")
    logf.write(
        "\nÉtudiants Apogée non trouvés dans ScoDoc:\n"
        + "\n".join(
            ["%s\t%s\t%s" % (e["nip"], e["nom"], e["prenom"]) for e in apo_non_scodoc]
        )
    )
    logf.write(
        "\nÉtudiants Apogée non inscrits sur ScoDoc dans cette étape:\n"
        + "\n".join(
            [
                "%s\t%s\t%s" % (e["nip"], e["nom"], e["prenom"])
                for e in apo_non_scodoc_inscrits
            ]
        )
    )

    logf.write(
        "\n\nElements Apogee inconnus dans ces semestres ScoDoc:\n"
        + "\n".join(apo_data.list_unknown_elements())
    )
    if adsup_xls:
        logf.write(f"\n\nADSUP générés: {len(adsup_table)}\n")
    log(logf.getvalue())  # sortie aussi sur le log ScoDoc

    # Write data to ZIP
    dest_zip.writestr(csv_filename, csv_data)
    dest_zip.writestr(log_filename, logf.getvalue())
    if nar_xls:
        dest_zip.writestr(nar_filename, nar_xls)
    dest_zip.writestr(cr_filename, cr_xls)
    if adsup_xls:
        dest_zip.writestr(adsup_filename, adsup_xls)

    if my_zip:
        dest_zip.close()
        data.seek(0)
        return send_file(
            data,
            mimetype="application/zip",
            download_name=scu.sanitize_filename(basename + "-scodoc.zip"),
            as_attachment=True,
        )
    else:
        return None  # zip modified in place