ScoDoc/app/scodoc/sco_abs_notification.py

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

##############################################################################
#
# Gestion scolarite IUT
#
# Copyright (c) 1999 - 2024 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@viennet.net
#
##############################################################################

"""Système de notification par mail des excès d'absences
(see ticket #147)


Il suffit d'appeler abs_notify() après chaque ajout d'absence.
"""

import datetime
from typing import Optional

from flask import flash, g, url_for
from flask_mail import Message

from app import db
from app import email
from app import log
from app.auth.models import User
from app.models.absences import AbsenceNotification
from app.models.etudiants import Identite
from app.models.events import Scolog
from app.models.formsemestre import FormSemestre
import app.scodoc.notesdb as ndb
from app.scodoc import sco_preferences
from app.scodoc import sco_utils as scu


def abs_notify(etudid: int, date: str | datetime.datetime):
    """Check if notifications are requested and send them
    Considère le nombre d'absence dans le semestre courant
    (s'il n'y a pas de semestre courant, ne fait rien,
    car l'etudiant n'est pas inscrit au moment de l'absence!).

    NE FAIT RIEN EN MODE DEBUG.
    """
    from app.scodoc import sco_assiduites

    # if current_app and current_app.config["DEBUG"]:
    #    return

    formsemestre = retreive_current_formsemestre(etudid, date)
    if not formsemestre:
        return  # non inscrit a la date, pas de notification

    _, nbabsjust, nbabs = sco_assiduites.get_assiduites_count_in_interval(
        etudid,
        metrique=scu.translate_assiduites_metric(
            sco_preferences.get_preference(
                "assi_metrique", formsemestre.formsemestre_id
            )
        ),
        date_debut=datetime.datetime.combine(
            formsemestre.date_debut, datetime.datetime.min.time()
        ),
        date_fin=datetime.datetime.combine(
            formsemestre.date_fin, datetime.datetime.min.time()
        ),
    )
    do_abs_notify(formsemestre, etudid, date, nbabs, nbabsjust)


def do_abs_notify(
    formsemestre: FormSemestre,
    etudid: int,
    date: str | datetime.datetime,
    nbabs: int,
    nbabsjust: int,
):
    """Given new counts of absences, check if notifications are requested and send them."""
    # prefs fallback to global pref if sem is None:
    if formsemestre:
        formsemestre_id = formsemestre.id
    else:
        formsemestre_id = None
    prefs = sco_preferences.SemPreferences(formsemestre_id=formsemestre_id)

    destinations = abs_notify_get_destinations(
        formsemestre, prefs, etudid, date, nbabs, nbabsjust
    )

    msg = abs_notification_message(formsemestre, prefs, etudid, nbabs, nbabsjust)
    if not msg:
        return  # abort

    # Vérification fréquence (pour ne pas envoyer de mails trop souvent)
    abs_notify_max_freq = sco_preferences.get_preference("abs_notify_max_freq")
    destinations_filtered = []
    for email_addr in destinations:
        nbdays_since_last_notif = user_nbdays_since_last_notif(email_addr, etudid)
        if (nbdays_since_last_notif is None) or (
            nbdays_since_last_notif >= abs_notify_max_freq
        ):
            destinations_filtered.append(email_addr)

    if destinations_filtered:
        abs_notify_send(
            destinations_filtered,
            etudid,
            msg,
            nbabs,
            nbabsjust,
            formsemestre_id,
        )


def abs_notify_send(destinations, etudid, msg, nbabs, nbabsjust, formsemestre_id):
    """Actually send the notification by email, and register it in database"""
    log(f"abs_notify: sending notification to {destinations}")
    for dest_addr in destinations:
        msg.recipients = [dest_addr]
        email.send_message(msg)
        notification = AbsenceNotification(
            etudid=etudid,
            email=dest_addr,
            nbabs=nbabs,
            nbabsjust=nbabsjust,
            formsemestre_id=formsemestre_id,
        )
        db.session.add(notification)

    Scolog.logdb(
        method="abs_notify",
        etudid=etudid,
        msg=f"sent to {destinations} (nbabs={nbabs})",
        commit=True,
    )


def abs_notify_get_destinations(
    formsemestre: FormSemestre,
    prefs: dict,
    etudid: int,
    date: str | datetime.datetime,
    nbabs: int,
    nbabsjust: int,
) -> set[str]:
    """Returns set of destination emails to be notified"""

    destinations = []  # list of email address to notify

    if abs_notify_is_above_threshold(etudid, nbabs, nbabsjust, formsemestre.id):
        if prefs["abs_notify_respsem"]:
            # notifie chaque responsable du semestre
            for responsable in formsemestre.responsables:
                if responsable.email:
                    destinations.append(responsable.email)
        if prefs["abs_notify_chief"] and prefs["email_chefdpt"]:
            destinations.append(prefs["email_chefdpt"])
        if prefs["abs_notify_email"]:
            destinations.append(prefs["abs_notify_email"])
        if prefs["abs_notify_etud"]:
            etud = Identite.get_etud(etudid)
            adresse = etud.adresses.first()
            if adresse:
                # Mail à utiliser pour les envois vers l'étudiant:
                # choix qui pourrait être controlé par une preference
                # ici priorité au mail institutionnel:
                email_default = adresse.email or adresse.emailperso
                if email_default:
                    destinations.append(email_default)

    # Notification (à chaque fois) des resp. de modules ayant des évaluations
    # à cette date
    # nb: on pourrait prevoir d'utiliser un autre format de message pour ce cas
    if prefs["abs_notify_respeval"]:
        mods = mod_with_evals_at_date(date, etudid)
        for mod in mods:
            u: User = db.session.get(User, mod["responsable_id"])
            if u is not None and u.is_active and u.email:
                destinations.append(u.email)

    # uniq
    destinations = set(destinations)

    return destinations


