ScoDoc/app/scodoc/sco_evaluation_db.py

# -*- mode: python -*-
# -*- coding: utf-8 -*-

##############################################################################
#
# Gestion scolarite IUT
#
# Copyright (c) 1999 - 2022 Emmanuel Viennet.  All rights reserved.
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
#
#   Emmanuel Viennet      emmanuel.viennet@gmail.com
#
##############################################################################

"""Gestion evaluations (ScoDoc7, sans SQlAlchemy)
"""

import datetime
import pprint

import flask
from flask import url_for, g
from flask_login import current_user

from app import log
import app.scodoc.sco_utils as scu
import app.scodoc.notesdb as ndb
from app.scodoc.sco_exceptions import AccessDenied, ScoValueError

from app.scodoc import sco_cache
from app.scodoc import sco_edit_module
from app.scodoc import sco_formsemestre
from app.scodoc import sco_moduleimpl
from app.scodoc import sco_news
from app.scodoc import sco_permissions_check


_evaluationEditor = ndb.EditableTable(
    "notes_evaluation",
    "evaluation_id",
    (
        "evaluation_id",
        "moduleimpl_id",
        "jour",
        "heure_debut",
        "heure_fin",
        "description",
        "note_max",
        "coefficient",
        "visibulletin",
        "publish_incomplete",
        "evaluation_type",
        "numero",
    ),
    sortkey="numero desc, jour desc, heure_debut desc",  # plus recente d'abord
    output_formators={
        "jour": ndb.DateISOtoDMY,
        "numero": ndb.int_null_is_zero,
    },
    input_formators={
        "jour": ndb.DateDMYtoISO,
        "heure_debut": ndb.TimetoISO8601,  # converti par evaluation_enrich_dict
        "heure_fin": ndb.TimetoISO8601,  # converti par evaluation_enrich_dict
        "visibulletin": bool,
        "publish_incomplete": bool,
        "evaluation_type": int,
    },
)


def evaluation_enrich_dict(e):
    """add or convert some fileds in an evaluation dict"""
    # For ScoDoc7 compat
    heure_debut_dt = e["heure_debut"] or datetime.time(
        8, 00
    )  # au cas ou pas d'heure (note externe?)
    heure_fin_dt = e["heure_fin"] or datetime.time(8, 00)
    e["heure_debut"] = ndb.TimefromISO8601(e["heure_debut"])
    e["heure_fin"] = ndb.TimefromISO8601(e["heure_fin"])
    e["jouriso"] = ndb.DateDMYtoISO(e["jour"])
    heure_debut, heure_fin = e["heure_debut"], e["heure_fin"]
    d = ndb.TimeDuration(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 heure_debut_dt < datetime.time(12, 00):
        e["matin"] = 1
    else:
        e["matin"] = 0
    if heure_fin_dt > datetime.time(12, 00):
        e["apresmidi"] = 1
    else:
        e["apresmidi"] = 0
    return e


def do_evaluation_list(args, sortkey=None):
    """List evaluations, sorted by numero (or most recent date first).

    Ajoute les champs:
    'duree' : '2h30'
    'matin' : 1 (commence avant 12:00) ou 0
    'apresmidi' : 1 (termine après 12:00) ou 0
    'descrheure' : ' de 15h00 à 16h30'
    """
    # Attention: transformation fonction ScoDc7 en SQLAlchemy
    cnx = ndb.GetDBConnexion()
    evals = _evaluationEditor.list(cnx, args, sortkey=sortkey)
    # calcule duree (chaine de car.) de chaque evaluation et ajoute jouriso, matin, apresmidi
    for e in evals:
        evaluation_enrich_dict(e)

    return evals


def do_evaluation_list_in_formsemestre(formsemestre_id):
    "list evaluations in this formsemestre"
    mods = sco_moduleimpl.moduleimpl_list(formsemestre_id=formsemestre_id)
    evals = []
    for modimpl in mods:
        evals += do_evaluation_list(args={"moduleimpl_id": modimpl["moduleimpl_id"]})
    return evals


def _check_evaluation_args(args):
    "Check coefficient, dates and duration, raises exception if invalid"
    moduleimpl_id = args["moduleimpl_id"]
    # check bareme
    note_max = args.get("note_max", None)
    if note_max is None:
        raise ScoValueError("missing note_max")
    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)")
    # check coefficient
    coef = args.get("coefficient", None)
    if coef is None:
        raise ScoValueError("missing coefficient")
    try:
        coef = float(coef)
    except ValueError:
        raise ScoValueError("Invalid coefficient value")
    if coef < 0:
        raise ScoValueError("Invalid coefficient value (must be positive or null)")
    # check date
    jour = args.get("jour", None)
    args["jour"] = jour
    if jour:
        M = sco_moduleimpl.moduleimpl_list(moduleimpl_id=moduleimpl_id)[0]
        sem = sco_formsemestre.get_formsemestre(M["formsemestre_id"])
        d, m, y = [int(x) for x in sem["date_debut"].split("/")]
        date_debut = datetime.date(y, m, d)
        d, m, y = [int(x) for x in sem["date_fin"].split("/")]
        date_fin = datetime.date(y, m, d)
        # passe par ndb.DateDMYtoISO pour avoir date pivot
        y, m, d = [int(x) for x in ndb.DateDMYtoISO(jour).split("-")]
        jour = datetime.date(y, m, d)
        if (jour > date_fin) or (jour < date_debut):
            raise ScoValueError(
                "La date de l'évaluation (%s/%s/%s) n'est pas dans le semestre !"
                % (d, m, y)
            )
    heure_debut = args.get("heure_debut", None)
    args["heure_debut"] = heure_debut
    heure_fin = args.get("heure_fin", None)
    args["heure_fin"] = heure_fin
    if jour and ((not heure_debut) or (not heure_fin)):
        raise ScoValueError("Les heures doivent être précisées")
    d = ndb.TimeDuration(heure_debut, heure_fin)
    if d and ((d < 0) or (d > 60 * 12)):
        raise ScoValueError("Heures de l'évaluation incohérentes !")


