ScoDoc/tools/migrate_abs_to_assiduites.py

# Script de migration des données de la base "absences" -> "assiduites"/"justificatifs"
from app import db
from app.profiler import Profiler
from app.models import (
    Assiduite,
    Justificatif,
    Absence,
    Identite,
    ModuleImplInscription,
    Departement,
)
from app.scodoc.sco_utils import (
    EtatAssiduite,
    EtatJustificatif,
    localize_datetime,
    ProgressBarColors,
    printProgressBar,
)
from datetime import time, datetime, date
from json import dump, dumps


class _glob:
    DUPLICATIONS_ASSIDUITES: dict[tuple[date, bool, int], Assiduite] = {}
    DUPLICATIONS_JUSTIFICATIFS: dict[tuple[date, bool, int], Justificatif] = {}
    DUPLICATED: list[Justificatif] = []
    PROBLEMS: dict[int, list[str]] = {}
    CURRENT_ETU: list = []
    MODULES: list[tuple[int, int]] = []
    COMPTE: list[int, int] = []
    ERR_ETU: list[int] = []


class _Statistics:
    def __init__(self) -> None:
        self.object: dict = {"total": 0}
        self.year: int = None

    def __set_year(self, year: int):
        if year not in self.object:
            self.object[year] = {
                "etuds_inexistant": [],
                "abs_invalide": {},
            }
        self.year = year
        return self

    def __add_etud(self, etudid: int):
        if etudid not in self.object[self.year]["etuds_inexistant"]:
            self.object[self.year]["etuds_inexistant"].append(etudid)
        return self

    def __add_abs(self, abs: int, err: str):
        if abs not in self.object[self.year]["abs_invalide"]:
            self.object[self.year]["abs_invalide"][abs] = [err]
        else:
            self.object[self.year]["abs_invalide"][abs].append(err)

        return self

    def add_problem(self, abs: Absence, err: str):
        abs.jour: date
        pivot: date = date(abs.jour.year, 9, 15)
        year: int = abs.jour.year
        if pivot < abs.jour:
            year += 1
        self.__set_year(year)

        if err == "Etudiant inexistant":
            self.__add_etud(abs.etudid)
        else:
            self.__add_abs(abs.id, err)

        self.object["total"] += 1

    def compute_stats(self) -> dict:
        stats: dict = {"total": self.object["total"]}
        for year in self.object:
            if year == "total":
                continue
            stats[year] = {}
            stats[year]["etuds_inexistant"] = len(self.object[year]["etuds_inexistant"])
            stats[year]["abs_invalide"] = len(self.object[year]["abs_invalide"])

        return stats

    def export(self, file):
        dump(self.object, file, indent=2)


def migrate_abs_to_assiduites(
    dept: str = "", morning: str = None, noon: str = None, evening: str = None
):
    """
    une absence à 3 états:

    |.estabs|.estjust|
    |1|0| -> absence non justifiée
    |1|1| -> absence justifiée
    |0|1| -> justifié

    dualité des temps :

    .matin: bool (0:00 -> time_pref | time_pref->23:59:59)
    .jour : date (jour de l'absence/justificatif)
    .moduleimpl_id: relation -> moduleimpl_id
    description:str -> motif abs / raision justif

    .entry_date: datetime -> timestamp d'entrée de l'abs
    .etudid: relation -> Identite
    """
    Profiler.clear()
    stats: _Statistics = _Statistics()

    time_elapsed: Profiler = Profiler("migration")
    time_elapsed.start()

    if morning is None:
        pref_time_morning = time(8, 0)
    else:
        morning: list[str] = morning.split("h")
        pref_time_morning = time(int(morning[0]), int(morning[1]))

    if noon is None:
        pref_time_noon = time(12, 0)
    else:
        noon: list[str] = noon.split("h")
        pref_time_noon = time(int(noon[0]), int(noon[1]))

    if evening is None:
        pref_time_evening = time(18, 0)
    else:
        evening: list[str] = evening.split("h")
        pref_time_evening = time(int(evening[0]), int(evening[1]))

    absences_query = Absence.query
    if dept is not None:

        dept: Departement = Departement.query.filter_by(acronym=dept).first()
        if dept is not None:
            etuds_id: list[int] = [etud.id for etud in dept.etudiants]
            absences_query = absences_query.filter(Absence.etudid.in_(etuds_id))
    absences: Absence = absences_query.order_by(Absence.etudid)

    _glob.DUPLICATED = []
    _glob.DUPLICATIONS_ASSIDUITES = {}
    _glob.DUPLICATIONS_JUSTIFICATIFS = {}
    _glob.CURRENT_ETU = []
    _glob.MODULES = []
    _glob.COMPTE = [0, 0]
    _glob.ERR_ETU = []

    absences_len: int = absences.count()

    print(
        f"{ProgressBarColors.BLUE}{absences_len} absences vont être migrées{ProgressBarColors.RESET}"
    )

    printProgressBar(0, absences_len, "Progression", "effectué", autosize=True)

    for i, abs in enumerate(absences):
        try:
            if abs.estabs:
                generated = _from_abs_to_assiduite(
                    abs, pref_time_morning, pref_time_noon, pref_time_evening
                )
            if not isinstance(generated, str):
                db.session.add(generated)
                _glob.COMPTE[0] += 1
        except Exception as e:
            stats.add_problem(abs, e.args[0])

        try:
            if abs.estjust:
                generated = _from_abs_to_justificatif(
                    abs, pref_time_morning, pref_time_noon, pref_time_evening
                )
                if not isinstance(generated, str):
                    db.session.add(generated)
                    _glob.COMPTE[1] += 1

        except Exception as e:
            stats.add_problem(abs, e.args[0])

        if i % 10 == 0:
            printProgressBar(
                i,
                absences_len,
                "Progression",
                "effectué",
                autosize=True,
            )

        if i % 1000 == 0:
            printProgressBar(
                i,
                absences_len,
                "Progression",
                "effectué",
                autosize=True,
            )
            db.session.commit()

    dup_assi = _glob.DUPLICATED
    assi: Assiduite
    for assi in dup_assi:
        assi.moduleimpl_id = None
        db.session.add(assi)

    db.session.commit()

    printProgressBar(
        absences_len,
        absences_len,
        "Progression",
        "effectué",
        autosize=True,
    )

    time_elapsed.stop()

    statistiques: dict = stats.compute_stats()
    print(
        f"{ProgressBarColors.GREEN}La migration a pris {time_elapsed.elapsed():.2f} secondes {ProgressBarColors.RESET}"
    )

    print(
        f"{ProgressBarColors.RED}{statistiques['total']} absences qui n'ont pas pu être migrée."
    )
    print(
        f"Vous retrouverez un fichier json {ProgressBarColors.GREEN}/tmp/scodoc_migration_abs.json{ProgressBarColors.RED} contenant les problèmes de migrations"
    )
    with open("/tmp/scodoc_migration_abs.json", "w", encoding="utf-8") as file:
        stats.export(file)

    print(
        f"{ProgressBarColors.CYAN}{_glob.COMPTE[0]} assiduités et {_glob.COMPTE[1]} justificatifs ont été générés.{ProgressBarColors.RESET}"
    )

    print(dumps(statistiques, indent=2))


