ScoDoc/app/models/modules.py

"""ScoDoc 9 models : Modules
"""

import http
from flask import current_app, g, url_for

from app import db, log
from app import models
from app.models import APO_CODE_STR_LEN
from app.models.but_refcomp import (
    ApcAppCritique,
    ApcParcours,
    ApcReferentielCompetences,
    app_critiques_modules,
    parcours_modules,
)
from app.models.events import ScolarNews
from app.scodoc import sco_utils as scu
from app.scodoc.codes_cursus import UE_SPORT
from app.scodoc.sco_exceptions import (
    ScoValueError,
    ScoLockedFormError,
    ScoNonEmptyFormationObject,
)
from app.scodoc.sco_utils import ModuleType


class Module(models.ScoDocModel):
    """Module"""

    __tablename__ = "notes_modules"

    id = db.Column(db.Integer, primary_key=True)
    module_id = db.synonym("id")
    titre = db.Column(db.Text())
    abbrev = db.Column(db.Text())  # nom court
    # certains départements ont des codes infiniment longs: donc Text !
    code = db.Column(db.Text(), nullable=False)
    "code module, chaine non nullable"
    heures_cours = db.Column(db.Float)
    heures_td = db.Column(db.Float)
    heures_tp = db.Column(db.Float)
    coefficient = db.Column(db.Float)  # coef PPN (sauf en APC)
    ects = db.Column(db.Float)  # Crédits ECTS
    ue_id = db.Column(db.Integer, db.ForeignKey("notes_ue.id"), index=True)
    formation_id = db.Column(db.Integer, db.ForeignKey("notes_formations.id"))
    matiere_id = db.Column(db.Integer, db.ForeignKey("notes_matieres.id"))
    # pas un id mais le numéro du semestre: 1, 2, ...
    # note: en APC, le semestre qui fait autorité est celui de l'UE
    semestre_id = db.Column(db.Integer, nullable=False, default=1, server_default="1")
    numero = db.Column(db.Integer, nullable=False, default=0)  # ordre de présentation
    code_apogee = db.Column(db.String(APO_CODE_STR_LEN))
    "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)"
    # Type: ModuleType.STANDARD, MALUS, RESSOURCE, SAE (enum)
    module_type = db.Column(db.Integer, nullable=False, default=0, server_default="0")
    # Relations:
    modimpls = db.relationship(
        "ModuleImpl", backref="module", lazy="dynamic", cascade="all, delete-orphan"
    )
    ues_apc = db.relationship("UniteEns", secondary="module_ue_coef", viewonly=True)
    tags = db.relationship(
        "NotesTag",
        secondary="notes_modules_tags",
        lazy=True,
        backref=db.backref("modules", lazy=True),
    )
    # BUT
    parcours = db.relationship(
        "ApcParcours",
        secondary=parcours_modules,
        lazy="subquery",
        backref=db.backref("modules", lazy=True),
        order_by="ApcParcours.numero, ApcParcours.code",
    )

    app_critiques = db.relationship(
        "ApcAppCritique",
        secondary=app_critiques_modules,
        lazy="subquery",
        backref=db.backref("modules", lazy=True),
    )

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

    def __init__(self, **kwargs):
        self.ue_coefs = []
        super().__init__(**kwargs)

    def __repr__(self):
        return f"""<Module{ModuleType(self.module_type or ModuleType.STANDARD).name
            } id={self.id} code={self.code!r} semestre_id={self.semestre_id}>"""

    @classmethod
    def convert_dict_fields(cls, args: dict) -> dict:
        """Convert fields in the given dict. No other side effect.
        returns: dict to store in model's db.
        """
        # s'assure que ects etc est non ''
        fs_empty_stored_as_nulls = {
            "coefficient",
            "ects",
            "heures_cours",
            "heures_td",
            "heures_tp",
        }
        args_dict = {}
        for key, value in args.items():
            if hasattr(cls, key) and not isinstance(getattr(cls, key, None), property):
                if key in fs_empty_stored_as_nulls and value == "":
                    value = None
                args_dict[key] = value
            if key == "app_critiques":  # peut être liste d'ApcAppCritique ou d'ids
                args_dict[key] = cls.convert_app_critiques(value)

        return args_dict

    @staticmethod
    def convert_app_critiques(
        app_crits: list, ref_comp: ApcReferentielCompetences | None = None
    ) -> list[ApcAppCritique]:
        """ """
        res = []
        for x in app_crits:
            app_crit = (
                x
                if isinstance(x, ApcAppCritique)
                else db.session.get(ApcAppCritique, x)
            )
            if app_crit is None:
                raise ScoValueError("app_critiques invalid")
            if ref_comp and app_crit.niveau.competence.referentiel_id != ref_comp.id:
                raise ScoValueError("app_critique hors référentiel !")
            res.append(app_crit)
        return res

    @classmethod
    def filter_model_attributes(cls, args: dict, excluded: set[str] = None) -> dict:
        """Returns a copy of dict with only the keys belonging to the Model and not in excluded.
        Add 'id' to excluded."""
        # on ne peut pas affecter directement parcours
        return super().filter_model_attributes(args, (excluded or set()) | {"parcours"})

    @classmethod
    def check_module_code_unicity(cls, code, formation_id, module_id=None) -> bool:
        "true si code module unique dans la formation"
        from app.models import Formation

        formation = Formation.get_formation(formation_id)
        query = formation.modules.filter_by(code=code)
        if module_id is not None:  # edition: supprime le module en cours
            query = query.filter(Module.id != module_id)
        return query.count() == 0

    def from_dict(self, args: dict, excluded: set[str] | None = None) -> bool:
        """Update object's fields given in dict. Add to session but don't commit.
        True if modification.
        - can't change ue nor formation
        - can change matiere_id, iff new matiere in same ue
        - can change parcours: parcours list of ApcParcour id or instances.
        Ne modifie pas les coefficients APC ue_coefs
        """
        args = args.copy()
        if "ue_coefs" in args:
            del args["ue_coefs"]
        if self.is_locked():
            # formation verrouillée: empeche de modifier coefficient, matiere, and semestre_id
            protected_fields = ("coefficient", "matiere_id", "semestre_id")
            for f in protected_fields:
                if f in args:
                    del args[f]
        # Unicité du code
        if "code" in args and not Module.check_module_code_unicity(
            args["code"], self.formation_id, self.id
        ):
            raise ScoValueError("code module déjà utilisé")
        # Vérifie les changements de matiere
        new_matiere_id = args.get("matiere_id", self.matiere_id)
        if new_matiere_id != self.matiere_id:
            # exists ?
            from app.models import Matiere

            matiere = db.session.get(Matiere, new_matiere_id)
            if matiere is None or matiere.ue_id != self.ue_id:
                raise ScoValueError("invalid matiere")

        modified = super().from_dict(
            args, excluded=(excluded or set()) | {"formation_id", "ue_id"}
        )

        existing_parcours = {p.id for p in self.parcours}
        new_parcours = args.get("parcours", []) or []
        if existing_parcours != set(new_parcours):
            self.set_parcours_from_list(new_parcours)
            return True
        return modified

    @classmethod
    def create_from_dict(
        cls,
        data: dict,
        inval_cache=False,
        news=False,
    ) -> "Module":
        """Create from given dict, add parcours.
        Flush session.
        Si news, commit and log news.
        """
        from app.models.formations import Formation

        # check required arguments
        for required_arg in ("code", "formation_id", "ue_id"):
            if required_arg not in data:
                raise ScoValueError(f"missing argument: {required_arg}")
        if not data["code"]:
            raise ScoValueError("module code must be non empty")
        # Check formation
        formation = Formation.get_formation(data["formation_id"])
        ue = UniteEns.get_ue(data["ue_id"])
        # refuse de créer un module APC avec semestres semestre du module != semestre de l'UE
        if formation.is_apc():
            if int(data.get("semestre_id", 1)) != ue.semestre_idx:
                raise ScoValueError(
                    "Formation incompatible: indices UE et module différents"
                )
        module = super().create_from_dict(data)
        db.session.flush()
        module.set_parcours_from_list(data.get("parcours", []) or [])
        log(f"module_create: created {module.id} with {data}")
        if news:
            db.session.commit()
            db.session.refresh(module)
            ScolarNews.add(
                typ=ScolarNews.NEWS_FORM,
                obj=formation.id,
                text=f"Modification de la formation {formation.acronyme}",
            )
        if inval_cache:
            formation.invalidate_cached_sems()

        return module

    def is_locked(self) -> bool:
        """True if module cannot be modified
        because it is used in a locked formsemestre.
        """
        from app.models import FormSemestre, ModuleImpl

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

    def can_be_deleted(self) -> bool:
        """True if module can be deleted"""
        return self.modimpls.count() == 0

    def delete(self):
        "Delete module. News, inval cache."
        if self.is_locked():
            raise ScoLockedFormError()
        if not self.can_be_deleted():
            raise ScoNonEmptyFormationObject(
                "Module",
                msg=self.titre or self.code,
                dest_url=url_for(
                    "notes.ue_table",
                    scodoc_dept=g.scodoc_dept,
                    formation_id=self.formation_id,
                    semestre_idx=self.ue.semestre_idx,
                ),
            )
        formation = self.formation
        db.session.delete(self)
        log(f"Module.delete({self.id})")
        db.session.commit()
        # news
        ScolarNews.add(
            typ=ScolarNews.NEWS_FORM,
            obj=formation.id,
            text=f"Modification de la formation {formation.acronyme}",
        )
        formation.invalidate_cached_sems()

    def set_parcours_from_list(self, parcours: list[ApcParcours | int]):
        """Ajoute ces parcours à la liste des parcours du module.
        Chaque élément est soit un objet parcours soit un id.
        S'assure que chaque parcours est dans le référentiel de compétence
        associé à la formation du module.
        """
        for p in parcours:
            if isinstance(p, ApcParcours):
                parcour: ApcParcours = p
                if p.referentiel_id != self.formation.referentiel_competence.id:
                    raise ScoValueError("Parcours hors référentiel du module")
            else:
                try:
                    pid = int(p)
                except ValueError as exc:
                    raise ScoValueError("id de parcours invalide") from exc
                query = (
                    ApcParcours.query.filter_by(id=pid)
                    .join(ApcReferentielCompetences)
                    .filter_by(id=self.formation.referentiel_competence.id)
                )
                if g.scodoc_dept:
                    query = query.filter_by(dept_id=g.scodoc_dept_id)
                parcour: ApcParcours = query.first()
                if parcour is None:
                    raise ScoValueError("Parcours invalide")
            self.parcours.append(parcour)

    def clone(self):
        """Create a new copy of this module."""
        mod = Module(
            titre=self.titre,
            abbrev=self.abbrev,
            code=self.code + "-copie",
            heures_cours=self.heures_cours,
            heures_td=self.heures_td,
            heures_tp=self.heures_tp,
            coefficient=self.coefficient,
            ects=self.ects,
            ue_id=self.ue_id,
            matiere_id=self.matiere_id,
            formation_id=self.formation_id,
            semestre_id=self.semestre_id,
            numero=self.numero,  # il est conseillé de renuméroter
            code_apogee="",  # volontairement vide pour éviter les erreurs
            module_type=self.module_type,
        )

        # Les tags:
        for tag in self.tags:
            mod.tags.append(tag)
        # Les parcours
        for parcour in self.parcours:
            mod.parcours.append(parcour)
        # Les AC
        for app_critique in self.app_critiques:
            mod.app_critiques.append(app_critique)
        return mod

    def to_dict(
        self,
        convert_objects=False,
        with_matiere=False,
        with_ue=False,
        with_parcours_ids=False,
    ) -> dict:
        """If convert_objects, convert all attributes to native types
        (suitable jor json encoding).
        If convert_objects and with_parcours_ids, give parcours as a list of id (API)
        """
        d = dict(self.__dict__)
        d.pop("_sa_instance_state", None)
        d.pop("formation", None)
        if convert_objects:
            if with_parcours_ids:
                d["parcours"] = [p.id for p in self.parcours]
            else:
                d["parcours"] = [p.to_dict() for p in self.parcours]
            d["ue_coefs"] = [
                c.to_dict(convert_objects=False)
                for c in self.ue_coefs
                # note: don't convert_objects: we do wan't the details of the UEs here
            ]
            d["app_critiques"] = {x.code: x.to_dict() for x in self.app_critiques}
        if not with_matiere:
            d.pop("matiere", None)
        if not with_ue:
            d.pop("ue", None)
        if convert_objects and with_matiere:
            d["matiere"] = self.matiere.to_dict(convert_objects=True)
        if convert_objects and with_ue:
            d["ue"] = self.ue.to_dict(convert_objects=True)

        # ScoDoc7 output_formators: (backward compat)
        d["module_id"] = self.id
        d["heures_cours"] = 0.0 if self.heures_cours is None else self.heures_cours
        d["heures_td"] = 0.0 if self.heures_td is None else self.heures_td
        d["heures_tp"] = 0.0 if self.heures_tp is None else self.heures_tp
        d["numero"] = 0 if self.numero is None else self.numero
        d["coefficient"] = 0.0 if self.coefficient is None else self.coefficient
        d["module_type"] = 0 if self.module_type is None else self.module_type
        d["code_apogee"] = d["code_apogee"] or ""  # pas de None
        return d

    def is_apc(self):
        "True si module SAÉ ou Ressource"
        return self.module_type and scu.ModuleType(self.module_type) in {
            scu.ModuleType.RESSOURCE,
            scu.ModuleType.SAE,
        }

    def type_name(self) -> str:
        "Le nom du type de module, pour les humains (avec majuscules et accents)"
        return scu.MODULE_TYPE_NAMES[self.module_type]

    def type_abbrv(self) -> str:
        """Le nom du type de module, pour les styles CSS.
        "mod", "malus", "res", "sae"
        """
        return scu.ModuleType.get_abbrev(self.module_type)

    def titre_str(self) -> str:
        "Identifiant du module à afficher : abbrev ou titre ou code"
        return self.abbrev or self.titre or self.code

    def sort_key(self) -> tuple:
        """Clé de tri pour formations classiques"""
        return self.numero or 0, self.code

    def sort_key_apc(self) -> tuple:
        """Clé de tri pour avoir
        présentation par type (res, sae), parcours, type, numéro
        """
        if (
            self.formation.referentiel_competence is None
            or len(self.parcours)
            == self.formation.referentiel_competence.parcours.count()
            or len(self.parcours) == 0
        ):
            key_parcours = ""
        else:
            key_parcours = "/".join([p.code for p in self.parcours])
        return self.module_type, key_parcours, self.numero or 0

    def set_ue_coef(self, ue, coef: float) -> None:
        """Set coef module vers cette UE"""
        self.update_ue_coef_dict({ue.id: coef})

    def set_ue_coef_dict(self, ue_coef_dict: dict) -> None:
        """set coefs vers les UE (remplace existants)
        ue_coef_dict = { ue_id : coef }
        Les coefs nuls (zéro) ne sont pas stockés: la relation est supprimée.
        """
        if self.formation.has_locked_sems(self.ue.semestre_idx):
            current_app.logger.info(
                "set_ue_coef_dict: locked formation, ignoring request"
            )
            raise ScoValueError("Formation verrouillée")
        changed = False
        for ue_id, coef in ue_coef_dict.items():
            # Existant ?
            coefs = [c for c in self.ue_coefs if c.ue_id == ue_id]
            if coefs:
                ue_coef = coefs[0]
                if coef == 0.0:  # supprime ce coef
                    db.session.delete(ue_coef)
                    changed = True
                elif coef != ue_coef.coef:
                    ue_coef.coef = coef
                    db.session.add(ue_coef)
                    changed = True
            else:
                # crée nouveau coef:
                if coef != 0.0:
                    ue = db.session.get(UniteEns, ue_id)
                    ue_coef = ModuleUECoef(module=self, ue=ue, coef=coef)
                    db.session.add(ue_coef)
                    self.ue_coefs.append(ue_coef)
                    changed = True
        if changed:
            self.formation.invalidate_module_coefs()

    def update_ue_coef_dict(self, ue_coef_dict: dict):
        """update coefs vers UE (ajoute aux existants)"""
        if self.formation.has_locked_sems(self.ue.semestre_idx):
            current_app.logger.info(
                "update_ue_coef_dict: locked formation, ignoring request"
            )
            raise ScoValueError("Formation verrouillée")
        current = self.get_ue_coef_dict()
        current.update(ue_coef_dict)
        self.set_ue_coef_dict(current)

    def get_ue_coef_dict(self):
        """returns { ue_id : coef }"""
        return {p.ue.id: p.coef for p in self.ue_coefs}

    def get_ue_coef_dict_acronyme(self):
        """returns { ue_acronyme : coef }"""
        return {p.ue.acronyme: p.coef for p in self.ue_coefs}

    def delete_ue_coef(self, ue):
        """delete coef"""
        if self.formation.has_locked_sems(self.ue.semestre_idx):
            current_app.logger.info(
                "delete_ue_coef: locked formation, ignoring request"
            )
            raise ScoValueError("Formation verrouillée")
        ue_coef = db.session.get(ModuleUECoef, (self.id, ue.id))
        if ue_coef:
            db.session.delete(ue_coef)
            self.formation.invalidate_module_coefs()

    def get_ue_coefs_sorted(self):
        "les coefs d'UE, trié par numéro et acronyme d'UE"
        # je n'ai pas su mettre un order_by sur le backref sans avoir
        # à redéfinir les relationships...
        return sorted(self.ue_coefs, key=lambda uc: (uc.ue.numero, uc.ue.acronyme))

    def ue_coefs_list(
        self, include_zeros=True, ues: list["UniteEns"] = None
    ) -> list[tuple["UniteEns", float]]:
        """Liste des coefs vers les UE (pour les modules APC).
        Si ues est spécifié, restreint aux UE indiquées.
        Sinon si include_zeros, liste aussi les UE sans coef (donc nul) de ce semestre,
        sauf UE bonus sport.
        Result: List of tuples [ (ue, coef) ]
        """
        if not self.is_apc():
            return []
        if include_zeros and ues is None:
            # Toutes les UE du même semestre:
            ues = (
                self.formation.ues.filter_by(semestre_idx=self.ue.semestre_idx)
                .filter(UniteEns.type != UE_SPORT)
                .order_by(UniteEns.numero, UniteEns.acronyme)
                .all()
            )
            if not ues:
                return []
        if ues:
            coefs_dict = self.get_ue_coef_dict()
            coefs_list = []
            for ue in ues:
                coefs_list.append((ue, coefs_dict.get(ue.id, 0.0)))
            return coefs_list
        # Liste seulement les coefs définis:
        return [(c.ue, c.coef) for c in self.get_ue_coefs_sorted()]

    def get_ue_coefs_descr(self) -> str:
        """Description des coefficients vers les UEs (APC)"""
        coefs_descr = ", ".join(
            [
                f"{ue.acronyme}: {co}"
                for ue, co in self.ue_coefs_list()
                if isinstance(co, float) and co > 0
            ]
        )
        if coefs_descr:
            descr = "Coefs: " + coefs_descr
        else:
            descr = "(pas de coefficients) "
        return descr

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

    def get_edt_ids(self) -> list[str]:
        "les ids pour l'emploi du temps: à défaut, le 1er code 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 []
        ]

    def get_parcours(self) -> list[ApcParcours]:
        """Les parcours utilisant ce module.
        Si tous les parcours, liste vide (!).
        """
        ref_comp = self.formation.referentiel_competence
        if not ref_comp:
            return []
        tous_parcours_ids = {p.id for p in ref_comp.parcours}
        parcours_ids = {p.id for p in self.parcours}
        if tous_parcours_ids == parcours_ids:
            return []
        return self.parcours

    def add_tag(self, tag: "NotesTag"):
        """Add tag to module. Check if already has it."""
        if tag.id in {t.id for t in self.tags}:
            return
        self.tags.append(tag)
        db.session.add(self)
        db.session.flush()

    def set_tags(self, taglist: str | list[str] | None = None):
        """taglist may either be:
        a string with tag names separated by commas ("un,deux")
        or a list of strings (["un", "deux"])
        Remplace les tags existants
        """
        # TODO refactoring ScoTag
        # TODO code à moderniser (+ revoir classe ScoTag, utiliser modèle)
        # TODO Voir ItemSuiviTag et api etud_suivi
        from app.scodoc.sco_tag_module import ScoTag, ModuleTag

        taglist = taglist or []
        if isinstance(taglist, str):
            taglist = taglist.split(",")
        taglist = [t.strip() for t in taglist]
        taglist = [t for t in taglist if t]
        log(f"module.set_tags: module_id={self.id} taglist={taglist}")
        # Check tags syntax
        for tag in taglist:
            if not ScoTag.check_tag_title(tag):
                log(f"module.set_tags({self.id}): invalid tag title")
                return scu.json_error(404, "invalid tag")

        newtags = set(taglist)
        oldtags = set(t.title for t in self.tags)
        to_del = oldtags - newtags
        to_add = newtags - oldtags

        # should be atomic, but it's not.
        for tagname in to_add:
            t = ModuleTag(tagname, object_id=self.id)
        for tagname in to_del:
            t = ModuleTag(tagname)
            t.remove_tag_from_object(self.id)

        return "", http.HTTPStatus.NO_CONTENT