def abs_notify_is_above_threshold(etudid, nbabs, nbabsjust, formsemestre_id):
    """True si il faut notifier les absences (indépendemment du destinataire)

    nbabs: nombre d'absence (de tous types, unité de compte = demi-journée)
    nbabsjust: nombre d'absences justifiées

    (nbabs > abs_notify_abs_threshold)
    (nbabs - nbabs_last_notified) > abs_notify_abs_increment

    TODO Mettre à jour avec le module assiduité  + fonctionnement métrique
    """
    abs_notify_abs_threshold = sco_preferences.get_preference(
        "abs_notify_abs_threshold", formsemestre_id
    )
    abs_notify_abs_increment = sco_preferences.get_preference(
        "abs_notify_abs_increment", formsemestre_id
    )
    nbabs_last_notified = etud_nbabs_last_notified(etudid, formsemestre_id)

    if nbabs_last_notified == 0:
        if nbabs > abs_notify_abs_threshold:
            return True  # first notification
        else:
            return False
    else:
        if (nbabs - nbabs_last_notified) >= abs_notify_abs_increment:
            return True
    return False


def etud_nbabs_last_notified(etudid: int, formsemestre_id: int = None):
    """nbabs lors de la dernière notification envoyée pour cet étudiant dans ce semestre
    ou sans semestre (ce dernier cas est nécessaire pour la transition au nouveau code)
    """
    notifications = (
        AbsenceNotification.query.filter_by(etudid=etudid)
        .filter(
            (AbsenceNotification.formsemestre_id == formsemestre_id)
            | (AbsenceNotification.formsemestre_id.is_(None))
        )
        .order_by(AbsenceNotification.notification_date.desc())
    )
    last_notif = notifications.first()
    return last_notif.nbabs if last_notif else 0


def user_nbdays_since_last_notif(email_addr, etudid) -> Optional[int]:
    """nb days since last notification to this email, or None if no previous notification"""
    notifications = AbsenceNotification.query.filter_by(
        etudid=etudid, email=email_addr
    ).order_by(AbsenceNotification.notification_date.desc())
    last_notif = notifications.first()
    if last_notif:
        now = datetime.datetime.now(last_notif.notification_date.tzinfo)
        return (now - last_notif.notification_date).days
    return None


def abs_notification_message(
    formsemestre: FormSemestre, prefs, etudid, nbabs, nbabsjust
):
    """Mime notification message based on template.
    returns a Message instance
    or None if sending should be canceled (empty template).
    """
    from app.scodoc import sco_bulletins

    etud = Identite.get_etud(etudid)

    # Variables accessibles dans les balises du template: %(nom_variable)s :
    values = sco_bulletins.make_context_dict(
        formsemestre, etud.to_dict_scodoc7(with_inscriptions=True)
    )

    values["nbabs"] = nbabs
    values["nbabsjust"] = nbabsjust
    values["nbabsnonjust"] = nbabs - nbabsjust
    values["url_ficheetud"] = url_for(
        "scolar.fiche_etud", scodoc_dept=g.scodoc_dept, etudid=etudid, _external=True
    )

    # Formsemestre concerné (ex: "BUT Informatique semestre 2")
    values["semestre"] = formsemestre.titre_num()

    template = prefs["abs_notification_mail_tmpl"]
    txt = ""
    if template:
        try:
            txt = prefs["abs_notification_mail_tmpl"] % values
        except KeyError:
            flash("Mail non envoyé: format invalide (voir paramétrage)")
            log("abs_notification_message: invalid key in abs_notification_mail_tmpl")
            txt = ""
    else:
        log("abs_notification_message: empty template, not sending message")
    if not txt:
        return None

    subject = f"""[ScoDoc] Trop d'absences pour {etud.nomprenom}"""
    msg = Message(subject, sender=email.get_from_addr(formsemestre.departement.acronym))
    msg.body = txt
    return msg


def retreive_current_formsemestre(
    etudid: int, cur_date: str | datetime.date
) -> Optional[FormSemestre]:
    """Get formsemestre dans lequel etudid est (ou était) inscrit a la date indiquée
    date est une chaine au format ISO (yyyy-mm-dd) ou un datetime.date

    Result: FormSemestre ou None si pas inscrit à la date indiquée
    """
    req = """SELECT i.formsemestre_id
    FROM notes_formsemestre_inscription i, notes_formsemestre sem
    WHERE sem.id = i.formsemestre_id AND i.etudid = %(etudid)s
    AND (%(cur_date)s >= sem.date_debut) AND (%(cur_date)s <= sem.date_fin)
    """

    r = ndb.SimpleDictFetch(req, {"etudid": etudid, "cur_date": cur_date})
    if not r:
        return None
    # s'il y a plusieurs semestres, prend le premier (rarissime et non significatif):
    formsemestre = FormSemestre.get_formsemestre(r[0]["formsemestre_id"])
    return formsemestre


def mod_with_evals_at_date(
    date_abs: str | datetime.datetime, etudid: int
) -> list[dict]:
    """Liste des moduleimpls avec des evaluations à la date indiquée"""
    req = """
    SELECT m.id AS moduleimpl_id, m.*
    FROM notes_moduleimpl m, notes_evaluation e, notes_moduleimpl_inscription i
    WHERE m.id = e.moduleimpl_id
    AND e.moduleimpl_id = i.moduleimpl_id
    AND i.etudid = %(etudid)s
    AND e.date_debut <= %(date_abs)s
    AND e.date_fin >= %(date_abs)s
    """
    return ndb.SimpleDictFetch(req, {"etudid": etudid, "date_abs": date_abs})