def do_evaluation_create(
    moduleimpl_id=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 arguments excedentaires de tf #sco8
):
    """Create an evaluation"""
    if not sco_permissions_check.can_edit_evaluation(moduleimpl_id=moduleimpl_id):
        raise AccessDenied(
            "Modification évaluation impossible pour %s" % current_user.get_nomplogin()
        )
    args = locals()
    log("do_evaluation_create: args=" + str(args))
    _check_evaluation_args(args)
    # Check numeros
    module_evaluation_renumber(moduleimpl_id, only_if_unumbered=True)
    if not "numero" in args or args["numero"] is None:
        n = None
        # determine le numero avec la date
        # Liste des eval existantes triees par date, la plus ancienne en tete
        mod_evals = do_evaluation_list(
            args={"moduleimpl_id": moduleimpl_id},
            sortkey="jour asc, heure_debut asc",
        )
        if args["jour"]:
            next_eval = None
            t = (
                ndb.DateDMYtoISO(args["jour"], null_is_empty=True),
                ndb.TimetoISO8601(args["heure_debut"], null_is_empty=True),
            )
            for e in mod_evals:
                if (
                    ndb.DateDMYtoISO(e["jour"], null_is_empty=True),
                    ndb.TimetoISO8601(e["heure_debut"], null_is_empty=True),
                ) > t:
                    next_eval = e
                    break
            if next_eval:
                n = module_evaluation_insert_before(mod_evals, next_eval)
            else:
                n = None  # a placer en fin
        if n is None:  # pas de date ou en fin:
            if mod_evals:
                log(pprint.pformat(mod_evals[-1]))
                n = mod_evals[-1]["numero"] + 1
            else:
                n = 0  # the only one
        # log("creating with numero n=%d" % n)
        args["numero"] = n

    #
    cnx = ndb.GetDBConnexion()
    r = _evaluationEditor.create(cnx, args)

    # news
    M = sco_moduleimpl.moduleimpl_list(moduleimpl_id=moduleimpl_id)[0]
    mod = sco_edit_module.module_list(args={"module_id": M["module_id"]})[0]
    mod["moduleimpl_id"] = M["moduleimpl_id"]
    mod["url"] = "Notes/moduleimpl_status?moduleimpl_id=%(moduleimpl_id)s" % mod
    sco_news.add(
        typ=sco_news.NEWS_NOTE,
        object=moduleimpl_id,
        text='Création d\'une évaluation dans <a href="%(url)s">%(titre)s</a>' % mod,
        url=mod["url"],
    )

    return r