class ModuleUECoef(db.Model):
    """Coefficients des modules vers les UE (APC, BUT)
    En mode APC, ces coefs remplacent le coefficient "PPN" du module.
    """

    __tablename__ = "module_ue_coef"

    module_id = db.Column(
        db.Integer,
        db.ForeignKey("notes_modules.id", ondelete="CASCADE"),
        primary_key=True,
    )
    ue_id = db.Column(
        db.Integer,
        db.ForeignKey("notes_ue.id", ondelete="CASCADE"),
        primary_key=True,
    )
    coef = db.Column(
        db.Float,
        nullable=False,
    )
    module = db.relationship(
        Module,
        backref=db.backref(
            "ue_coefs",
            passive_deletes=True,
            cascade="save-update, merge, delete, delete-orphan",
        ),
    )
    ue = db.relationship(
        "UniteEns",
        backref=db.backref(
            "module_ue_coefs",
            passive_deletes=True,
            cascade="save-update, merge, delete, delete-orphan",
        ),
    )

    def to_dict(self, convert_objects=False) -> dict:
        """If convert_objects, convert all attributes to native types
        (suitable for json encoding).
        """
        d = dict(self.__dict__)
        d.pop("_sa_instance_state", None)
        if convert_objects:
            d["ue"] = self.ue.to_dict(with_module_ue_coefs=False, convert_objects=True)
        return d


