ScoDoc/app/models/assiduites.py

# -*- coding: UTF-8 -*
"""Gestion de l'assiduité (assiduités + justificatifs)
"""
from datetime import datetime

from app import db, log
from app.models import ModuleImpl, Scolog
from app.models.etudiants import Identite
from app.auth.models import User
from app.scodoc.sco_exceptions import ScoValueError
from app.scodoc.sco_utils import (
    EtatAssiduite,
    EtatJustificatif,
    localize_datetime,
)

from flask_sqlalchemy.query import Query


class Assiduite(db.Model):
    """
    Représente une assiduité:
        - une plage horaire lié à un état et un étudiant
        - un module si spécifiée
        - une description si spécifiée
    """

    __tablename__ = "assiduites"

    id = db.Column(db.Integer, primary_key=True, nullable=False)
    assiduite_id = db.synonym("id")

    date_debut = db.Column(
        db.DateTime(timezone=True), server_default=db.func.now(), nullable=False
    )
    date_fin = db.Column(
        db.DateTime(timezone=True), server_default=db.func.now(), nullable=False
    )

    moduleimpl_id = db.Column(
        db.Integer,
        db.ForeignKey("notes_moduleimpl.id", ondelete="SET NULL"),
    )
    etudid = db.Column(
        db.Integer,
        db.ForeignKey("identite.id", ondelete="CASCADE"),
        index=True,
        nullable=False,
    )
    etat = db.Column(db.Integer, nullable=False)

    description = db.Column(db.Text)

    entry_date = db.Column(db.DateTime(timezone=True), server_default=db.func.now())

    user_id = db.Column(
        db.Integer,
        db.ForeignKey("user.id", ondelete="SET NULL"),
        nullable=True,
    )

    est_just = db.Column(db.Boolean, server_default="false", nullable=False)

    external_data = db.Column(db.JSON, nullable=True)

    # Déclare la relation "joined" car on va très souvent vouloir récupérer
    # l'étudiant en même tant que l'assiduité (perf.: évite nouvelle requete SQL)
    etudiant = db.relationship("Identite", back_populates="assiduites", lazy="joined")

    def to_dict(self, format_api=True) -> dict:
        """Retourne la représentation json de l'assiduité"""
        etat = self.etat
        username = self.user_id
        if format_api:
            etat = EtatAssiduite.inverse().get(self.etat).name
            if self.user_id is not None:
                user: User = db.session.get(User, self.user_id)

                if user is None:
                    username = "Non renseigné"
                else:
                    username = user.get_prenomnom()
        data = {
            "assiduite_id": self.id,
            "etudid": self.etudid,
            "code_nip": self.etudiant.code_nip,
            "moduleimpl_id": self.moduleimpl_id,
            "date_debut": self.date_debut,
            "date_fin": self.date_fin,
            "etat": etat,
            "desc": self.description,
            "entry_date": self.entry_date,
            "user_id": username,
            "est_just": self.est_just,
            "external_data": self.external_data,
        }
        return data

    def __str__(self) -> str:
        "chaine pour journaux et debug (lisible par humain français)"
        try:
            etat_str = EtatAssiduite(self.etat).name.lower().capitalize()
        except ValueError:
            etat_str = "Invalide"
        return f"""{etat_str} {
                "just." if self.est_just else "non just."
            } de {
                self.date_debut.strftime("%d/%m/%Y %Hh%M")
            } à {
                self.date_fin.strftime("%d/%m/%Y %Hh%M")
            }"""

    @classmethod
    def create_assiduite(
        cls,
        etud: Identite,
        date_debut: datetime,
        date_fin: datetime,
        etat: EtatAssiduite,
        moduleimpl: ModuleImpl = None,
        description: str = None,
        entry_date: datetime = None,
        user_id: int = None,
        est_just: bool = False,
        external_data: dict = None,
    ) -> "Assiduite":
        """Créer une nouvelle assiduité pour l'étudiant"""
        # Vérification de non duplication des périodes
        assiduites: Query = etud.assiduites
        if is_period_conflicting(date_debut, date_fin, assiduites, Assiduite):
            raise ScoValueError(
                "Duplication des assiduités (la période rentrée rentre en conflit avec une assiduité enregistrée)"
            )

        if not est_just:
            est_just = (
                len(
                    get_justifs_from_date(etud.etudid, date_debut, date_fin, valid=True)
                )
                > 0
            )

        if moduleimpl is not None:
            # Vérification de l'existence du module pour l'étudiant
            if moduleimpl.est_inscrit(etud):
                nouv_assiduite = Assiduite(
                    date_debut=date_debut,
                    date_fin=date_fin,
                    etat=etat,
                    etudiant=etud,
                    moduleimpl_id=moduleimpl.id,
                    description=description,
                    entry_date=entry_date,
                    user_id=user_id,
                    est_just=est_just,
                    external_data=external_data,
                )
            else:
                raise ScoValueError("L'étudiant n'est pas inscrit au module")
        else:
            nouv_assiduite = Assiduite(
                date_debut=date_debut,
                date_fin=date_fin,
                etat=etat,
                etudiant=etud,
                description=description,
                entry_date=entry_date,
                user_id=user_id,
                est_just=est_just,
                external_data=external_data,
            )
        db.session.add(nouv_assiduite)
        log(f"create_assiduite: {etud.id} {nouv_assiduite}")
        Scolog.logdb(
            method="create_assiduite",
            etudid=etud.id,
            msg=f"assiduité: {nouv_assiduite}",
        )
        return nouv_assiduite


