ScoDoc/app/models/formations.py

"""ScoDoc 9 models : Formations
"""

from flask import abort, g
from flask_sqlalchemy.query import Query

import app
from app import db, log
from app.comp import df_cache
from app.models import ScoDocModel, SHORT_STR_LEN
from app.models.but_refcomp import (
    ApcAnneeParcours,
    ApcCompetence,
    ApcParcours,
    ApcParcoursNiveauCompetence,
)
from app.models.events import ScolarNews
from app.models.modules import Module
from app.models.moduleimpls import ModuleImpl
from app.models.ues import UniteEns, UEParcours
from app.scodoc import sco_cache
from app.scodoc import codes_cursus
from app.scodoc import sco_utils as scu
from app.scodoc.codes_cursus import UE_STANDARD
from app.scodoc.sco_exceptions import ScoNonEmptyFormationObject, ScoValueError


class Formation(ScoDocModel):
    """Programme pédagogique d'une formation"""

    __tablename__ = "notes_formations"
    __table_args__ = (db.UniqueConstraint("dept_id", "acronyme", "titre", "version"),)

    id = db.Column(db.Integer, primary_key=True)
    formation_id = db.synonym("id")
    dept_id = db.Column(db.Integer, db.ForeignKey("departement.id"), index=True)

    acronyme = db.Column(db.Text(), nullable=False)
    titre = db.Column(db.Text(), nullable=False)
    titre_officiel = db.Column(db.Text(), nullable=False)
    version = db.Column(db.Integer, default=1, server_default="1")
    commentaire = db.Column(db.Text())
    formation_code = db.Column(
        db.String(SHORT_STR_LEN),
        server_default=db.text("notes_newid_fcod()"),
        nullable=False,
    )
    # nb: la fonction SQL notes_newid_fcod doit être créée à part
    type_parcours = db.Column(db.Integer, default=0, server_default="0")
    code_specialite = db.Column(db.String(SHORT_STR_LEN))

    # Optionnel, pour les formations type BUT
    referentiel_competence_id = db.Column(
        db.Integer, db.ForeignKey("apc_referentiel_competences.id", ondelete="SET NULL")
    )
    ues = db.relationship("UniteEns", backref="formation", lazy="dynamic")
    formsemestres = db.relationship("FormSemestre", lazy="dynamic", backref="formation")
    ues = db.relationship(
        "UniteEns", lazy="dynamic", backref="formation", order_by="UniteEns.numero"
    )
    modules = db.relationship("Module", lazy="dynamic", backref="formation")

    def __repr__(self):
        return f"""<{self.__class__.__name__}(id={self.id}, dept_id={
                self.dept_id}, acronyme={self.acronyme!r}, version={self.version})>"""

    def html(self) -> str:
        "titre complet pour affichage"
        return f"""Formation {self.titre} ({self.acronyme}) version {self.version
            } code <tt>{self.formation_code}</tt>"""

    @classmethod
    def get_formation(cls, formation_id: int | str, dept_id: int = None) -> "Formation":
        """Formation ou 404, cherche uniquement dans le département spécifié
        ou le courant (g.scodoc_dept)"""
        if not isinstance(formation_id, int):
            try:
                formation_id = int(formation_id)
            except (TypeError, ValueError):
                abort(404, "formation_id invalide")
        if g.scodoc_dept:
            dept_id = dept_id if dept_id is not None else g.scodoc_dept_id
        if dept_id is not None:
            return cls.query.filter_by(id=formation_id, dept_id=dept_id).first_or_404()
        return cls.query.filter_by(id=formation_id).first_or_404()

    def to_dict(self, with_refcomp_attrs=False, with_departement=True):
        """As a dict.
        Si with_refcomp_attrs, ajoute attributs permettant de retrouver le ref. de comp.
        """
        e = dict(self.__dict__)
        e.pop("_sa_instance_state", None)
        if "referentiel_competence" in e:
            e.pop("referentiel_competence")
        e["code_specialite"] = e["code_specialite"] or ""
        e["commentaire"] = e["commentaire"] or ""
        if with_departement and self.departement:
            e["departement"] = self.departement.to_dict()
        else:
            e.pop("departement", None)
        e["formation_id"] = self.id  # ScoDoc7 backward compat
        if with_refcomp_attrs and self.referentiel_competence:
            e["refcomp_version_orebut"] = self.referentiel_competence.version_orebut
            e["refcomp_specialite"] = self.referentiel_competence.specialite
            e["refcomp_type_titre"] = self.referentiel_competence.type_titre

        return e

    def get_cursus(self) -> codes_cursus.TypeCursus:
        """get l'instance de TypeCursus de cette formation
        (le TypeCursus définit le genre de formation, à ne pas confondre
        avec les parcours du BUT).
        """
        return codes_cursus.get_cursus_from_code(self.type_parcours)

    def get_titre_version(self) -> str:
        """Titre avec version"""
        return f"{self.acronyme} {self.titre} v{self.version}"

    def is_apc(self):
        "True si formation APC avec SAE (BUT)"
        return self.get_cursus().APC_SAE

    def get_module_coefs(self, semestre_idx: int = None):
        """Les coefs des modules vers les UE (accès via cache)"""
        from app.comp import moy_ue

        if semestre_idx is None:
            key = f"{self.id}"
        else:
            key = f"{self.id}.{semestre_idx}"

        modules_coefficients = df_cache.ModuleCoefsCache.get(key)
        if modules_coefficients is None:
            modules_coefficients, _, _ = moy_ue.df_load_module_coefs(
                self.id, semestre_idx
            )
            df_cache.ModuleCoefsCache.set(key, modules_coefficients)
        return modules_coefficients

    def has_locked_sems(self, semestre_idx: int = None):
        """True if there is a locked formsemestre in this formation.
        If semestre_idx is specified, check only this index.
        """
        query = self.formsemestres.filter_by(etat=False)
        if semestre_idx is not None:
            query = query.filter_by(semestre_id=semestre_idx)
        return len(query.all()) > 0

    def invalidate_module_coefs(self, semestre_idx: int = None):
        """Invalide le cache des coefficients de modules.
        Si semestre_idx est None, invalide les coefs de tous les semestres,
        sinon invalide le semestre indiqué et le cache de la formation.

        Dans tous les cas, invalide tous les formsemestres.
        """
        if semestre_idx is None:
            keys = {f"{self.id}.{m.semestre_id}" for m in self.modules}
        else:
            keys = f"{self.id}.{semestre_idx}"
        df_cache.ModuleCoefsCache.delete_many(keys | {f"{self.id}"})
        # Invalidate aussi les poids de toutes les évals de la formation
        for modimpl in ModuleImpl.query.filter(
            ModuleImpl.module_id == Module.id,
            Module.formation_id == self.id,
        ):
            modimpl.invalidate_evaluations_poids()

        sco_cache.invalidate_formsemestre()

    def invalidate_cached_sems(self):
        "Invalide caches de tous les formssemestres de la formation"
        for sem in self.formsemestres:
            sco_cache.invalidate_formsemestre(formsemestre_id=sem.id)

    def sanitize_old_formation(self) -> None:
        """
        Corrige si nécessaire certains champs issus d'anciennes versions de ScoDoc:
        Pour les formations APC (BUT) seulement:
        - affecte à chaque module de cette formation le semestre de son UE de rattachement,
        si elle en a une.
        - si le module_type n'est pas renseigné, le met à STANDARD.
        - on n'utilise pas de matières: réaffecte tous les modules
          à la première matière de leur UE.

        Devrait être appelé lorsqu'on change le type de formation vers le BUT,
        et aussi lorsqu'on change le semestre d'une UE BUT.
        Utile pour la migration des anciennes formations vers le BUT.

        En cas de changement, invalide les caches coefs/poids.
        """
        if not self.is_apc():
            return
        change = False
        for mod in self.modules:
            # --- Indices de semestres:
            if (
                mod.ue.semestre_idx is not None
                and mod.ue.semestre_idx > 0
                and mod.semestre_id != mod.ue.semestre_idx
            ):
                mod.semestre_id = mod.ue.semestre_idx
                db.session.add(mod)
                change = True
            # --- Types de modules
            if mod.module_type is None:
                mod.module_type = scu.ModuleType.STANDARD
                db.session.add(mod)
                change = True
        # --- Numéros de modules
        if Module.query.filter_by(formation_id=self.id, numero=None).count() > 0:
            scu.objects_renumber(db, self.modules.all())

        # --- Types d'UE (avant de rendre le type non nullable)
        ues_sans_type = UniteEns.query.filter_by(formation_id=self.id, type=None)
        if ues_sans_type.count() > 0:
            for ue in ues_sans_type:
                ue.type = 0
                db.session.add(ue)

        # --- Réaffectation des matières:
        for ue in self.ues:
            mat = ue.matieres.first()
            if mat is None:
                mat = Matiere()
                ue.matieres.append(mat)
                db.session.add(mat)
            for module in ue.modules:
                if module.matiere_id != mat.id:
                    module.matiere = mat
                    db.session.add(module)
                    change = True

        db.session.commit()
        if change:
            app.clear_scodoc_cache()

    def query_ues_parcour(
        self, parcour: ApcParcours, with_sport: bool = False
    ) -> Query:
        """Les UEs (sans bonus, sauf si with_sport) d'un parcours de la formation
        (déclarée comme faisant partie du parcours ou du tronc commun, sans aucun parcours)
        Si parcour est None, les UE sans parcours.
        Exemple: pour avoir les UE du semestre 3, faire
        `formation.query_ues_parcour(parcour).filter(UniteEns.semestre_idx == 3)`
        """
        if with_sport:
            query_f = UniteEns.query.filter_by(formation=self)
        else:
            query_f = UniteEns.query.filter_by(formation=self, type=UE_STANDARD)
        # Les UE sans parcours:
        query_no_parcours = query_f.outerjoin(UEParcours).filter(
            UEParcours.parcours_id == None
        )
        if parcour is None:
            return query_no_parcours.order_by(UniteEns.numero)
        # Ajoute les UE du parcours sélectionné:
        return query_no_parcours.union(
            query_f.join(UEParcours).filter_by(parcours_id=parcour.id)
        ).order_by(UniteEns.numero)
        # return UniteEns.query.filter_by(formation=self, type=UE_STANDARD).filter(
        #     UniteEns.niveau_competence_id == ApcNiveau.id,
        #     (UniteEns.parcour_id == parcour.id) | (UniteEns.parcour_id == None),
        #     ApcParcoursNiveauCompetence.competence_id == ApcNiveau.competence_id,
        #     ApcParcoursNiveauCompetence.annee_parcours_id == ApcAnneeParcours.id,
        #     ApcAnneeParcours.parcours_id == parcour.id,
        # )

    def query_competences_parcour(self, parcour: ApcParcours) -> Query:
        """Les ApcCompetences d'un parcours de la formation.
        None si pas de référentiel de compétences.
        """
        if self.referentiel_competence_id is None:
            return None
        return (
            ApcCompetence.query.filter_by(referentiel_id=self.referentiel_competence_id)
            .join(
                ApcParcoursNiveauCompetence,
                ApcParcoursNiveauCompetence.competence_id == ApcCompetence.id,
            )
            .join(
                ApcAnneeParcours,
                ApcParcoursNiveauCompetence.annee_parcours_id == ApcAnneeParcours.id,
            )
            .filter(ApcAnneeParcours.parcours_id == parcour.id)
        )

    def refcomp_desassoc(self):
        """Désassocie la formation de son ref. de compétence"""
        self.referentiel_competence = None
        db.session.add(self)
        # Niveaux des UE
        for ue in self.ues:
            ue.niveau_competence = None
            db.session.add(ue)
        # Parcours et AC des modules
        for mod in self.modules:
            mod.parcours = []
            mod.app_critiques = []
            db.session.add(mod)
        db.session.commit()


