cmsmanage/paperwork/certification_emails.py

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)