class NotesTag(models.ScoDocModel):
    """Tag sur un module"""

    __tablename__ = "notes_tags"
    __table_args__ = (db.UniqueConstraint("title", "dept_id"),)

    id = db.Column(db.Integer, primary_key=True)
    tag_id = db.synonym("id")

    dept_id = db.Column(db.Integer, db.ForeignKey("departement.id"), index=True)
    title = db.Column(db.Text(), nullable=False)

    def __repr__(self):
        return f"<Tag {self.id} {self.title!r}>"

    @classmethod
    def get_or_create(cls, title: str, dept_id: int | None = None) -> "NotesTag":
        """Get tag, or create it if it doesn't yet exists.
        If dept_id unspecified, use current dept.
        """
        dept_id = dept_id if dept_id is not None else g.scodoc_dept_id
        tag = NotesTag.query.filter_by(dept_id=dept_id, title=title).first()
        if tag is None:
            tag = NotesTag(dept_id=dept_id, title=title)
            db.session.add(tag)
            db.session.flush()
        return tag


# Association tag <-> module
notes_modules_tags = db.Table(
    "notes_modules_tags",
    db.Column(
        "tag_id",
        db.Integer,
        db.ForeignKey("notes_tags.id", ondelete="CASCADE"),
    ),
    db.Column(
        "module_id", db.Integer, db.ForeignKey("notes_modules.id", ondelete="CASCADE")
    ),
)

from app.models.ues import UniteEns