597 lines
23 KiB
Python
597 lines
23 KiB
Python
# models.py
|
|
import os
|
|
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/<event_slug>/overview<.extension>
|
|
return "{0}/overview{1}".format(instance.slug, os.path.splitext(filename)[1])
|
|
|
|
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 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/<event_slug>/overview<.extension>
|
|
return "{0}/{1}{2}".format(instance.event.slug, instance.id, os.path.splitext(filename)[1])
|
|
|
|
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 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}" |