# models.py import uuid from typing import Self from datetime import timedelta from django.db import models from django.conf import settings from django.utils import timezone from django.core.exceptions import ValidationError from django.utils.translation import gettext_lazy as _ import markdown from events.utils.pageutils import is_next_page, is_previous_page User = settings.AUTH_USER_MODEL class TimestampedModel(models.Model): created_at = models.DateTimeField(auto_now_add=True) updated_at = models.DateTimeField(auto_now=True) class Meta: abstract = True def event_image_path(instance, filename): # file will be uploaded to MEDIA_ROOT/user_/ return "events/static/img/{0}/{1}".format(instance.slug, filename) class Event(TimestampedModel): """The main event which contains activities. An event can have multiple activities, invites, RSVPs, and guests. """ id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False) owner = models.ForeignKey(User, null=True, blank=True, related_name="+", on_delete=models.SET_NULL) title = models.CharField(max_length=255) slug = models.SlugField(max_length=255, db_index=True) description = models.TextField(blank=True) # markdown-supported image = models.ImageField(upload_to=event_image_path, null=True, blank=True) # Behavioural settings allow_rsvp_without_invite = models.BooleanField(default=False) allow_comments = models.BooleanField(default=True) default_email_updates_optin = models.BooleanField(default=True) # Limits (optional) guest_limit = models.PositiveIntegerField(null=True, blank=True) # Response locking and cookie behaviour response_lock_before = models.DurationField(default=timedelta(hours=24)) cookie_expire_after_event = models.DurationField(default=timedelta(days=7)) class Meta: ordering = ['-id'] def __str__(self): return self.title @property def markdown(self) -> str: return markdown.markdown(self.description) @property def static_image_url(self) -> str: image_str = str(self.image) if len(image_str) > 0: return image_str[14:] return "img/placeholder.jpg" @property def start_time(self): first = self.activities.order_by('start_time').first() # pyright: ignore[reportAttributeAccessIssue] return first.start_time if first else None @property def end_time(self): last = self.activities.order_by('-end_time').first() # pyright: ignore[reportAttributeAccessIssue] return last.end_time if last else None @property def cookie_expire_datetime(self): expiretime = self.end_time + self.cookie_expire_after_event return expiretime def is_locked(self) -> bool: last = self.activities.order_by('-end_time').first() # pyright: ignore[reportAttributeAccessIssue] if not last: return False return timezone.now() >= (last.start_time - self.response_lock_before) def total_attendees(self) -> int: return Guest.objects.filter(rsvp__event=self).count() def remaining_capacity(self) -> int | None: if self.guest_limit is None: return None return max(self.guest_limit - self.total_attendees(), 0) class ActivityGroup(TimestampedModel): """Optional grouping for activities; groups can enforce single-choice selection.""" id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False) event = models.ForeignKey(Event, related_name='activity_groups', on_delete=models.CASCADE) title = models.CharField(max_length=255) single_choice = models.BooleanField(default=False) def __str__(self): return f"{self.event.title} — {self.title}" def activity_image_path(instance, filename): # file will be uploaded to MEDIA_ROOT/user_/ return "events/static/img/{0}/{1}".format(instance.slug, filename) class Activity(TimestampedModel): """An individual activity within an event. Activities can be grouped into ActivityGroups. This is the main object that guests RSVP to. """ id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False) event = models.ForeignKey(Event, related_name='activities', on_delete=models.CASCADE) group = models.ForeignKey(ActivityGroup, null=True, blank=True, related_name='activities', on_delete=models.SET_NULL) title = models.CharField(max_length=255) description = models.TextField(blank=True) location = models.CharField(max_length=255) start_time = models.DateTimeField() end_time = models.DateTimeField(null=True, blank=True) guest_limit = models.PositiveIntegerField(null=True, blank=True) image = models.ImageField(upload_to=activity_image_path, null=True, blank=True) allow_rsvp = models.BooleanField(default=True) def __str__(self): if self.group is not None: return f"{self.event.title} — {self.group.title} — {self.title}" return f"{self.event.title} — {self.title}" @property def markdown(self) -> str: return markdown.markdown(self.description) def static_image_url(self) -> str: image_str = str(self.image) if len(image_str) > 0: return image_str[14:] return "img/placeholder_activity.jpg" def is_locked(self) -> bool: if not self.allow_rsvp: return True return timezone.now() >= (self.start_time - self.event.response_lock_before) def remaining_capacity(self) -> int: if self.guest_limit is None: return 999 return max(self.guest_limit - self.attendee_count(), 0) def attendee_count(self) -> int: return Guest.objects.filter(rsvp__activity_selections__activity=self).count() def attendee_list(self) -> list[list[str]]: """ Create a json list of names who are joining this activity Context: - An activity is linked to RSVPs in ActivitySelection - An RSVP has a guests property which is a list of Guest objects for that group - A guest can have the main_invitee property set, but not neceserilly - Each Guest has a name property - A name can be: "John", or "John Doe", or "John Peter Doe", or for foreign guests: "John van Doe" Steps: 1. Get the RSVPs which are coming to Activity (using ActivitySelection) 2. For each RSVP, get the guest list 2.1. Sort the list so that the Guest with property main_invitee is first, then based on the age group (old to young) 2.2. Transform all names for privacy reasons to be the first name only 2.3. If a first name is duplicate in another rsvp guest group, then add the last name letter For example: "John D.", or "John van D." 3. Create a list of lists: [['John D.', 'Jane', 'Jack H.'], ['Maria'], ['John van D.', 'Peter'], ['Jack K.']] 4. Return this list """ selected_rsvps = ActivitySelection.objects.filter(activity=self).values_list('rsvp', flat=True) if not selected_rsvps: return [] age_order = { 'adult': 0, 'child': 1, 'baby': 2 } def sort_key(guest): # main_invitee first (False = 0, True = 1) main_invitee_priority = 0 if guest.is_main_invitee else 1 # Then by age group (old to young) age_priority = age_order.get(guest.age_group, 2) return (main_invitee_priority, age_priority) # Store all first names first_names_tracker = {} for rsvp_id in selected_rsvps: guests = Guest.objects.filter(rsvp=rsvp_id) for guest in guests: name_parts = guest.name.split() first_name = name_parts[0] if first_name in first_names_tracker: first_names_tracker[first_name] += 1 else: first_names_tracker[first_name] = 1 # Create the lists guest_list = [] for rsvp_id in selected_rsvps: guests_non_sort = Guest.objects.filter(rsvp=rsvp_id, ) guests = sorted(guests_non_sort, key=sort_key) rsvp_list = [] for guest in guests: name_parts = guest.name.split() first_name = name_parts[0] # If no duplicate and no last name, just first_name name_to_store = first_name # Check if name is duplicate if first_name in first_names_tracker and first_names_tracker[first_name] > 1: if len(name_parts) == 2: name_to_store = first_name + " " + name_parts[-1][0] + "." elif len(name_parts) > 2: name_to_store = first_name for name_part in name_parts[1:-1]: name_to_store += " " + name_part name_to_store += " " + name_parts[-1][0] + "." rsvp_list += [name_to_store] guest_list += [rsvp_list] # Step 5: Return as JSON string return guest_list class RSVP(TimestampedModel): class Page(models.TextChoices): EMAIL = 'email', _('Email') GUESTS = 'guests', _('Guests') ACTIVITIES = 'activities', _('Activities') QUESTIONS = 'questions', _('Questions') NOTES = 'notes', _('Notes') COMPLETED = 'completed', _('Completed') class Status(models.TextChoices): UNOPENED = 'unopened', _('Not Opened') NOT_STARTED = 'not_started', _('Not Started') INCOMPLETE = 'incomplete', _('Incomplete') NOT_COMING = 'not_coming', _('Not coming') YES_JOINING = 'joining', _('Yes, joining') id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False) event = models.ForeignKey(Event, related_name='rsvps', on_delete=models.CASCADE) invite = models.SlugField(max_length=255, db_index=True, unique=True) edit_token = models.UUIDField(default=uuid.uuid4, editable=False, unique=True) responder_name = models.CharField(max_length=255, blank=True) responder_email = models.EmailField(blank=False) send_email_updates = models.BooleanField(default=True) personalized_message = models.TextField(blank=True) allow_bring_guests = models.BooleanField(default=False) page = models.CharField(max_length=16, choices=Page.choices, default=Page.EMAIL) furthest_page = models.CharField(max_length=16, choices=Page.choices, default=Page.EMAIL) status = models.CharField(max_length=16, choices=Status.choices, default=Status.UNOPENED) notes = models.TextField(blank=True) # Activities selected by this RSVP (through ActivitySelection) activities = models.ManyToManyField(Activity, through='ActivitySelection', related_name='rsvps') def __str__(self): return f"RSVP {self.edit_token} — {self.event.title} — {self.responder_email} — {self.status}" @property def guest_count(self) -> int: return self.guests.count() # pyright: ignore[reportAttributeAccessIssue] def selected_activities_ids(self): selected_activities = ( ActivitySelection.objects .filter(rsvp=self) ) selected_activities_ids = [act.activity.id for act in selected_activities] return selected_activities_ids @property def name(self) -> str: main_guest = self.guests.filter(is_main_invitee=True).first() # pyright: ignore[reportAttributeAccessIssue] if(main_guest): return main_guest.name if(self.responder_name): return self.responder_name return "friend" PAGE_ORDER = [ 'email', 'guests', 'activities', 'questions', 'notes', 'completed', ] def next_page(self) -> Page: """Move to the next page and save to model.""" new_page = self.page try: current_index = self.PAGE_ORDER.index(self.page) if current_index < len(self.PAGE_ORDER) - 1: new_page = self.Page(self.PAGE_ORDER[current_index + 1]) except ValueError: # If page is not in the list (e.g., INVALID), start from EMAIL pass if is_next_page(new_page, self.Page(self.furthest_page)): self.furthest_page = new_page if self.status == self.Status.NOT_STARTED: #TODO send an email here self.status = self.Status.INCOMPLETE self.page = new_page if self.page == self.Page.COMPLETED: self.status = self.Status.YES_JOINING self.save() return new_page def previous_page(self) -> Page: """Move to the previous page and save to model""" new_page = self.page try: current_index = self.PAGE_ORDER.index(self.page) if current_index > 0: new_page = self.Page(self.PAGE_ORDER[current_index - 1]) except ValueError: # If page is not in the list (e.g., INVALID), stay at INVALID pass self.page = new_page self.save() return new_page def set_page(self, page:str) -> Page: """Move to the specific page and save to model""" new_page = self.page try: new_page = self.Page(page) except ValueError: pass self.page = new_page self.save() return new_page def is_page_finished(self, page:Page) -> bool: """return `True` if page is previous to the furthest reached page""" return is_previous_page(page, self.Page(self.furthest_page)) def is_next_page_allowed(self) -> bool | str: # Activities page: require at least one selected activity if self.page == self.Page.ACTIVITIES: if not self.activities.exists(): return "You must choose at least one activity." return True # Questions page: ensure every required question for selected activities is answered if self.page == self.Page.QUESTIONS: selected_activities = self.activities.all() required_questions = Question.objects.filter(activity__in=selected_activities, required=True) for q in required_questions: if q.scope == "rsvp": resp_qs = Response.objects.filter(rsvp=self, question=q) if not resp_qs.exists(): return f'Please answer required question: "{q.text}"' for resp in resp_qs: if q.kind in ("text", "integer", "boolean"): if not (resp.value_text and resp.value_text.strip()): return f'Please answer required question: "{q.text}"' else: # single / multiple choice if resp.choices.count() == 0: return f'Please answer required question: "{q.text}"' elif q.scope == "guest": for guest in self.guests.all(): # pyright: ignore[reportAttributeAccessIssue] resp = Response.objects.filter(guest=guest, question=q).first() if not resp: return f'Please answer required question "{q.text}" for all guests.' if q.kind in ("text", "integer", "boolean"): if not (resp.value_text and resp.value_text.strip()): return f'Please answer required question "{q.text}" for all guests.' else: if resp.choices.count() == 0: return f'Please answer required question "{q.text}" for all guests.' return True # Default: allow moving forward return True class ActivitySelection(models.Model): """Join table: which activities an RSVP selected. NOTE: single-choice enforcement for groups is handled at the application/form level (clean/transaction) because cross-table unique constraints involving a related table's column can't be expressed simply as a DB constraint here. """ id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False) rsvp = models.ForeignKey(RSVP, related_name='activity_selections', on_delete=models.CASCADE) activity = models.ForeignKey(Activity, related_name='activity_selections', on_delete=models.CASCADE) class Meta: unique_together = (('rsvp', 'activity'),) def clean(self): # If the activity belongs to a group that is single_choice, ensure the RSVP has no other selections group = self.activity.group if group and group.single_choice: other = ActivitySelection.objects.filter(rsvp=self.rsvp, activity__group=group).exclude(pk=self.pk) if other.exists(): raise ValidationError(f"You may only select one activity from the group '{group.title}'.") class Guest(TimestampedModel): AGE_GROUP = [ ("baby", "Baby (0-2)"), ("child", "Child (3-15)"), ("adult", "Adult (15+)") ] id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False) event = models.ForeignKey(Event, related_name='guests', on_delete=models.CASCADE) rsvp = models.ForeignKey(RSVP, related_name='guests', on_delete=models.CASCADE) name = models.CharField(max_length=255) age_group = models.CharField(max_length=10, choices=AGE_GROUP, default='adult') is_main_invitee = models.BooleanField(default=False) created_by_host = models.BooleanField(default=False) def __str__(self): return f"{self.name} ({self.age_group})" @property def firstname(self) -> str: return self.name.split(" ")[0] @property def lastname(self) -> str: return self.name.split(" ")[1:] class Question(TimestampedModel): KIND_CHOICES = [ ('text', 'Text'), ('single', 'Single choice'), ('multiple', 'Multiple choice'), ('integer', 'Integer'), ('boolean', 'Yes/No') ] SCOPE_CHOICES = [ ("rsvp", "Once per RSVP"), ("guest", "Once per Guest"), ] id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False) activity = models.ForeignKey(Activity, related_name='questions', on_delete=models.CASCADE) text = models.CharField(max_length=1000) help_text = models.CharField(max_length=1000, blank=True) kind = models.CharField(max_length=10, choices=KIND_CHOICES, default='text') scope = models.CharField(max_length=10, choices=SCOPE_CHOICES, default="rsvp") required = models.BooleanField(default=False) order = models.PositiveIntegerField(default=0) def __str__(self): return f"Q: {self.text} (on `{self.activity} - {self.scope}` )" class QuestionChoice(models.Model): question = models.ForeignKey(Question, related_name='choices', on_delete=models.CASCADE) value = models.CharField(max_length=255) label = models.CharField(max_length=255) def __str__(self): return self.label class Response(TimestampedModel): """Answer to a Question. Questions live on Activities. A Response must link to the appropriate owner according to the question context: - An activity-level question that applies to the entire RSVP: store `rsvp`. - A per-person question: store `guest`. To keep the model simple, question-level scoping is enforced at form/validation time. """ id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False) activity = models.ForeignKey(Activity, related_name='responses', on_delete=models.CASCADE) question = models.ForeignKey(Question, related_name='responses', on_delete=models.CASCADE) rsvp = models.ForeignKey(RSVP, null=True, blank=True, related_name='responses', on_delete=models.CASCADE) guest = models.ForeignKey(Guest, null=True, blank=True, related_name='responses', on_delete=models.CASCADE) # Choice (single) if applicable choices = models.ManyToManyField(QuestionChoice, through="ResponseChoice", related_name="responses", blank=True) value_text = models.TextField(blank=True) def clean(self): scope = self.question.scope if scope == "rsvp": if not self.rsvp or self.guest: raise ValidationError("This question must be answered once per RSVP.") if scope == "guest": if not self.guest or self.rsvp: raise ValidationError("This question must be answered once per Guest.") kind = self.question.kind # For choice-backed questions, validation happens relative to choices M2M if kind == "single": # Expect exactly one selected choice (or at least one) if not self.pk: # if not saved yet, can't reliably check M2M, so skip strict check here pass else: if self.choices.count() != 1: raise ValidationError("Single-choice question requires exactly one choice.") if kind == "multiple": if not self.pk: pass # else: # if self.choices.count() < 1: # raise ValidationError("Multiple-choice question requires at least one choice.") if kind in ("text", "integer", "boolean") and self.choices.exists(): raise ValidationError("This question does not use choices.") if self.question.activity.id != self.activity.id: raise ValidationError("Question must belong to the same activity.") def __str__(self): whom = f"{self.question.scope}:" if(self.question.scope == "rsvp"): whom += " " + self.rsvp.responder_name elif(self.question.scope == "guest"): whom += " " + self.guest.name if self.question.kind in ("text", "integer", "boolean"): return f"Response to {self.question} by {whom}: {self.value_text}" if self.question.kind in ("single", "multiple") and self.choices.count() > 0: return f"Response to {self.question} by {whom}: {self.choices.all}" class ResponseChoice(TimestampedModel): id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False) response = models.ForeignKey("Response", related_name="response_choices", on_delete=models.CASCADE) choice = models.ForeignKey(QuestionChoice, related_name="response_choices", on_delete=models.CASCADE) class Meta: unique_together = (("response", "choice"),) def __str__(self): return f"{self.choice} for {self.response}" class Comment(TimestampedModel): event = models.ForeignKey(Event, related_name='comments', on_delete=models.CASCADE) guest = models.ForeignKey(Guest, related_name='comments', on_delete=models.CASCADE) text = models.TextField() notify_subscribers = models.BooleanField(default=False) def __str__(self): return f"Comment by {self.guest.name} on {self.event.title}"