class Justificatif(db.Model):
    """
    Représente un justificatif:
        - une plage horaire lié à un état et un étudiant
        - une raison si spécifiée
        - un fichier si spécifié
    """

    __tablename__ = "justificatifs"

    id = db.Column(db.Integer, primary_key=True)
    justif_id = db.synonym("id")

    date_debut = db.Column(
        db.DateTime(timezone=True), server_default=db.func.now(), nullable=False
    )
    date_fin = db.Column(
        db.DateTime(timezone=True), server_default=db.func.now(), nullable=False
    )

    etudid = db.Column(
        db.Integer,
        db.ForeignKey("identite.id", ondelete="CASCADE"),
        index=True,
        nullable=False,
    )
    etat = db.Column(
        db.Integer,
        nullable=False,
    )

    entry_date = db.Column(db.DateTime(timezone=True), server_default=db.func.now())

    user_id = db.Column(
        db.Integer,
        db.ForeignKey("user.id", ondelete="SET NULL"),
        nullable=True,
        index=True,
    )

    raison = db.Column(db.Text())

    # Archive_id -> sco_archives_justificatifs.py
    fichier = db.Column(db.Text())

    # Déclare la relation "joined" car on va très souvent vouloir récupérer
    # l'étudiant en même tant que le justificatif (perf.: évite nouvelle requete SQL)
    etudiant = db.relationship(
        "Identite", back_populates="justificatifs", lazy="joined"
    )

    external_data = db.Column(db.JSON, nullable=True)

    def to_dict(self, format_api: bool = False) -> dict:
        """transformation de l'objet en dictionnaire sérialisable"""

        etat = self.etat
        username = self.user_id

        if format_api:
            etat = EtatJustificatif.inverse().get(self.etat).name
            if self.user_id is not None:
                user: User = db.session.get(User, self.user_id)
                if user is None:
                    username = "Non renseigné"
                else:
                    username = user.get_prenomnom()

        data = {
            "justif_id": self.justif_id,
            "etudid": self.etudid,
            "code_nip": self.etudiant.code_nip,
            "date_debut": self.date_debut,
            "date_fin": self.date_fin,
            "etat": etat,
            "raison": self.raison,
            "fichier": self.fichier,
            "entry_date": self.entry_date,
            "user_id": username,
            "external_data": self.external_data,
        }
        return data

    def __str__(self) -> str:
        "chaine pour journaux et debug (lisible par humain français)"
        try:
            etat_str = EtatJustificatif(self.etat).name
        except ValueError:
            etat_str = "Invalide"
        return f"""Justificatif {etat_str} de {
                self.date_debut.strftime("%d/%m/%Y %Hh%M")
            } à {
                self.date_fin.strftime("%d/%m/%Y %Hh%M")
            }"""

    @classmethod
    def create_justificatif(
        cls,
        etud: Identite,
        date_debut: datetime,
        date_fin: datetime,
        etat: EtatJustificatif,
        raison: str = None,
        entry_date: datetime = None,
        user_id: int = None,
        external_data: dict = None,
    ) -> "Justificatif":
        """Créer un nouveau justificatif pour l'étudiant"""
        nouv_justificatif = Justificatif(
            date_debut=date_debut,
            date_fin=date_fin,
            etat=etat,
            etudiant=etud,
            raison=raison,
            entry_date=entry_date,
            user_id=user_id,
            external_data=external_data,
        )

        db.session.add(nouv_justificatif)

        log(f"create_justificatif: {etud.id} {nouv_justificatif}")
        Scolog.logdb(
            method="create_justificatif",
            etudid=etud.id,
            msg=f"justificatif: {nouv_justificatif}",
        )
        return nouv_justificatif