def do_evaluation_edit(args):
    "edit an evaluation"
    evaluation_id = args["evaluation_id"]
    the_evals = do_evaluation_list({"evaluation_id": evaluation_id})
    if not the_evals:
        raise ValueError("evaluation inexistante !")
    moduleimpl_id = the_evals[0]["moduleimpl_id"]
    if not sco_permissions_check.can_edit_evaluation(moduleimpl_id=moduleimpl_id):
        raise AccessDenied(
            "Modification évaluation impossible pour %s" % current_user.get_nomplogin()
        )
    args["moduleimpl_id"] = moduleimpl_id
    _check_evaluation_args(args)

    cnx = ndb.GetDBConnexion()
    _evaluationEditor.edit(cnx, args)
    # inval cache pour ce semestre
    M = sco_moduleimpl.moduleimpl_list(moduleimpl_id=moduleimpl_id)[0]
    sco_cache.invalidate_formsemestre(formsemestre_id=M["formsemestre_id"])


def do_evaluation_delete(evaluation_id):
    "delete evaluation"
    the_evals = do_evaluation_list({"evaluation_id": evaluation_id})
    if not the_evals:
        raise ValueError("evaluation inexistante !")
    moduleimpl_id = the_evals[0]["moduleimpl_id"]
    if not sco_permissions_check.can_edit_evaluation(moduleimpl_id=moduleimpl_id):
        raise AccessDenied(
            "Modification évaluation impossible pour %s" % current_user.get_nomplogin()
        )
    NotesDB = do_evaluation_get_all_notes(evaluation_id)  # { etudid : value }
    notes = [x["value"] for x in NotesDB.values()]
    if notes:
        raise ScoValueError(
            "Impossible de supprimer cette évaluation: il reste des notes"
        )

    cnx = ndb.GetDBConnexion()

    _evaluationEditor.delete(cnx, evaluation_id)
    # inval cache pour ce semestre
    M = sco_moduleimpl.moduleimpl_list(moduleimpl_id=moduleimpl_id)[0]
    sco_cache.invalidate_formsemestre(formsemestre_id=M["formsemestre_id"])
    # news

    mod = sco_edit_module.module_list(args={"module_id": M["module_id"]})[0]
    mod["moduleimpl_id"] = M["moduleimpl_id"]
    mod["url"] = (
        scu.NotesURL() + "/moduleimpl_status?moduleimpl_id=%(moduleimpl_id)s" % mod
    )
    sco_news.add(
        typ=sco_news.NEWS_NOTE,
        object=moduleimpl_id,
        text='Suppression d\'une évaluation dans <a href="%(url)s">%(titre)s</a>' % mod,
        url=mod["url"],
    )


# ancien _notes_getall
def do_evaluation_get_all_notes(
    evaluation_id, table="notes_notes", filter_suppressed=True, by_uid=None
):
    """Toutes les notes pour une evaluation: { etudid : { 'value' : value, 'date' : date ... }}
    Attention: inclut aussi les notes des étudiants qui ne sont plus inscrits au module.
    """
    do_cache = (
        filter_suppressed and table == "notes_notes" and (by_uid is None)
    )  # pas de cache pour (rares) appels via undo_notes ou specifiant un enseignant
    if do_cache:
        r = sco_cache.EvaluationCache.get(evaluation_id)
        if r != None:
            return r
    cnx = ndb.GetDBConnexion()
    cursor = cnx.cursor(cursor_factory=ndb.ScoDocCursor)
    cond = " where evaluation_id=%(evaluation_id)s"
    if by_uid:
        cond += " and uid=%(by_uid)s"

    cursor.execute(
        "select * from " + table + cond,
        {"evaluation_id": evaluation_id, "by_uid": by_uid},
    )
    res = cursor.dictfetchall()
    d = {}
    if filter_suppressed:
        for x in res:
            if x["value"] != scu.NOTES_SUPPRESS:
                d[x["etudid"]] = x
    else:
        for x in res:
            d[x["etudid"]] = x
    if do_cache:
        status = sco_cache.EvaluationCache.set(evaluation_id, d)
        if not status:
            log(f"Warning: EvaluationCache.set: {evaluation_id}\t{status}")
    return d


