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
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.notesdb as ndb
import app.scodoc.sco_utils as scu

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"
    )
    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,
        jour=None,
        heure_debut=None,
        heure_fin=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."""
        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,
        )
        ScolarNews.add(
            typ=ScolarNews.NEWS_NOTE,
            obj=moduleimpl.id,
            text=f"""Création d'une évaluation dans <a href="{url}">{
                    moduleimpl.module.titre or '(module sans titre)'}</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 > 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 to_dict(self) -> dict:
        "Représentation dict (riche, compat ScoDoc 7)"
        e = dict(self.__dict__)
        e.pop("_sa_instance_state", None)
        # ScoDoc7 output_formators
        e["evaluation_id"] = self.id
        e["date_debut"] = e.date_debut.isoformat() if e.date_debut else None
        e["date_fin"] = e.date_debut.isoformat() if e.date_fin else None
        e["numero"] = ndb.int_null_is_zero(e["numero"])
        e["poids"] = self.get_ue_poids_dict()  # { ue_id : poids }

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

        return evaluation_enrich_dict(e)

    def to_dict_api(self) -> dict:
        "Représentation dict pour API JSON"
        return {
            "coefficient": self.coefficient,
            "date_debut": self.date_debut.isoformat(),
            "date_fin": self.date_fin.isoformat(),
            "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,
            "visi_bulletin": self.visibulletin,
        }

    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() + 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"
        if self.heure_debut and (
            not self.heure_fin or self.heure_fin == self.heure_debut
        ):
            return f"""à {self.heure_debut.strftime("%Hh%M")}"""
        elif self.heure_debut and self.heure_fin:
            return f"""de {self.heure_debut.strftime("%Hh%M")} à {self.heure_fin.strftime("%Hh%M")}"""
        else:
            return ""

    def descr_duree(self) -> str:
        "Description de la durée pour affichages"
        if self.heure_debut is None and self.heure_fin is None:
            return ""
        debut = self.heure_debut or DEFAULT_EVALUATION_TIME
        fin = self.heure_fin or DEFAULT_EVALUATION_TIME
        d = (fin.hour * 60 + fin.minute) - (debut.hour * 60 + debut.minute)
        duree = f"{d//60}h"
        if d % 60:
            duree += f"{d%60:02d}"
        return duree

    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)
            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 à refactorer
def evaluation_enrich_dict(e: dict):
    """add or convert some fields in an evaluation dict"""
    # For ScoDoc7 compat
    heure_debut_dt = e["date_debut"].time()
    heure_fin_dt = e["date_fin"].time()
    e["heure_debut"] = heure_debut_dt.strftime("%Hh%M")
    e["heure_fin"] = heure_fin_dt.strftime("%Hh%M")
    e["jour_iso"] = e["date_debut"].isoformat()  # XXX
    heure_debut, heure_fin = e["heure_debut"], e["heure_fin"]
    d = _time_duration_HhM(heure_debut, heure_fin)
    if d is not None:
        m = d % 60
        e["duree"] = "%dh" % (d / 60)
        if m != 0:
            e["duree"] += "%02d" % m
    else:
        e["duree"] = ""
    if heure_debut and (not heure_fin or heure_fin == heure_debut):
        e["descrheure"] = " à " + heure_debut
    elif heure_debut and heure_fin:
        e["descrheure"] = " de %s à %s" % (heure_debut, heure_fin)
    else:
        e["descrheure"] = ""
    # matin, apresmidi: utile pour se referer aux absences:

    if e["jour"] and heure_debut_dt < datetime.time(12, 00):
        e["matin"] = 1
    else:
        e["matin"] = 0
    if e["jour"] and heure_fin_dt > datetime.time(12, 00):
        e["apresmidi"] = 1
    else:
        e["apresmidi"] = 0
    return e


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
    description = data.get("description", "")
    if len(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:
        raise ScoValueError("Invalid evaluation_type value")

    # --- note_max (bareme)
    note_max = data.get("note_max", 20.0) or 20.0
    try:
        note_max = float(note_max)
    except ValueError:
        raise ScoValueError("Invalid note_max value")
    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", 1.0) or 1.0
    try:
        coef = float(coef)
    except ValueError:
        raise ScoValueError("Invalid coefficient value")
    if coef < 0:
        raise ScoValueError("Invalid coefficient value (must be positive or null)")
    data["coefficient"] = coef
    # --- jour (date de l'évaluation)
    jour = data.get("jour", None)
    if jour and not isinstance(jour, datetime.date):
        if date_format == "dmy":
            y, m, d = [int(x) for x in ndb.DateDMYtoISO(jour).split("-")]
            jour = datetime.date(y, m, d)
        else:  # ISO
            jour = datetime.date.fromisoformat(jour)
        formsemestre = moduleimpl.formsemestre
        if (jour > formsemestre.date_fin) or (jour < formsemestre.date_debut):
            raise ScoValueError(
                f"""La date de l'évaluation ({jour.strftime("%d/%m/%Y")}) n'est pas dans le semestre !""",
                dest_url="javascript:history.back();",
            )
    data["jour"] = jour
    # --- 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)
    if jour and ((not heure_debut) or (not heure_fin)):
        raise ScoValueError("Les heures doivent être précisées")
    if heure_debut and heure_fin:
        duration = ((data["heure_fin"].hour * 60) + data["heure_fin"].minute) - (
            (data["heure_debut"].hour * 60) + data["heure_debut"].minute
        )
        if duration < 0 or duration > 60 * 12:
            raise ScoValueError("Heures de l'évaluation incohérentes !")


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