134 lines
4.8 KiB
Python
134 lines
4.8 KiB
Python
from abc import abstractmethod
|
|
from collections.abc import Iterable, Iterator
|
|
from itertools import groupby
|
|
|
|
from django.contrib.auth import get_user_model
|
|
from django.core import mail
|
|
from django.core.mail.message import sanitize_address
|
|
from django.db.models import QuerySet
|
|
|
|
from cmsmanage.email import TemplatedMultipartEmail
|
|
from membershipworks.models import Member
|
|
from paperwork.models import Certification, Department
|
|
|
|
|
|
class CertificationEmailBase(TemplatedMultipartEmail):
|
|
from_email = "Claremont MakerSpace Member Certification System <Certifications@ClaremontMakerSpace.org>"
|
|
reply_to = ["Claremont MakerSpace <Info@ClaremontMakerSpace.org>"]
|
|
|
|
@classmethod
|
|
@abstractmethod
|
|
def render_for_certifications(
|
|
cls, ordered_certifications: QuerySet[Certification]
|
|
) -> Iterable[mail.EmailMessage]:
|
|
raise NotImplementedError
|
|
|
|
|
|
class DepartmentCertificationEmail(CertificationEmailBase):
|
|
template = "paperwork/email/department_certifications.dj.html"
|
|
|
|
@property
|
|
def subject(self) -> str:
|
|
certification_count = len(self.context["certifications"])
|
|
return f"{certification_count} new CMS Certifications issued for {self.context['department']}"
|
|
|
|
@classmethod
|
|
def render_for_department(
|
|
cls, department: Department, certifications: Iterator[Certification]
|
|
) -> Iterable[mail.EmailMessage]:
|
|
if department.shop_lead_flag is not None:
|
|
shop_leads = department.shop_lead_flag.members.values_list(
|
|
"first_name", "account_name", "email", named=True
|
|
)
|
|
yield cls.render(
|
|
{
|
|
"shop_lead_names": [
|
|
shop_lead.first_name for shop_lead in shop_leads
|
|
],
|
|
"department": department,
|
|
"certifications": list(certifications),
|
|
},
|
|
to=[
|
|
sanitize_address((shop_lead.account_name, shop_lead.email), "ascii")
|
|
for shop_lead in shop_leads
|
|
],
|
|
)
|
|
|
|
@classmethod
|
|
def render_for_certifications(
|
|
cls, ordered_certifications: QuerySet[Certification]
|
|
) -> Iterable[mail.EmailMessage]:
|
|
certifications_by_department = groupby(
|
|
ordered_certifications,
|
|
lambda c: c.certification_version.definition.department,
|
|
)
|
|
|
|
for department, certifications in certifications_by_department:
|
|
yield from cls.render_for_department(department, certifications)
|
|
|
|
|
|
class MemberCertificationEmail(CertificationEmailBase):
|
|
template = "paperwork/email/member_certifications.dj.html"
|
|
|
|
@property
|
|
def subject(self) -> str:
|
|
return f"You have been issued {len(self.context['certifications'])} new CMS Certifications"
|
|
|
|
@classmethod
|
|
def render_for_member(
|
|
cls, member: Member, certifications: list[Certification]
|
|
) -> mail.EmailMessage:
|
|
return cls.render(
|
|
{"member": member, "certifications": certifications},
|
|
to=[sanitize_address((member.account_name, member.email), "ascii")],
|
|
)
|
|
|
|
@classmethod
|
|
def render_for_certifications(
|
|
cls, ordered_certifications: QuerySet[Certification]
|
|
) -> Iterable[mail.EmailMessage]:
|
|
certifications_by_member = groupby(
|
|
ordered_certifications.filter(member__isnull=False), lambda c: c.member
|
|
)
|
|
|
|
for member, certifications in certifications_by_member:
|
|
yield cls.render_for_member(member, list(certifications))
|
|
|
|
|
|
class AdminCertificationEmail(CertificationEmailBase):
|
|
template = "paperwork/email/admin_certifications.dj.html"
|
|
|
|
@property
|
|
def subject(self) -> str:
|
|
return f"{len(self.context['certifications'])} new CMS Certifications issued"
|
|
|
|
@classmethod
|
|
def render_for_certifications(
|
|
cls, ordered_certifications: QuerySet[Certification]
|
|
) -> Iterable[mail.EmailMessage]:
|
|
yield cls.render(
|
|
{"certifications": ordered_certifications},
|
|
to=get_user_model()
|
|
.objects.with_perm(
|
|
"paperwork.receive_certification_emails",
|
|
include_superusers=False,
|
|
# TODO: backend should not be hardcoded
|
|
# TODO: LDAPBackend does not support with_perm() directly
|
|
backend="django.contrib.auth.backends.ModelBackend",
|
|
)
|
|
.values_list("email", flat=True),
|
|
)
|
|
|
|
|
|
def all_certification_emails(queryset: QuerySet[Certification]):
|
|
ordered_queryset = queryset.select_related(
|
|
"certification_version__definition"
|
|
).order_by("certification_version__definition__department")
|
|
|
|
for email_type in [
|
|
DepartmentCertificationEmail,
|
|
MemberCertificationEmail,
|
|
AdminCertificationEmail,
|
|
]:
|
|
yield from email_type.render_for_certifications(ordered_queryset)
|