ScoDoc/app/scodoc/sco_assiduites.py

"""
Ecrit par Matthias Hartmann.
"""
from datetime import date, datetime, time, timedelta
from pytz import UTC

import app.scodoc.sco_utils as scu
from app.models.assiduites import Assiduite, Justificatif
from app.models.etudiants import Identite
from app.models.formsemestre import FormSemestre, FormSemestreInscription
from app.scodoc import sco_formsemestre_inscriptions
from app.scodoc import sco_preferences
from app.scodoc import sco_cache
from app.scodoc import sco_etud


class CountCalculator:
    """Classe qui gére le comptage des assiduités"""

    def __init__(
        self,
        morning: time = time(8, 0),
        noon: time = time(12, 0),
        after_noon: time = time(14, 00),
        evening: time = time(18, 0),
        skip_saturday: bool = True,
    ) -> None:
        self.morning: time = morning
        self.noon: time = noon
        self.after_noon: time = after_noon
        self.evening: time = evening
        self.skip_saturday: bool = skip_saturday

        delta_total: timedelta = datetime.combine(date.min, evening) - datetime.combine(
            date.min, morning
        )
        delta_lunch: timedelta = datetime.combine(
            date.min, after_noon
        ) - datetime.combine(date.min, noon)

        self.hour_per_day: float = (delta_total - delta_lunch).total_seconds() / 3600

        self.days: list[date] = []
        self.half_days: list[tuple[date, bool]] = []  # tuple -> (date, morning:bool)
        self.hours: float = 0.0

        self.count: int = 0

    def reset(self):
        """Remet à zero le compteur"""
        self.days = []
        self.half_days = []
        self.hours = 0.0
        self.count = 0

    def add_half_day(self, day: date, is_morning: bool = True):
        """Ajoute une demi journée dans le comptage"""
        key: tuple[date, bool] = (day, is_morning)
        if key not in self.half_days:
            self.half_days.append(key)

    def add_day(self, day: date):
        """Ajoute un jour dans le comptage"""
        if day not in self.days:
            self.days.append(day)

    def check_in_morning(self, period: tuple[datetime, datetime]) -> bool:
        """Vérifiée si la période donnée fait partie du matin
        (Test sur la date de début)
        """

        interval_morning: tuple[datetime, datetime] = (
            scu.localize_datetime(datetime.combine(period[0].date(), self.morning)),
            scu.localize_datetime(datetime.combine(period[0].date(), self.noon)),
        )

        in_morning: bool = scu.is_period_overlapping(
            period, interval_morning, bornes=False
        )
        return in_morning

    def check_in_evening(self, period: tuple[datetime, datetime]) -> bool:
        """Vérifie si la période fait partie de l'aprèm
        (test sur la date de début)
        """

        interval_evening: tuple[datetime, datetime] = (
            scu.localize_datetime(datetime.combine(period[0].date(), self.after_noon)),
            scu.localize_datetime(datetime.combine(period[0].date(), self.evening)),
        )

        in_evening: bool = scu.is_period_overlapping(period, interval_evening)

        return in_evening

    def compute_long_assiduite(self, assi: Assiduite):
        """Calcule les métriques sur une assiduité longue (plus d'un jour)"""

        pointer_date: date = assi.date_debut.date() + timedelta(days=1)
        start_hours: timedelta = assi.date_debut - scu.localize_datetime(
            datetime.combine(assi.date_debut, self.morning)
        )
        finish_hours: timedelta = assi.date_fin - scu.localize_datetime(
            datetime.combine(assi.date_fin, self.morning)
        )

        self.add_day(assi.date_debut.date())
        self.add_day(assi.date_fin.date())

        start_period: tuple[datetime, datetime] = (
            assi.date_debut,
            scu.localize_datetime(
                datetime.combine(assi.date_debut.date(), self.evening)
            ),
        )

        finish_period: tuple[datetime, datetime] = (
            scu.localize_datetime(datetime.combine(assi.date_fin.date(), self.morning)),
            assi.date_fin,
        )
        hours = 0.0
        for period in (start_period, finish_period):
            if self.check_in_evening(period):
                self.add_half_day(period[0].date(), False)
            if self.check_in_morning(period):
                self.add_half_day(period[0].date())

        while pointer_date < assi.date_fin.date():
            if pointer_date.weekday() < (6 - self.skip_saturday):
                self.add_day(pointer_date)
                self.add_half_day(pointer_date)
                self.add_half_day(pointer_date, False)
                self.hours += self.hour_per_day
                hours += self.hour_per_day

            pointer_date += timedelta(days=1)

        self.hours += finish_hours.total_seconds() / 3600
        self.hours += self.hour_per_day - (start_hours.total_seconds() / 3600)

    def compute_assiduites(self, assiduites: Assiduite):
        """Calcule les métriques pour la collection d'assiduité donnée"""
        assi: Assiduite
        assiduites: list[Assiduite] = (
            assiduites.all() if isinstance(assiduites, Assiduite) else assiduites
        )
        for assi in assiduites:
            self.count += 1
            delta: timedelta = assi.date_fin - assi.date_debut

            if delta.days > 0:
                # raise Exception(self.hours)
                self.compute_long_assiduite(assi)

                continue

            period: tuple[datetime, datetime] = (assi.date_debut, assi.date_fin)
            deb_date: date = assi.date_debut.date()
            if self.check_in_morning(period):
                self.add_half_day(deb_date)
            if self.check_in_evening(period):
                self.add_half_day(deb_date, False)

            self.add_day(deb_date)

            self.hours += delta.total_seconds() / 3600

    def to_dict(self) -> dict[str, object]:
        """Retourne les métriques sous la forme d'un dictionnaire"""
        return {
            "compte": self.count,
            "journee": len(self.days),
            "demi": len(self.half_days),
            "heure": round(self.hours, 2),
        }