def is_period_conflicting(
    date_debut: datetime,
    date_fin: datetime,
    collection: Query,
    collection_cls: Assiduite | Justificatif,
) -> bool:
    """
    Vérifie si une date n'entre pas en collision
    avec les justificatifs ou assiduites déjà présentes
    """

    date_debut = localize_datetime(date_debut)
    date_fin = localize_datetime(date_fin)

    if (
        collection.filter_by(date_debut=date_debut, date_fin=date_fin).first()
        is not None
    ):
        return True

    count: int = collection.filter(
        collection_cls.date_debut < date_fin, collection_cls.date_fin > date_debut
    ).count()

    return count > 0


def compute_assiduites_justified(
    etudid: int, justificatifs: list[Justificatif] = None, reset: bool = False
) -> list[int]:
    """
    compute_assiduites_justified_faster

    Args:
        etudid (int): l'identifiant de l'étudiant
        justificatifs (list[Justificatif]): La liste des justificatifs qui seront utilisés
        reset (bool, optional): remet à false les assiduites non justifiés. Defaults to False.

    Returns:
        list[int]: la liste des assiduités qui ont été justifiées.
    """
    if justificatifs is None:
        justificatifs: Justificatif = Justificatif.query.filter_by(etudid=etudid).all()

    justificatifs = [j for j in justificatifs if j.etat == EtatJustificatif.VALIDE]

    assiduites: Assiduite = Assiduite.query.filter_by(etudid=etudid)

    assiduites_justifiees: list[int] = []

    for assi in assiduites:
        if assi.etat == EtatAssiduite.PRESENT:
            continue

        assi_justificatifs = Justificatif.query.filter(
            Justificatif.etudid == assi.etudid,
            Justificatif.date_debut <= assi.date_debut,
            Justificatif.date_fin >= assi.date_fin,
            Justificatif.etat == EtatJustificatif.VALIDE,
        ).all()

        if any(
            assi.date_debut >= j.date_debut and assi.date_fin <= j.date_fin
            for j in justificatifs + assi_justificatifs
        ):
            assi.est_just = True
            assiduites_justifiees.append(assi.assiduite_id)
            db.session.add(assi)
        elif reset:
            assi.est_just = False
            db.session.add(assi)
    db.session.commit()
    return assiduites_justifiees


def get_assiduites_justif(assiduite_id: int, long: bool):
    assi: Assiduite = Assiduite.query.get_or_404(assiduite_id)
    return get_justifs_from_date(assi.etudid, assi.date_debut, assi.date_fin, long)


def get_justifs_from_date(
    etudid: int,
    date_debut: datetime,
    date_fin: datetime,
    long: bool = False,
    valid: bool = False,
):
    justifs: Query = Justificatif.query.filter(
        Justificatif.etudid == etudid,
        Justificatif.date_debut <= date_debut,
        Justificatif.date_fin >= date_fin,
    )

    if valid:
        justifs = justifs.filter(Justificatif.etat == EtatJustificatif.VALIDE)

    return [j.justif_id if not long else j.to_dict(True) for j in justifs]