919 lines
32 KiB
Python
919 lines
32 KiB
Python
import uuid
|
|
from datetime import datetime, timedelta
|
|
from decimal import Decimal
|
|
from typing import TYPE_CHECKING, TypedDict
|
|
|
|
import django.core.mail.message
|
|
from django.conf import settings
|
|
from django.contrib.auth.models import AbstractBaseUser
|
|
from django.db import connection, models
|
|
from django.db.models import (
|
|
Case,
|
|
Count,
|
|
Exists,
|
|
ExpressionWrapper,
|
|
F,
|
|
OuterRef,
|
|
Q,
|
|
QuerySet,
|
|
Subquery,
|
|
Sum,
|
|
Value,
|
|
When,
|
|
)
|
|
from django.db.models.functions import Cast, Coalesce
|
|
from django.urls import reverse
|
|
from django.utils import timezone
|
|
|
|
import nh3
|
|
from django_db_views.db_view import DBView
|
|
from django_stubs_ext import WithAnnotations
|
|
from simple_history.models import HistoricalRecords, HistoricForeignKey
|
|
|
|
from reservations.models import Reservation
|
|
|
|
|
|
class BaseModel(models.Model):
|
|
_api_names_override: dict[str, str] = {}
|
|
_date_fields: dict[str, str | None] = {}
|
|
_allowed_missing_fields: list[str] = []
|
|
|
|
class Meta:
|
|
abstract = True
|
|
|
|
@classmethod
|
|
def _remap_headers(cls, data):
|
|
for field in cls._meta.get_fields():
|
|
# TODO: more robust filtering of fields that don't have a column
|
|
if (
|
|
field.auto_created
|
|
or field.many_to_many
|
|
or not field.concrete
|
|
or field.generated
|
|
):
|
|
continue
|
|
|
|
field_name, api_name = field.get_attname_column()
|
|
|
|
if field_name in cls._api_names_override:
|
|
api_name = cls._api_names_override[field_name]
|
|
|
|
yield field_name, data[api_name]
|
|
|
|
@classmethod
|
|
def from_api_dict(cls, data):
|
|
data = data.copy()
|
|
|
|
for field in cls._allowed_missing_fields:
|
|
if field not in data:
|
|
data[field] = None
|
|
|
|
# parse date fields to datetime objects
|
|
for field, fmt in cls._date_fields.items():
|
|
if data[field]:
|
|
if fmt is None:
|
|
# can't use '%s' format string, have to use the special function
|
|
data[field] = datetime.fromtimestamp(
|
|
data[field], tz=timezone.get_current_timezone()
|
|
)
|
|
else:
|
|
data[field] = datetime.strptime(str(data[field]), fmt)
|
|
else:
|
|
# convert empty string to None to make NULL in SQL
|
|
data[field] = None
|
|
|
|
return cls(**dict(cls._remap_headers(data)))
|
|
|
|
|
|
class Flag(BaseModel):
|
|
id = models.CharField(max_length=24, primary_key=True)
|
|
name = models.TextField(null=True, blank=True)
|
|
type = models.CharField(max_length=6)
|
|
|
|
history = HistoricalRecords()
|
|
|
|
class Meta:
|
|
db_table = "flag"
|
|
ordering = ("name",)
|
|
|
|
def __str__(self):
|
|
return f"{self.name} ({self.type})"
|
|
|
|
|
|
class MemberQuerySet(models.QuerySet):
|
|
# TODO: maybe rename to reflect EXISTS?
|
|
@staticmethod
|
|
def has_flag(flag_type: str, flag_name: str):
|
|
return Exists(
|
|
Flag.objects.filter(type=flag_type, name=flag_name, members=OuterRef("pk"))
|
|
)
|
|
|
|
# TODO: it should be fairly easy to reduce the number of EXISTS by
|
|
# merging the ORed flags
|
|
def with_is_active(self):
|
|
return self.annotate(
|
|
is_active=(
|
|
self.has_flag("folder", "Members")
|
|
| self.has_flag("folder", "CMS Staff")
|
|
)
|
|
& ~(
|
|
self.has_flag("label", "Account On Hold")
|
|
| self.has_flag("level", "CMS Membership on hold")
|
|
| self.has_flag("folder", "Former Members")
|
|
)
|
|
)
|
|
|
|
|
|
class Member(BaseModel):
|
|
uid = models.CharField(max_length=24, primary_key=True)
|
|
year_of_birth = models.TextField(db_column="Year of Birth", null=True, blank=True)
|
|
account_name = models.TextField(db_column="Account Name", null=True, blank=True)
|
|
first_name = models.TextField(db_column="First Name", null=True, blank=True)
|
|
last_name = models.TextField(db_column="Last Name", null=True, blank=True)
|
|
phone = models.TextField(db_column="Phone", null=True, blank=True)
|
|
email = models.TextField(db_column="Email", null=True, blank=True)
|
|
volunteer_email = models.TextField(
|
|
db_column="Volunteer Email", null=True, blank=True
|
|
)
|
|
address_street = models.TextField(
|
|
db_column="Address (Street)", null=True, blank=True
|
|
)
|
|
address_city = models.TextField(db_column="Address (City)", null=True, blank=True)
|
|
address_state_province = models.TextField(
|
|
db_column="Address (State/Province)", null=True, blank=True
|
|
)
|
|
address_postal_code = models.TextField(
|
|
db_column="Address (Postal Code)", null=True, blank=True
|
|
)
|
|
address_country = models.TextField(
|
|
db_column="Address (Country)", null=True, blank=True
|
|
)
|
|
profile_description = models.TextField(
|
|
db_column="Profile description", null=True, blank=True
|
|
)
|
|
website = models.TextField(db_column="Website", null=True, blank=True)
|
|
fax = models.TextField(db_column="Fax", null=True, blank=True)
|
|
contact_person = models.TextField(db_column="Contact Person", null=True, blank=True)
|
|
password = models.TextField(db_column="Password", null=True, blank=True)
|
|
position_relation = models.TextField(
|
|
db_column="Position/relation", null=True, blank=True
|
|
)
|
|
parent_account_id = models.TextField(
|
|
db_column="Parent Account ID", null=True, blank=True
|
|
)
|
|
closet_storage = models.TextField(
|
|
db_column="Closet Storage #", null=True, blank=True
|
|
)
|
|
storage_shelf = models.TextField(db_column="Storage Shelf #", null=True, blank=True)
|
|
personal_studio_space = models.TextField(
|
|
db_column="Personal Studio Space #", null=True, blank=True
|
|
)
|
|
access_permitted_shops_during_extended_hours = models.BooleanField(
|
|
db_column="Access Permitted Shops During Extended Hours?"
|
|
)
|
|
access_front_door_and_studio_space_during_extended_hours = models.BooleanField(
|
|
db_column="Access Front Door and Studio Space During Extended Hours?"
|
|
)
|
|
access_wood_shop = models.BooleanField(db_column="Access Wood Shop?")
|
|
access_metal_shop = models.BooleanField(db_column="Access Metal Shop?")
|
|
access_storage_closet = models.BooleanField(db_column="Access Storage Closet?")
|
|
access_studio_space = models.BooleanField(db_column="Access Studio Space?")
|
|
access_front_door = models.BooleanField(db_column="Access Front Door?")
|
|
access_card_number = models.TextField(
|
|
db_column="Access Card Number", null=True, blank=True
|
|
)
|
|
access_card_facility_code = models.TextField(
|
|
db_column="Access Card Facility Code", null=True, blank=True
|
|
)
|
|
nfc_card_number = models.TextField(
|
|
db_column="NFC Card Number", null=True, blank=True
|
|
)
|
|
auto_billing_id = models.TextField(
|
|
db_column="Auto Billing ID", null=True, blank=True
|
|
)
|
|
billing_method = models.TextField(db_column="Billing Method", null=True, blank=True)
|
|
renewal_date = models.DateField(db_column="Renewal Date", null=True, blank=True)
|
|
join_date = models.DateField(db_column="Join Date", null=True, blank=True)
|
|
admin_note = models.TextField(db_column="Admin note", null=True, blank=True)
|
|
profile_gallery_image_url = models.TextField(
|
|
db_column="Profile gallery image URL", null=True, blank=True
|
|
)
|
|
business_card_image_url = models.TextField(
|
|
db_column="Business card image URL", null=True, blank=True
|
|
)
|
|
instagram = models.TextField(db_column="Instagram", null=True, blank=True)
|
|
pinterest = models.TextField(db_column="Pinterest", null=True, blank=True)
|
|
youtube = models.TextField(db_column="Youtube", null=True, blank=True)
|
|
yelp = models.TextField(db_column="Yelp", null=True, blank=True)
|
|
google = models.TextField(db_column="Google+", null=True, blank=True)
|
|
bbb = models.TextField(db_column="BBB", null=True, blank=True)
|
|
twitter = models.TextField(db_column="Twitter", null=True, blank=True)
|
|
facebook = models.TextField(db_column="Facebook", null=True, blank=True)
|
|
linked_in = models.TextField(db_column="LinkedIn", null=True, blank=True)
|
|
do_not_show_street_address_in_profile = models.TextField(
|
|
db_column="Do not show street address in profile", null=True, blank=True
|
|
)
|
|
do_not_list_in_directory = models.TextField(
|
|
db_column="Do not list in directory", null=True, blank=True
|
|
)
|
|
how_did_you_hear = models.TextField(
|
|
db_column="HowDidYouHear", null=True, blank=True
|
|
)
|
|
authorize_charge = models.TextField(
|
|
db_column="authorizeCharge", null=True, blank=True
|
|
)
|
|
policy_agreement = models.TextField(
|
|
db_column="policyAgreement", null=True, blank=True
|
|
)
|
|
waiver_form_signed_and_on_file_date = models.DateField(
|
|
db_column="Waiver form signed and on file date.", null=True, blank=True
|
|
)
|
|
membership_agreement_signed_and_on_file_date = models.DateField(
|
|
db_column="Membership Agreement signed and on file date.", null=True, blank=True
|
|
)
|
|
ip_address = models.TextField(db_column="IP Address", null=True, blank=True)
|
|
audit_date = models.DateField(db_column="Audit Date", null=True, blank=True)
|
|
agreement_version = models.TextField(
|
|
db_column="Agreement Version", null=True, blank=True
|
|
)
|
|
paperwork_status = models.TextField(
|
|
db_column="Paperwork status", null=True, blank=True
|
|
)
|
|
membership_agreement_dated = models.BooleanField(
|
|
db_column="Membership agreement dated"
|
|
)
|
|
membership_agreement_acknowledgement_page_filled_out = models.BooleanField(
|
|
db_column="Membership Agreement Acknowledgement Page Filled Out"
|
|
)
|
|
membership_agreement_signed = models.BooleanField(
|
|
db_column="Membership Agreement Signed"
|
|
)
|
|
liability_form_filled_out = models.BooleanField(
|
|
db_column="Liability Form Filled Out"
|
|
)
|
|
flags = models.ManyToManyField(Flag, through="MemberFlag", related_name="members")
|
|
|
|
history = HistoricalRecords()
|
|
|
|
_api_names_override = {
|
|
"uid": "Account ID",
|
|
"how_did_you_hear": "Please tell us how you heard about the Claremont MakerSpace and what tools or shops you are most excited to start using:",
|
|
"authorize_charge": "Yes - I authorize TwinState MakerSpaces, Inc. to charge my credit card for the membership and other options that I have selected.",
|
|
"policy_agreement": "I have read the Claremont MakerSpace Membership Agreement & Policies, and agree to all terms stated therein.",
|
|
}
|
|
|
|
_date_fields = {
|
|
"Join Date": "%b %d, %Y",
|
|
"Renewal Date": "%b %d, %Y",
|
|
"Audit Date": "%m/%d/%Y",
|
|
"Membership Agreement signed and on file date.": "%m/%d/%Y",
|
|
"Waiver form signed and on file date.": "%m/%d/%Y",
|
|
}
|
|
|
|
objects = MemberQuerySet.as_manager()
|
|
|
|
class Meta:
|
|
db_table = "members"
|
|
ordering = ("first_name", "last_name")
|
|
indexes = [
|
|
models.Index(fields=["account_name"], name="account_name_idx"),
|
|
models.Index(fields=["first_name"], name="first_name_idx"),
|
|
models.Index(fields=["last_name"], name="last_name_idx"),
|
|
]
|
|
|
|
def __str__(self):
|
|
return f"{self.account_name}"
|
|
|
|
@classmethod
|
|
def from_user(cls, user) -> "Member | None":
|
|
if hasattr(user, "ldap_user"):
|
|
return cls.objects.get(uid=user.ldap_user.attrs["employeeNumber"][0])
|
|
return None
|
|
|
|
def sanitized_mailbox(self, use_volunteer=False) -> str:
|
|
if use_volunteer and self.volunteer_email:
|
|
email = self.volunteer_email
|
|
elif self.email:
|
|
email = self.email
|
|
else:
|
|
raise Exception(f"No Email Address for user: {self.uid}")
|
|
|
|
if not self.account_name:
|
|
return email
|
|
|
|
return django.core.mail.message.sanitize_address(
|
|
(self.account_name, email), settings.DEFAULT_CHARSET
|
|
)
|
|
|
|
|
|
class MemberFlag(BaseModel):
|
|
member = HistoricForeignKey(
|
|
Member, on_delete=models.PROTECT, db_column="uid", db_constraint=False
|
|
)
|
|
flag = HistoricForeignKey(Flag, on_delete=models.PROTECT)
|
|
|
|
history = HistoricalRecords()
|
|
|
|
class Meta:
|
|
db_table = "memberflag"
|
|
constraints = [
|
|
models.UniqueConstraint(
|
|
fields=["member", "flag"], name="unique_member_flag"
|
|
)
|
|
]
|
|
|
|
def __str__(self):
|
|
return f"{self.member} - {self.flag}"
|
|
|
|
|
|
class Transaction(BaseModel):
|
|
sid = models.CharField(max_length=256, null=True, blank=True)
|
|
member = models.ForeignKey(
|
|
Member,
|
|
on_delete=models.PROTECT,
|
|
db_column="uid",
|
|
db_constraint=False,
|
|
related_name="transactions",
|
|
null=True,
|
|
blank=True,
|
|
)
|
|
timestamp = models.DateTimeField()
|
|
type = models.TextField(null=True, blank=True)
|
|
sum = models.DecimalField(max_digits=13, decimal_places=4, null=True, blank=True)
|
|
fee = models.DecimalField(max_digits=13, decimal_places=4, null=True, blank=True)
|
|
event_id = models.TextField(null=True, blank=True)
|
|
for_what = models.TextField(db_column="For", null=True, blank=True)
|
|
items = models.TextField(db_column="Items", null=True, blank=True)
|
|
discount_code = models.TextField(db_column="Discount Code", null=True, blank=True)
|
|
note = models.TextField(db_column="Note", null=True, blank=True)
|
|
name = models.TextField(db_column="Name", null=True, blank=True)
|
|
contact_person = models.TextField(db_column="Contact Person", null=True, blank=True)
|
|
full_address = models.TextField(db_column="Full Address", null=True, blank=True)
|
|
street = models.TextField(db_column="Street", null=True, blank=True)
|
|
city = models.TextField(db_column="City", null=True, blank=True)
|
|
state_province = models.TextField(db_column="State/Province", null=True, blank=True)
|
|
postal_code = models.TextField(db_column="Postal Code", null=True, blank=True)
|
|
country = models.TextField(db_column="Country", null=True, blank=True)
|
|
phone = models.TextField(db_column="Phone", null=True, blank=True)
|
|
email = models.TextField(db_column="Email", null=True, blank=True)
|
|
|
|
_allowed_missing_fields = [
|
|
"sid",
|
|
"uid",
|
|
"eid",
|
|
"fee",
|
|
"sum",
|
|
]
|
|
_api_names_override = {
|
|
"event_id": "eid",
|
|
"timestamp": "_dp",
|
|
"type": "Transaction Type",
|
|
"for_what": "Reference",
|
|
}
|
|
_date_fields = {
|
|
"_dp": None,
|
|
}
|
|
|
|
class Meta:
|
|
db_table = "transactions"
|
|
constraints = [
|
|
models.UniqueConstraint("sid", "timestamp", name="unique_sid_timestamp")
|
|
]
|
|
|
|
def __str__(self):
|
|
return f"{self.type} [{self.member if self.member else self.name}] {self.timestamp}"
|
|
|
|
|
|
class EventCategory(models.Model):
|
|
id = models.IntegerField(primary_key=True)
|
|
title = models.TextField()
|
|
|
|
def __str__(self):
|
|
return self.title
|
|
|
|
@classmethod
|
|
def from_api_dict(cls, id_: int, data):
|
|
return cls(id=id_, title=data["ttl"])
|
|
|
|
|
|
class Event(BaseModel):
|
|
class EventCalendar(models.IntegerChoices):
|
|
HIDDEN = 0
|
|
GREEN = 1
|
|
RED = 2
|
|
YELLOW = 3
|
|
BLUE = 4
|
|
PURPLE = 5
|
|
MAGENTA = 6
|
|
GREY = 7
|
|
TEAL = 8
|
|
|
|
eid = models.CharField(max_length=255, primary_key=True)
|
|
url = models.TextField()
|
|
title = models.TextField()
|
|
start = models.DateTimeField()
|
|
end = models.DateTimeField(null=True, blank=True)
|
|
cap = models.IntegerField(null=True, blank=True)
|
|
count = models.IntegerField()
|
|
category = models.ForeignKey(EventCategory, on_delete=models.PROTECT)
|
|
calendar = models.IntegerField(choices=EventCalendar)
|
|
venue = models.TextField(null=True, blank=True)
|
|
occurred = models.GeneratedField(
|
|
expression=~(Q(cap=0) | Q(count=0) | Q(calendar=EventCalendar.HIDDEN)),
|
|
output_field=models.BooleanField(),
|
|
db_persist=True,
|
|
)
|
|
# TODO:
|
|
# "lgo": {
|
|
# "l": "https://d1tif55lvfk8gc.cloudfront.net/656e3842ae3975908b05e304.jpg?1673405126",
|
|
# "s": "https://d1tif55lvfk8gc.cloudfront.net/656e3842ae3975908b05e304s.jpg?1673405126"
|
|
# },
|
|
|
|
_api_names_override = {
|
|
"title": "ttl",
|
|
"category_id": "grp",
|
|
"start": "sdp",
|
|
"end": "edp",
|
|
"count": "cnt",
|
|
"calendar": "cal",
|
|
"venue": "adn",
|
|
}
|
|
|
|
_date_fields = {
|
|
"sdp": None,
|
|
"edp": None,
|
|
}
|
|
|
|
_allowed_missing_fields = ["cap", "edp", "adn"]
|
|
|
|
def __str__(self):
|
|
return self.unescaped_title
|
|
|
|
@property
|
|
def unescaped_title(self):
|
|
return nh3.clean(self.title, tags=set())
|
|
|
|
|
|
class EventInstructor(models.Model):
|
|
name = models.TextField(blank=True)
|
|
member = models.OneToOneField(
|
|
Member, on_delete=models.PROTECT, null=True, blank=True, db_constraint=False
|
|
)
|
|
|
|
def __str__(self):
|
|
return str(self.member) if self.member else self.name
|
|
|
|
|
|
class EventExtQuerySet(models.QuerySet["EventExtAnnotated"]):
|
|
def summarize(self, aggregate: bool = False):
|
|
method = self.aggregate if aggregate else self.annotate
|
|
return method(
|
|
count__sum=Sum("count", filter=F("occurred")),
|
|
instructor__count=Count("instructor", distinct=True, filter=F("occurred")),
|
|
meetings__sum=Sum("meetings", filter=F("occurred")),
|
|
duration__sum=Sum("duration", filter=F("occurred")),
|
|
person_hours__sum=Sum("person_hours", filter=F("occurred")),
|
|
event_count=Count("eid", filter=F("occurred")),
|
|
canceled_event_count=Count("eid", filter=~F("occurred")),
|
|
gross_revenue__sum=Sum("gross_revenue", filter=F("occurred")),
|
|
total_due_to_instructor__sum=Sum(
|
|
"total_due_to_instructor", filter=F("occurred")
|
|
),
|
|
net_revenue__sum=Sum("net_revenue", filter=F("occurred")),
|
|
)
|
|
|
|
def with_financials(self) -> "QuerySet[EventExtAnnotatedWithFinancials]":
|
|
return self.annotate(
|
|
**{
|
|
field: F(f"ticket_aggregates__{field}")
|
|
for field in [
|
|
"quantity",
|
|
"amount",
|
|
"materials",
|
|
"amount_without_materials",
|
|
"instructor_revenue",
|
|
"instructor_amount",
|
|
]
|
|
},
|
|
total_due_to_instructor=(
|
|
F("instructor_amount") + F("instructor_flat_rate")
|
|
),
|
|
gross_revenue=Coalesce(
|
|
F("attendee_stats__gross_revenue"),
|
|
0,
|
|
output_field=models.DecimalField(),
|
|
),
|
|
net_revenue=F("gross_revenue") - F("total_due_to_instructor"),
|
|
)
|
|
|
|
def with_meeting_times_match_event(self):
|
|
return self.annotate(
|
|
meeting_times_match_event=(
|
|
Q(
|
|
start=Subquery(
|
|
EventMeetingTime.objects.filter(event=OuterRef("pk"))
|
|
.order_by("start")
|
|
.values("start")[:1],
|
|
output_field=models.DateTimeField(),
|
|
)
|
|
)
|
|
& Q(
|
|
end=Subquery(
|
|
EventMeetingTime.objects.filter(event=OuterRef("pk"))
|
|
.order_by("-end")
|
|
.values("end")[:1],
|
|
output_field=models.DateTimeField(),
|
|
)
|
|
)
|
|
),
|
|
)
|
|
|
|
|
|
class EventExtManager(models.Manager):
|
|
def get_queryset(self) -> EventExtQuerySet:
|
|
return EventExtQuerySet(self.model, using=self._db).annotate(
|
|
meetings=Subquery(
|
|
EventMeetingTime.objects.filter(event=OuterRef("pk"))
|
|
.values("event__pk")
|
|
.annotate(d=Count("pk"))
|
|
.values("d")[:1],
|
|
output_field=models.IntegerField(),
|
|
),
|
|
next_meeting_start=Subquery(
|
|
EventMeetingTime.objects.filter(
|
|
event=OuterRef("pk"), end__gt=timezone.now()
|
|
)
|
|
.order_by("start")
|
|
.values("start")[:1]
|
|
),
|
|
duration=Subquery(
|
|
EventMeetingTime.objects.filter(event=OuterRef("pk"))
|
|
.values("event__pk")
|
|
.annotate(d=Sum("duration"))
|
|
.values("d")[:1],
|
|
output_field=models.DurationField(),
|
|
),
|
|
person_hours=ExpressionWrapper(
|
|
ExpressionWrapper(F("duration"), models.IntegerField()) * F("count"),
|
|
models.DurationField(),
|
|
),
|
|
)
|
|
|
|
|
|
class EventExt(Event):
|
|
"""Extension of `Event` to capture some fields not supported in MembershipWorks"""
|
|
|
|
instructor = models.ForeignKey(
|
|
EventInstructor, on_delete=models.PROTECT, null=True, blank=True
|
|
)
|
|
materials_fee = models.DecimalField(
|
|
max_digits=13,
|
|
decimal_places=4,
|
|
null=True,
|
|
blank=True,
|
|
help_text="Please enter 0 if there was no materials fee",
|
|
)
|
|
materials_fee_included_in_price = models.BooleanField(
|
|
null=True,
|
|
help_text="Did CMS charge the materials fee via MembershipWorks? Use 'No' if the materials fee was collected via cash or other means. Not used if materials fee is 0.",
|
|
)
|
|
instructor_percentage = models.DecimalField(
|
|
max_digits=5, decimal_places=4, default=0.5
|
|
)
|
|
instructor_flat_rate = models.DecimalField(
|
|
max_digits=13, decimal_places=4, default=0
|
|
)
|
|
details = models.JSONField(null=True, blank=True)
|
|
details_timestamp = models.GeneratedField(
|
|
expression=models.Func(
|
|
Cast(models.F("details___ts"), models.IntegerField()),
|
|
function="to_timestamp",
|
|
),
|
|
output_field=models.DateTimeField(),
|
|
db_persist=True,
|
|
verbose_name="Last details fetch",
|
|
)
|
|
|
|
registrations = models.JSONField(null=True, blank=True)
|
|
|
|
should_survey = models.BooleanField(default=False)
|
|
survey_email_sent = models.BooleanField(default=False)
|
|
|
|
objects = EventExtManager.from_queryset(EventExtQuerySet)()
|
|
|
|
class Meta:
|
|
verbose_name = "event"
|
|
ordering = ["-start"]
|
|
|
|
def get_absolute_url(self) -> str:
|
|
return reverse("membershipworks:event-detail", kwargs={"eid": self.eid})
|
|
|
|
def user_is_instructor(self, user: AbstractBaseUser) -> bool:
|
|
if self.instructor is None:
|
|
return False
|
|
member = Member.from_user(user)
|
|
if member is not None:
|
|
return self.instructor.member == member
|
|
return False
|
|
|
|
@property
|
|
def missing_for_invoice(self) -> list[str]:
|
|
reasons = {
|
|
"Missing instructor": self.instructor is None,
|
|
"Instructor not linked to a member": (
|
|
self.instructor is not None and self.instructor.member is None
|
|
),
|
|
"Missing materials fee": self.materials_fee is None,
|
|
"Materials fee is not 0 and materials_fee_included_in_price not defined": (
|
|
self.materials_fee != 0 and self.materials_fee_included_in_price is None
|
|
),
|
|
"total_due_to_instructor is None (this can have several causes)": (
|
|
self.total_due_to_instructor is None
|
|
),
|
|
}
|
|
return [k for k, v in reasons.items() if v]
|
|
|
|
@property
|
|
def ready_for_invoice(self) -> bool:
|
|
return len(self.missing_for_invoice) == 0
|
|
|
|
|
|
if TYPE_CHECKING:
|
|
from decimal import Decimal
|
|
|
|
class EventExtAnnotations(TypedDict):
|
|
meetings: int
|
|
duration: timedelta
|
|
person_hours: timedelta
|
|
details_timestamp: datetime
|
|
|
|
class EventExtFinancialAnnotations(TypedDict):
|
|
quantity: Decimal
|
|
amount: Decimal
|
|
materials: Decimal
|
|
amount_without_materials: Decimal
|
|
instructor_revenue: Decimal
|
|
instructor_amount: Decimal
|
|
total_due_to_instructor: Decimal
|
|
gross_revenue: Decimal
|
|
net_revenue: Decimal
|
|
|
|
EventExtAnnotated = WithAnnotations[EventExt, EventExtAnnotations]
|
|
EventExtAnnotatedWithFinancials = WithAnnotations[EventExt, EventExtAnnotations]
|
|
else:
|
|
EventExtAnnotated = WithAnnotations[EventExt]
|
|
EventExtAnnotatedWithFinancials = WithAnnotations[EventExt]
|
|
|
|
|
|
class EventMeetingTime(Reservation):
|
|
event = models.ForeignKey(
|
|
EventExt, on_delete=models.CASCADE, related_name="meeting_times"
|
|
)
|
|
|
|
def get_title(self) -> str:
|
|
return self.event.unescaped_title
|
|
|
|
# TODO: should probably do some validation in python to enforce
|
|
# - uniqueness and non-overlapping (per event)
|
|
# - min/max start/end time == event start end
|
|
|
|
def make_google_calendar_event(self):
|
|
status = (
|
|
"confirmed"
|
|
if self.event.cap > 0 and self.event.calendar != Event.EventCalendar.HIDDEN
|
|
else "cancelled"
|
|
)
|
|
|
|
return super().make_google_calendar_event() | {
|
|
# TODO: add event description and links
|
|
"summary": self.event.unescaped_title,
|
|
"status": status,
|
|
}
|
|
|
|
|
|
class EventInvoice(models.Model):
|
|
uuid = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
|
|
event = models.OneToOneField(
|
|
EventExt, on_delete=models.PROTECT, related_name="invoice"
|
|
)
|
|
date_submitted = models.DateField()
|
|
date_paid = models.DateField(blank=True, null=True)
|
|
pdf = models.FileField(upload_to="protected/invoices/%Y/%m/%d/")
|
|
amount = models.DecimalField(max_digits=13, decimal_places=4)
|
|
|
|
def __str__(self) -> str:
|
|
return f'"{self.event}" submitted={self.date_submitted} paid:{self.date_paid} ${self.amount}'
|
|
|
|
|
|
class EventTicketTypeQuerySet(models.QuerySet["EventTicketType"]):
|
|
def group_by_ticket_type(self):
|
|
return self.values("price_group").annotate(
|
|
label=F("price_group"),
|
|
actual_price=F("actual_price"),
|
|
**{
|
|
field: Sum(field)
|
|
for field in [
|
|
"quantity",
|
|
"materials",
|
|
"amount",
|
|
"amount_without_materials",
|
|
"instructor_revenue",
|
|
"instructor_amount",
|
|
]
|
|
},
|
|
)
|
|
|
|
|
|
class EventTicketTypeManager(models.Manager["EventTicketType"]):
|
|
def get_queryset(self) -> models.QuerySet["EventTicketType"]:
|
|
qs = super().get_queryset()
|
|
return qs.annotate(
|
|
# Before 2024-07-01, use Members ticket price for any
|
|
# restricted ticket, but list price for unrestricted
|
|
# (Non-Members) ticket. After, use Members ticket price
|
|
# for all tickets except where members ticket is free.
|
|
price_group=Case(
|
|
When(
|
|
Q(members_price=0)
|
|
| Q(
|
|
event__start__lt=datetime(
|
|
year=2024,
|
|
month=7,
|
|
day=1,
|
|
tzinfo=timezone.get_default_timezone(),
|
|
)
|
|
),
|
|
Case(
|
|
When(Q(restrict_to__isnull=True), Value("Non-Members")),
|
|
default=Value("Members"),
|
|
),
|
|
),
|
|
default=Value("Attendee"),
|
|
),
|
|
actual_price=Case(
|
|
# Price group will be "Non-Members" iff we are using
|
|
# the list price for that ticket type. In all other
|
|
# cases (special program discounts or non-members
|
|
# tickets after 2024-07-01), use the members price
|
|
When(
|
|
Q(price_group=Value("Non-Members")),
|
|
"list_price",
|
|
),
|
|
default="members_price",
|
|
),
|
|
materials=Case(
|
|
When(
|
|
(
|
|
Q(event__materials_fee_included_in_price=True)
|
|
| Q(event__materials_fee=0)
|
|
& Q(event__materials_fee__isnull=False)
|
|
),
|
|
ExpressionWrapper(
|
|
F("event__materials_fee") * F("quantity"),
|
|
output_field=models.DecimalField(),
|
|
),
|
|
),
|
|
When(Q(event__materials_fee_included_in_price__isnull=True), None),
|
|
default=0,
|
|
output_field=models.DecimalField(),
|
|
),
|
|
amount=ExpressionWrapper(
|
|
F("actual_price") * F("quantity"),
|
|
output_field=models.DecimalField(),
|
|
),
|
|
amount_without_materials=ExpressionWrapper(
|
|
F("amount") - F("materials"), output_field=models.DecimalField()
|
|
),
|
|
instructor_revenue=ExpressionWrapper(
|
|
F("amount_without_materials") * F("event__instructor_percentage"),
|
|
output_field=models.DecimalField(),
|
|
),
|
|
instructor_amount=ExpressionWrapper(
|
|
F("instructor_revenue") + F("materials"),
|
|
output_field=models.DecimalField(),
|
|
),
|
|
)
|
|
|
|
|
|
class EventTicketType(DBView):
|
|
objects = EventTicketTypeManager.from_queryset(EventTicketTypeQuerySet)()
|
|
|
|
event = models.ForeignKey(
|
|
EventExt, on_delete=models.DO_NOTHING, related_name="ticket_types"
|
|
)
|
|
label = models.TextField(db_column="lbl")
|
|
list_price = models.DecimalField(db_column="amt", max_digits=13, decimal_places=4)
|
|
members_price = models.DecimalField(max_digits=13, decimal_places=4)
|
|
quantity = models.IntegerField(db_column="cnt")
|
|
restrict_to = models.JSONField(db_column="dsp")
|
|
|
|
view_definition = f"""
|
|
SELECT
|
|
row_number() over () as id,
|
|
eventext.event_ptr_id as event_id,
|
|
tkt.*,
|
|
jsonb_path_query_first(
|
|
eventext.details,
|
|
'$.tkt[*] ? (exists (@.dsp ? (@[*] == "{settings.MW_MEMBERS_FOLDER_ID}"))).amt'
|
|
)::numeric as members_price
|
|
FROM membershipworks_eventext AS eventext,
|
|
jsonb_to_recordset(eventext.details -> 'tkt') AS tkt (
|
|
lbl TEXT,
|
|
amt NUMERIC,
|
|
cnt INT,
|
|
dsp JSONB
|
|
)
|
|
"""
|
|
|
|
def __str__(self) -> str:
|
|
return f"{self.label}: {self.quantity} * {self.list_price}"
|
|
|
|
class Meta:
|
|
managed = False
|
|
base_manager_name = "objects"
|
|
|
|
|
|
class EventTicketAggregate(DBView):
|
|
event = models.OneToOneField(
|
|
EventExt,
|
|
on_delete=models.DO_NOTHING,
|
|
related_name="ticket_aggregates",
|
|
primary_key=True,
|
|
)
|
|
quantity = models.IntegerField()
|
|
amount = models.DecimalField(max_digits=13, decimal_places=4)
|
|
materials = models.DecimalField(max_digits=13, decimal_places=4)
|
|
amount_without_materials = models.DecimalField(max_digits=13, decimal_places=4)
|
|
instructor_revenue = models.DecimalField(max_digits=13, decimal_places=4)
|
|
instructor_amount = models.DecimalField(max_digits=13, decimal_places=4)
|
|
|
|
@staticmethod
|
|
def view_definition():
|
|
qs = EventTicketType.objects.values("event").annotate(
|
|
**{
|
|
field: Sum(field)
|
|
for field in [
|
|
"quantity",
|
|
"amount",
|
|
"materials",
|
|
"amount_without_materials",
|
|
"instructor_revenue",
|
|
"instructor_amount",
|
|
]
|
|
},
|
|
)
|
|
|
|
with connection.cursor() as cursor:
|
|
return cursor.mogrify(*qs.query.sql_with_params())
|
|
|
|
def __str__(self) -> str:
|
|
return f"{self.event}: {self.quantity}, {self.amount}"
|
|
|
|
class Meta:
|
|
managed = False
|
|
|
|
|
|
class EventAttendeeStats(DBView):
|
|
event = models.ForeignKey(
|
|
EventExt, on_delete=models.DO_NOTHING, related_name="attendee_stats"
|
|
)
|
|
gross_revenue = models.DecimalField(max_digits=13, decimal_places=4)
|
|
|
|
view_definition = """
|
|
SELECT eventext.event_ptr_id as event_id, SUM(usr.sum) as gross_revenue
|
|
FROM
|
|
membershipworks_eventext as eventext,
|
|
jsonb_to_recordset(eventext.details -> 'usr') AS usr (
|
|
sum NUMERIC
|
|
)
|
|
GROUP BY event_id
|
|
"""
|
|
|
|
class Meta:
|
|
managed = False
|
|
|
|
|
|
class EventAttendee(DBView):
|
|
event = models.ForeignKey(
|
|
EventExt, on_delete=models.DO_NOTHING, related_name="attendees"
|
|
)
|
|
uid = models.ForeignKey(Member, on_delete=models.DO_NOTHING)
|
|
name = models.CharField(max_length=256, db_column="nam")
|
|
email = models.CharField(max_length=256, db_column="eml")
|
|
sum = models.DecimalField(max_digits=13, decimal_places=4)
|
|
|
|
view_definition = """
|
|
SELECT eventext.event_ptr_id as event_id, usr.*
|
|
FROM
|
|
membershipworks_eventext AS eventext,
|
|
jsonb_to_recordset(eventext.details -> 'usr') AS usr (
|
|
uid TEXT,
|
|
nam TEXT,
|
|
eml TEXT,
|
|
sum NUMERIC
|
|
)
|
|
"""
|
|
|
|
class Meta:
|
|
managed = False
|