def get_assiduites_stats(
    assiduites: Assiduite, metric: str = "all", filtered: dict[str, object] = None
) -> Assiduite:
    """Compte les assiduités en fonction des filtres"""

    if filtered is not None:
        deb, fin = None, None
        for key in filtered:
            if key == "etat":
                assiduites = filter_assiduites_by_etat(assiduites, filtered[key])
            elif key == "date_fin":
                fin = filtered[key]
            elif key == "date_debut":
                deb = filtered[key]
            elif key == "moduleimpl_id":
                assiduites = filter_by_module_impl(assiduites, filtered[key])
            elif key == "formsemestre":
                assiduites = filter_by_formsemestre(assiduites, filtered[key])
            elif key == "est_just":
                assiduites = filter_assiduites_by_est_just(assiduites, filtered[key])
            elif key == "user_id":
                assiduites = filter_by_user_id(assiduites, filtered[key])
        if (deb, fin) != (None, None):
            assiduites = filter_by_date(assiduites, Assiduite, deb, fin)

    calculator: CountCalculator = CountCalculator()
    calculator.compute_assiduites(assiduites)
    count: dict = calculator.to_dict()

    metrics: list[str] = metric.split(",")

    output: dict = {}

    for key, val in count.items():
        if key in metrics:
            output[key] = val
    return output if output else count


def filter_assiduites_by_etat(assiduites: Assiduite, etat: str) -> Assiduite:
    """
    Filtrage d'une collection d'assiduites en fonction de leur état
    """
    etats: list[str] = list(etat.split(","))
    etats = [scu.EtatAssiduite.get(e, -1) for e in etats]
    return assiduites.filter(Assiduite.etat.in_(etats))


def filter_assiduites_by_est_just(
    assiduites: Assiduite, est_just: bool
) -> Justificatif:
    """
    Filtrage d'une collection d'assiduites en fonction de s'ils sont justifiés
    """
    return assiduites.filter_by(est_just=est_just)


def filter_by_user_id(
    collection: Assiduite or Justificatif,
    user_id: int,
) -> Justificatif:
    """
    Filtrage d'une collection en fonction de l'user_id
    """
    return collection.filter_by(user_id=user_id)


