Source code for intranet.apps.eighth.models

# pylint: disable=too-many-lines; Allow more than 1000 lines
import datetime
import logging
import string
from typing import Collection, Iterable, List, Optional, Union

from cacheops import invalidate_obj
from sentry_sdk import add_breadcrumb, capture_exception
from simple_history.models import HistoricalRecords

from django.conf import settings
from django.contrib.auth import get_user_model
from django.contrib.auth.models import Group as DjangoGroup
from django.core import validators
from django.core.cache import cache
from django.core.exceptions import NON_FIELD_ERRORS, ValidationError
from django.db import models, transaction
from django.db.models import Count, Manager, Q, QuerySet
from django.http.request import HttpRequest
from django.utils import formats, timezone

from ...utils.date import get_date_range_this_year, is_current_year
from ...utils.deletion import set_historical_user
from ...utils.locking import lock_on
from ..notifications.tasks import email_send_task
from . import exceptions as eighth_exceptions

logger = logging.getLogger(__name__)


[docs]class AbstractBaseEighthModel(models.Model): """Abstract base model that includes created and last modified times. Attributes: created_time (datetime): The time the model was created last_modified_time (datetime): The time the model was last modified. """ created_time = models.DateTimeField(auto_now_add=True, null=True) last_modified_time = models.DateTimeField(auto_now=True, null=True)
[docs] class Meta: abstract = True
[docs]class EighthSponsor(AbstractBaseEighthModel): """Represents a sponsor for an eighth period activity. A sponsor could be linked to an actual user or just a name. Attributes: first_name (str): The first name of the sponsor. last_name (str): The last name of the sponsor. user (User): A :class:`users.User` object linked to the sponsor. department (str): The sponsor's department. full_time (bool): Whether the sponsor is full-time. online_attendance (bool): Whether the sponsor takes. attendance online. contracted_eighth (bool): Whether the sponsor is contracted to supervise 8th periods. show_full_name (bool): Whether to always show the sponsor's full name (e.x. because there are two teachers named Lewis). """ # Hard-coded list of departments in the school DEPARTMENTS = ( ("general", "General"), ("math_cs", "Math/CS"), ("english", "English"), ("social_studies", "Social Studies"), ("fine_arts", "Fine Arts"), ("health_pe", "Health/PE"), ("scitech", "Science/Technology"), ) first_name = models.CharField(max_length=50) last_name = models.CharField(max_length=50) user = models.OneToOneField(settings.AUTH_USER_MODEL, null=True, blank=True, on_delete=set_historical_user) department = models.CharField(max_length=20, choices=DEPARTMENTS, default="general") full_time = models.BooleanField(default=True) online_attendance = models.BooleanField(default=True) contracted_eighth = models.BooleanField(default=True) show_full_name = models.BooleanField(default=False) history = HistoricalRecords() class Meta: unique_together = (("first_name", "last_name", "user", "online_attendance", "full_time", "department"),) ordering = ("last_name", "first_name") @property def name(self) -> str: """If show_full_name is set, returns "last name, first name". Otherwise, returns last name only. Returns: The name to display for the sponsor, omitting the first name unless explicitly requested. """ if self.show_full_name and self.first_name: return self.last_name + ", " + self.first_name else: return self.last_name @property def to_be_assigned(self) -> bool: """Returns True if the sponsor's name contains "to be assigned" or similar. Returns: Whether the sponsor is a "to be assigned" sponsor. """ return any(x in self.name.lower() for x in ["to be assigned", "to be determined", "to be announced"]) def __str__(self): return self.name
[docs]class EighthRoom(AbstractBaseEighthModel): """Represents a room in which an eighth period activity can be held. Attributes: name (str): The name of the room. capacity (int): The maximum capacity of the room (-1 for unlimited, 0 to prevent student signup) available_for_eighth (bool): Whether the room is available for eighth period signups. """ name = models.CharField(max_length=100) capacity = models.SmallIntegerField(default=28) available_for_eighth = models.BooleanField(default=True) history = HistoricalRecords() unique_together = (("name", "capacity"),)
[docs] @staticmethod def total_capacity_of_rooms(rooms: Iterable["EighthRoom"]) -> int: """Returns the total capacity of the provided rooms. Args: rooms: Rooms to determine total capacity for. Returns: The total capacity of the provided rooms. """ capacity = 0 for r in rooms: if r.capacity == -1: return -1 capacity += r.capacity return capacity
@property def formatted_name(self) -> str: """The formatted name of the room. If it looks like the room is a numbered room -- the name starts with either a number or with the text "Room" -- returns "Rm. <room number>." Returns: The formatted name of the Room. """ if self.name[0] in string.digits: # All rooms starting with an integer will be prefixed return "Rm. {}".format(self.name) if self.name.startswith("Room"): # Some room names are prefixed with 'Room'; for consistency return "Rm. {}".format(self.name[5:]) return self.name @property def to_be_determined(self) -> bool: """Whether the Room needs to be assigned.""" return any(x in self.name.lower() for x in ["to be assigned", "to be determined", "to be announced"]) def __str__(self): return "{} ({})".format(self.name, self.capacity) class Meta: ordering = ("name",)
[docs]class EighthActivityExcludeDeletedManager(models.Manager):
[docs] def get_queryset(self): return super().get_queryset().exclude(deleted=True)
[docs]class EighthActivity(AbstractBaseEighthModel): """Represents an eighth period activity. Attributes: name(str): The name of the activity, max length 100 characters. description (str): The description of the activity, shown on the signup page below the other information. Information on an EighthScheduledActivity basis can be found in the "comments" field of that model. Max length 2000 characters. sponsors (:obj:`list` of :obj:`EighthSponsor`): The default activity-level sponsors for the activity. On an EighthScheduledActivity basis, you should NOT query this field. Instead, use scheduled_activity.get_true_sponsors() rooms (:obj:`list` of :obj:`EighthRoom`): The default activity-level rooms for the activity. On an EighthScheduledActivity basis, you should NOT query this field. Use scheduled_activity.get_true_rooms() default_capacity (int): The default capacity, which overrides the sum of the default rooms when scheduling the activity. By default, this has a null value and is ignored. presign (bool): If True, the activity can only be signed up for within 48 hours of the day that the activity is scheduled. one_a_day (bool): If True, a student can only sign up for one instance of this activity per day. both_blocks (bool): If True, a signup for an EighthScheduledActivity during an A or B block will enforce and automatically trigger a signup on the other block. Does not enforce signups for blocks other than A and B. sticky (bool): If True, then students who sign up or are placed in this activity cannot switch out of it. A sticky activity should also be restricted, unless you're mean. special (bool): If True, then the activity receives a special designation on the signup list, and is stuck to the top of the list. administrative (bool): If True, then students cannot see the activity in their signup list. However, the activity still exists in the system and can be seen by administrators. Students can still sign up for the activity through the API -- this does not prevent students from signing up for it, and just merely hides it from view. An administrative activity should be restricted. finance (str): The account name of the club with the Finance Office. If blank or null, there is no account. restricted (str): Whether the signups for the activity are restricted to certain users/groups or if there are blacklisted users. users_allowed (:obj:`list` of :obj:`User`): Individual users allowed to sign up for this activity. Extensive use of this is discouraged; make a group instead through the "Add and Assign Empty Group" button on the Edit Activity page. Only takes effect if the activity is restricted. groups_allowed (:obj:`list` of :obj:`Group`): Individual groups allowed to sign up for this activity. Only takes effect if the activity is restricted. users_blacklisted (:obj:`list` of :obj:`User`): Individual users who are not allowed to sign up for this activity. Only takes effect if the activity is not restricted. freshman_allowed (bool): Whether freshmen are allowed to sign up for this activity. Only takes effect if the activity is restricted. sophomores_allowed (bool): Whether sophomores are allowed to sign up for this activity. Only takes effect if the activity is restricted. juniors_allowed (bool): Whether juniors are allowed to sign up for this activity. Only takes effect if the activity is restricted. seniors_allowed (bool): Whether seniors are allowed to sign up for this activity. Only takes effect if the activity is restricted. wed_a (bool): Whether the activity generally meets on Wednesday A blocks. Does not affect schedule, is just information for the Eighth Office. wed_b (bool): Whether the activity generally meets on Wednesday B blocks. Does not affect schedule, is just information for the Eighth Office. fri_a (bool): Whether the activity generally meets on Friday A blocks. Does not affect schedule, is just information for the Eighth Office. fri_b (bool): Whether the activity generally meets on Friday B blocks. Does not affect schedule, is just information for the Eighth Office. admin_comments (str): Notes for the Eighth Office. favorites (:obj:`list` of :obj:`User`): A ManyToManyField of User objects who have favorited the activity. similarities (:obj:`list` of :obj:`EighthActivitySimilarity`): A ManyToManyField of EighthActivitySimilarity objects which are similar to this activity. deleted (bool): Whether the activity still technically exists in the system, but was marked to be deleted. """ objects = models.Manager() undeleted_objects = EighthActivityExcludeDeletedManager() name = models.CharField(max_length=100, validators=[validators.MinLengthValidator(4)]) # This should really be unique description = models.CharField(max_length=2000, blank=True) sponsors = models.ManyToManyField(EighthSponsor, blank=True) rooms = models.ManyToManyField(EighthRoom, blank=True) default_capacity = models.SmallIntegerField(null=True, blank=True) presign = models.BooleanField(default=False) one_a_day = models.BooleanField(default=False) both_blocks = models.BooleanField(default=False) sticky = models.BooleanField(default=False) special = models.BooleanField(default=False) administrative = models.BooleanField(default=False) finance = models.CharField(max_length=100, blank=True, null=True) restricted = models.BooleanField(default=False) users_allowed = models.ManyToManyField(settings.AUTH_USER_MODEL, related_name="restricted_activity_set", blank=True) groups_allowed = models.ManyToManyField(DjangoGroup, related_name="restricted_activity_set", blank=True) users_blacklisted = models.ManyToManyField(settings.AUTH_USER_MODEL, blank=True) freshmen_allowed = models.BooleanField(default=False) sophomores_allowed = models.BooleanField(default=False) juniors_allowed = models.BooleanField(default=False) seniors_allowed = models.BooleanField(default=False) wed_a = models.BooleanField("Meets Wednesday A", default=False) wed_b = models.BooleanField("Meets Wednesday B", default=False) fri_a = models.BooleanField("Meets Friday A", default=False) fri_b = models.BooleanField("Meets Friday B", default=False) admin_comments = models.CharField(max_length=1000, blank=True) favorites = models.ManyToManyField(settings.AUTH_USER_MODEL, related_name="favorited_activity_set", blank=True) similarities = models.ManyToManyField("EighthActivitySimilarity", related_name="activity_set", blank=True) deleted = models.BooleanField(blank=True, default=False) history = HistoricalRecords()
[docs] def capacity(self) -> int: """Returns 'default_capacity' if it is set. Otherwise, returns the total capacity of all the activity's rooms. Returns: The activity's capacity. """ # Note that this is the default capacity if the # rooms/capacity are not overridden for a particular block. return self.default_capacity or EighthRoom.total_capacity_of_rooms(self.rooms.all())
@property def aid(self) -> int: """The publicly visible activity ID. Returns: The public activity ID. """ return self.id @property def name_with_flags(self) -> str: """Return the activity name with special, both blocks, restricted, administrative, sticky, and deleted flags. Returns: The activity name with all flags. """ return self._name_with_flags(True) @property def name_with_flags_no_restricted(self) -> str: """Returns the activity's name with flags. These flags indicate whether the activity is special, both blocks, administrative, sticky, and/or cancelled. Returns: The activity name with all flags except the "restricted" flag. """ return self._name_with_flags(False) def _name_with_flags(self, include_restricted: bool, title: Optional[str] = None) -> str: """Generates the activity's name with flags. Args: include_restricted: Whether to include the "restricted" flag. title: An optional title to add after the activity name (but before the flags) Returns: The activity name with all the appropriate flags. """ name = "Special: " if self.special else "" name += self.name if title: name += " - {}".format(title) name += " (R)" if include_restricted and self.restricted else "" name += " (BB)" if self.both_blocks else "" name += " (A)" if self.administrative else "" name += " (S)" if self.sticky else "" name += " (Deleted)" if self.deleted else "" return name
[docs] @classmethod def restricted_activities_available_to_user(cls, user: "get_user_model()") -> List[int]: """Finds the restricted activities available to the given user. Args: user: The User to find the restricted activities for. Returns: The restricted activities available to the user. """ if not user: return [] q = Q(users_allowed=user.id) | Q(groups_allowed__user=user.id) if user and user.grade and user.grade.number and user.grade.name and 9 <= user.grade.number <= 12: q |= Q(**{"{}_allowed".format(user.grade.name_plural): True}) return EighthActivity.objects.filter(q).values_list("id", flat=True)
[docs] @classmethod def available_ids(cls) -> List[int]: """Returns all available IDs not used by an EighthActivity. Returns: A list of the available activity IDs. """ id_min = 1 id_max = 3200 nums = set(range(id_min, id_max)) used = {row[0] for row in EighthActivity.objects.values_list("id")} avail = nums - used return list(avail)
[docs] def get_active_schedulings(self) -> Union[QuerySet, Collection["EighthScheduledActivity"]]: # pylint: disable=unsubscriptable-object """Returns all EighthScheduledActivitys scheduled this year for this activity. Returns: EighthScheduledActivitys of this activity occurring this year. """ date_start, date_end = get_date_range_this_year() return EighthScheduledActivity.objects.filter(activity=self, block__date__gte=date_start, block__date__lte=date_end)
@property def is_active(self) -> bool: """Returns whether an activity is "active." An activity is considered to be active if it has been scheduled at all this year. Returns: Whether the activity is active. """ return self.get_active_schedulings().exists() @property def frequent_users(self) -> Union[QuerySet, Collection["get_user_model()"]]: # pylint: disable=unsubscriptable-object """Return a QuerySet of user id's and counts that have signed up for this activity more than `settings.SIMILAR_THRESHOLD` times. This is used for suggesting activities to users. Returns: A QuerySet of users who attend this activity frequently. """ key = "eighthactivity_{}:frequent_users".format(self.id) cached = cache.get(key) if cached: return cached freq_users = ( self.eighthscheduledactivity_set.exclude(Q(eighthsignup_set__user=None) | Q(administrative=True) | Q(special=True) | Q(restricted=True)) .values("eighthsignup_set__user") .annotate(count=Count("eighthsignup_set__user")) .filter(count__gte=settings.SIMILAR_THRESHOLD) .order_by("-count") ) cache.set(key, freq_users, timeout=60 * 60 * 24 * 7) return freq_users @property def is_popular(self) -> bool: """Returns whether this activity has more than ``settings.SIMILAR_THRESHOLD * 2`` frequent_users. Returns: Whether this activity has more than ``settings.SIMILAR_THRESHOLD * 2`` frequent_users. """ return self.frequent_users.count() > (settings.SIMILAR_THRESHOLD * 2) class Meta: verbose_name_plural = "eighth activities" def __str__(self): return self.name_with_flags
[docs]class EighthBlockQuerySet(models.query.QuerySet):
[docs] def this_year(self) -> Union[QuerySet, Collection["EighthBlock"]]: # pylint: disable=unsubscriptable-object """Get EighthBlocks from this school year only. Returns: A QuerySet containing all of the blocks selected by this QuerySet that occur during this school year. """ start_date, end_date = get_date_range_this_year() return self.filter(date__gte=start_date, date__lte=end_date)
[docs] def filter_today(self) -> Union[QuerySet, Collection["EighthBlock"]]: # pylint: disable=unsubscriptable-object """Gets EighthBlocks that occur today. Returns: A QuerySet containing all of the blocks selected by this QuerySet that occur today. """ return self.filter(date=timezone.localdate())
[docs]class EighthBlockManager(models.Manager):
[docs] def get_queryset(self): return EighthBlockQuerySet(self.model, using=self._db)
[docs] def get_upcoming_blocks(self, max_number: int = -1) -> Union[QuerySet, Collection["EighthBlock"]]: # pylint: disable=unsubscriptable-object """Gets the given number of upcoming blocks that will take place in the future. If there is no block in the future, the most recent block will be returned. Returns: A QuerySet of the X upcoming ``EighthBlock`` objects. """ now = timezone.localtime() # Show same day if it's before 17:00 if now.hour < 17: now = now.replace(hour=0, minute=0, second=0, microsecond=0) blocks = self.order_by("date", "block_letter").filter(date__gte=now) if max_number == -1: return blocks return blocks[:max_number]
[docs] def get_first_upcoming_block(self) -> "EighthBlock": """Gets the first upcoming block (the first block that will take place in the future). Returns: The first upcoming ``EighthBlock`` object, or ``None`` if there are none upcoming. """ return self.get_upcoming_blocks().first()
[docs] def get_next_upcoming_blocks(self) -> Union[QuerySet, Collection["EighthBlock"]]: # pylint: disable=unsubscriptable-object """Gets the next upccoming blocks. It finds the other blocks that are occurring on the day of the first upcoming block. Returns: A QuerySet of the next upcoming ``EighthBlock`` objects. """ next_block = EighthBlock.objects.get_first_upcoming_block() if not next_block: return EighthBlock.objects.none() next_blocks = EighthBlock.objects.filter(date=next_block.date) return next_blocks
[docs] def get_blocks_this_year(self) -> Union[QuerySet, Collection["EighthBlock"]]: # pylint: disable=unsubscriptable-object """Gets a QuerySet of blocks that occur this school year. Returns: A QuerySet of all the blocks that occur during this school year. """ date_start, date_end = get_date_range_this_year() return EighthBlock.objects.filter(date__gte=date_start, date__lte=date_end)
[docs] def get_blocks_today(self) -> Union[QuerySet, Collection["EighthBlock"]]: # pylint: disable=unsubscriptable-object """Gets a QuerySet of blocks that occur today. Returns: A QuerySet of all the blocks that occur today. """ return self.filter(date=timezone.localdate())
[docs]class EighthBlock(AbstractBaseEighthModel): r"""Represents an eighth period block. Attributes: date The date of the block. signup_time The recommended time at which all users should sign up. This does *not* prevent people from signing up at this time, however students will see the amount of time left to sign up. Defaults to 12:40. block_letter The block letter (e.g. A, B, A1, A2, SOL). Despite its name, it can now be more than just a letter. locked Whether signups are closed. activities List of :class:`EighthScheduledActivity`\s for the block. override_blocks List of :class:`EighthBlock`\s that the block overrides. This allows the half-blocks used during Techlab visits to be easily managed. If a student should only be allowed to sign up for either only block A or both blocks A1 and A2, then block A would override blocks A1 and A2, and blocks A1 and A2 would override block A. comments A short comments field displayed next to the block letter. """ objects = EighthBlockManager() date = models.DateField(null=False) signup_time = models.TimeField(default=datetime.time(12, 40)) block_letter = models.CharField(max_length=10) locked = models.BooleanField(default=False) activities = models.ManyToManyField(EighthActivity, through="EighthScheduledActivity", blank=True) comments = models.CharField(max_length=100, blank=True) override_blocks = models.ManyToManyField("EighthBlock", blank=True) history = HistoricalRecords()
[docs] def save(self, *args, **kwargs): # pylint: disable=signature-differs """Capitalize the first letter of the block name.""" letter = getattr(self, "block_letter", None) if letter and len(letter) >= 1: self.block_letter = letter[:1].upper() + letter[1:] super().save(*args, **kwargs)
[docs] def next_blocks(self, quantity: int = -1) -> Union[QuerySet, Collection["EighthBlock"]]: # pylint: disable=unsubscriptable-object """Gets future blocks this school year in order. Args: quantity: The number of blocks to list after this block, or -1 for all following blocks. Returns: A QuerySet with the specified number of future blocks this school year. If ``quantity`` is passed, then the QuerySet will not be filter()-able. """ blocks = ( EighthBlock.objects.get_blocks_this_year() .order_by("date", "block_letter") .filter(Q(date__gt=self.date) | (Q(date=self.date) & Q(block_letter__gt=self.block_letter))) ) if quantity == -1: return blocks return blocks[:quantity]
[docs] def previous_blocks(self, quantity: int = -1) -> Union[QuerySet, Collection["EighthBlock"]]: # pylint: disable=unsubscriptable-object """Gets the previous blocks this school year in order. Args: quantity: The number of blocks to list before this block, or -1 for all previous blocks. Returns: If `quantity` is not passed, retuns a QuerySet with all the blocks this school year before this block. If `quantity` is passed, returns a list with the specified number of blocks this school year before this block. """ blocks = ( EighthBlock.objects.get_blocks_this_year() .order_by("-date", "-block_letter") .filter(Q(date__lt=self.date) | (Q(date=self.date) & Q(block_letter__lt=self.block_letter))) ) if quantity == -1: return blocks.reverse() return blocks[:quantity:-1]
[docs] def is_today(self) -> bool: """Returns whether the block occurs today. Returns: Whether the block occurs today. """ return timezone.localdate() == self.date
[docs] def signup_time_future(self) -> bool: """Returns whether the block closes in the future. Returns: Whether the block's signup time is in the future. """ now = timezone.localtime() return now.date() < self.date or (self.date == now.date() and self.signup_time > now.time())
[docs] def date_in_past(self) -> bool: """Returns whether the block has already happened. Returns: Whether the block's date is in the past. """ return timezone.localdate() > self.date
[docs] def in_clear_absence_period(self) -> bool: """Returns whether today's date is within the block's clear absence period. This determines whether the option to clear an eighth period absence should be shown. If the block is not within the clear absence period, an absence cannot be cleared. Returns: Whether the current date is in the block's clear absence period. """ now = timezone.localtime() two_weeks = self.date + datetime.timedelta(days=settings.CLEAR_ABSENCE_DAYS) return now.date() <= two_weeks
[docs] def attendance_locked(self) -> bool: """Returns whether the block's attendance is locked. If the block's attendance is locked, non-eighth admins cannot change attendance. Returns: Whether the block's attendance is locked. """ now = timezone.localtime() return now.date() > self.date or (now.date() == self.date and now.time() > datetime.time(settings.ATTENDANCE_LOCK_HOUR, 0))
[docs] def num_signups(self) -> int: """Gets how many people have signed up for an activity this block. Returns: The number of people who have signed up for an activity during this block. """ return EighthSignup.objects.filter(scheduled_activity__block=self, user__in=get_user_model().objects.get_students()).count()
[docs] def num_no_signups(self) -> int: """Gets how many people have not signed up for an activity this block. Returns: The number of people who have not signed up for an activity during this block. """ signup_users_count = get_user_model().objects.get_students().count() return signup_users_count - self.num_signups()
[docs] def get_unsigned_students(self) -> Union[QuerySet, Collection["get_user_model()"]]: # pylint: disable=unsubscriptable-object """Return a QuerySet of people who haven't signed up for an activity during this block. Returns: The users who have not signed up for an activity during this block. """ return get_user_model().objects.get_students().exclude(eighthsignup__scheduled_activity__block=self)
[docs] def get_hidden_signups(self) -> Union[QuerySet, Collection["EighthSignup"]]: # pylint: disable=unsubscriptable-object """Returns a QuerySet of EighthSignups whose users are not students but have signed up for an activity. This is usually a list of signups for the z-Withdrawn from TJ activity. Returns: A QuerySet of users who are not current students but have signed up for an activity this block. """ return EighthSignup.objects.filter(scheduled_activity__block=self).exclude(user__in=get_user_model().objects.get_students())
@property def letter_width(self) -> int: """Returns the width in pixels that should be allocated for the block_letter on the signup page. Return: The width in pixels of the block letter. """ return (len(self.block_letter) - 1) * 6 + 15 @property def short_text(self) -> str: """Returns the date and block letter for this block. It is returned in the format of MM/DD B, like "9/1 B" Returns: The date and block letter. """ return "{} {}".format(self.date.strftime("%m/%d"), self.block_letter) ####### @property def hybrid_text(self) -> str: """Returns the user friendly name of a hybrid block. See Hybrid-README.rst. * - P1 and * - P2 are returned as * (In-Person) * - Virt is returned as * (Virtual) Any other names are returned as themselves. Returns: The user friendly name of a hybrid block. """ name = self.block_letter if "Virt" in name: name = name[0] + " (Virtual)" elif "P1" in name or "P2" in name: name = name[0] + " (In-Person)" return name ####### @property def is_this_year(self) -> bool: """Return whether the block occurs during this school year. Returns: Whether the block occurs during this school year. """ return is_current_year(datetime.datetime.combine(self.date, datetime.time())) @property def formatted_date(self) -> str: """Returns the block date, formatted according to ``settings.EIGHTH_BLOCK_DATE_FORMAT``. Returns: The formatted block date. """ return formats.date_format(self.date, settings.EIGHTH_BLOCK_DATE_FORMAT) def __str__(self): ####### if settings.ENABLE_HYBRID_EIGHTH: return "{} ({})".format(self.formatted_date, self.hybrid_text) else: ####### return "{} ({})".format(self.formatted_date, self.block_letter) class Meta: unique_together = (("date", "block_letter"),) ordering = ("date", "block_letter")
[docs]class EighthScheduledActivityManager(Manager): """Model Manager for EighthScheduledActivity."""
[docs] def for_sponsor( self, sponsor: EighthSponsor, include_cancelled: bool = False ) -> Union[QuerySet, Collection["EighthScheduledActivity"]]: # pylint: disable=unsubscriptable-object """Returns a QuerySet of EighthScheduledActivities where the given EighthSponsor is sponsoring. If a sponsorship is defined in an EighthActivity, it may be overridden on a block by block basis in an EighthScheduledActivity. Sponsors from the EighthActivity do not carry over. EighthScheduledActivities that are deleted or cancelled are also not counted. Args: sponsor: The sponsor to search for. include_cancelled: Whether to include cancelled activities. Deleted activities are always excluded. Returns: A QuerySet of EighthScheduledActivities where the given EighthSponsor is sponsoring. """ sponsoring_filter = Q(sponsors=sponsor) | (Q(sponsors=None) & Q(activity__sponsors=sponsor)) sched_acts = EighthScheduledActivity.objects.exclude(activity__deleted=True).filter(sponsoring_filter).distinct() if not include_cancelled: sched_acts = sched_acts.exclude(cancelled=True) return sched_acts
[docs]class EighthScheduledActivity(AbstractBaseEighthModel): r"""Represents the relationship between an activity and a block in which it has been scheduled. Attributes: block : :class:`EighthBlock` The :class:`EighthBlock` during which an :class:`EighthActivity` has been scheduled activity The scheduled :class:`EighthActivity` members The :class:`User<intranet.apps.users.models.User>`\s who have signed up for an :class:`EighthBlock` both_blocks If True, a signup for an EighthScheduledActivity during an A or B block will enforce and automatically trigger a signup on the other block. Does not enforce signups for blocks other than A and B. comments Notes for students admin_comments Notes for the Eighth Office sponsors :class:`EighthSponsor`\s that will override the :class:`EighthActivity`'s default sponsors rooms :class:`EighthRoom`\s that will override the :class:`EighthActivity`'s default rooms attendance_taken Whether the :class:`EighthSponsor` for the scheduled :class:`EighthActivity` has taken attendance yet special Whether this scheduled instance of the activity is special. If not set, falls back on the EighthActivity's special setting. cancelled whether the :class:`EighthScheduledActivity` has been cancelled """ # Use model manager objects = EighthScheduledActivityManager() block = models.ForeignKey(EighthBlock, on_delete=models.CASCADE) activity = models.ForeignKey(EighthActivity, on_delete=models.CASCADE) members = models.ManyToManyField(settings.AUTH_USER_MODEL, through="EighthSignup", related_name="eighthscheduledactivity_set") waitlist = models.ManyToManyField(settings.AUTH_USER_MODEL, through="EighthWaitlist", related_name="%(class)s_scheduledactivity_set") admin_comments = models.CharField(max_length=1000, blank=True) title = models.CharField(max_length=1000, blank=True) comments = models.CharField(max_length=1000, blank=True) # Overridden attributes sponsors = models.ManyToManyField(EighthSponsor, blank=True) rooms = models.ManyToManyField(EighthRoom, blank=True) capacity = models.SmallIntegerField(null=True, blank=True) both_blocks = models.BooleanField(default=False) special = models.BooleanField(default=False) administrative = models.BooleanField(default=False) restricted = models.BooleanField(default=False) sticky = models.BooleanField(default=False) attendance_taken = models.BooleanField(default=False) cancelled = models.BooleanField(default=False) archived_member_count = models.SmallIntegerField(null=True, blank=True) history = HistoricalRecords()
[docs] def get_all_associated_rooms(self) -> Union[QuerySet, Collection["EighthRoom"]]: # pylint: disable=unsubscriptable-object """Returns a QuerySet of all the rooms associated with either this EighthScheduledActivity or its EighthActivity. Returns: A QuerySet of all the rooms associated with either this EighthScheduledActivity or its EighthActivity. """ return (self.rooms.all() | self.activity.rooms.all()).distinct()
@property def full_title(self) -> str: """Gets the full title for the activity, appending the title of the scheduled activity to the activity's name. Returns: The full title for the scheduled activity, without flags. """ cancelled_str = " (Cancelled)" if self.cancelled else "" act_name = self.activity.name + cancelled_str if self.special and not self.activity.special: act_name = "Special: " + act_name return act_name if not self.title else "{} - {}".format(act_name, self.title) @property def title_with_flags(self) -> str: """Gets the title for the activity, appending the title of the scheduled activity to the activity's name and flags. Returns: The full title for the scheduled activity, with flags. """ cancelled_str = " (Cancelled)" if self.cancelled else "" name_with_flags = self.activity._name_with_flags(True, self.title) + cancelled_str # pylint: disable=protected-access if self.special and not self.activity.special: name_with_flags = "Special: " + name_with_flags return name_with_flags
[docs] def get_true_sponsors(self) -> Union[QuerySet, Collection[EighthSponsor]]: # pylint: disable=unsubscriptable-object """Retrieves the sponsors for the scheduled activity, taking into account activity defaults and overrides. Returns: The sponsors for this scheduled activity. """ return self.sponsors.all() or self.activity.sponsors.all()
[docs] def user_is_sponsor(self, user: "get_user_model()") -> bool: """Returns whether the given user is a sponsor of the activity. Args: user: The user to check for sponsorship of this activity. Returns: Whether the given user is a sponsor of the activity. """ return self.get_true_sponsors().filter(user=user).exists()
[docs] def get_true_rooms(self) -> Union[QuerySet, Collection[EighthRoom]]: # pylint: disable=unsubscriptable-object """Retrieves the rooms for the scheduled activity, taking into account activity defaults and overrides. Returns: The true room list of the scheduled activity. """ return self.rooms.all() or self.activity.rooms.all()
[docs] def get_true_capacity(self) -> int: """Retrieves the capacity for the scheduled activity, taking into account activity defaults and overrides. Returns: The true capacity of the scheduled activity. """ if self.capacity is not None: return self.capacity rooms = self.rooms.all() if rooms: return EighthRoom.total_capacity_of_rooms(rooms) if self.activity.default_capacity: # use activity-level override return self.activity.default_capacity return EighthRoom.total_capacity_of_rooms(self.activity.rooms.all())
[docs] def is_both_blocks(self) -> bool: """Gets whether this scheduled activity runs both blocks. Returns: Whether this scheduled activity runs both blocks. """ return self.both_blocks or self.activity.both_blocks
[docs] def get_restricted(self) -> bool: """Gets whether this scheduled activity is restricted. Returns: Whether this scheduled activity is restricted. """ return self.restricted or self.activity.restricted
[docs] def get_sticky(self) -> bool: """Gets whether this scheduled activity is sticky. Returns: Whether this scheduled activity is sticky. """ return self.sticky or self.activity.sticky
[docs] def get_finance(self) -> str: """Retrieves the name of this activity's account with the finance office, if any. Returns: The name of this activity's account with the finance office. """ return self.activity.finance
[docs] def get_administrative(self) -> bool: """Returns whether this scheduled activity is administrative. Returns: Whether this activity is administrative """ return self.administrative or self.activity.administrative
[docs] def get_special(self) -> bool: """Returns whether this scheduled activity is special. Returns: Whether this scheduled activity is special. """ return self.special or self.activity.special
[docs] def is_full(self, nocache: bool = False) -> bool: """Returns whether the scheduled activity is full. Args: nocache: Whether to disable caching for the query that checks the scheduled activity's current capacity. Returns: Whether this scheduled activity is full. """ capacity = self.get_true_capacity() # We don't disable caching because this changes much less frequently signups = self.eighthsignup_set.all() if nocache: signups = signups.nocache() return capacity != -1 and signups.count() >= capacity
[docs] def is_almost_full(self) -> bool: """Returns whether the scheduled activity is almost full (>=90%). Returns: Whether this scheduled activity is at least 90% full. """ capacity = self.get_true_capacity() return capacity != -1 and self.eighthsignup_set.count() >= (0.9 * capacity)
[docs] def is_overbooked(self) -> bool: """Returns whether the activity is overbooked (>100%) capacity. Returns: Whether this scheduled activity is overbooked. """ capacity = self.get_true_capacity() return capacity != -1 and self.eighthsignup_set.count() > capacity
[docs] def is_too_early_to_signup(self, now: Optional[datetime.datetime] = None) -> (bool, datetime): """Returns whether it is too early to sign up for the activity if it is a presign. This contains the 48 hour presign logic. Args: now: A datetime object to use for the check instead of the current time. Returns: Whether it is too early to sign up for this scheduled activity and when the activity opens for signups. """ if now is None: now = timezone.localtime() activity_date = datetime.datetime.combine(self.block.date, datetime.time(0, 0, 0)) if now.tzinfo is not None: activity_date = timezone.make_aware(activity_date, now.tzinfo) # Presign activities can only be signed up for a set time in advance. presign_period = datetime.timedelta(minutes=settings.EIGHTH_PRESIGNUP_HOURS * 60 + settings.EIGHTH_PRESIGNUP_MINUTES) return (now < (activity_date - presign_period), activity_date - presign_period)
[docs] def has_open_passes(self) -> bool: """Returns whether there are passes that have not been acknowledged. Returns: Whether this activity has open passes. """ return self.eighthsignup_set.filter(after_deadline=True, pass_accepted=False).exists()
def _get_viewable_members( self, user: "get_user_model()" ) -> Union[QuerySet, Collection["get_user_model()"]]: # pylint: disable=unsubscriptable-object """Get an unsorted QuerySet of the members that you have permission to view. Args: user: The user who is attempting to view the member list. Returns: Unsorted QuerySet of the members that you have permssion to view. """ if user and (user.is_eighth_admin or user.is_eighthoffice or user.is_teacher): return self.members.all() else: q = Q(properties__parent_show_eighth=True, properties__self_show_eighth=True) if user: q |= Q(id=user.id) return self.members.filter(q)
[docs] def get_viewable_members( self, user: "get_user_model()" = None ) -> Union[QuerySet, Collection["get_user_model()"]]: # pylint: disable=unsubscriptable-object """Returns a QuerySet of the members that you have permission to view, sorted alphabetically. Args: user: The user who is attempting to view the member list. Returns: QuerySet of the members that you have permission to view, sorted alphabetically. """ return self._get_viewable_members(user).order_by("last_name", "first_name")
[docs] def get_viewable_members_serializer(self, request) -> Union[QuerySet, Collection["get_user_model()"]]: # pylint: disable=unsubscriptable-object """Given a request, returns an unsorted QuerySet of the members that the requesting user has permission to view. Args: request: The request object associated with the member list query. Returns: Unsorted QuerySet of the members that you have permssion to view. """ return self._get_viewable_members(request.user)
[docs] def get_hidden_members( self, user: "get_user_model()" = None ) -> Union[QuerySet, Collection["get_user_model()"]]: # pylint: disable=unsubscriptable-object """Returns a QuerySet of the members that you do not have permission to view. Args: user: The user who is attempting to view the member list. Returns: Unsorted QuerySet of the members that you do not have permission to view. """ if user and (user.is_eighth_admin or user.is_eighthoffice or user.is_teacher): return get_user_model().objects.none() else: hidden_members = self.members.exclude(properties__parent_show_eighth=True, properties__self_show_eighth=True) if user: hidden_members = hidden_members.exclude(id=user.id) return hidden_members
[docs] def get_both_blocks_sibling(self) -> Optional["EighthScheduledActivity"]: """If this is a both-blocks activity, get the other EighthScheduledActivity object that occurs on the other block. both_blocks means A and B block, NOT all of the blocks on that day. Returns: EighthScheduledActivity object if found. ``None``, if the activity cannot have a sibling. """ if not self.is_both_blocks(): return None if self.block.block_letter not in ["A", "B"]: # both_blocks is not currently implemented for blocks other than A and B return None other_block_letter = "A" if self.block.block_letter == "B" else "B" try: return EighthScheduledActivity.objects.exclude(pk=self.pk).get( activity=self.activity, block__date=self.block.date, block__block_letter=other_block_letter ) except EighthScheduledActivity.DoesNotExist: return None
[docs] def notify_waitlist(self, waitlists: Iterable["EighthWaitlist"]): """Notifies all users on the given EighthWaitlist objects that the activity they are on the waitlist for has an open spot. Args: waitlists: The EighthWaitlist objects whose users should be notified that the activity has an open slot. """ for waitlist in waitlists: email_send_task.delay( "eighth/emails/waitlist.txt", "eighth/emails/waitlist.html", {"activity": waitlist.scheduled_activity}, "Open Spot Notification", [waitlist.user.primary_email_address], )
[docs] @transaction.atomic # This MUST be run in a transaction. Do NOT remove this decorator. def add_user( self, user: "get_user_model()", request: Optional[HttpRequest] = None, force: bool = False, no_after_deadline: bool = False, add_to_waitlist: bool = False, ): """Signs up a user to this scheduled activity if possible. This is where the magic happens. Raises an exception if there's a problem signing the user up unless the signup is forced and the requesting user has permission. Args: user: The user to add to the scheduled activity. request: The request object associated with the signup action. Should always be passed if applicable, as some information is extracted from the request. force: Whether to force the signup. no_after_deadline: Whether to mark the user as not having signed up after the deadline, regardless of whether they did or not. add_to_waitlist: Explicitly add the user to the waitlist. """ if request is not None: force = (force or ("force" in request.GET)) and request.user.is_eighth_admin add_to_waitlist = (add_to_waitlist or ("add_to_waitlist" in request.GET)) and request.user.is_eighth_admin exception = eighth_exceptions.SignupException() if user.grade and user.grade.number > 12: exception.SignupForbidden = True all_sched_act = [self] all_blocks = [self.block] # Finds the other scheduling of the same activity on the same day # See note above in get_both_blocks_sibling() sibling = self.get_both_blocks_sibling() if sibling: all_sched_act.append(sibling) all_blocks.append(sibling.block) # Lock on the User and the EighthScheduledActivity to prevent duplicates. logger.debug("Locking on user %d and scheduled activity %d", user.id, self.id) lock_on([user, self]) logger.debug("Successfully locked on user %d and scheduled activity %d", user.id, self.id) waitlist = None if force: EighthWaitlist.objects.filter(scheduled_activity_id=self.id, user_id=user.id, block_id=self.block.id).delete() else: # Check if the user who sent the request has the permissions # to change the target user's signups if request is not None: if user != request.user and not request.user.is_eighth_admin: exception.SignupForbidden = True # Check if the activity has been deleted if self.activity.deleted: exception.ActivityDeleted = True # Check if the user is already stickied into an activity if ( EighthSignup.objects.filter(user=user, scheduled_activity__block__in=all_blocks) .filter(Q(scheduled_activity__activity__sticky=True) | Q(scheduled_activity__sticky=True)) .filter(Q(scheduled_activity__cancelled=False)) .exists() ): exception.Sticky = True for sched_act in all_sched_act: # Check if the block has been locked if sched_act.block.locked: exception.BlockLocked = True # Check if the scheduled activity has been cancelled if sched_act.cancelled: exception.ScheduledActivityCancelled = True # Check if the activity is full # pylint: disable=too-many-boolean-expressions if settings.ENABLE_WAITLIST and ( add_to_waitlist or ( sched_act.is_full() and not self.is_both_blocks() and (request is not None and not request.user.is_eighth_admin and request.user.is_student) ) ): # pylint: enable=too-many-boolean-expressions if user.primary_email_address: if EighthWaitlist.objects.filter(user_id=user.id, block_id=self.block.id).exists(): EighthWaitlist.objects.filter(user_id=user.id, block_id=self.block.id).delete() waitlist = EighthWaitlist.objects.create(user=user, block=self.block, scheduled_activity=sched_act) else: exception.PrimaryEmailNotSet = True elif sched_act.is_full(nocache=True): exception.ActivityFull = True # Check if it's too early to sign up for the activity if self.activity.presign: early = self.is_too_early_to_signup() if early[0]: exception.Presign = True exception.desc_errors["Presign"] = [early[1].strftime("%A, %B %-d at %-I:%M %p")] # Check if signup would violate one-a-day constraint if not self.is_both_blocks() and self.activity.one_a_day: in_act = ( EighthSignup.objects.exclude(scheduled_activity__block=self.block) .filter(user=user, scheduled_activity__block__date=self.block.date, scheduled_activity__activity=self.activity) .exists() ) if in_act: exception.OneADay = True # Check if user is allowed in the activity if it's restricted if self.get_restricted(): acts = EighthActivity.restricted_activities_available_to_user(user) if self.activity.id not in acts: exception.Restricted = True # Check if user is blacklisted from activity if self.activity.users_blacklisted.filter(username=user).exists(): exception.Blacklisted = True if self.get_sticky(): EighthWaitlist.objects.filter(user_id=user.id, block_id=self.block.id).delete() success_message = "Successfully added to waitlist for activity." if waitlist else "Successfully signed up for activity." # If we've collected any errors, raise the exception and abort # the signup attempt if exception.errors: print(exception) logger.debug(exception) logger.debug("test") raise exception # Everything's good to go - complete the signup. If we've gotten to # this point, the signup is either before deadline or performed by # an eighth period admin, so previous absences and passes are cleared. after_deadline = self.block.locked # If we're doing an eighth admin action (like signing up a group), # don't make an after deadline signup, which creates a pass. if no_after_deadline: after_deadline = False if not waitlist: if not self.is_both_blocks(): try: existing_signup = EighthSignup.objects.nocache().get(user=user, scheduled_activity__block=self.block) except EighthSignup.DoesNotExist: add_breadcrumb( category="eighth-signup", message="Signing user {} up for single-block activity {} in block {}".format(user.id, self.activity.id, self.block.id), level="debug", ) logger.debug("Signing user %d up for single-block activity %d in block %d", user.id, self.activity.id, self.block.id) signup = EighthSignup.objects.create_signup( user=user, scheduled_activity=self, after_deadline=after_deadline, own_signup=(request is not None and user == request.user) ) logger.debug( "Result of new signup existence query: %s", EighthSignup.objects.filter(user=user, scheduled_activity__block=self.block, pk=signup.pk).exists(), ) logger.debug( "Result of conflict existence query: %s", EighthSignup.objects.filter(user=user, scheduled_activity__block=self.block).exclude(pk=signup.pk).exists(), ) logger.debug("Successfully signed user %d up for single-block activity %d in block %d", user.id, self.activity.id, self.block.id) if signup.has_conflict(nocache=True): try: signup.save() except ValidationError as e: logger.error( "Newly created signup %d (user %d, single-block activity %d, block %d, scheduled activity %d) failed the " "post-creation duplicate check in add_user()", signup.id, user.id, self.activity.id, self.block.id, self.id, ) capture_exception(e) else: previous_activity_name = existing_signup.scheduled_activity.activity.name_with_flags prev_sponsors = existing_signup.scheduled_activity.get_true_sponsors() previous_activity_sponsors = ", ".join(map(str, prev_sponsors)) previous_activity = existing_signup.scheduled_activity if not existing_signup.scheduled_activity.is_both_blocks(): add_breadcrumb( category="eighth-signup", message="Switching user {} from single-block activity {} to single-block activity {} in block {}".format( user.id, existing_signup.scheduled_activity.activity.id, self.activity.id, self.block.id ), level="debug", ) logger.debug( "Switching user %d from single-block activity %d to single-block activity %d in block %d", user.id, existing_signup.scheduled_activity.activity.id, self.activity.id, self.block.id, ) existing_signup.scheduled_activity = self existing_signup.after_deadline = after_deadline existing_signup.was_absent = False existing_signup.absence_acknowledged = False existing_signup.pass_accepted = False existing_signup.previous_activity_name = previous_activity_name existing_signup.previous_activity_sponsors = previous_activity_sponsors existing_signup.own_signup = request is not None and user == request.user existing_signup.save() invalidate_obj(existing_signup) logger.debug( "Successfully switched user %d from single-block activity %d to single-block activity %d in block %d", user.id, existing_signup.scheduled_activity.activity.id, self.activity.id, self.block.id, ) if existing_signup.has_conflict(): try: existing_signup.save() except ValidationError as e: logger.error( "Reused signup %d (user %d, single-block activity %d, block %d, scheduled activity %d) failed the post-creation " "duplicate check in add_user()", existing_signup.id, user.id, self.activity.id, self.block.id, self.id, ) capture_exception(e) else: # Clear out the other signups for this block if the user is # switching out of a both-blocks activity sibling = existing_signup.scheduled_activity.get_both_blocks_sibling() existing_blocks = [existing_signup.scheduled_activity.block] if sibling: existing_blocks.append(sibling.block) add_breadcrumb( category="eighth-signup", message="Switching user {} from dual-block activity {} to single-block activity {} in block {}".format( user.id, existing_signup.scheduled_activity.activity.id, self.activity.id, self.block.id ), level="debug", ) logger.debug( "Switching user %d from dual-block activity %d to single-block activity %d in block %d", user.id, existing_signup.scheduled_activity.activity.id, self.activity.id, self.block.id, ) EighthSignup.objects.filter(user=user, scheduled_activity__block__in=existing_blocks).delete() EighthWaitlist.objects.filter(user=user, scheduled_activity=self).delete() signup = EighthSignup.objects.create_signup( user=user, scheduled_activity=self, after_deadline=after_deadline, previous_activity_name=previous_activity_name, previous_activity_sponsors=previous_activity_sponsors, own_signup=(request is not None and user == request.user), ) logger.debug( "Successfully switched user %d from dual-block activity %d to single-block activity %d in block %d", user.id, existing_signup.scheduled_activity.activity.id, self.activity.id, self.block.id, ) if signup.has_conflict(): try: signup.save() except ValidationError as e: logger.error( "Newly created signup %d created to switch out of dual-block activity (user %d, single-block activity %d, block " "%d, scheduled activity %d) failed the post-creation duplicate check in add_user()", existing_signup.id, user.id, self.activity.id, self.block.id, self.id, ) capture_exception(e) if settings.ENABLE_WAITLIST and ( previous_activity.waitlist.all().exists() and not self.block.locked and request is not None and not request.session.get("disable_waitlist_transactions", False) ): if not previous_activity.is_full(): waitlists = EighthWaitlist.objects.get_next_waitlist(previous_activity) self.notify_waitlist(waitlists) else: existing_signups = EighthSignup.objects.filter(user=user, scheduled_activity__block__in=all_blocks) first_signup = existing_signups.first() prev_sched_act = first_signup.scheduled_activity if first_signup is not None else None if prev_sched_act is None: logger.debug( "Signing user %d up for double-block activity %d during block %d and its sibling", user.id, self.activity.id, self.block.id ) else: logger.debug( "Switching user %d from activity %d to double-block activity %d during block %d (and also its sibling)", user.id, prev_sched_act.activity.id, self.activity.id, self.block.id, ) prev_data = {} for signup in existing_signups: add_breadcrumb( category="eighth-signup", message="User {}: original activity for block {}: {}".format( user.id, signup.scheduled_activity.block.id, signup.scheduled_activity.activity.id ), level="debug", ) prev_sponsors = signup.scheduled_activity.get_true_sponsors() prev_data[signup.scheduled_activity.block.block_letter] = { "name": signup.scheduled_activity.activity.name_with_flags, "sponsors": ", ".join(map(str, prev_sponsors)), } existing_signups.delete() for sched_act in all_sched_act: letter = sched_act.block.block_letter if letter in prev_data: previous_activity_name = prev_data[letter]["name"] previous_activity_sponsors = prev_data[letter]["sponsors"] else: previous_activity_name = None previous_activity_sponsors = None add_breadcrumb( category="eighth-signup", message="Switching user {} to double-block activity {} in block {}".format(user.id, sched_act.id, self.block.id), level="debug", ) signup = EighthSignup.objects.create_signup( user=user, scheduled_activity=sched_act, after_deadline=after_deadline, previous_activity_name=previous_activity_name, previous_activity_sponsors=previous_activity_sponsors, own_signup=(request is not None and user == request.user), ) if signup.has_conflict(): try: signup.save() except ValidationError as e: logger.error( "Newly created signup %d created to sign up for dual-block activity (user %d, activity %d, block %d, scheduled " "activity %d) failed the post-creation duplicate check in add_user()", signup.id, user.id, sched_act.activity.id, sched_act.block.id, sched_act.id, ) capture_exception(e) # signup.previous_activity_name = signup.activity.name_with_flags # signup.previous_activity_sponsors = ", ".join(map(str, signup.get_true_sponsors())) if prev_sched_act is None: logger.debug( "Successfully signed user %d up for double-block activity %d during block %d and its sibling", user.id, self.activity.id, self.block.id, ) else: logger.debug( "Successfully switched user %d from activity %d to double-block activity %d during block %d (and also its sibling)", user.id, prev_sched_act.activity.id, self.activity.id, self.block.id, ) return success_message
[docs] def cancel(self): """Cancel an EighthScheduledActivity and send a notification email to signed-up students. This method should be always be called instead of setting the 'cancelled' flag manually. (Note: To avoid spamming students signed up for both-block activities, an email is not sent for the B-block activity in both-block activities.) """ if not self.cancelled: self.cancelled = True self.save(update_fields=["cancelled"]) if not self.is_both_blocks or self.block.block_letter != "B": from .notifications import activity_cancelled_email # pylint: disable=import-outside-toplevel,cyclic-import activity_cancelled_email(self)
[docs] def uncancel(self): """Uncancel an EighthScheduledActivity. This does nothing besides unset the cancelled flag and save the object. """ if self.cancelled: self.cancelled = False self.save(update_fields=["cancelled"])
[docs] def save(self, *args, **kwargs): # pylint: disable=signature-differs super().save(*args, **kwargs)
class Meta: unique_together = (("block", "activity"),) verbose_name_plural = "eighth scheduled activities" def __str__(self): cancelled_str = " (Cancelled)" if self.cancelled else "" suff = " - {}".format(self.title) if self.title else "" return "{}{} on {}{}".format(self.activity, suff, self.block, cancelled_str)
[docs]class EighthSignupManager(Manager): """Model manager for EighthSignup."""
[docs] def create_signup(self, user: "get_user_model()", scheduled_activity: "EighthScheduledActivity", **kwargs) -> "EighthSignup": """Creates an EighthSignup for the given user in the given activity after checking for duplicate signups. This raises an error if there are duplicate signups. Args: user: The user to create the EighthSignup for. scheduled_activity: The EighthScheduledActivity to sign the user up for. """ if EighthSignup.objects.filter(user=user, scheduled_activity__block=scheduled_activity.block).nocache().exists(): logger.error( "Duplicate signup before creating signup for user %d in activity %d, block %d, scheduled activity %d", user.id, scheduled_activity.activity.id, scheduled_activity.block.id, scheduled_activity.id, ) raise ValidationError("EighthSignup already exists for this user on this block.") signup = self.create(user=user, scheduled_activity=scheduled_activity, **kwargs) if EighthSignup.objects.exclude(pk=signup.pk).filter(user=user, scheduled_activity__block=scheduled_activity.block).nocache().exists(): logger.error( "Duplicate signup after creating signup %d to sign up user %d in activity %d, block %d, scheduled activity %d; deleting", signup.id, user.id, scheduled_activity.activity.id, scheduled_activity.block.id, scheduled_activity.id, ) signup.delete() raise ValidationError("EighthSignup already exists for this user on this block.") return signup
[docs] def get_absences(self) -> Union[QuerySet, Collection["EighthSignup"]]: # pylint: disable=unsubscriptable-object """Returns all EighthSignups for which the student was marked as absent. Returns: A QuerySet of all the EighthSignups for which the student was marked as absent. """ return EighthSignup.objects.filter(was_absent=True, scheduled_activity__attendance_taken=True)
[docs]class EighthSignup(AbstractBaseEighthModel): """Represents a signup/membership in an eighth period activity. Attributes: user The :class:`User<intranet.apps.users.models.User>` who has signed up. scheduled_activity The :class:`EighthScheduledActivity` for which the user has signed up. after_deadline Whether the signup was after deadline. previous_activity_name The name of the activity the student was previously signed up for (used for passes) previous_activity_sponsors The sponsors of the activity the student was previously signed up for. pass_accepted Whether the pass was accepted was_absent Whether the student was absent. absence_acknowledged Whether the student has dismissed the absence notification. absence_emailed Whether the student has been emailed about the absence. """ objects = EighthSignupManager() time = models.DateTimeField(auto_now=True) user = models.ForeignKey(settings.AUTH_USER_MODEL, null=False, on_delete=set_historical_user) scheduled_activity = models.ForeignKey( EighthScheduledActivity, related_name="eighthsignup_set", null=False, db_index=True, on_delete=models.CASCADE ) # An after-deadline signup is assumed to be a pass after_deadline = models.BooleanField(default=False) previous_activity_name = models.CharField(max_length=200, blank=True, null=True, default=None) previous_activity_sponsors = models.CharField(max_length=10000, blank=True, null=True, default=None) pass_accepted = models.BooleanField(default=False, blank=True) was_absent = models.BooleanField(default=False, blank=True) absence_acknowledged = models.BooleanField(default=False, blank=True) absence_emailed = models.BooleanField(default=False, blank=True) archived_was_absent = models.BooleanField(default=False, blank=True)
[docs] def save(self, *args, **kwargs): # pylint: disable=signature-differs if self.has_conflict(nocache=True): logger.error( "Duplicate signup while saving signup %d for user %d in activity %d, block %d, scheduled activity %d", self.id, self.user.id, self.scheduled_activity.activity.id, self.scheduled_activity.block.id, self.scheduled_activity.id, ) raise ValidationError("EighthSignup already exists for this user on this block.") super().save(*args, **kwargs)
own_signup = models.BooleanField(default=False) history = HistoricalRecords()
[docs] def validate_unique(self, *args, **kwargs): # pylint: disable=signature-differs """Checked whether more than one EighthSignup exists for a User on a given EighthBlock.""" super().validate_unique(*args, **kwargs) if self.has_conflict(nocache=True): raise ValidationError({NON_FIELD_ERRORS: ("EighthSignup already exists for the User and the EighthScheduledActivity's block",)})
[docs] def has_conflict(self, nocache: bool = False) -> bool: """Returns True if another EighthSignup object exists for the same user in the same block. Args: nocache: Whether to explicitly disable caching for this check. Returns: Whether there is another EighthSignup for the same user in the same block. """ q = EighthSignup.objects.exclude(pk=self.pk).filter(user=self.user, scheduled_activity__block=self.scheduled_activity.block) if nocache: q = q.nocache() return q.exists()
[docs] def remove_signup(self, user: "get_user_model()" = None, force: bool = False, dont_run_waitlist: bool = False) -> str: """Attempts to remove the EighthSignup if the user has permission to do so. Args: user: The user who is attempting to remove the EighthSignup. force: Whether to force removal. dont_run_waitlist: Whether to skip notifying users on the activity's waitlist. Returns: A message to be displayed to the user indicating successful removal. """ exception = eighth_exceptions.SignupException() if user is not None: if user != self.user and not user.is_eighth_admin: exception.SignupForbidden = True # Check if the block has been locked if self.scheduled_activity.block.locked: exception.BlockLocked = True # Check if the scheduled activity has been cancelled if self.scheduled_activity.cancelled: exception.ScheduledActivityCancelled = True # Check if the activity has been deleted if self.scheduled_activity.activity.deleted: exception.ActivityDeleted = True # Check if the user is already stickied into an activity if self.scheduled_activity.activity and self.scheduled_activity.activity.sticky and not self.scheduled_activity.cancelled: exception.Sticky = True if exception.messages() and not force: raise exception else: block = self.scheduled_activity.block self.delete() if settings.ENABLE_WAITLIST and self.scheduled_activity.waitlist.all().exists() and not block.locked and not dont_run_waitlist: if not self.scheduled_activity.is_full(): waitlists = EighthWaitlist.objects.get_next_waitlist(self.scheduled_activity) self.scheduled_activity.notify_waitlist(waitlists) return "Successfully removed signup for {}.".format(block)
[docs] def accept_pass(self): """Accepts an eighth period pass for the EighthSignup object.""" self.was_absent = False self.pass_accepted = True self.save(update_fields=["was_absent", "pass_accepted"])
[docs] def reject_pass(self): """Rejects an eighth period pass for the EighthSignup object.""" self.was_absent = True self.pass_accepted = True self.save(update_fields=["was_absent", "pass_accepted"])
[docs] def in_clear_absence_period(self) -> bool: """Returns whether the block for this signup is in the clear absence period. Returns: Whether the block for this signup is in the clear absence period. """ return self.scheduled_activity.block.in_clear_absence_period()
[docs] def archive_remove_absence(self): """If user was absent, archives the absence.""" if self.was_absent: self.was_absent = False self.archived_was_absent = True self.save(update_fields=["was_absent", "archived_was_absent"])
def __str__(self): return "{}: {}".format(self.user, self.scheduled_activity) class Meta: unique_together = (("user", "scheduled_activity"),)
[docs]class EighthWaitlistManager(Manager): """Model manager for EighthWaitlist."""
[docs] def get_next_waitlist( self, activity: EighthScheduledActivity ) -> Union[QuerySet, Collection["EighthWaitlist"]]: # pylint: disable=unsubscriptable-object """Returns a QuerySet of all the EighthWaitlist objects for the given activity, ordered by signup time. Args: activity: The activity to list the EighthWaitlist objects for. Returns: A QuerySet of all the EighthWaitlist objects for the given activity, ordered by signup time. """ return self.filter(scheduled_activity_id=activity.id).order_by("time")
[docs] def check_for_prescence(self, activity: EighthScheduledActivity, user: "get_user_model()") -> bool: """Returns whether the given user is in a waitlist for the given activity. Args: activity: The activity for which the waitlist should be queried. user: The user who should be searched for in the activity's waitlist. Returns: Whether the given user is in a waitlist for the given activity. """ return self.filter(scheduled_activity_id=activity.id, user=user).exists()
[docs] def position_in_waitlist(self, aid: int, uid: int) -> int: """Given an activity ID and a user ID, returns the user's position in the waitlist (starting at 1). If the user is not in the waitlist, returns 0. Args: aid: The ID of the EighthScheduledActivity for which the waitlist should be queried. uid: The ID of the user whose position in the waitlist should be found. Returns: The user's position in the waitlist, starting at 1, or 0 if the user is not in the waitlist. """ try: return self.filter(scheduled_activity_id=aid, time__lt=self.get(scheduled_activity_id=aid, user_id=uid).time).count() + 1 except EighthWaitlist.DoesNotExist: return 0
[docs]class EighthWaitlist(AbstractBaseEighthModel): objects = EighthWaitlistManager() time = models.DateTimeField(auto_now=True) user = models.ForeignKey(settings.AUTH_USER_MODEL, null=False, on_delete=set_historical_user) block = models.ForeignKey(EighthBlock, null=False, on_delete=models.CASCADE) scheduled_activity = models.ForeignKey( EighthScheduledActivity, related_name="eighthwaitlist_set", null=False, db_index=True, on_delete=models.CASCADE ) def __str__(self): return "{}: {}".format(self.user, self.scheduled_activity)
[docs]class EighthActivitySimilarity(AbstractBaseEighthModel): count = models.IntegerField() weighted = models.FloatField() @property def _weighted(self) -> float: cap = self.activity_set.first().capacity() + self.activity_set.last().capacity() if cap == 0: cap = 100 return self.count / cap
[docs] def update_weighted(self): """Recalculates the 'weighted' field based on changes to similar activities.""" self.weighted = self._weighted self.save(update_fields=["weighted"])
def __str__(self): act_set = self.activity_set.all() return "{} and {}: {}".format(act_set.first(), act_set.last(), self.count)