Show paid status in admin and filter CSV exports by year (#30)

## Summary
Closes #30.
- Green check on membership/person admin when a current-year payment email matches
- Year picker on membership and payment CSV downloads
- Payments admin search by email, plus year list filters

schasite/admin.py

@@ -1,4 +1,13 @@
 from django.contrib import admin
+from django.db.models import Exists, OuterRef, Q
+from django.http import HttpResponse
+from django.shortcuts import render
+from django.urls import reverse
+from django.utils import timezone
+from datetime import datetime
+import csv
+import io
+
 from .models import (
     UsefulLinks,
     Membership,
@@ -11,11 +20,116 @@ from .models import (
     Payments,
     SCHAOfficer,
 )
-from django.http import HttpResponse
-from datetime import datetime
 from .forms import PaymentImport
 
-# Register your models here.
+
+def _current_year():
+    return timezone.now().year
+
+
+def person_payment_exists(year=None):
+    """True when this MembershipPerson has a payment in `year` (email or FK)."""
+    year = _current_year() if year is None else year
+    email_match = (
+        Q(email__iexact=OuterRef("email")) & Q(email__isnull=False) & ~Q(email="")
+    )
+    return Exists(
+        Payments.objects.filter(date__year=year).filter(
+            Q(person_id=OuterRef("pk")) | email_match
+        )
+    )
+
+
+def membership_payment_exists(year=None):
+    """True when any person on this membership has a payment in `year`."""
+    return Exists(
+        MembershipPerson.objects.filter(membership_id=OuterRef("pk")).filter(
+            person_payment_exists(year)
+        )
+    )
+
+
+def _available_years(model, field_name):
+    years = {d.year for d in model.objects.dates(field_name, "year")}
+    years.add(_current_year())
+    return sorted(years, reverse=True)
+
+
+def _apply_year_filter(queryset, field_name, year):
+    if not year or year == "all":
+        return queryset
+    try:
+        year_int = int(year)
+    except (TypeError, ValueError):
+        return queryset
+    return queryset.filter(**{f"{field_name}__year": year_int})
+
+
+def _csv_response(filename_suffix, header, rows):
+    buf = io.StringIO()
+    writer = csv.writer(buf)
+    writer.writerow(header)
+    writer.writerows(rows)
+    filename = datetime.now().strftime("%Y_%m_%d_%H_%M_%S") + filename_suffix
+    response = HttpResponse(buf.getvalue(), content_type="text/csv")
+    response["Content-Disposition"] = f"attachment; filename={filename}"
+    return response
+
+
+def _year_suffix(year):
+    if year and year != "all":
+        return f"_{year}"
+    return ""
+
+
+def _csv_year_form_response(
+    modeladmin, request, *, action_name, year_field, help_text
+):
+    opts = modeladmin.model._meta
+    return render(
+        request,
+        "admin/csv_year_export.html",
+        {
+            "opts": opts,
+            "title": "Download CSV",
+            "action_name": action_name,
+            "years": _available_years(modeladmin.model, year_field),
+            "default_year": _current_year(),
+            "help_text": help_text,
+            "selected_ids": request.POST.getlist("_selected_action"),
+            "select_across": request.POST.get("select_across", "0"),
+            "index": request.POST.get("index", "0"),
+            "changelist_url": reverse(
+                f"admin:{opts.app_label}_{opts.model_name}_changelist"
+            ),
+        },
+    )
+
+
+class CreatedYearListFilter(admin.SimpleListFilter):
+    title = "year"
+    parameter_name = "year"
+
+    def lookups(self, request, model_admin):
+        return [(year, str(year)) for year in _available_years(model_admin.model, "created")]
+
+    def queryset(self, request, queryset):
+        if self.value():
+            return queryset.filter(created__year=self.value())
+        return queryset
+
+
+class PaymentYearListFilter(admin.SimpleListFilter):
+    title = "year"
+    parameter_name = "year"
+
+    def lookups(self, request, model_admin):
+        return [(year, str(year)) for year in _available_years(model_admin.model, "date")]
+
+    def queryset(self, request, queryset):
+        if self.value():
+            return queryset.filter(date__year=self.value())
+        return queryset
 
 
 class UsefulLinksAdmin(admin.ModelAdmin):
@@ -31,7 +145,17 @@ class MembershipAddressInline(admin.TabularInline):
 class MembershipPersonInline(admin.TabularInline):
     model = MembershipPerson
     extra = 1
-    readonly_fields = ("id",)
+    readonly_fields = ("id", "has_payment")
+
+    def get_queryset(self, request):
+        qs = super().get_queryset(request)
+        return qs.annotate(_has_payment=person_payment_exists())
+
+    @admin.display(description="Paid this year", boolean=True)
+    def has_payment(self, obj):
+        if not obj.pk:
+            return False
+        return bool(getattr(obj, "_has_payment", False))
 
 
 class MembershipCommiteeInline(admin.TabularInline):
@@ -47,101 +171,104 @@ class MembershipServicesInline(admin.TabularInline):
 
 
 def download_csv_by_members(modelAdmin, request, queryset):
-    import csv
+    if "apply" not in request.POST:
-    import io as StringIO
+        return _csv_year_form_response(
-
+            modelAdmin,
-    def stream_csv(queryset):
+            request,
-        csvfile = StringIO.StringIO()
+            action_name="download_csv_by_members",
-        writer = csv.writer(csvfile)
+            year_field="created",
-        writer.writerow(
+            help_text="Choose a year to include. Memberships are filtered by the year the record was created.",
-            [
-                "address_1",
-                "city",
-                "state",
-                "zip_code",
-                "first_name",
-                "last_name",
-                "email",
-                "phone_number",
-            ]
-        )
-        for q in queryset:
-            people = [
-                item for item in MembershipPerson.objects.filter(membership_id=q.id)
-            ]
-            for person in people:
-                writer.writerow(
-                    [
-                        q.addressmodel1.address_1,
-                        q.addressmodel1.city,
-                        q.addressmodel1.state,
-                        q.addressmodel1.zip_code,
-                        person.first_name,
-                        person.last_name,
-                        person.email,
-                        person.phone_number,
-                    ]
-                )
-                yield csvfile.getvalue()
-
-    now = datetime.now()
-    filename = now.strftime("%Y_%m_%d_%H_%M_%S") + "_scha_member_by_member.csv"
-    response = HttpResponse(stream_csv(queryset), content_type="text/csv")
-    response["Content-Disposition"] = "attachment; filename={}".format(filename)
-    return response
-
-
-def download_csv_by_address(modeladmin, request, queryset):
-    import csv
-    import io as StringIO
-
-    def stream_csv(queryset):
-        csvfile = StringIO.StringIO()
-        writer = csv.writer(csvfile)
-        writer.writerow(
-            [
-                "address_1",
-                "city",
-                "state",
-                "zip_code",
-                "person_1_email",
-                "person_1_phone",
-                "person_1_first_name",
-                "person_1_last_name",
-                "person_2_email",
-                "person_2_phone",
-                "person_2_first_name",
-                "person_2_last_name",
-            ]
         )
 
-        for q in queryset:
+    year = request.POST.get("year", "all")
-            people = [
+    queryset = _apply_year_filter(queryset, "created", year)
-                item for item in MembershipPerson.objects.filter(membership_id=q.id)
+    rows = []
-            ]
+    for q in queryset:
-            writer.writerow(
+        people = MembershipPerson.objects.filter(membership_id=q.id)
+        for person in people:
+            rows.append(
                 [
                     q.addressmodel1.address_1,
                     q.addressmodel1.city,
                     q.addressmodel1.state,
                     q.addressmodel1.zip_code,
-                    people[0].email if len(people) > 0 else "",
+                    person.first_name,
-                    people[0].phone_number if len(people) > 0 else "",
+                    person.last_name,
-                    people[0].first_name if len(people) > 0 else "",
+                    person.email,
-                    people[0].last_name if len(people) > 0 else "",
+                    person.phone_number,
-                    people[1].email if len(people) > 1 else "",
-                    people[1].phone_number if len(people) > 1 else "",
-                    people[1].first_name if len(people) > 1 else "",
-                    people[1].last_name if len(people) > 1 else "",
                 ]
             )
-            yield csvfile.getvalue()
+    return _csv_response(
+        f"_scha_member_by_member{_year_suffix(year)}.csv",
+        [
+            "address_1",
+            "city",
+            "state",
+            "zip_code",
+            "first_name",
+            "last_name",
+            "email",
+            "phone_number",
+        ],
+        rows,
+    )
 
-    now = datetime.now()
+
-    filename = now.strftime("%Y_%m_%d_%H_%M_%S") + "_scha_member_by_address.csv"
+download_csv_by_members.short_description = "Download CSV by member"
-    response = HttpResponse(stream_csv(queryset), content_type="text/csv")
+
-    response["Content-Disposition"] = "attachment; filename={}".format(filename)
+
-    return response
+def download_csv_by_address(modeladmin, request, queryset):
+    if "apply" not in request.POST:
+        return _csv_year_form_response(
+            modeladmin,
+            request,
+            action_name="download_csv_by_address",
+            year_field="created",
+            help_text="Choose a year to include. Memberships are filtered by the year the record was created.",
+        )
+
+    year = request.POST.get("year", "all")
+    queryset = _apply_year_filter(queryset, "created", year)
+    rows = []
+    for q in queryset:
+        people = list(MembershipPerson.objects.filter(membership_id=q.id))
+        rows.append(
+            [
+                q.addressmodel1.address_1,
+                q.addressmodel1.city,
+                q.addressmodel1.state,
+                q.addressmodel1.zip_code,
+                people[0].email if len(people) > 0 else "",
+                people[0].phone_number if len(people) > 0 else "",
+                people[0].first_name if len(people) > 0 else "",
+                people[0].last_name if len(people) > 0 else "",
+                people[1].email if len(people) > 1 else "",
+                people[1].phone_number if len(people) > 1 else "",
+                people[1].first_name if len(people) > 1 else "",
+                people[1].last_name if len(people) > 1 else "",
+            ]
+        )
+    return _csv_response(
+        f"_scha_member_by_address{_year_suffix(year)}.csv",
+        [
+            "address_1",
+            "city",
+            "state",
+            "zip_code",
+            "person_1_email",
+            "person_1_phone",
+            "person_1_first_name",
+            "person_1_last_name",
+            "person_2_email",
+            "person_2_phone",
+            "person_2_first_name",
+            "person_2_last_name",
+        ],
+        rows,
+    )
+
+
+download_csv_by_address.short_description = "Download CSV by address"
 
 
 class MembershipAdmin(admin.ModelAdmin):
@@ -151,8 +278,18 @@ class MembershipAdmin(admin.ModelAdmin):
         MembershipCommiteeInline,
         MembershipServicesInline,
     ]
+    list_display = ["get_address_str", "get_person_1", "has_payment"]
+    list_filter = [CreatedYearListFilter]
     actions = [download_csv_by_address, download_csv_by_members]
 
+    def get_queryset(self, request):
+        qs = super().get_queryset(request)
+        return qs.annotate(_has_payment=membership_payment_exists())
+
+    @admin.display(description="Paid this year", boolean=True, ordering="_has_payment")
+    def has_payment(self, obj):
+        return bool(getattr(obj, "_has_payment", False))
+
 
 class CalendarEventAddressInline(admin.TabularInline):
     model = CalendarEventAddressModel
@@ -189,7 +326,17 @@ class AddressModelAdmin(admin.ModelAdmin):
 
 
 class MembershipPersonAdmin(admin.ModelAdmin):
-    pass
+    list_display = ["first_name", "last_name", "email", "phone_number", "has_payment"]
+    search_fields = ["email", "first_name", "last_name"]
+    list_filter = [CreatedYearListFilter]
+
+    def get_queryset(self, request):
+        qs = super().get_queryset(request)
+        return qs.annotate(_has_payment=person_payment_exists())
+
+    @admin.display(description="Paid this year", boolean=True, ordering="_has_payment")
+    def has_payment(self, obj):
+        return bool(getattr(obj, "_has_payment", False))
 
 
 class MembershipCommitteeAdmin(admin.ModelAdmin):
@@ -205,51 +352,58 @@ class CalendarEventAddressModelAdmin(admin.ModelAdmin):
 
 
 def download_payments(modelAdmin, request, queryset):
-    import csv
+    if "apply" not in request.POST:
-    import io as StringIO
+        return _csv_year_form_response(
-
+            modelAdmin,
-    def stream_payment_csv(queryset):
+            request,
-        csvfile = StringIO.StringIO()
+            action_name="download_payments",
-        writer = csv.writer(csvfile)
+            year_field="date",
-        writer.writerow(
+            help_text="Choose a year to include. Payments are filtered by payment date.",
-            ["email", "date", "status", "first_name", "last_name", "phone_number"]
         )
-        for q in queryset:
-            first_name = ""
-            last_name = ""
-            phone_number = ""
-            if q.person:
-                first_name = q.person.first_name if q.person.first_name else ""
-                last_name = q.person.last_name if q.person.last_name else ""
-                phone_number = q.person.phone_number if q.person.phone_number else ""
-            writer.writerow(
-                [
-                    q.email,
-                    q.date,
-                    q.status,
-                    first_name,
-                    last_name,
-                    phone_number,
-                ]
-            )
-            yield csvfile.getvalue()
 
-    now = datetime.now()
+    year = request.POST.get("year", "all")
-    filename = now.strftime("%Y_%m_%d_%H_%M_%S") + "_scha_payments_by_member.csv"
+    queryset = _apply_year_filter(queryset, "date", year)
-    response = HttpResponse(stream_payment_csv(queryset), content_type="text/csv")
+    rows = []
-    response["Content-Disposition"] = "attachment; filename={}".format(filename)
+    for q in queryset:
-    return response
+        first_name = ""
+        last_name = ""
+        phone_number = ""
+        if q.person:
+            first_name = q.person.first_name if q.person.first_name else ""
+            last_name = q.person.last_name if q.person.last_name else ""
+            phone_number = q.person.phone_number if q.person.phone_number else ""
+        rows.append(
+            [
+                q.email,
+                q.date,
+                q.status,
+                first_name,
+                last_name,
+                phone_number,
+            ]
+        )
+    return _csv_response(
+        f"_scha_payments_by_member{_year_suffix(year)}.csv",
+        ["email", "date", "status", "first_name", "last_name", "phone_number"],
+        rows,
+    )
+
+
+download_payments.short_description = "Download payments CSV"
 
 
 class PaymentsAdmin(admin.ModelAdmin):
     list_display = ["date", "status", "email"]
     search_fields = ["email"]
+    list_filter = [PaymentYearListFilter]
     actions = [download_payments]
     form = PaymentImport
 
+
 class SCHAOfficerAdmin(admin.ModelAdmin):
     list_display = ["position", "name", "email"]
 
+
 admin.site.register(UsefulLinks, UsefulLinksAdmin)
 admin.site.register(Membership, MembershipAdmin)
 admin.site.register(CalendarEvent, CalendarEventAdmin)
@@ -260,4 +414,3 @@ admin.site.register(MembershipServices, MembershipServicesAdmin)
 admin.site.register(CalendarEventAddressModel, CalendarEventAddressModelAdmin)
 admin.site.register(Payments, PaymentsAdmin)
 admin.site.register(SCHAOfficer, SCHAOfficerAdmin)
-

schasite/templates/admin/csv_year_export.html

@@ -0,0 +1,29 @@
+{% extends "admin/base_site.html" %}
+{% load i18n %}
+
+{% block content %}
+<form action="" method="post">{% csrf_token %}
+  <p>{{ help_text }}</p>
+  <fieldset class="module aligned">
+    <div class="form-row">
+      <label for="id_year">Year:</label>
+      <select name="year" id="id_year">
+        <option value="all">All years</option>
+        {% for year in years %}
+        <option value="{{ year }}"{% if year == default_year %} selected{% endif %}>{{ year }}</option>
+        {% endfor %}
+      </select>
+    </div>
+  </fieldset>
+  {% for selected_id in selected_ids %}
+  <input type="hidden" name="_selected_action" value="{{ selected_id }}">
+  {% endfor %}
+  <input type="hidden" name="action" value="{{ action_name }}">
+  <input type="hidden" name="select_across" value="{{ select_across }}">
+  <input type="hidden" name="index" value="{{ index }}">
+  <div class="submit-row">
+    <input type="submit" name="apply" class="default" value="Download CSV">
+    <a href="{{ changelist_url }}" class="button cancel-link">{% translate "Cancel" %}</a>
+  </div>
+</form>
+{% endblock %}

schasite/tests.py

@@ -6,12 +6,17 @@ from django.urls import reverse
 
 from scha.settings.base import build_csrf_trusted_origins
 
+from django.contrib.auth.models import User
+from django.utils import timezone
+
+from .admin import membership_payment_exists, person_payment_exists
 from .forms import AddressForm, PeopleForm
 from .models import (
     AddressModel1,
     CalendarEvent,
     Membership,
     MembershipPerson,
+    Payments,
     SCHAOfficer,
     UsefulLinks,
 )
@@ -469,3 +474,183 @@ class AuthViewTests(TestCase):
         response = self.views.member_posts(request)
         self.assertEqual(response.status_code, 302)
         self.assertIn("/login", response.url)
+
+
+def _membership_with_person(email="alex@example.com"):
+    membership = Membership.objects.create()
+    AddressModel1.objects.create(
+        membership=membership,
+        address_1="123 Main St",
+        city="Wheaton",
+        state="IL",
+        zip_code="60189",
+    )
+    person = MembershipPerson.objects.create(
+        membership=membership,
+        first_name="Alex",
+        last_name="Resident",
+        email=email,
+        phone_number="+16305559876",
+    )
+    return membership, person
+
+
+class AdminPaidIndicatorTests(TestCase):
+    def setUp(self):
+        self.client = Client()
+        self.admin_user = User.objects.create_superuser(
+            "admin", "admin@example.com", "pass"
+        )
+        self.client.force_login(self.admin_user)
+        self.this_year = timezone.now().year
+        self.membership, self.person = _membership_with_person("paid@example.com")
+
+    def test_person_marked_paid_when_email_matches_current_year_payment(self):
+        Payments.objects.create(
+            email="PAID@example.com",
+            date=datetime.date(self.this_year, 3, 1),
+        )
+
+        paid = MembershipPerson.objects.annotate(
+            _has_payment=person_payment_exists()
+        ).get(pk=self.person.pk)
+        self.assertTrue(paid._has_payment)
+
+        unpaid_membership, unpaid_person = _membership_with_person("unpaid@example.com")
+        unpaid = MembershipPerson.objects.annotate(
+            _has_payment=person_payment_exists()
+        ).get(pk=unpaid_person.pk)
+        self.assertFalse(unpaid._has_payment)
+
+        membership = Membership.objects.annotate(
+            _has_payment=membership_payment_exists()
+        ).get(pk=self.membership.pk)
+        self.assertTrue(membership._has_payment)
+        other = Membership.objects.annotate(
+            _has_payment=membership_payment_exists()
+        ).get(pk=unpaid_membership.pk)
+        self.assertFalse(other._has_payment)
+
+    def test_prior_year_payment_does_not_count_as_paid(self):
+        Payments.objects.create(
+            email="paid@example.com",
+            date=datetime.date(self.this_year - 1, 6, 1),
+        )
+        person = MembershipPerson.objects.annotate(
+            _has_payment=person_payment_exists()
+        ).get(pk=self.person.pk)
+        self.assertFalse(person._has_payment)
+
+    def test_person_changelist_shows_paid_icon(self):
+        Payments.objects.create(
+            email="paid@example.com",
+            date=datetime.date(self.this_year, 4, 15),
+        )
+        url = reverse("admin:schasite_membershipperson_changelist")
+        response = self.client.get(url)
+        self.assertEqual(response.status_code, 200)
+        self.assertContains(response, "Paid this year")
+        self.assertContains(response, "icon-yes.svg")
+
+
+class AdminPaymentSearchTests(TestCase):
+    def setUp(self):
+        self.client = Client()
+        self.admin_user = User.objects.create_superuser(
+            "admin", "admin@example.com", "pass"
+        )
+        self.client.force_login(self.admin_user)
+        Payments.objects.create(email="dues@example.com", date=datetime.date(2026, 1, 10))
+        Payments.objects.create(email="other@example.com", date=datetime.date(2026, 2, 10))
+
+    def test_payments_admin_search_by_email(self):
+        url = reverse("admin:schasite_payments_changelist")
+        response = self.client.get(url, {"q": "dues@example.com"})
+        self.assertEqual(response.status_code, 200)
+        self.assertContains(response, "dues@example.com")
+        self.assertNotContains(response, "other@example.com")
+
+
+class AdminCsvYearFilterTests(TestCase):
+    def setUp(self):
+        self.client = Client()
+        self.admin_user = User.objects.create_superuser(
+            "admin", "admin@example.com", "pass"
+        )
+        self.client.force_login(self.admin_user)
+
+        self.current, self.current_person = _membership_with_person("now@example.com")
+        self.old, self.old_person = _membership_with_person("old@example.com")
+        Membership.objects.filter(pk=self.old.pk).update(
+            created=timezone.make_aware(datetime.datetime(2024, 5, 1, 12, 0, 0))
+        )
+
+        Payments.objects.create(email="now@example.com", date=datetime.date(2026, 3, 1))
+        Payments.objects.create(email="old@example.com", date=datetime.date(2024, 3, 1))
+
+    def test_membership_csv_prompts_for_year(self):
+        url = reverse("admin:schasite_membership_changelist")
+        response = self.client.post(
+            url,
+            {
+                "action": "download_csv_by_members",
+                "_selected_action": [str(self.current.pk), str(self.old.pk)],
+                "index": "0",
+            },
+        )
+        self.assertEqual(response.status_code, 200)
+        self.assertContains(response, "Year:")
+        self.assertContains(response, "Download CSV")
+
+    def test_membership_csv_filters_by_created_year(self):
+        url = reverse("admin:schasite_membership_changelist")
+        response = self.client.post(
+            url,
+            {
+                "action": "download_csv_by_members",
+                "_selected_action": [str(self.current.pk), str(self.old.pk)],
+                "index": "0",
+                "apply": "Download CSV",
+                "year": "2024",
+            },
+        )
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(response["Content-Type"], "text/csv")
+        content = response.content.decode()
+        self.assertIn("old@example.com", content)
+        self.assertNotIn("now@example.com", content)
+
+    def test_membership_csv_by_address_filters_by_created_year(self):
+        url = reverse("admin:schasite_membership_changelist")
+        response = self.client.post(
+            url,
+            {
+                "action": "download_csv_by_address",
+                "_selected_action": [str(self.current.pk), str(self.old.pk)],
+                "index": "0",
+                "apply": "Download CSV",
+                "year": "2024",
+            },
+        )
+        self.assertEqual(response.status_code, 200)
+        content = response.content.decode()
+        self.assertIn("old@example.com", content)
+        self.assertNotIn("now@example.com", content)
+
+    def test_payments_csv_filters_by_payment_year(self):
+        url = reverse("admin:schasite_payments_changelist")
+        payments = list(Payments.objects.values_list("pk", flat=True))
+        response = self.client.post(
+            url,
+            {
+                "action": "download_payments",
+                "_selected_action": [str(pk) for pk in payments],
+                "index": "0",
+                "apply": "Download CSV",
+                "year": "2024",
+            },
+        )
+        self.assertEqual(response.status_code, 200)
+        content = response.content.decode()
+        self.assertIn("old@example.com", content)
+        self.assertNotIn("now@example.com", content)