ScoDoc/app/models/moduleimpls.py

# -*- coding: UTF-8 -*
"""ScoDoc models: moduleimpls
"""
import pandas as pd
from flask import abort, g
from flask_login import current_user
from flask_sqlalchemy.query import Query

import app
from app import db
from app.auth.models import User
from app.comp import df_cache
from app.models import APO_CODE_STR_LEN, ScoDocModel
from app.models.etudiants import Identite
from app.models.evaluations import Evaluation
from app.models.modules import Module
from app.scodoc.sco_exceptions import AccessDenied, ScoLockedSemError
from app.scodoc.sco_permissions import Permission
from app.scodoc import sco_utils as scu


class ModuleImpl(ScoDocModel):
    """Mise en oeuvre d'un module pour une annee/semestre"""

    __tablename__ = "notes_moduleimpl"
    __table_args__ = (db.UniqueConstraint("formsemestre_id", "module_id"),)

    id = db.Column(db.Integer, primary_key=True)
    code_apogee = db.Column(db.String(APO_CODE_STR_LEN), index=True, nullable=True)
    "id de l'element pedagogique Apogee correspondant"
    edt_id: str | None = db.Column(db.Text(), index=True, nullable=True)
    "identifiant emplois du temps (unicité non imposée)"
    moduleimpl_id = db.synonym("id")
    module_id = db.Column(db.Integer, db.ForeignKey("notes_modules.id"), nullable=False)
    formsemestre_id = db.Column(
        db.Integer,
        db.ForeignKey("notes_formsemestre.id"),
        index=True,
        nullable=False,
    )
    responsable_id = db.Column(
        "responsable_id", db.Integer, db.ForeignKey("user.id", ondelete="SET NULL")
    )
    responsable = db.relationship("User", back_populates="modimpls")
    # formule de calcul moyenne:
    computation_expr = db.Column(db.Text())

    evaluations = db.relationship(
        "Evaluation",
        lazy="dynamic",
        backref="moduleimpl",
        order_by=(Evaluation.numero, Evaluation.date_debut),
    )
    "évaluations, triées par numéro et dates croissants, donc la plus ancienne d'abord."
    enseignants = db.relationship(
        "User",
        secondary="notes_modules_enseignants",
        lazy="dynamic",
        backref="moduleimpl",
    )
    "enseignants du module (sans le responsable)"

    _sco_dept_relations = ("FormSemestre",)  # accès au dept_id

    def __repr__(self):
        return f"<{self.__class__.__name__} {self.id} module={repr(self.module)}>"

    def get_codes_apogee(self) -> set[str]:
        """Les codes Apogée (codés en base comme "VRT1,VRT2").
        (si non renseigné, ceux du module)
        """
        if self.code_apogee:
            return {x.strip() for x in self.code_apogee.split(",") if x}
        return self.module.get_codes_apogee()

    def get_edt_ids(self) -> list[str]:
        "les ids pour l'emploi du temps: à défaut, les codes Apogée"
        return [
            scu.normalize_edt_id(x)
            for x in scu.split_id(self.edt_id) or scu.split_id(self.code_apogee)
        ] or self.module.get_edt_ids()

    def get_evaluations_poids(self) -> pd.DataFrame:
        """Les poids des évaluations vers les UEs (accès via cache redis).
        Toutes les évaluations sont considérées (normales, bonus, rattr., etc.)
        """
        evaluations_poids = df_cache.EvaluationsPoidsCache.get(self.id)
        if evaluations_poids is None:
            from app.comp import moy_mod

            evaluations_poids, _ = moy_mod.load_evaluations_poids(self.id)
            df_cache.EvaluationsPoidsCache.set(self.id, evaluations_poids)
        return evaluations_poids

    @classmethod
    def get_modimpl(cls, moduleimpl_id: int | str, dept_id: int = None) -> "ModuleImpl":
        """ModuleImpl ou 404, cherche uniquement dans le département spécifié ou le courant."""
        from app.models.formsemestre import FormSemestre

        if not isinstance(moduleimpl_id, int):
            try:
                moduleimpl_id = int(moduleimpl_id)
            except (TypeError, ValueError):
                abort(404, "moduleimpl_id invalide")
        if g.scodoc_dept:
            dept_id = dept_id if dept_id is not None else g.scodoc_dept_id
        query = cls.query.filter_by(id=moduleimpl_id)
        if dept_id is not None:
            query = query.join(FormSemestre).filter_by(dept_id=dept_id)
        return query.first_or_404()

    def invalidate_evaluations_poids(self):
        """Invalide poids cachés"""
        df_cache.EvaluationsPoidsCache.delete(self.id)

    def check_apc_conformity(
        self, res: "ResultatsSemestreBUT", evaluation_type=Evaluation.EVALUATION_NORMALE
    ) -> bool:
        """true si les poids des évaluations du type indiqué (normales par défaut)
        du module permettent de satisfaire les coefficients du PN.
        """
        # appelé par formsemestre_status, liste notes, et moduleimpl_status
        if not self.module.formation.get_cursus().APC_SAE or (
            self.module.module_type
            not in {scu.ModuleType.RESSOURCE, scu.ModuleType.SAE}
        ):
            return True  # Non BUT, toujours conforme
        from app.comp import moy_mod

        mod_results = res.modimpls_results.get(self.id)
        if mod_results is None:
            app.critical_error("check_apc_conformity: err 1")

        selected_evaluations_ids = [
            eval_id
            for eval_id, eval_type in mod_results.evals_type.items()
            if eval_type == evaluation_type
        ]
        if not selected_evaluations_ids:
            return True  # conforme si pas d'évaluations
        selected_evaluations_poids = self.get_evaluations_poids().loc[
            selected_evaluations_ids
        ]
        return moy_mod.moduleimpl_is_conforme(
            self,
            selected_evaluations_poids,
            res.modimpl_coefs_df,
        )

    def to_dict(self, convert_objects=False, with_module=True):
        """as a dict, with the same conversions as in ScoDoc7, including module.
        If convert_objects, convert all attributes to native types
        (suitable jor json encoding).
        """
        d = dict(self.__dict__)
        d.pop("_sa_instance_state", None)
        if convert_objects:
            # on n'exporte pas le formsemestre et les inscriptions
            d.pop("formsemestre", None)
            d.pop("inscriptions", None)
        # ScoDoc7 output_formators: (backward compat)
        d["moduleimpl_id"] = self.id
        d["ens"] = [
            {"moduleimpl_id": self.id, "ens_id": e.id} for e in self.enseignants
        ]
        if with_module:
            d["module"] = self.module.to_dict(convert_objects=convert_objects)
        else:
            d.pop("module", None)
        d["code_apogee"] = d["code_apogee"] or ""  # pas de None
        return d

    def can_edit_evaluation(self, user) -> bool:
        """True if this user can create, delete or edit and evaluation in this modimpl
        (nb: n'implique pas le droit de saisir ou modifier des notes)
        """
        # acces pour resp. moduleimpl et resp. form semestre (dir etud)
        if (
            user.has_permission(Permission.EditAllEvals)
            or user.id == self.responsable_id
            or user.id in (r.id for r in self.formsemestre.responsables)
        ):
            return True
        elif self.formsemestre.ens_can_edit_eval:
            if user.id in (e.id for e in self.enseignants):
                return True

        return False

    def can_edit_notes(self, user: "User", allow_ens=True) -> bool:
        """True if authuser can enter or edit notes in this module.
        If allow_ens, grant access to all ens in this module

        Si des décisions de jury ont déjà été saisies dans ce semestre,
        seul le directeur des études peut saisir des notes (et il ne devrait pas).
        """
        # was sco_permissions_check.can_edit_notes
        from app.scodoc import sco_cursus_dut

        if not self.formsemestre.etat:
            return False  # semestre verrouillé
        is_dir_etud = user.id in (u.id for u in self.formsemestre.responsables)
        can_edit_all_notes = user.has_permission(Permission.EditAllNotes)
        if sco_cursus_dut.formsemestre_has_decisions(self.formsemestre_id):
            # il y a des décisions de jury dans ce semestre !
            return can_edit_all_notes or is_dir_etud
        if (
            not can_edit_all_notes
            and user.id != self.responsable_id
            and not is_dir_etud
        ):
            # enseignant (chargé de TD) ?
            return allow_ens and user.id in (ens.id for ens in self.enseignants)
        return True

    def can_change_responsable(self, user: User, raise_exc=False) -> bool:
        """Check if user can modify module resp.
        If raise_exc, raises exception (AccessDenied or ScoLockedSemError) if not.
        = Admin, et dir des etud. (si option l'y autorise)
        """
        if not self.formsemestre.etat:
            if raise_exc:
                raise ScoLockedSemError("Modification impossible: semestre verrouille")
            return False
        # -- check access
        # admin ou resp. semestre avec flag resp_can_change_resp
        if user.has_permission(Permission.EditFormSemestre):
            return True
        if (
            user.id in [resp.id for resp in self.formsemestre.responsables]
        ) and self.formsemestre.resp_can_change_ens:
            return True
        if raise_exc:
            raise AccessDenied(f"Modification impossible pour {user}")
        return False

    def can_change_ens(self, user: User | None = None, raise_exc=True) -> bool:
        """check if user can modify ens list (raise exception if not)"
        if user is None, current user.
        """
        user = current_user if user is None else user
        if not self.formsemestre.etat:
            if raise_exc:
                raise ScoLockedSemError("Modification impossible: semestre verrouille")
            return False
        # -- check access
        # admin, resp. module ou resp. semestre
        if (
            user.id != self.responsable_id
            and not user.has_permission(Permission.EditFormSemestre)
            and user.id not in (u.id for u in self.formsemestre.responsables)
        ):
            if raise_exc:
                raise AccessDenied(f"Modification impossible pour {user}")
            return False
        return True

    def can_change_inscriptions(self, user: User | None = None, raise_exc=True) -> bool:
        """check si user peut inscrire/désinsincrire des étudiants à ce module.
        Autorise ScoEtudInscrit ou responsables semestre.
        """
        user = current_user if user is None else user
        if not self.formsemestre.etat:
            if raise_exc:
                raise ScoLockedSemError("Modification impossible: semestre verrouille")
            return False
        # -- check access
        # resp. module ou ou perm. EtudInscrit ou resp. semestre
        if (
            user.id != self.responsable_id
            and not user.has_permission(Permission.EtudInscrit)
            and user.id not in (u.id for u in self.formsemestre.responsables)
        ):
            if raise_exc:
                raise AccessDenied(f"Modification impossible pour {user}")
            return False
        return True

    def est_inscrit(self, etud: Identite):
        """
        Vérifie si l'étudiant est bien inscrit au moduleimpl (même si DEM ou DEF au semestre).
        (lent, pas de cache: pour un accès rapide, utiliser nt.modimpl_inscr_df).
        Retourne ModuleImplInscription si inscrit au module, False sinon.
        """
        # vérifie inscrit au moduleimpl ET au formsemestre
        from app.models.formsemestre import FormSemestre, FormSemestreInscription

        inscription = (
            ModuleImplInscription.query.filter_by(etudid=etud.id, moduleimpl_id=self.id)
            .join(ModuleImpl)
            .join(FormSemestre)
            .join(FormSemestreInscription)
            .filter_by(etudid=etud.id)
            .first()
        )

        return inscription or False

    def query_inscriptions(self) -> Query:
        """Query ModuleImplInscription: inscrits au moduleimpl et au formsemestre
        (pas de cache: pour un accès rapide, utiliser nt.modimpl_inscr_df).
        """
        from app.models.formsemestre import FormSemestre, FormSemestreInscription

        return (
            ModuleImplInscription.query.filter_by(moduleimpl_id=self.id)
            .join(ModuleImpl)
            .join(FormSemestre)
            .join(FormSemestreInscription)
            .filter_by(etudid=ModuleImplInscription.etudid)
        )


