mp2i-vms.fr
422 lines
14 KiB
Python
422 lines
14 KiB
Python
from datetime import date, datetime, timedelta
|
|
|
|
from asgiref.sync import async_to_sync
|
|
from pytz import timezone
|
|
|
|
import aiohttp
|
|
|
|
from django.db import models
|
|
from django.db.models import F, Q, Count, QuerySet
|
|
from django.contrib.auth.models import User
|
|
from django.conf import settings
|
|
|
|
from discord import Webhook
|
|
|
|
calendar = {
|
|
"C": [
|
|
(date(2023, 10, 21), date(2023, 11, 6)),
|
|
(date(2023, 12, 23), date(2024, 1, 8)),
|
|
(date(2024, 2, 10), date(2024, 2, 26)),
|
|
(date(2024, 4, 6), date(2024, 4, 22)),
|
|
(date(2024, 5, 6), date(2024, 5, 11)), # pont un peu gratté
|
|
]
|
|
}
|
|
|
|
|
|
class School(models.Model):
|
|
uai = models.CharField(max_length=10)
|
|
description = models.CharField(max_length=100)
|
|
vacation = models.CharField(max_length=1)
|
|
|
|
def __str__(self) -> str:
|
|
return self.description
|
|
|
|
|
|
class Class(models.Model):
|
|
school = models.ForeignKey(School, on_delete=models.CASCADE)
|
|
description = models.CharField(max_length=20)
|
|
year = models.IntegerField()
|
|
day_zero = models.DateField()
|
|
|
|
def week_number(self, day: date) -> int:
|
|
"""
|
|
Entrées :
|
|
- self
|
|
- day (datetime.date)
|
|
|
|
Sortie :
|
|
- Le numéro de la semaine contenant day, sans compter les vacation.
|
|
Renvoie un numéro non spécifiée si le day est pendant une période de vacation
|
|
"""
|
|
|
|
zone = self.school.vacation
|
|
vacation = calendar[zone]
|
|
day0 = self.day_zero
|
|
|
|
n = 1 + (day - day0).days // 7
|
|
for debut, fin in vacation:
|
|
if day > debut:
|
|
n -= round((fin - debut) / timedelta(weeks=1))
|
|
return n
|
|
|
|
def today_number(self) -> int:
|
|
"""
|
|
Entrée:
|
|
- self
|
|
|
|
Sortie:
|
|
- Le numéro de la semaine courante
|
|
"""
|
|
|
|
return self.week_number(date.today())
|
|
|
|
def week_beginning_date(self, n: int) -> date:
|
|
"""
|
|
Entrée:
|
|
- self
|
|
- n (int) <-> numéro de semaine
|
|
|
|
Sortie:
|
|
- Le date du lundi de la semaine n
|
|
"""
|
|
|
|
zone = self.school.vacation
|
|
vacation = calendar[zone]
|
|
day0 = self.day_zero
|
|
|
|
day = day0 + (n - 1) * timedelta(weeks=1)
|
|
|
|
for begin, end in vacation:
|
|
if day >= begin:
|
|
day += round((end - begin) / timedelta(weeks=1)) * timedelta(weeks=1)
|
|
|
|
return day
|
|
|
|
def term_of_date(self, day: date):
|
|
"""
|
|
Entrées :
|
|
- self
|
|
- day (datetime.date)
|
|
|
|
Sortie :
|
|
- La période (si elle existe et est unique) contenant day
|
|
|
|
Exceptions:
|
|
- Le day n'est pas dans une période
|
|
- Le day est au chevauchement de deux périodes
|
|
"""
|
|
return Term.objects.get(cls=self, debut__lte=day, fin__gte=day)
|
|
|
|
def current_term(self):
|
|
"""
|
|
On prend la période non révolue la plus récente
|
|
"""
|
|
|
|
return (Term.objects
|
|
.filter(cls=self, end__gte=date.today())
|
|
.order_by("-begin")
|
|
.first())
|
|
|
|
def __str__(self):
|
|
return f"{self.description} ({self.lycee.description})"
|
|
|
|
|
|
class Term(models.Model):
|
|
cls = models.ForeignKey(Class, on_delete=models.CASCADE)
|
|
description = models.CharField(max_length=100)
|
|
begin = models.DateField()
|
|
end = models.DateField()
|
|
|
|
class Meta:
|
|
ordering = ["begin"]
|
|
|
|
def range_weeks(self) -> range:
|
|
"""
|
|
Entrée:
|
|
- self
|
|
|
|
Sortie:
|
|
- Un range des numéros de semaine
|
|
"""
|
|
return range(self.cls.week_number(self.begin), self.cls.week_number(self.end) + 1)
|
|
|
|
def query_colles(self) -> QuerySet:
|
|
return (Colle.objects
|
|
.select_related("slot", "slot__term")
|
|
.prefetch_related("swap_set")
|
|
.filter(slot__term=self)
|
|
.annotate(adt_plus=Count("swap", filter=Q(swap__enroll=1)))
|
|
.annotate(adt_minus=Count("swap", filter=Q(swap__enroll=0)))
|
|
.annotate(volume=F("slot__capacity") + F("adt_plus") - F("adt_minus")))
|
|
|
|
def query_colles_of_student(self, student) -> QuerySet:
|
|
return (Colle.objects
|
|
.select_related("slot", "slot__term")
|
|
.prefetch_related("swap_set")
|
|
.filter(slot__term=self)
|
|
.filter((Q(groups__student=student)
|
|
& ~Q(swap__enroll=0, swap__student=student))
|
|
| Q(swap__enroll=1, swap__student=student))
|
|
.annotate(adt_plus=Count("swap", filter=Q(swap__enroll=1)))
|
|
.annotate(adt_minus=Count("swap", filter=Q(swap__enroll=0)))
|
|
.annotate(volume=F("slot__capacity") + F("adt_plus") - F("adt_minus")))
|
|
|
|
def query_colles_not_full(self) -> QuerySet:
|
|
return (self.query_colles()
|
|
.filter(volume__lt=F("slot__capacity"), date__gte=date.today()))
|
|
|
|
def __str__(self) -> str:
|
|
return self.description
|
|
|
|
|
|
class Subject(models.Model):
|
|
cls = models.ForeignKey(Class, on_delete=models.CASCADE)
|
|
description = models.CharField(max_length=100)
|
|
code = models.CharField(max_length=20)
|
|
|
|
def __str__(self):
|
|
return self.description
|
|
|
|
|
|
class GroupType(models.Model):
|
|
term = models.ForeignKey(Term, on_delete=models.CASCADE)
|
|
description = models.CharField(max_length=100)
|
|
|
|
def __str__(self):
|
|
return self.description
|
|
|
|
|
|
class Group(models.Model):
|
|
class Meta:
|
|
ordering = ["term__cls__description", "term__description", "description"]
|
|
|
|
term = models.ForeignKey(Term, on_delete=models.CASCADE)
|
|
type = models.ForeignKey(GroupType, null=True, on_delete=models.CASCADE)
|
|
description = models.CharField(max_length=100)
|
|
|
|
members = models.ManyToManyField("Student", through="Member")
|
|
|
|
def __str__(self):
|
|
return self.description
|
|
|
|
"""def get_colles(self):
|
|
return Rotation.objects.filter(slot__term=self.term,
|
|
Q(groupes=self) || Q(a)).order_by("date")
|
|
|
|
def get_colles_par_sem(self):
|
|
semaines = ((s, self.term.cls.week_beginning_date(s)) for s in self.term.range_semaines())
|
|
colles_flat = self.get_colles()
|
|
|
|
return [
|
|
(sem, lundi,
|
|
colles_flat.filter(date__gte=lundi, date__lt=lundi + timedelta(weeks=1)))
|
|
for sem, lundi in semaines
|
|
]"""
|
|
|
|
|
|
class Student(models.Model):
|
|
class Meta:
|
|
ordering = ["cls", "last_name", "first_name"]
|
|
|
|
cls = models.ForeignKey(Class, on_delete=models.CASCADE)
|
|
first_name = models.CharField(max_length=100)
|
|
last_name = models.CharField(max_length=100)
|
|
groups = models.ManyToManyField("Group", through="Member")
|
|
|
|
def is_member(self, group):
|
|
"""
|
|
Renvoie si self appartient au groupe.
|
|
"""
|
|
return group.members.contains(self)
|
|
|
|
def group_of_type(self, term, type_):
|
|
"""
|
|
Renvoie le groupe du critère auquel self appartient.
|
|
"""
|
|
if isinstance(type_, str):
|
|
type_ = GroupType.objects.get(term=term, description=type_)
|
|
|
|
return Member.objects.get(group__term=term, student=self, group__type=type_).group
|
|
|
|
def colle_group(self, term):
|
|
"""
|
|
Renvoie le groupe de colle de self pendant term.
|
|
"""
|
|
return self.group_of_type(term, "colle")
|
|
|
|
def __str__(self):
|
|
return f"{self.first_name} {self.last_name}"
|
|
|
|
|
|
class Member(models.Model):
|
|
student = models.ForeignKey(Student, on_delete=models.CASCADE)
|
|
group = models.ForeignKey(Group, on_delete=models.CASCADE)
|
|
|
|
|
|
class Colleur(models.Model):
|
|
gender = models.CharField(max_length=1)
|
|
name = models.CharField(max_length=100)
|
|
|
|
def __str__(self):
|
|
if self.gender == "M":
|
|
return f"M. {self.name}"
|
|
else:
|
|
return f"Mme {self.name}"
|
|
|
|
def get_classes(self):
|
|
return (x.term_of_date.cls for x in Slot.objects.filter(colleur=self).select_related("term__cls"))
|
|
|
|
|
|
class Slot(models.Model):
|
|
term = models.ForeignKey(Term, on_delete=models.CASCADE)
|
|
day = models.IntegerField()
|
|
time = models.TimeField()
|
|
duration = models.DurationField()
|
|
room = models.CharField(max_length=20)
|
|
subject = models.ForeignKey(Subject, on_delete=models.CASCADE)
|
|
colleur = models.ForeignKey(Colleur, on_delete=models.CASCADE)
|
|
type = models.ForeignKey(GroupType, on_delete=models.CASCADE)
|
|
capacity = models.IntegerField()
|
|
|
|
class Meta:
|
|
verbose_name_plural = "slots"
|
|
|
|
def __str__(self):
|
|
days = ["dimanche", "lundi", "mardi", "mercredi", "jeudi", "vendredi", "samedi"]
|
|
|
|
return f"Colle {self.subject} avec {self.colleur} {days[self.day]} {self.time}"
|
|
|
|
|
|
class Colle(models.Model):
|
|
class Meta:
|
|
ordering = ["slot__term__cls", "slot__subject__description", "slot__colleur__name", "date",
|
|
"slot__time"]
|
|
|
|
slot = models.ForeignKey(Slot, on_delete=models.CASCADE)
|
|
groups = models.ManyToManyField(Group)
|
|
date = models.DateField()
|
|
|
|
def initial_group(self):
|
|
"""
|
|
Renvoie les étudiants inscrits à la colle sans prendre en compte les swaps.
|
|
"""
|
|
return Student.objects.filter(id__in=Member.objects.filter(groupe__in=self.groups.all()))
|
|
|
|
def final_group(self):
|
|
"""
|
|
Renvoie les étudiants inscrits à la colle en tenant compte des swaps.
|
|
"""
|
|
swaps = Swap.objects.filter(colle=self)
|
|
|
|
return Student.objects.filter(
|
|
(Q(id__in=Member.objects.filter(group__in=self.groups.all()))
|
|
| Q(id__in=swaps.filter(enroll=True).values("student_id")))
|
|
& ~Q(id__in=swaps.filter(enroll=False).values("student_id"))
|
|
)
|
|
|
|
def volume(self):
|
|
"""
|
|
Renvoie le nombre d'étudiants inscrits à la colle en tenant compte des swaps.
|
|
"""
|
|
n_base = sum(len(group.members.count()) for group in self.groups.all())
|
|
n_plus = len(Swap.objects.filter(enroll=True, colle=self))
|
|
n_moins = len(Swap.objects.filter(enroll=False, colle=self))
|
|
|
|
return n_base + n_plus - n_moins
|
|
|
|
def is_full(self):
|
|
"""
|
|
Renvoie si la colle est pleine.
|
|
"""
|
|
eff = self.volume()
|
|
return eff >= self.slot.capacity
|
|
|
|
def is_edited(self):
|
|
"""
|
|
Renvoie si la colle a été amendée.
|
|
"""
|
|
return Swap.objects.filter(colle=self).exists()
|
|
|
|
def amend(self, student, enroll, notify=False):
|
|
"""
|
|
Amende la colle en (des)inscrivant student à la colle self, selon enroll.
|
|
"""
|
|
|
|
if Swap.objects.filter(colle=self, student=student, enroll=enroll).exists():
|
|
raise Exception("Duplication")
|
|
elif Swap.objects.filter(colle=self, student=student, enroll=not enroll).exists():
|
|
# les swaps complémentaires s'annulent
|
|
Swap.objects.get(colle=self, student=student, enroll=not enroll).delete()
|
|
elif enroll and any(student.is_member(group) for group in self.groups.all()):
|
|
# on ne peut pas s'ajouter si on est dans le groupe de base
|
|
raise Exception("Vous êtes déjà dans le groupe")
|
|
elif not enroll and all(not student.is_member(group) for group in self.groups.all()):
|
|
raise Exception("Vous n'êtes pas dans le groupe")
|
|
elif enroll and self.is_full():
|
|
raise Exception("Capacité dépassée")
|
|
else:
|
|
swap = Swap(colle=self, student=student, enroll=enroll)
|
|
swap.save()
|
|
|
|
#if notify:
|
|
# func = async_to_sync(swap.notify)
|
|
# func()
|
|
|
|
def __str__(self):
|
|
return f"{self.slot} le {self.date} avec groupes {'+'.join(str(groupe) for groupe in self.groups.all())}"
|
|
|
|
def datetime(self):
|
|
return datetime.combine(self.date, self.slot.time, tzinfo=timezone("Europe/Paris"))
|
|
|
|
|
|
class Swap(models.Model):
|
|
enroll = models.BooleanField()
|
|
colle = models.ForeignKey(Colle, on_delete=models.CASCADE)
|
|
student = models.ForeignKey(Student, on_delete=models.CASCADE)
|
|
|
|
async def notify(self):
|
|
async with aiohttp.ClientSession() as session:
|
|
webhook = Webhook.from_url(settings.DISCORD_NOTIFY_WEBHOOK_URL, session=session)
|
|
|
|
if self.enroll:
|
|
await webhook.send(f"Colle réservée : {self.colle}", username=settings.DISCORD_NOTIFY_WEBHOOK_USERNAME)
|
|
else:
|
|
await webhook.send(f"Colle libérée : {self.colle}", username=settings.DISCORD_NOTIFY_WEBHOOK_USERNAME)
|
|
|
|
|
|
class Profile(models.Model):
|
|
user = models.OneToOneField(User, on_delete=models.CASCADE, primary_key=True)
|
|
student = models.ForeignKey(Student, null=True, blank=True, on_delete=models.SET_NULL)
|
|
colleur = models.ForeignKey(Colleur, null=True, blank=True, on_delete=models.SET_NULL)
|
|
|
|
def __str__(self):
|
|
return f"Profil {self.user} : {self.student} ; {self.colleur}"
|
|
|
|
@staticmethod
|
|
def from_request(request, preprocess=lambda query: query):
|
|
user = request.user
|
|
session = request.session
|
|
|
|
match session.get("profile"):
|
|
case "student":
|
|
profil = preprocess(Profile.objects.filter(user=user)).get()
|
|
return profil.student
|
|
case "colleur":
|
|
profil = preprocess(Profile.objects.filter(user=user)).get()
|
|
return profil.colleur
|
|
case _:
|
|
raise ValueError("profil non choisi")
|
|
|
|
|
|
class CalendarLink(models.Model):
|
|
key = models.CharField(max_length=32, unique=True)
|
|
student = models.ForeignKey(Student, on_delete=models.CASCADE)
|
|
term = models.ForeignKey(Term, on_delete=models.CASCADE)
|
|
|
|
class Meta:
|
|
constraints = [
|
|
models.UniqueConstraint(
|
|
fields=['student', 'term'], name='unique_student_term_combination'
|
|
)
|
|
]
|