def module_evaluation_renumber(moduleimpl_id, only_if_unumbered=False, redirect=0):
    """Renumber evaluations in this module, according to their date. (numero=0: oldest one)
    Needed because previous versions of ScoDoc did not have eval numeros
    Note: existing numeros are ignored
    """
    redirect = int(redirect)
    # log('module_evaluation_renumber( moduleimpl_id=%s )' % moduleimpl_id )
    # List sorted according to date/heure, ignoring numeros:
    # (note that we place  evaluations with NULL date at the end)
    mod_evals = do_evaluation_list(
        args={"moduleimpl_id": moduleimpl_id},
        sortkey="jour asc, heure_debut asc",
    )

    all_numbered = False not in [x["numero"] > 0 for x in mod_evals]
    if all_numbered and only_if_unumbered:
        return  # all ok

    # Reset all numeros:
    i = 1
    for e in mod_evals:
        e["numero"] = i
        do_evaluation_edit(e)
        i += 1

    # If requested, redirect to moduleimpl page:
    if redirect:
        return flask.redirect(
            url_for(
                "notes.moduleimpl_status",
                scodoc_dept=g.scodoc_dept,
                moduleimpl_id=moduleimpl_id,
            )
        )


def module_evaluation_insert_before(mod_evals, next_eval):
    """Renumber evals 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 not n:
            log("renumbering old evals")
            module_evaluation_renumber(next_eval["moduleimpl_id"])
            next_eval = do_evaluation_list(
                args={"evaluation_id": next_eval["evaluation_id"]}
            )[0]
            n = next_eval["numero"]
    else:
        n = 1
    # log('inserting at position numero %s' % n )
    # all numeros >= n are incremented
    for e in mod_evals:
        if e["numero"] >= n:
            e["numero"] += 1
            # log('incrementing %s to %s' % (e['evaluation_id'], e['numero']))
            do_evaluation_edit(e)

    return n


def module_evaluation_move(evaluation_id, after=0, redirect=1):
    """Move before/after previous one (decrement/increment numero)
    (published)
    """
    e = do_evaluation_list(args={"evaluation_id": evaluation_id})[0]
    redirect = int(redirect)
    # access: can change eval ?
    if not sco_permissions_check.can_edit_evaluation(moduleimpl_id=e["moduleimpl_id"]):
        raise AccessDenied(
            "Modification évaluation impossible pour %s" % current_user.get_nomplogin()
        )

    module_evaluation_renumber(e["moduleimpl_id"], only_if_unumbered=True)
    e = do_evaluation_list(args={"evaluation_id": evaluation_id})[0]

    after = int(after)  # 0: deplace avant, 1 deplace apres
    if after not in (0, 1):
        raise ValueError('invalid value for "after"')
    mod_evals = do_evaluation_list({"moduleimpl_id": e["moduleimpl_id"]})
    if len(mod_evals) > 1:
        idx = [p["evaluation_id"] for p in mod_evals].index(evaluation_id)
        neigh = None  # object to swap with
        if after == 0 and idx > 0:
            neigh = mod_evals[idx - 1]
        elif after == 1 and idx < len(mod_evals) - 1:
            neigh = mod_evals[idx + 1]
        if neigh:  #
            if neigh["numero"] == e["numero"]:
                log("Warning: module_evaluation_move: forcing renumber")
                module_evaluation_renumber(e["moduleimpl_id"], only_if_unumbered=False)
            else:
                # swap numero with neighbor
                e["numero"], neigh["numero"] = neigh["numero"], e["numero"]
                do_evaluation_edit(e)
                do_evaluation_edit(neigh)
    # redirect to moduleimpl page:
    if redirect:
        return flask.redirect(
            url_for(
                "notes.moduleimpl_status",
                scodoc_dept=g.scodoc_dept,
                moduleimpl_id=e["moduleimpl_id"],
            )
        )