ScoDoc/app/models/evaluations.py

# -*- coding: UTF-8 -*

"""ScoDoc models: evaluations
"""
import datetime
from operator import attrgetter

from flask import g, url_for
from flask_login import current_user
import sqlalchemy as sa

from app import db, log
from app.models.etudiants import Identite
from app.models.events import ScolarNews
from app.models.moduleimpls import ModuleImpl
from app.models.notes import NotesNotes
from app.models.ues import UniteEns

from app.scodoc import sco_cache
from app.scodoc.sco_exceptions import AccessDenied, ScoValueError
import app.scodoc.sco_utils as scu
from app.scodoc.sco_xml import quote_xml_attr

MAX_EVALUATION_DURATION = datetime.timedelta(days=365)
NOON = datetime.time(12, 00)
DEFAULT_EVALUATION_TIME = datetime.time(8, 0)

VALID_EVALUATION_TYPES = {0, 1, 2}


class Evaluation(db.Model):
    """Evaluation (contrôle, examen, ...)"""

    __tablename__ = "notes_evaluation"

    id = db.Column(db.Integer, primary_key=True)
    evaluation_id = db.synonym("id")
    moduleimpl_id = db.Column(
        db.Integer, db.ForeignKey("notes_moduleimpl.id"), index=True
    )
    date_debut = db.Column(db.DateTime(timezone=True), nullable=True)
    date_fin = db.Column(db.DateTime(timezone=True), nullable=True)
    description = db.Column(db.Text)
    note_max = db.Column(db.Float)
    coefficient = db.Column(db.Float)
    visibulletin = db.Column(
        db.Boolean, nullable=False, default=True, server_default="true"
    )
    "visible sur les bulletins version intermédiaire"
    publish_incomplete = db.Column(
        db.Boolean, nullable=False, default=False, server_default="false"
    )
    # type d'evaluation: 0 normale, 1 rattrapage, 2 "2eme session"
    evaluation_type = db.Column(
        db.Integer, nullable=False, default=0, server_default="0"
    )
    # ordre de presentation (par défaut, le plus petit numero
    # est la plus ancienne eval):
    numero = db.Column(db.Integer, nullable=False, default=0)
    ues = db.relationship("UniteEns", secondary="evaluation_ue_poids", viewonly=True)

    def __repr__(self):
        return f"""<Evaluation {self.id} {
            self.date_debut.isoformat() if self.date_debut else ''} "{
            self.description[:16] if self.description else ''}">"""

    @classmethod
    def create(
        cls,
        moduleimpl: ModuleImpl = None,
        date_debut: datetime.datetime = None,
        date_fin: datetime.datetime = None,
        description=None,
        note_max=None,
        coefficient=None,
        visibulletin=None,
        publish_incomplete=None,
        evaluation_type=None,
        numero=None,
        **kw,  # ceci pour absorber les éventuel arguments excedentaires
    ):
        """Create an evaluation. Check permission and all arguments.
        Ne crée pas les poids vers les UEs.
        """
        if not moduleimpl.can_edit_evaluation(current_user):
            raise AccessDenied(
                f"Modification évaluation impossible pour {current_user.get_nomplogin()}"
            )
        args = locals()
        del args["cls"]
        del args["kw"]
        check_convert_evaluation_args(moduleimpl, args)
        # Check numeros
        Evaluation.moduleimpl_evaluation_renumber(moduleimpl, only_if_unumbered=True)
        if not "numero" in args or args["numero"] is None:
            args["numero"] = cls.get_new_numero(moduleimpl, args["date_debut"])
        #
        evaluation = Evaluation(**args)
        sco_cache.invalidate_formsemestre(formsemestre_id=moduleimpl.formsemestre_id)
        url = url_for(
            "notes.moduleimpl_status",
            scodoc_dept=g.scodoc_dept,
            moduleimpl_id=moduleimpl.id,
        )
        log(f"created evaluation in {moduleimpl.module.titre_str()}")
        ScolarNews.add(
            typ=ScolarNews.NEWS_NOTE,
            obj=moduleimpl.id,
            text=f"""Création d'une évaluation dans <a href="{url}">{
                    moduleimpl.module.titre_str()}</a>""",
            url=url,
        )
        return evaluation

    @classmethod
    def get_new_numero(
        cls, moduleimpl: ModuleImpl, date_debut: datetime.datetime
    ) -> int:
        """Get a new numero for an evaluation in this moduleimpl
        If necessary, renumber existing evals to make room for a new one.
        """
        n = None
        # Détermine le numero grâce à la date
        # Liste des eval existantes triées par date, la plus ancienne en tete
        evaluations = moduleimpl.evaluations.order_by(Evaluation.date_debut).all()
        if date_debut is not None:
            next_eval = None
            t = date_debut
            for e in evaluations:
                if e.date_debut and e.date_debut > t:
                    next_eval = e
                    break
            if next_eval:
                n = _moduleimpl_evaluation_insert_before(evaluations, next_eval)
            else:
                n = None  # à placer en fin
        if n is None:  # pas de date ou en fin:
            if evaluations:
                n = evaluations[-1].numero + 1
            else:
                n = 0  # the only one
        return n

    def delete(self):
        "delete evaluation (commit) (check permission)"
        from app.scodoc import sco_evaluation_db

        modimpl: ModuleImpl = self.moduleimpl
        if not modimpl.can_edit_evaluation(current_user):
            raise AccessDenied(
                f"Modification évaluation impossible pour {current_user.get_nomplogin()}"
            )
        notes_db = sco_evaluation_db.do_evaluation_get_all_notes(
            self.id
        )  # { etudid : value }
        notes = [x["value"] for x in notes_db.values()]
        if notes:
            raise ScoValueError(
                "Impossible de supprimer cette évaluation: il reste des notes"
            )
        log(f"deleting evaluation {self}")
        db.session.delete(self)
        db.session.commit()

        # inval cache pour ce semestre
        sco_cache.invalidate_formsemestre(formsemestre_id=modimpl.formsemestre_id)
        # news
        url = url_for(
            "notes.moduleimpl_status",
            scodoc_dept=g.scodoc_dept,
            moduleimpl_id=modimpl.id,
        )
        ScolarNews.add(
            typ=ScolarNews.NEWS_NOTE,
            obj=modimpl.id,
            text=f"""Suppression d'une évaluation dans <a href="{
                url
            }">{modimpl.module.titre}</a>""",
            url=url,
        )

    def to_dict(self) -> dict:
        "Représentation dict (riche, compat ScoDoc 7)"
        e_dict = dict(self.__dict__)
        e_dict.pop("_sa_instance_state", None)
        # ScoDoc7 output_formators
        e_dict["evaluation_id"] = self.id
        e_dict["date_debut"] = self.date_debut.isoformat() if self.date_debut else None
        e_dict["date_fin"] = self.date_debut.isoformat() if self.date_fin else None
        e_dict["numero"] = self.numero or 0
        e_dict["poids"] = self.get_ue_poids_dict()  # { ue_id : poids }

        # Deprecated
        e_dict["jour"] = self.date_debut.strftime("%d/%m/%Y") if self.date_debut else ""

        return evaluation_enrich_dict(self, e_dict)

    def to_dict_api(self) -> dict:
        "Représentation dict pour API JSON"
        return {
            "coefficient": self.coefficient,
            "date_debut": self.date_debut.isoformat() if self.date_debut else "",
            "date_fin": self.date_fin.isoformat() if self.date_fin else "",
            "description": self.description,
            "evaluation_type": self.evaluation_type,
            "id": self.id,
            "moduleimpl_id": self.moduleimpl_id,
            "note_max": self.note_max,
            "numero": self.numero,
            "poids": self.get_ue_poids_dict(),
            "publish_incomplete": self.publish_incomplete,
            "visibulletin": self.visibulletin,
            # Deprecated (supprimer avant #sco9.7)
            "date": self.date_debut.date().isoformat() if self.date_debut else "",
            "heure_debut": self.date_debut.time().isoformat()
            if self.date_debut
            else "",
            "heure_fin": self.date_fin.time().isoformat() if self.date_fin else "",
        }

    def to_dict_bul(self) -> dict:
        "dict pour les bulletins json"
        # c'est la version API avec quelques champs legacy en plus
        e_dict = self.to_dict_api()
        # Pour les bulletins (json ou xml), quote toujours la description
        e_dict["description"] = quote_xml_attr(self.description or "")
        # deprecated fields:
        e_dict["evaluation_id"] = self.id
        e_dict["jour"] = e_dict["date_debut"]  # chaine iso
        e_dict["heure_debut"] = (
            self.date_debut.time().isoformat() if self.date_debut else ""
        )
        e_dict["heure_fin"] = self.date_fin.time().isoformat() if self.date_fin else ""

        return e_dict

    def from_dict(self, data):
        """Set evaluation attributes from given dict values."""
        check_convert_evaluation_args(self.moduleimpl, data)
        if data.get("numero") is None:
            data["numero"] = Evaluation.get_max_numero(self.moduleimpl.id) + 1
        for k in self.__dict__.keys():
            if k != "_sa_instance_state" and k != "id" and k in data:
                setattr(self, k, data[k])

    @classmethod
    def get_max_numero(cls, moduleimpl_id: int) -> int:
        """Return max numero among evaluations in this
        moduleimpl (0 if None)
        """
        max_num = (
            db.session.query(sa.sql.functions.max(Evaluation.numero))
            .filter_by(moduleimpl_id=moduleimpl_id)
            .first()[0]
        )
        return max_num or 0

    @classmethod
    def moduleimpl_evaluation_renumber(
        cls, moduleimpl: ModuleImpl, only_if_unumbered=False
    ):
        """Renumber evaluations in this moduleimpl, according to their date. (numero=0: oldest one)
        Needed because previous versions of ScoDoc did not have eval numeros
        Note: existing numeros are ignored
        """
        # Liste des eval existantes triées par date, la plus ancienne en tete
        evaluations = moduleimpl.evaluations.order_by(
            Evaluation.date_debut, Evaluation.numero
        ).all()
        all_numbered = all(e.numero is not None for e in evaluations)
        if all_numbered and only_if_unumbered:
            return  # all ok

        # Reset all numeros:
        i = 1
        for e in evaluations:
            e.numero = i
            db.session.add(e)
            i += 1
        db.session.commit()

    def descr_heure(self) -> str:
        "Description de la plage horaire pour affichages ('de 13h00 à 14h00')"
        if self.date_debut and (not self.date_fin or self.date_fin == self.date_debut):
            return f"""à {self.date_debut.strftime("%Hh%M")}"""
        elif self.date_debut and self.date_fin:
            return f"""de {self.date_debut.strftime("%Hh%M")
                           } à {self.date_fin.strftime("%Hh%M")}"""
        else:
            return ""

    def descr_duree(self) -> str:
        "Description de la durée pour affichages ('3h' ou '2h30')"
        if self.date_debut is None or self.date_fin is None:
            return ""
        minutes = (self.date_fin - self.date_debut).seconds // 60
        duree = f"{minutes // 60}h"
        minutes = minutes % 60
        if minutes != 0:
            duree += f"{minutes:02d}"
        return duree

    def descr_date(self) -> str:
        """Description de la date pour affichages
        'sans date'
        'le 21/9/2021 à 13h'
        'le 21/9/2021 de 13h à 14h30'
        'du 21/9/2021 à 13h30 au 23/9/2021 à 15h'
        """
        if self.date_debut is None:
            return "sans date"

        def _h(dt: datetime.datetime) -> str:
            if dt.minute:
                return dt.strftime("%Hh%M")
            return f"{dt.hour}h"

        if self.date_fin is None:
            return f"le {self.date_debut.strftime('%d/%m/%Y')} à {_h(self.date_debut)}"
        if self.date_debut.date() == self.date_fin.date():  # même jour
            if self.date_debut.time() == self.date_fin.time():
                return (
                    f"le {self.date_debut.strftime('%d/%m/%Y')} à {_h(self.date_debut)}"
                )
            return f"""le {self.date_debut.strftime('%d/%m/%Y')} de {
                _h(self.date_debut)} à {_h(self.date_fin)}"""
        # évaluation sur plus d'une journée
        return f"""du {self.date_debut.strftime('%d/%m/%Y')} à {
            _h(self.date_debut)} au {self.date_fin.strftime('%d/%m/%Y')} à {_h(self.date_fin)}"""

    def heure_debut(self) -> str:
        """L'heure de début (sans la date), en ISO.
        Chaine vide si non renseignée."""
        return self.date_debut.time().isoformat("minutes") if self.date_debut else ""

    def heure_fin(self) -> str:
        """L'heure de fin (sans la date), en ISO.
        Chaine vide si non renseignée."""
        return self.date_fin.time().isoformat("minutes") if self.date_fin else ""

    def clone(self, not_copying=()):
        """Clone, not copying the given attrs
        Attention: la copie n'a pas d'id avant le prochain commit
        """
        d = dict(self.__dict__)
        d.pop("id")  # get rid of id
        d.pop("_sa_instance_state")  # get rid of SQLAlchemy special attr
        for k in not_copying:
            d.pop(k)
        copy = self.__class__(**d)
        db.session.add(copy)
        return copy

    def is_matin(self) -> bool:
        "Evaluation commençant le matin (faux si pas de date)"
        if not self.date_debut:
            return False
        return self.date_debut.time() < NOON

    def is_apresmidi(self) -> bool:
        "Evaluation commençant l'après midi (faux si pas de date)"
        if not self.date_debut:
            return False
        return self.date_debut.time() >= NOON

    def set_default_poids(self) -> bool:
        """Initialize les poids vers les UE à leurs valeurs par défaut
        C'est à dire à 1 si le coef. module/UE est non nul, 0 sinon.
        Les poids existants ne sont pas modifiés.
        Return True if (uncommited) modification, False otherwise.
        """
        ue_coef_dict = self.moduleimpl.module.get_ue_coef_dict()
        sem_ues = self.moduleimpl.formsemestre.get_ues(with_sport=False)
        modified = False
        for ue in sem_ues:
            existing_poids = EvaluationUEPoids.query.filter_by(
                ue=ue, evaluation=self
            ).first()
            if existing_poids is None:
                coef_ue = ue_coef_dict.get(ue.id, 0.0) or 0.0
                if coef_ue > 0:
                    poids = 1.0  # par défaut au départ
                else:
                    poids = 0.0
                self.set_ue_poids(ue, poids)
                modified = True
        return modified

    def set_ue_poids(self, ue, poids: float) -> None:
        """Set poids évaluation vers cette UE"""
        self.update_ue_poids_dict({ue.id: poids})

    def set_ue_poids_dict(self, ue_poids_dict: dict) -> None:
        """set poids vers les UE (remplace existants)
        ue_poids_dict = { ue_id : poids }
        """
        L = []
        for ue_id, poids in ue_poids_dict.items():
            ue = db.session.get(UniteEns, ue_id)
            if ue is None:
                raise ScoValueError("poids vers une UE inexistante")
            ue_poids = EvaluationUEPoids(evaluation=self, ue=ue, poids=poids)
            L.append(ue_poids)
            db.session.add(ue_poids)
        self.ue_poids = L  # backref # pylint:disable=attribute-defined-outside-init
        self.moduleimpl.invalidate_evaluations_poids()  # inval cache

    def update_ue_poids_dict(self, ue_poids_dict: dict) -> None:
        """update poids vers UE (ajoute aux existants)"""
        current = self.get_ue_poids_dict()
        current.update(ue_poids_dict)
        self.set_ue_poids_dict(current)

    def get_ue_poids_dict(self, sort=False) -> dict:
        """returns { ue_id : poids }
        Si sort, trie par UE
        """
        if sort:
            return {
                p.ue.id: p.poids
                for p in sorted(
                    self.ue_poids, key=attrgetter("ue.numero", "ue.acronyme")
                )
            }

        return {p.ue.id: p.poids for p in self.ue_poids}

    def get_ue_poids_str(self) -> str:
        """string describing poids, for excel cells and pdfs
        Note: si les poids ne sont pas initialisés (poids par défaut),
        ils ne sont pas affichés.
        """
        # restreint aux UE du semestre dans lequel est cette évaluation
        # au cas où le module ait changé de semestre et qu'il reste des poids
        evaluation_semestre_idx = self.moduleimpl.module.semestre_id
        return ", ".join(
            [
                f"{p.ue.acronyme}: {p.poids}"
                for p in sorted(
                    self.ue_poids, key=lambda p: (p.ue.numero or 0, p.ue.acronyme)
                )
                if evaluation_semestre_idx == p.ue.semestre_idx
            ]
        )

    def get_etud_note(self, etud: Identite) -> NotesNotes:
        """La note de l'étudiant, ou None si pas noté.
        (nb: pas de cache, lent, ne pas utiliser pour des calculs)
        """
        return NotesNotes.query.filter_by(etudid=etud.id, evaluation_id=self.id).first()