def filter_by_date(
    collection: Assiduite or Justificatif,
    collection_cls: Assiduite or Justificatif,
    date_deb: datetime = None,
    date_fin: datetime = None,
    strict: bool = False,
):
    """
    Filtrage d'une collection d'assiduites en fonction d'une date
    """
    if date_deb is None:
        date_deb = datetime.min
    if date_fin is None:
        date_fin = datetime.max

    date_deb = scu.localize_datetime(date_deb)
    date_fin = scu.localize_datetime(date_fin)
    if not strict:
        return collection.filter(
            collection_cls.date_debut <= date_fin, collection_cls.date_fin >= date_deb
        )
    return collection.filter(
        collection_cls.date_debut < date_fin, collection_cls.date_fin > date_deb
    )


def filter_justificatifs_by_etat(
    justificatifs: Justificatif, etat: str
) -> Justificatif:
    """
    Filtrage d'une collection de justificatifs en fonction de leur état
    """
    etats: list[str] = list(etat.split(","))
    etats = [scu.EtatJustificatif.get(e, -1) for e in etats]
    return justificatifs.filter(Justificatif.etat.in_(etats))


def filter_by_module_impl(
    assiduites: Assiduite, module_impl_id: int or None
) -> Assiduite:
    """
    Filtrage d'une collection d'assiduites en fonction de l'ID du module_impl
    """
    return assiduites.filter(Assiduite.moduleimpl_id == module_impl_id)


def filter_by_formsemestre(assiduites_query: Assiduite, formsemestre: FormSemestre):
    """
    Filtrage d'une collection d'assiduites en fonction d'un formsemestre
    """

    if formsemestre is None:
        return assiduites_query.filter(False)

    assiduites_query = (
        assiduites_query.join(Identite, Assiduite.etudid == Identite.id)
        .join(
            FormSemestreInscription,
            Identite.id == FormSemestreInscription.etudid,
        )
        .filter(FormSemestreInscription.formsemestre_id == formsemestre.id)
    )

    assiduites_query = assiduites_query.filter(
        Assiduite.date_debut >= formsemestre.date_debut
    )
    return assiduites_query.filter(Assiduite.date_fin <= formsemestre.date_fin)


def justifies(justi: Justificatif, obj: bool = False) -> list[int]:
    """
    Retourne la liste des assiduite_id qui sont justifié par la justification
    Une assiduité est justifiée si elle est COMPLETEMENT ou PARTIELLEMENT comprise dans la plage du justificatif
        et que l'état du justificatif est "valide"
    renvoie des id si obj == False, sinon les Assiduités
    """

    if justi.etat != scu.EtatJustificatif.VALIDE:
        return []

    assiduites_query: Assiduite = Assiduite.query.join(
        Justificatif, Assiduite.etudid == Justificatif.etudid
    ).filter(
        Assiduite.date_debut <= justi.date_fin,
        Assiduite.date_fin >= justi.date_debut,
    )

    if not obj:
        return [assi.id for assi in assiduites_query.all()]

    return assiduites_query


def get_all_justified(
    etudid: int, date_deb: datetime = None, date_fin: datetime = None
) -> list[Assiduite]:
    """Retourne toutes les assiduités justifiées sur une période"""

    if date_deb is None:
        date_deb = datetime.min
    if date_fin is None:
        date_fin = datetime.max

    date_deb = scu.localize_datetime(date_deb)
    date_fin = scu.localize_datetime(date_fin)
    justified = Assiduite.query.filter_by(est_just=True, etudid=etudid)
    after = filter_by_date(
        justified,
        Assiduite,
        date_deb,
        date_fin,
    )
    return after


# Gestion du cache
def get_assiduites_count(etudid, sem):
    """Les comptes d'absences de cet étudiant dans ce semestre:
    tuple (nb abs non justifiées, nb abs justifiées)
    Utilise un cache.
    """
    metrique = sco_preferences.get_preference("assi_metrique", sem["formsemestre_id"])
    return get_assiduites_count_in_interval(
        etudid,
        sem["date_debut_iso"],
        sem["date_fin_iso"],
        scu.translate_assiduites_metric(metrique),
    )


