Add enrollment preprocessing CLI tool

evap/evaluation/tools.py

@@ -97,7 +97,7 @@ def vote_end_datetime(vote_end_date):
     return date_to_datetime(vote_end_date) + datetime.timedelta(hours=24 + settings.EVALUATION_END_OFFSET_HOURS)
 
 
-def get_parameter_from_url_or_session(request, parameter, default=False):
+def get_parameter_from_url_or_session(request, parameter, default=False) -> bool:
     result = request.GET.get(parameter, None)
     if result is None:  # if no parameter is given take session value
         result = request.session.get(parameter, default)

evap/staff/tests/test_views.py

@@ -1,7 +1,9 @@
+import csv
 import datetime
 import os
 from abc import ABC, abstractmethod
 from io import BytesIO
+from itertools import cycle
 from typing import Literal
 from unittest.mock import PropertyMock, patch
 
@@ -529,6 +531,48 @@ def test_wrong_files_dont_crash(self):
         self.assertIn("An error happened when processing the file", reply)
 
 
+class TestUserExportView(WebTestStaffMode):
+    url = "/staff/user/export"
+
+    @classmethod
+    def setUpTestData(cls) -> None:
+        cls.manager = make_manager()
+        baker.make(UserProfile, _quantity=5, _fill_optional=["first_name_given", "last_name", "email", "title"])
+        baker.make(
+            UserProfile,
+            _quantity=5,
+            _fill_optional=["first_name_given", "last_name", "email", "title"],
+            is_active=False,
+        )
+        # the titles are not filled by baker
+        titles = cycle(("", "Some", "Custom", "Titles"))
+        for user in UserProfile.objects.iterator():
+            user.title = next(titles)
+            user.save()
+
+    def assertUsersExported(self, response, user_objects):
+        reader = csv.reader(response.text.strip().split("\n"))
+        # skip header
+        next(reader)
+        self.assertSetEqual({tuple(row) for row in reader}, user_objects)
+
+    def test_export_all(self):
+        user_objects = {
+            (user.title or "", user.last_name or "", user.first_name or "", user.email or "")
+            for user in UserProfile.objects.iterator()
+        }
+        response = self.app.get(self.url, user=self.manager)
+        self.assertUsersExported(response, user_objects)
+
+    def test_export_filtered(self):
+        user_objects = {
+            (user.title or "", user.last_name or "", user.first_name or "", user.email or "")
+            for user in UserProfile.objects.exclude(is_active=False).iterator()
+        }
+        response = self.app.get(self.url, {"filter_users": True}, user=self.manager)
+        self.assertUsersExported(response, user_objects)
+
+
 class TestUserImportView(WebTestStaffMode):
     url = "/staff/user/import"
 

evap/staff/urls.py

@@ -73,6 +73,7 @@
     path("user/", views.user_index, name="user_index"),
     path("user/create", views.user_create, name="user_create"),
     path("user/import", views.user_import, name="user_import"),
+    path("user/export", views.user_export, name="user_export"),
     path("user/<int:user_id>/edit", views.user_edit, name="user_edit"),
     path("user/list", views.user_list, name="user_list"),
     path("user/delete", views.user_delete, name="user_delete"),

evap/staff/views.py

@@ -1,9 +1,11 @@
+import codecs
 import csv
 import itertools
 from collections import OrderedDict, defaultdict, namedtuple
 from collections.abc import Container
 from dataclasses import dataclass
 from datetime import date, datetime
+from io import BytesIO
 from typing import Any, cast
 
 import openpyxl
@@ -15,7 +17,7 @@
 from django.dispatch import receiver
 from django.forms import formset_factory
 from django.forms.models import inlineformset_factory, modelformset_factory
-from django.http import Http404, HttpRequest, HttpResponse, HttpResponseBadRequest, HttpResponseRedirect
+from django.http import FileResponse, Http404, HttpRequest, HttpResponse, HttpResponseBadRequest, HttpResponseRedirect
 from django.shortcuts import get_object_or_404, redirect, render
 from django.urls import reverse
 from django.utils.html import format_html
@@ -2116,6 +2118,25 @@ def user_list(request):
     return render(request, "staff_user_list.html", {"users": users, "filter_users": filter_users})
 
 
+@manager_required
+def user_export(request):
+    filter_users = get_parameter_from_url_or_session(request, "filter_users")
+
+    users = UserProfile.objects.all()
+    if filter_users:
+        users = users.exclude(is_active=False)
+
+    io = BytesIO()
+    Writer = codecs.getwriter("utf-8")
+    writer = csv.writer(Writer(io))
+    row = (_("Title"), _("Last name"), _("First name"), _("Email"))
+    writer.writerow(row)
+    writer.writerows((user.title, user.last_name, user.first_name, user.email) for user in users)
+
+    io.seek(0)
+    return FileResponse(io, as_attachment=True, filename="exported_users.csv")
+
+
 @manager_required
 def user_create(request):
     form = UserForm(request.POST or None, instance=UserProfile())

pyproject.toml

@@ -15,7 +15,7 @@ extend-exclude = """\
 
 [tool.isort]
 profile = "black"
-src_paths = ["evap"]
+src_paths = ["evap", "tools"]
 line_length = 120
 skip_gitignore = true
 extend_skip_glob = ["**/migrations/*"]

tools/enrolment_preprocessor.py

@@ -0,0 +1,60 @@
+#!/usr/bin/env python3
+import csv
+import sys
+from argparse import ArgumentParser
+from dataclasses import dataclass
+
+from openpyxl import load_workbook
+from openpyxl.cell import Cell
+
+
+@dataclass
+class User:
+    title: str
+    last_name: str
+    first_name: str
+    email: str
+
+
+def fix_users(users: dict[str, User], title: Cell | None, last_name: Cell, first_name: Cell, email: Cell):
+    imported = User(last_name=last_name.value, first_name=first_name.value, email=email.value, title=title.value or "" if title else "")  # type: ignore[arg-type]  # if schema is correct, all values are strings
+    existing = users.setdefault(imported.email, imported)
+    if existing != imported:
+        print("There is a conflict in the user data.")
+        print(f"existing: {existing}.")
+        print(f"imported: {imported}.")
+        if input("Do you want to keep the existing user? (y/n)")[0].lower() == "n":
+            return
+
+        email.value = existing.email
+        last_name.value = existing.last_name
+        first_name.value = existing.first_name
+        if title:
+            title.value = existing.title
+
+
+if __name__ == "__main__":
+    args = sys.argv
+    if "python" in args[0]:
+        args = args[1:]
+    parser = ArgumentParser(description="Commandline tool to preprocess enrollment xlsx files.")
+    parser.add_argument(
+        "-ud", "--user-data", help="Path to a csv file containing an export of all existing users.", required=True
+    )
+    parser.add_argument(
+        "-ed", "--enrolment-data", help="Path to the enrolment data in xlsx format for import.", required=True
+    )
+    ns = parser.parse_args(args[1:])
+
+    workbook = load_workbook(ns.enrolment_data)
+    user_dict = {}
+    with open(ns.user_data, encoding="utf-8") as csvfile:
+        reader = csv.reader(csvfile)
+        for row in reader:
+            print(*row)
+            user_dict[row[0]] = User(*row)
+    for sheet_name in ["MA Belegungen", "BA Belegungen"]:
+        for row in workbook[sheet_name].iter_rows(min_row=2, min_col=2):
+            fix_users(user_dict, None, *row[:3])
+            fix_users(user_dict, *row[7:])
+    workbook.save(ns.enrolment_data)