# Enseignants (chargés de TD ou TP) d'un moduleimpl
notes_modules_enseignants = db.Table(
    "notes_modules_enseignants",
    db.Column(
        "moduleimpl_id",
        db.Integer,
        db.ForeignKey("notes_moduleimpl.id", ondelete="CASCADE"),
    ),
    db.Column("ens_id", db.Integer, db.ForeignKey("user.id", ondelete="CASCADE")),
    # ? db.UniqueConstraint("moduleimpl_id", "ens_id"),
)
# XXX il manque probablement une relation pour gérer cela


class ModuleImplInscription(ScoDocModel):
    """Inscription à un module  (etudiants,moduleimpl)"""

    __tablename__ = "notes_moduleimpl_inscription"
    __table_args__ = (db.UniqueConstraint("moduleimpl_id", "etudid"),)

    id = db.Column(db.Integer, primary_key=True)
    moduleimpl_inscription_id = db.synonym("id")
    moduleimpl_id = db.Column(
        db.Integer,
        db.ForeignKey("notes_moduleimpl.id"),
        index=True,
    )
    etudid = db.Column(db.Integer, db.ForeignKey("identite.id"), index=True)
    etud = db.relationship(
        Identite,
        backref=db.backref("moduleimpl_inscriptions", cascade="all, delete-orphan"),
    )
    modimpl = db.relationship(
        ModuleImpl,
        backref=db.backref("inscriptions", cascade="all, delete-orphan"),
    )

    def to_dict(self) -> dict:
        "dict repr."
        return {
            "id": self.id,
            "etudid": self.etudid,
            "moduleimpl_id": self.moduleimpl_id,
        }

    @classmethod
    def etud_modimpls_in_ue(
        cls, formsemestre_id: int, etudid: int, ue_id: int
    ) -> Query:
        """moduleimpls de l'UE auxquels l'étudiant est inscrit.
        (Attention: inutile en APC, il faut considérer les coefficients)
        """
        return ModuleImplInscription.query.filter(
            ModuleImplInscription.etudid == etudid,
            ModuleImplInscription.moduleimpl_id == ModuleImpl.id,
            ModuleImpl.formsemestre_id == formsemestre_id,
            ModuleImpl.module_id == Module.id,
            Module.ue_id == ue_id,
        )

    @classmethod
    def nb_inscriptions_dans_ue(
        cls, formsemestre_id: int, etudid: int, ue_id: int
    ) -> int:
        """Nombre de moduleimpls de l'UE auxquels l'étudiant est inscrit"""
        return cls.etud_modimpls_in_ue(formsemestre_id, etudid, ue_id).count()

    @classmethod
    def sum_coefs_modimpl_ue(
        cls, formsemestre_id: int, etudid: int, ue_id: int
    ) -> float:
        """Somme des coefficients des modules auxquels l'étudiant est inscrit
        dans l'UE du semestre indiqué.
        N'utilise que les coefficients, donc inadapté aux formations APC.
        """
        return sum(
            [
                inscr.modimpl.module.coefficient
                for inscr in cls.etud_modimpls_in_ue(formsemestre_id, etudid, ue_id)
            ]
        )