ScoDoc/app/models/groups.py

# -*- coding: UTF-8 -*
##############################################################################
# ScoDoc
# Copyright (c) 1999 - 2024 Emmanuel Viennet.  All rights reserved.
# See LICENSE
##############################################################################

"""ScoDoc models: Groups & partitions
"""
from operator import attrgetter
from sqlalchemy.exc import IntegrityError

from app import db, log
from app.models import ScoDocModel, GROUPNAME_STR_LEN, SHORT_STR_LEN
from app.models.etudiants import Identite
from app.models.events import Scolog
from app.scodoc import sco_cache
from app.scodoc import sco_utils as scu
from app.scodoc.sco_exceptions import AccessDenied, ScoValueError


class Partition(ScoDocModel):
    """Partition: découpage d'une promotion en groupes"""

    __table_args__ = (db.UniqueConstraint("formsemestre_id", "partition_name"),)

    id = db.Column(db.Integer, primary_key=True)
    partition_id = db.synonym("id")
    formsemestre_id = db.Column(
        db.Integer,
        db.ForeignKey("notes_formsemestre.id"),
        index=True,
    )
    # "TD", "TP", ... (NULL for 'all')
    partition_name = db.Column(db.String(SHORT_STR_LEN))
    # Numero = ordre de presentation)
    numero = db.Column(db.Integer, nullable=False, default=0)
    # Calculer le rang ?
    bul_show_rank = db.Column(
        db.Boolean(), nullable=False, default=False, server_default="false"
    )
    # Montrer quand on indique les groupes de l'étudiant ?
    show_in_lists = db.Column(
        db.Boolean(), nullable=False, default=True, server_default="true"
    )
    # Editable (créer/renommer groupes) ? (faux pour les groupes de parcours)
    groups_editable = db.Column(
        db.Boolean(), nullable=False, default=True, server_default="true"
    )
    groups = db.relationship(
        "GroupDescr",
        backref=db.backref("partition", lazy=True),
        lazy="dynamic",
        cascade="all, delete-orphan",
        order_by="GroupDescr.numero, GroupDescr.group_name",
    )
    _sco_dept_relations = ("FormSemestre",)

    def __init__(self, **kwargs):
        super(Partition, self).__init__(**kwargs)
        if self.numero is None:
            # génère numero à la création
            last_partition = Partition.query.order_by(Partition.numero.desc()).first()
            if last_partition:
                self.numero = last_partition.numero + 1
            else:
                self.numero = 1

    def __repr__(self):
        return f"""<{self.__class__.__name__} {self.id} "{self.partition_name or '(default)'}">"""

    @classmethod
    def check_name(
        cls, formsemestre: "FormSemestre", partition_name: str, existing=False
    ) -> bool:
        """check if a partition named 'partition_name' can be created in the given formsemestre.
        If existing is True, allow a partition_name already existing in the formsemestre.
        """
        if not isinstance(partition_name, str):
            return False
        if not (0 < len(partition_name.strip()) < SHORT_STR_LEN):
            return False
        if (not existing) and (
            partition_name in [p.partition_name for p in formsemestre.partitions]
        ):
            return False
        return True

    @classmethod
    def formsemestre_remove_etud(cls, formsemestre_id: int, etud: "Identite"):
        "retire l'étudiant de toutes les partitions de ce semestre"
        for group in GroupDescr.query.join(Partition).filter_by(
            formsemestre_id=formsemestre_id
        ):
            group.remove_etud(etud)

    def is_default(self) -> bool:
        "vrai si partition par défault (tous les étudiants)"
        return not self.partition_name

    def is_parcours(self) -> bool:
        "Vrai s'il s'agit de la partition de parcours"
        return self.partition_name == scu.PARTITION_PARCOURS

    def to_dict(self, with_groups=False, str_keys: bool = False) -> dict:
        """as a dict, with or without groups.
        If str_keys, convert integer dict keys to strings (useful for JSON)
        """
        d = dict(self.__dict__)
        d["partition_id"] = self.id
        d.pop("_sa_instance_state", None)
        d.pop("formsemestre", None)

        if with_groups:
            groups = sorted(self.groups, key=attrgetter("numero", "group_name"))
            # un dict et non plus une liste, pour JSON
            if str_keys:
                d["groups"] = {
                    str(group.id): group.to_dict(with_partition=False)
                    for group in groups
                }
            else:
                d["groups"] = {
                    group.id: group.to_dict(with_partition=False) for group in groups
                }
        return d

    def get_etud_group(self, etudid: int) -> "GroupDescr|None":
        "Le groupe de l'étudiant dans cette partition, ou None si pas présent"
        return (
            GroupDescr.query.filter_by(partition_id=self.id)
            .join(group_membership)
            .filter_by(etudid=etudid)
            .first()
        )

    def set_etud_group(self, etud: "Identite", group: "GroupDescr") -> bool:
        """Affect etudid to group_id in given partition.
        Raises IntegrityError si conflit,
        or ValueError si ce group_id n'est pas dans cette partition
        ou que l'étudiant n'est pas inscrit au semestre.
        Return True si changement, False s'il était déjà dans ce groupe.
        """
        if not group.id in (g.id for g in self.groups):
            raise ScoValueError(
                f"""Le groupe {group.id} n'est pas dans la partition {
                    self.partition_name or "tous"}"""
            )
        if etud.id not in (e.id for e in self.formsemestre.etuds):
            raise ScoValueError(
                f"""étudiant {etud.nomprenom} non inscrit au formsemestre du groupe {
                    group.group_name}"""
            )
        try:
            existing_row = (
                db.session.query(group_membership)
                .filter_by(etudid=etud.id)
                .join(GroupDescr)
                .filter_by(partition_id=self.id)
                .first()
            )
            if existing_row:
                existing_group_id = existing_row[1]
                if group.id == existing_group_id:
                    return False
                # Fait le changement avec l'ORM sinon risque élevé de blocage
                existing_group = db.session.get(GroupDescr, existing_group_id)
                db.session.commit()
                group.etuds.append(etud)
                existing_group.etuds.remove(etud)
                db.session.add(etud)
                db.session.add(existing_group)
                db.session.add(group)
            else:
                new_row = group_membership.insert().values(
                    etudid=etud.id, group_id=group.id
                )
                db.session.execute(new_row)
            db.session.commit()
        except IntegrityError:
            db.session.rollback()
            raise
        return True

    def create_group(self, group_name="", default=False) -> "GroupDescr":
        "Crée un groupe dans cette partition"
        if not self.formsemestre.can_change_groups():
            raise AccessDenied(
                """Vous n'avez pas le droit d'effectuer cette opération,
                ou bien le semestre est verrouillé !"""
            )
        if group_name:
            group_name = group_name.strip()
        if not group_name and not default:
            raise ValueError("invalid group name: ()")
        if not GroupDescr.check_name(self, group_name, default=default):
            raise ScoValueError(
                f"Le groupe {group_name} existe déjà dans cette partition"
            )
        numeros = [g.numero if g.numero is not None else 0 for g in self.groups]
        if len(numeros) > 0:
            new_numero = max(numeros) + 1
        else:
            new_numero = 0
        group = GroupDescr(partition=self, group_name=group_name, numero=new_numero)
        db.session.add(group)
        db.session.commit()
        log(f"create_group: created group_id={group.id}")
        #
        return group