class Matiere(ScoDocModel):
    """Matières: regroupe les modules d'une UE
    La matière a peu d'utilité en dehors de la présentation des modules
    d'une UE.
    """

    __tablename__ = "notes_matieres"
    __table_args__ = (db.UniqueConstraint("ue_id", "titre"),)

    id = db.Column(db.Integer, primary_key=True)
    matiere_id = db.synonym("id")
    ue_id = db.Column(db.Integer, db.ForeignKey("notes_ue.id"))
    titre = db.Column(db.Text())
    numero = db.Column(db.Integer, nullable=False, default=0)  # ordre de présentation

    modules = db.relationship(
        "Module", lazy="dynamic", backref="matiere", cascade="all, delete-orphan"
    )
    _sco_dept_relations = ("UniteEns", "Formation")  # accès au dept_id

    def __repr__(self):
        return f"""<{self.__class__.__name__}(id={self.id}, ue_id={
            self.ue_id}, titre='{self.titre!r}')>"""

    def to_dict(self):
        """as a dict, with the same conversions as in ScoDoc7"""
        e = dict(self.__dict__)
        e.pop("_sa_instance_state", None)
        # ScoDoc7 output_formators
        e["numero"] = e["numero"] if e["numero"] else 0
        e["matiere_id"] = self.id
        return e

    def is_locked(self) -> bool:
        """True if matiere can't modified
        because it contains modules used in a locked formsemestre.
        """
        from app.models.formsemestre import FormSemestre

        mat = (
            db.session.query(Matiere)
            .filter_by(id=self.id)
            .join(Module)
            .join(ModuleImpl)
            .join(FormSemestre)
            .filter_by(etat=False)
            .all()
        )
        return bool(mat)

    def can_be_deleted(self) -> bool:
        "True si la matiere n'est pas utilisée dans des formsemestres"
        locked = self.is_locked()
        if locked:
            return False
        if any(m.modimpls.all() for m in self.modules):
            return False
        return True

    def delete(self):
        "Delete matière. News, inval cache."
        formation = self.ue.formation
        log(f"matiere.delete: matiere_id={self.id}")
        if not self.can_be_deleted():
            # il y a au moins un modimpl dans un module de cette matière
            raise ScoNonEmptyFormationObject("Matière", self.titre)
        db.session.delete(self)
        db.session.commit()
        # news
        ScolarNews.add(
            typ=ScolarNews.NEWS_FORM,
            obj=formation.id,
            text=f"Modification de la formation {formation.acronyme}",
        )
        # cache
        formation.invalidate_cached_sems()

    @classmethod
    def create_from_dict(cls, data: dict) -> "Matiere":
        """Create matière from dict. Log, news, cache.
        data must include ue_id, a valid UE id.
        Commit session.
        """
        # check ue
        if data.get("ue_id") is None:
            raise ScoValueError("UE id missing")
        _ = UniteEns.get_ue(data["ue_id"])

        mat = super().create_from_dict(data)
        db.session.commit()
        db.session.refresh(mat)
        # news
        formation = mat.ue.formation
        ScolarNews.add(
            typ=ScolarNews.NEWS_FORM,
            obj=formation.id,
            text=f"Modification de la formation {formation.acronyme}",
        )
        formation.invalidate_cached_sems()
        return mat