def get_assiduites_count_in_interval(
    etudid, date_debut_iso, date_fin_iso, metrique="demi"
):
    """Les comptes d'absences de cet étudiant entre ces deux dates, incluses:
    tuple (nb abs, nb abs justifiées)
    Utilise un cache.
    """
    key = (
        str(etudid)
        + "_"
        + date_debut_iso
        + "_"
        + date_fin_iso
        + f"{metrique}_assiduites"
    )
    r = sco_cache.AbsSemEtudCache.get(key)
    if not r:
        date_debut: datetime.datetime = scu.is_iso_formated(date_debut_iso, True)
        date_fin: datetime.datetime = scu.is_iso_formated(date_fin_iso, True)

        assiduites: Assiduite = Assiduite.query.filter_by(etudid=etudid)
        assiduites = assiduites.filter(Assiduite.etat == scu.EtatAssiduite.ABSENT)
        justificatifs: Justificatif = Justificatif.query.filter_by(etudid=etudid)

        assiduites = filter_by_date(assiduites, Assiduite, date_debut, date_fin)
        justificatifs = filter_by_date(
            justificatifs, Justificatif, date_debut, date_fin
        )

        calculator: CountCalculator = CountCalculator()
        calculator.compute_assiduites(assiduites)
        nb_abs: dict = calculator.to_dict()[metrique]

        abs_just: list[Assiduite] = get_all_justified(etudid, date_debut, date_fin)

        calculator.reset()
        calculator.compute_assiduites(abs_just)
        nb_abs_just: dict = calculator.to_dict()[metrique]

        r = (nb_abs, nb_abs_just)
        ans = sco_cache.AbsSemEtudCache.set(key, r)
        if not ans:
            log("warning: get_assiduites_count failed to cache")
    return r


def invalidate_assiduites_count(etudid, sem):
    """Invalidate (clear) cached counts"""
    date_debut = sem["date_debut_iso"]
    date_fin = sem["date_fin_iso"]
    for met in ["demi", "journee", "compte", "heure"]:
        key = str(etudid) + "_" + date_debut + "_" + date_fin + f"{met}_assiduites"
        sco_cache.AbsSemEtudCache.delete(key)


def invalidate_assiduites_count_sem(sem):
    """Invalidate (clear) cached abs counts for all the students of this semestre"""
    inscriptions = (
        sco_formsemestre_inscriptions.do_formsemestre_inscription_listinscrits(
            sem["formsemestre_id"]
        )
    )
    for ins in inscriptions:
        invalidate_assiduites_count(ins["etudid"], sem)


def invalidate_assiduites_etud_date(etudid, date: datetime):
    """Doit etre appelé à chaque modification des assiduites pour cet étudiant et cette date.
    Invalide cache absence et caches semestre
    date: date au format ISO
    """
    from app.scodoc import sco_compute_moy

    # Semestres a cette date:
    etud = sco_etud.get_etud_info(etudid=etudid, filled=True)[0]
    sems = [
        sem
        for sem in etud["sems"]
        if scu.is_iso_formated(sem["date_debut_iso"], True).replace(tzinfo=UTC)
        <= date.replace(tzinfo=UTC)
        and scu.is_iso_formated(sem["date_fin_iso"], True).replace(tzinfo=UTC)
        >= date.replace(tzinfo=UTC)
    ]

    # Invalide les PDF et les absences:
    for sem in sems:
        # Inval cache bulletin et/ou note_table
        if sco_compute_moy.formsemestre_expressions_use_abscounts(
            sem["formsemestre_id"]
        ):
            # certaines formules utilisent les absences
            pdfonly = False
        else:
            # efface toujours le PDF car il affiche en général les absences
            pdfonly = True

        sco_cache.invalidate_formsemestre(
            formsemestre_id=sem["formsemestre_id"], pdfonly=pdfonly
        )

        # Inval cache compteurs absences:
        invalidate_assiduites_count(etudid, sem)


def simple_invalidate_cache(obj: dict, etudid: str or int = None):
    """Invalide le cache de l'étudiant et du / des semestres"""
    date_debut = (
        obj["date_debut"]
        if isinstance(obj["date_debut"], datetime)
        else scu.is_iso_formated(obj["date_debut"], True)
    )
    date_fin = (
        obj["date_fin"]
        if isinstance(obj["date_fin"], datetime)
        else scu.is_iso_formated(obj["date_fin"], True)
    )
    etudid = etudid if etudid is not None else obj["etudid"]
    invalidate_assiduites_etud_date(etudid, date_debut)
    invalidate_assiduites_etud_date(etudid, date_fin)