cmsmanage/paperwork/api.py

from typing import Required, TypedDict

from django.db.models import Q, QuerySet

from rest_framework import routers, serializers, viewsets
from rest_framework.decorators import action
from rest_framework.response import Response

from membershipworks.models import Member

from .models import (
    Certification,
    CertificationDefinition,
    CertificationVersion,
    Department,
)


class DepartmentSerializer(serializers.HyperlinkedModelSerializer):
    class Meta:
        model = Department
        fields = ["name", "parent", "shop_lead_flag", "list_reply_to_address"]


class ListConfig(TypedDict, total=False):
    real_name: str
    subject_prefix: str
    reply_to_address: str


class MailingList(TypedDict, total=False):
    config: ListConfig
    moderators: set[str]
    members: Required[set[str]]


class DepartmentViewSet(viewsets.ModelViewSet):
    queryset: QuerySet[Department] = Department.objects.all()
    serializer_class = DepartmentSerializer

    @action(detail=False, methods=["get"])
    def mailing_lists(self, request, format=None):
        """
        Generate a mailing list for each department, containing all
        certified users for tools in that department or child departments
        """
        departments = self.queryset.prefetch_related(
            "children",
            "shop_lead_flag__members",
        )
        lists: dict[str, MailingList] = {}
        for department in departments.filter(has_mailing_list=True):
            if department.shop_lead_flag is not None:
                moderator_emails = {
                    member.volunteer_email if member.volunteer_email else member.email
                    for member in department.shop_lead_flag.members.all()
                }
            else:
                moderator_emails = set()

            active_certified_members = {
                member.sanitized_mailbox()
                for member in Member.objects.with_is_active().filter(
                    is_active=True,
                    certification__certification_version__definition__department=department,
                )
            }

            # list_name can only be None if has_mailing_list is False
            assert department.list_name is not None

            lists[department.list_name] = {
                "config": {
                    "real_name": department.list_name,
                    "subject_prefix": f"[CMS {department.name}] ",
                    "reply_to_address": department.list_reply_to_address,
                },
                "moderators": moderator_emails,
                "members": active_certified_members,
            }

        # Add child departments' members to their parents
        # TODO: this seems wildly inefficient
        def recurse_children(department):
            for child in department.children.all():
                recurse_children(child)

                lists[department.list_name]["members"] |= lists[child.list_name][
                    "members"
                ]

        for department in departments.filter(has_mailing_list=True):
            if department.parent_id is None:
                recurse_children(department)

        shopleads: dict[Member, list[Department]] = {}
        for department in departments.filter(shop_lead_flag__isnull=False):
            for member in department.shop_lead_flag.members.all():
                if member not in shopleads:
                    shopleads[member] = []
                shopleads[member].append(department)

        # Add members to the Shop Leads mailing list, but don't configure it
        lists["ShopLeads"] = {
            "members": {
                shoplead.sanitized_mailbox(use_volunteer=True)
                for shoplead, departments in shopleads.items()
            },
        }

        # TODO: this isn't really in the domain of the `paperwork` app...
        deskers = (
            Member.objects.with_is_active()
            .filter(is_active=True)
            .filter(
                Member.objects.has_flag("label", "Volunteer: Desker")
                | Q(billing_method__startswith="Desker")
            )
        )
        lists["Deskers"] = {
            "members": {
                desker.sanitized_mailbox(use_volunteer=True) for desker in deskers
            }
        }

        return Response(lists)


class CertificationDefinitionSerializer(serializers.HyperlinkedModelSerializer):
    class Meta:
        model = CertificationDefinition
        fields = ["name", "department"]


class CertificationDefinitionViewSet(viewsets.ModelViewSet):
    queryset = CertificationDefinition.objects.all()
    serializer_class = CertificationDefinitionSerializer


class CertificationVersionSerializer(serializers.HyperlinkedModelSerializer):
    class Meta:
        model = CertificationVersion
        fields = ["definition", "version"]


class CertificationVersionViewSet(viewsets.ModelViewSet):
    queryset = CertificationVersion.objects.all()
    serializer_class = CertificationVersionSerializer


class CertificationSerializer(serializers.HyperlinkedModelSerializer):
    class Meta:
        model = Certification
        fields = [
            "name",
            "member",
            "certification_version",
            "certified_by",
            "date",
            "shop_lead_notified",
            "notes",
        ]


class CertificationViewSet(viewsets.ModelViewSet):
    queryset = Certification.objects.all()
    serializer_class = CertificationSerializer


router = routers.DefaultRouter()
router.register(r"paperwork/department", DepartmentViewSet)
router.register(r"paperwork/certification_definition", CertificationDefinitionViewSet)
router.register(r"paperwork/certification_version", CertificationVersionViewSet)
router.register(r"paperwork/certification", CertificationViewSet)