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(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.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()) 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"), date__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 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 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) 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", "date", "slot__time"] slot = models.ForeignKey(Slot, on_delete=models.CASCADE) groups = models.ManyToManyField(Group, blank=True) 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 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.date} {self.slot.time} {self.slot.room}. Groupe(s) {{{'; '.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' ) ] 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}")