346 lines
10 KiB
Python
346 lines
10 KiB
Python
# 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
|