class EvaluationUEPoids(db.Model):
    """Poids des évaluations (BUT)
    association many to many
    """

    evaluation_id = db.Column(
        db.Integer,
        db.ForeignKey("notes_evaluation.id", ondelete="CASCADE"),
        primary_key=True,
    )
    ue_id = db.Column(
        db.Integer,
        db.ForeignKey("notes_ue.id", ondelete="CASCADE"),
        primary_key=True,
    )
    poids = db.Column(
        db.Float,
        nullable=False,
    )
    evaluation = db.relationship(
        Evaluation,
        backref=db.backref("ue_poids", cascade="all, delete-orphan"),
    )
    ue = db.relationship(
        UniteEns,
        backref=db.backref("evaluation_ue_poids", cascade="all, delete-orphan"),
    )

    def __repr__(self):
        return f"<EvaluationUEPoids {self.evaluation} {self.ue} poids={self.poids}>"


# Fonction héritée de ScoDoc7
def evaluation_enrich_dict(e: Evaluation, e_dict: dict):
    """add or convert some fields in an evaluation dict"""
    # For ScoDoc7 compat
    e_dict["heure_debut"] = e.date_debut.strftime("%Hh%M") if e.date_debut else ""
    e_dict["heure_fin"] = e.date_fin.strftime("%Hh%M") if e.date_fin else ""
    e_dict["jour_iso"] = e.date_debut.isoformat() if e.date_debut else ""
    # Calcule durée en minutes
    e_dict["descrheure"] = e.descr_heure()
    e_dict["descrduree"] = e.descr_duree()
    # matin, apresmidi: utile pour se referer aux absences:
    # note août 2023: si l'évaluation s'étend sur plusieurs jours,
    # cet indicateur n'a pas grand sens
    if e.date_debut and e.date_debut.time() < datetime.time(12, 00):
        e_dict["matin"] = 1
    else:
        e_dict["matin"] = 0
    if e.date_fin and e.date_fin.time() > datetime.time(12, 00):
        e_dict["apresmidi"] = 1
    else:
        e_dict["apresmidi"] = 0
    return e_dict