class GroupDescr(ScoDocModel):
    """Description d'un groupe d'une partition"""

    __tablename__ = "group_descr"
    __table_args__ = (db.UniqueConstraint("partition_id", "group_name"),)

    id = db.Column(db.Integer, primary_key=True)
    group_id = db.synonym("id")
    partition_id = db.Column(db.Integer, db.ForeignKey("partition.id"))
    group_name = db.Column(db.String(GROUPNAME_STR_LEN))
    """nom du groupe: "A", "C2", ...  (NULL for 'all')"""
    edt_id: str | None = db.Column(db.Text(), index=True, nullable=True)
    "identifiant emplois du temps (unicité non imposée)"
    numero = db.Column(db.Integer, nullable=False, default=0)
    "Numero = ordre de presentation"

    _sco_dept_relations = (
        "Partition",
        "FormSemestre",
    )

    etuds = db.relationship(
        "Identite",
        secondary="group_membership",
        lazy="dynamic",
    )

    def __repr__(self):
        return (
            f"""<{self.__class__.__name__} {self.id} "{self.group_name or '(tous)'}">"""
        )

    @classmethod
    def filter_model_attributes(cls, data: dict, excluded: set[str] = None) -> dict:
        """Returns a copy of dict with only the keys belonging to the Model and not in excluded.
        Exclude `partition_id` : a group cannot be moved from a partition to another.
        """
        return super().filter_model_attributes(
            data,
            excluded=(excluded or set()) | {"partition_id"},
        )

    def get_nom_with_part(self, default="-") -> str:
        """Nom avec partition: 'TD A'
        Si groupe par défaut (tous), utilise default ou "-"
        """
        if self.partition.partition_name is None:
            return default
        return f"{self.partition.partition_name or ''} {self.group_name or '-'}"

    def to_dict(self, with_partition=True) -> dict:
        """as a dict, with or without partition"""
        if with_partition:
            partition_dict = self.partition.to_dict(with_groups=False)
        d = dict(self.__dict__)
        d.pop("_sa_instance_state", None)
        if with_partition:
            d["partition"] = partition_dict
        return d

    def get_edt_ids(self) -> list[str]:
        "les ids normalisés pour l'emploi du temps: à défaut, le nom scodoc du groupe"
        return [
            scu.normalize_edt_id(x)
            for x in scu.split_id(self.edt_id) or [self.group_name] or []
        ]

    def get_nb_inscrits(self) -> int:
        """Nombre inscrits à ce group et au formsemestre.
        C'est nécessaire car lors d'une désinscription, on conserve l'appartenance
        aux groupes pour faciliter une éventuelle ré-inscription.
        """
        from app.models.formsemestre import FormSemestreInscription

        return (
            Identite.query.join(group_membership)
            .filter_by(group_id=self.id)
            .join(FormSemestreInscription)
            .filter_by(formsemestre_id=self.partition.formsemestre.id)
            .count()
        )

    @classmethod
    def check_name(
        cls, partition: "Partition", group_name: str, existing=False, default=False
    ) -> bool:
        """check if a group named 'group_name' can be created in the given partition.
        If existing is True, allow a group_name already existing in the partition.
        If default, name must be empty and default group must not yet exists.
        """
        if not isinstance(group_name, str):
            return False
        if not default and not (0 < len(group_name.strip()) < GROUPNAME_STR_LEN):
            return False
        if (not existing) and (group_name in [g.group_name for g in partition.groups]):
            return False
        return True

    def set_name(self, group_name: str, dest_url: str = None):
        """Set group name, and optionally edt_id.
        Check permission (partition must be groups_editable)
        and invalidate caches. Commit session.
        dest_url is used for error messages.
        """
        if not self.partition.formsemestre.can_change_groups():
            raise AccessDenied("Vous n'avez pas le droit d'effectuer cette opération !")
        if self.group_name is None:
            raise ValueError("can't set a name to default group")
        if not self.partition.groups_editable:
            raise AccessDenied("Partition non éditable")
        if group_name:
            group_name = group_name.strip()
        if not group_name:
            raise ScoValueError("nom de groupe vide !", dest_url=dest_url)
        if group_name != self.group_name and not GroupDescr.check_name(
            self.partition, group_name
        ):
            raise ScoValueError(
                "Le nom de groupe existe déjà dans la partition", dest_url=dest_url
            )

        self.group_name = group_name
        db.session.add(self)
        db.session.commit()
        sco_cache.invalidate_formsemestre(
            formsemestre_id=self.partition.formsemestre_id
        )

    def set_edt_id(self, edt_id: str):
        "Set edt_id. Check permission. Commit session."
        if not self.partition.formsemestre.can_change_groups():
            raise AccessDenied("Vous n'avez pas le droit d'effectuer cette opération !")
        if isinstance(edt_id, str):
            edt_id = edt_id.strip() or None
        self.edt_id = edt_id
        db.session.add(self)
        db.session.commit()

    def remove_etud(self, etud: "Identite"):
        "Enlève l'étudiant de ce groupe s'il en fait partie (ne fait rien sinon)"
        if etud in self.etuds:
            self.etuds.remove(etud)
            Scolog.logdb(
                method="group_remove_etud",
                etudid=etud.id,
                msg=f"Retrait du groupe {self.group_name} de {self.partition.partition_name}",
            )
            db.session.commit()
            # Update parcours
            if self.partition.partition_name == scu.PARTITION_PARCOURS:
                self.partition.formsemestre.update_inscriptions_parcours_from_groups(
                    etudid=etud.id
                )
            sco_cache.invalidate_formsemestre(self.partition.formsemestre_id)


group_membership = db.Table(
    "group_membership",
    db.Column("etudid", db.Integer, db.ForeignKey("identite.id", ondelete="CASCADE")),
    db.Column("group_id", db.Integer, db.ForeignKey("group_descr.id")),
    db.UniqueConstraint("etudid", "group_id"),
)