def _from_abs_to_assiduite(
    _abs: Absence, morning: time, noon: time, evening: time
) -> Assiduite:
    etat = EtatAssiduite.ABSENT
    date_deb: datetime = None
    date_fin: datetime = None

    if _abs.matin:
        date_deb = datetime.combine(_abs.jour, morning)
        date_fin = datetime.combine(_abs.jour, noon)
    else:
        date_deb = datetime.combine(_abs.jour, noon)
        date_fin = datetime.combine(_abs.jour, evening)

    date_deb = localize_datetime(date_deb)
    date_fin = localize_datetime(date_fin)

    duplicata: Assiduite = _glob.DUPLICATIONS_ASSIDUITES.get(
        (_abs.jour, _abs.matin, _abs.etudid)
    )
    if duplicata is not None:
        _glob.DUPLICATED.append(duplicata)
        return "Duplicata"

    desc: str = _abs.description
    entry_date: datetime = _abs.entry_date

    if _abs.etudid not in _glob.CURRENT_ETU:
        etud: Identite = Identite.query.filter_by(id=_abs.etudid).first()
        if etud is None:
            raise Exception("Etudiant inexistant")
        _glob.CURRENT_ETU.append(_abs.etudid)

    moduleimpl_id: int = _abs.moduleimpl_id

    if (
        moduleimpl_id is not None
        and (_abs.etudid, _abs.moduleimpl_id) not in _glob.MODULES
    ):
        moduleimpl_inscription: ModuleImplInscription = (
            ModuleImplInscription.query.filter_by(
                moduleimpl_id=_abs.moduleimpl_id, etudid=_abs.etudid
            ).first()
        )
        if moduleimpl_inscription is None:
            raise Exception("Moduleimpl_id incorrect ou étudiant non inscrit")

    retour = Assiduite.fast_create_assiduite(
        etudid=_abs.etudid,
        date_debut=date_deb,
        date_fin=date_fin,
        etat=etat,
        moduleimpl_id=moduleimpl_id,
        description=desc,
        entry_date=entry_date,
    )
    _glob.DUPLICATIONS_ASSIDUITES[(_abs.jour, _abs.matin, _abs.etudid)] = retour
    return retour


def _from_abs_to_justificatif(
    _abs: Absence, morning: time, noon: time, evening: time
) -> Justificatif:

    etat = EtatJustificatif.VALIDE
    date_deb: datetime = None
    date_fin: datetime = None
    if _abs.matin:
        date_deb = datetime.combine(_abs.jour, morning)
        date_fin = datetime.combine(_abs.jour, noon)
    else:
        date_deb = datetime.combine(_abs.jour, noon)
        date_fin = datetime.combine(_abs.jour, evening)

    date_deb = localize_datetime(date_deb)
    date_fin = localize_datetime(date_fin)

    duplicata: Justificatif = _glob.DUPLICATIONS_JUSTIFICATIFS.get(
        (_abs.jour, _abs.matin, _abs.etudid)
    )
    if duplicata is not None:
        return "Duplicated"

    desc: str = _abs.description
    entry_date: datetime = _abs.entry_date

    if _abs.etudid not in _glob.CURRENT_ETU:
        etud: Identite = Identite.query.filter_by(id=_abs.etudid).first()
        if etud is None:
            raise Exception("Etudiant inexistant")
        _glob.CURRENT_ETU.append(_abs.etudid)

    retour = Justificatif.fast_create_justificatif(
        etudid=_abs.etudid,
        date_debut=date_deb,
        date_fin=date_fin,
        etat=etat,
        raison=desc,
        entry_date=entry_date,
    )

    _glob.DUPLICATIONS_JUSTIFICATIFS[(_abs.jour, _abs.matin, _abs.etudid)] = retour
    return retour