def check_convert_evaluation_args(moduleimpl: ModuleImpl, data: dict):
    """Check coefficient, dates and duration, raises exception if invalid.
    Convert date and time strings to date and time objects.

    Set required default value for unspecified fields.
    May raise ScoValueError.
    """
    # --- description
    data["description"] = data.get("description", "") or ""
    if len(data["description"]) > scu.MAX_TEXT_LEN:
        raise ScoValueError("description too large")

    # --- evaluation_type
    try:
        data["evaluation_type"] = int(data.get("evaluation_type", 0) or 0)
        if not data["evaluation_type"] in VALID_EVALUATION_TYPES:
            raise ScoValueError("invalid evaluation_type value")
    except ValueError as exc:
        raise ScoValueError("invalid evaluation_type value") from exc

    # --- note_max (bareme)
    note_max = data.get("note_max", 20.0) or 20.0
    try:
        note_max = float(note_max)
    except ValueError as exc:
        raise ScoValueError("invalid note_max value") from exc
    if note_max < 0:
        raise ScoValueError("invalid note_max value (must be positive or null)")
    data["note_max"] = note_max
    # --- coefficient
    coef = data.get("coefficient", None)
    if coef is None:
        coef = 1.0
    try:
        coef = float(coef)
    except ValueError as exc:
        raise ScoValueError("invalid coefficient value") from exc
    if coef < 0:
        raise ScoValueError("invalid coefficient value (must be positive or null)")
    data["coefficient"] = coef
    # --- date de l'évaluation
    formsemestre = moduleimpl.formsemestre
    date_debut = data.get("date_debut", None)
    if date_debut:
        if isinstance(date_debut, str):
            data["date_debut"] = datetime.datetime.fromisoformat(date_debut)
        if data["date_debut"].tzinfo is None:
            data["date_debut"] = scu.TIME_ZONE.localize(data["date_debut"])
        if not (
            formsemestre.date_debut
            <= data["date_debut"].date()
            <= formsemestre.date_fin
        ):
            raise ScoValueError(
                f"""La date de début de l'évaluation ({
                    data["date_debut"].strftime("%d/%m/%Y")
                }) n'est pas dans le semestre !""",
                dest_url="javascript:history.back();",
            )
    date_fin = data.get("date_fin", None)
    if date_fin:
        if isinstance(date_fin, str):
            data["date_fin"] = datetime.datetime.fromisoformat(date_fin)
        if data["date_fin"].tzinfo is None:
            data["date_fin"] = scu.TIME_ZONE.localize(data["date_fin"])
        if not (
            formsemestre.date_debut <= data["date_fin"].date() <= formsemestre.date_fin
        ):
            raise ScoValueError(
                f"""La date de fin de l'évaluation ({
                    data["date_fin"].strftime("%d/%m/%Y")
                }) n'est pas dans le semestre !""",
                dest_url="javascript:history.back();",
            )
    if date_debut and date_fin:
        duration = data["date_fin"] - data["date_debut"]
        if duration.total_seconds() < 0 or duration > MAX_EVALUATION_DURATION:
            raise ScoValueError("Heures de l'évaluation incohérentes !")
    # # --- heures
    # heure_debut = data.get("heure_debut", None)
    # if heure_debut and not isinstance(heure_debut, datetime.time):
    #     if date_format == "dmy":
    #         data["heure_debut"] = heure_to_time(heure_debut)
    #     else:  # ISO
    #         data["heure_debut"] = datetime.time.fromisoformat(heure_debut)
    # heure_fin = data.get("heure_fin", None)
    # if heure_fin and not isinstance(heure_fin, datetime.time):
    #     if date_format == "dmy":
    #         data["heure_fin"] = heure_to_time(heure_fin)
    #     else:  # ISO
    #         data["heure_fin"] = datetime.time.fromisoformat(heure_fin)


