mp2i-vms.fr
452 lines
15 KiB
Python
452 lines
15 KiB
Python
from datetime import date, datetime, timedelta
|
|
from pprint import pprint
|
|
|
|
from pytz import timezone
|
|
|
|
import aiohttp
|
|
|
|
from django.db import models
|
|
from django.db.models import F, Q, Count, QuerySet, Subquery, OuterRef, Sum
|
|
from django.contrib.auth.models import User
|
|
from django.conf import settings
|
|
|
|
from discord import Webhook
|
|
|
|
calendar = {
|
|
"C": [
|
|
(date(2024, 10, 19), date(2024, 11, 4)),
|
|
(date(2024, 12, 21), date(2025, 1, 6)),
|
|
(date(2025, 2, 15), date(2025, 3, 3)),
|
|
(date(2025, 4, 12), date(2025, 4, 28)),
|
|
]
|
|
}
|
|
|
|
|
|
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.school.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(base_vol=Count("groups__members"))
|
|
.annotate(swap_plus=Count("swap", filter=Q(swap__enroll=1), distinct=True))
|
|
.annotate(swap_minus=Count("swap", filter=Q(swap__enroll=0), distinct=True))
|
|
.annotate(volume=F("base_vol") + F("swap_plus") - F("swap_minus"))
|
|
.order_by("datetime", "slot__time"))
|
|
|
|
def query_colles_of_student(self, student) -> QuerySet:
|
|
has_student = ((Q(groups__student=student)
|
|
& ~Q(swap__enroll=0, swap__student=student))
|
|
| Q(swap__enroll=1, swap__student=student))
|
|
|
|
return (Colle.objects
|
|
.select_related("slot", "slot__term")
|
|
.prefetch_related("swap_set")
|
|
.filter(slot__term=self)
|
|
.annotate(base_vol=Count("groups__members", distinct=True))
|
|
.annotate(swap_plus=Count("pk", filter=Q(swap__enroll=1), distinct=True))
|
|
.annotate(swap_minus=Count("pk", filter=Q(swap__enroll=0), distinct=True))
|
|
.annotate(volume=F("base_vol") + F("swap_plus") - F("swap_minus"))
|
|
.filter(has_student)
|
|
.order_by()
|
|
)
|
|
|
|
def query_colles_not_full_excluding_student(self, student) -> QuerySet:
|
|
has_student = ((Q(groups__student=student)
|
|
& ~Q(swap__enroll=0, swap__student=student))
|
|
| Q(swap__enroll=1, swap__student=student))
|
|
|
|
return (self.query_colles()
|
|
.filter(volume__lt=F("slot__capacity"), datetime__gte=date.today())
|
|
.exclude(has_student))
|
|
|
|
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)
|
|
|
|
class Meta:
|
|
ordering = ["description"]
|
|
|
|
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 f"{self.description} ({self.type})"
|
|
|
|
"""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_fla.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 swap_score(self, term):
|
|
colles = term.query_colles_of_student(self)
|
|
|
|
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)
|
|
|
|
def __str__(self):
|
|
return f"{self.student} of {self.group}"
|
|
|
|
|
|
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:
|
|
ordering = ["subject", "colleur", "day", "time"]
|
|
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", "datetime"]
|
|
|
|
slot = models.ForeignKey(Slot, on_delete=models.CASCADE)
|
|
groups = models.ManyToManyField(Group, blank=True)
|
|
datetime = models.DateTimeField()
|
|
|
|
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 is_attendee(self, student):
|
|
return self.final_group().contains(student)
|
|
|
|
def get_volume(self):
|
|
"""
|
|
Renvoie le nombre d'étudiants inscrits à la colle en tenant compte des swaps.
|
|
"""
|
|
return (Student.objects
|
|
.filter(Q(groups__colle=self)
|
|
& ~Q(swap__colle=self, swap__enroll=False)
|
|
| Q(swap__colle=self, swap__enroll=True))
|
|
.distinct()
|
|
.count())
|
|
|
|
def is_full(self):
|
|
"""
|
|
Renvoie si la colle est pleine.
|
|
"""
|
|
return self.get_volume() >= 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"Colle {self.slot.subject} ({self.slot.colleur}); {self.datetime} {self.slot.time} {self.slot.room}. Groupe(s) {{{'; '.join(str(groupe) for groupe in self.groups.all())}}}"
|
|
|
|
|
|
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'
|
|
)
|
|
]
|
|
|
|
|
|
|
|
def test():
|
|
valentin = Student.objects.get(pk=25)
|
|
term = Term.objects.get(pk=3)
|
|
colles = term.query_colles_of_student(valentin).order_by("-volume")
|
|
|
|
for c in colles:
|
|
print(f"* {c.slot} {c.volume} : {c.base_vol} + {c.swap_plus} - {c.swap_minus}")
|