def heure_to_time(heure: str) -> datetime.time:
    "Convert external heure ('10h22' or '10:22') to a time"
    t = heure.strip().upper().replace("H", ":")
    h, m = t.split(":")[:2]
    return datetime.time(int(h), int(m))


def _time_duration_HhM(heure_debut: str, heure_fin: str) -> int:
    """duree (nb entier de minutes) entre deux heures a notre format
    ie 12h23
    """
    if heure_debut and heure_fin:
        h0, m0 = [int(x) for x in heure_debut.split("h")]
        h1, m1 = [int(x) for x in heure_fin.split("h")]
        d = (h1 - h0) * 60 + (m1 - m0)
        return d
    else:
        return None


def _moduleimpl_evaluation_insert_before(
    evaluations: list[Evaluation], next_eval: Evaluation
) -> int:
    """Renumber evaluations such that an evaluation with can be inserted before next_eval
    Returns numero suitable for the inserted evaluation
    """
    if next_eval:
        n = next_eval.numero
        if n is None:
            Evaluation.moduleimpl_evaluation_renumber(next_eval.moduleimpl)
            n = next_eval.numero
    else:
        n = 1
    # all numeros >= n are incremented
    for e in evaluations:
        if e.numero >= n:
            e.numero += 1
            db.session.add(e)
    db.session.commit()
    return n