"""WS5.3 — Staff user-management UI (FRREP-UR010 + FRREP-UR011). The PRD requires a staff-facing user-management interface (not Django admin) covering: * UR010 — view / search / filter / edit / activate / deactivate / delete * UR011 — bulk CSV import with per-row error report and activation emails * non-partner registration review (the ``pending_admin_review`` flag added in WS5.1 — without a queue UI, the flag lights up but no one acts on it) All actions are gated on the ``core.manage_users`` permission (or staff / superuser status). Every state change is audit-logged via ``audit_rep`` so the audit trail can answer "who deactivated this account and when." """ from __future__ import annotations import csv import io import logging from typing import Iterable from django.conf import settings from django.contrib import messages from django.contrib.auth import get_user_model from django.contrib.auth.mixins import LoginRequiredMixin from django.contrib.auth.models import Group, Permission from django.core.exceptions import PermissionDenied from django.core.signing import TimestampSigner from django.db.models import ProtectedError, Q from django.shortcuts import get_object_or_404, redirect, render from django.urls import reverse from django.http import HttpResponse from django.views import View from django.views.generic import ListView from apps.core.lists import CsvExportMixin from apps.core.authentication.models import Institution from apps.core.notifications.models import Notification from apps.core.notifications.services import send_notification from apps.core.permissions.access import iilmp_role_groups from apps.core.permissions.roles import UserRole from apps.core.audit.rep_helpers import action_create, action_delete, action_update, audit_rep User = get_user_model() logger = logging.getLogger(__name__) signer = TimestampSigner() def _can_manage_users(user) -> bool: if not user.is_authenticated: return False if user.is_superuser or user.is_staff: return True return user.has_perm("core.manage_users") class StaffUserManageMixin(LoginRequiredMixin): """Permission gate for the user-management surface.""" def dispatch(self, request, *args, **kwargs): if not request.user.is_authenticated: return self.handle_no_permission() if not _can_manage_users(request.user): raise PermissionDenied return super().dispatch(request, *args, **kwargs) # ------------------------------------------------------------------ list view def _profile_attr(attr, default=""): """Return a callable that safely reads an attribute from user.profile.""" def _get(u): try: return getattr(u.profile, attr) or default except Exception: return default return _get def _profile_display(attr, default=""): """Return a callable that reads a display value (e.g. get_gender_display) from user.profile.""" def _get(u): try: return getattr(u.profile, f"get_{attr}_display")() or default except Exception: return default return _get class UserListView(StaffUserManageMixin, CsvExportMixin, ListView): template_name = "authentication/manage/user_list.html" context_object_name = "users" paginate_by = 25 csv_filename = "iilmp_users" csv_columns = ( # ── Identity ─────────────────────────────────────────── ("First name", "first_name"), ("Last name", "last_name"), ("Email", "email"), ("Phone", lambda u: u.phone or ""), # ── Platform ─────────────────────────────────────────── ("Role", lambda u: u.get_role_display() if hasattr(u, "get_role_display") else u.role or ""), ("Account status", lambda u: "Active" if u.is_active else "Inactive"), ("Email verified", lambda u: "Yes" if u.email_verified else "No"), ("Year joined", lambda u: u.date_joined.year if u.date_joined else ""), ("Month joined", lambda u: u.date_joined.strftime("%B") if u.date_joined else ""), ("Date joined", lambda u: u.date_joined.strftime("%Y-%m-%d") if u.date_joined else ""), ("Last login", lambda u: u.last_login.strftime("%Y-%m-%d") if u.last_login else "Never"), # ── Institution ──────────────────────────────────────── ("Institution", lambda u: u.institution.name if u.institution_id else u.non_partner_institution_name or ""), ("Institution country", lambda u: u.institution.country if u.institution_id else ""), ("Institution type", lambda u: u.institution.get_institution_type_display() if u.institution_id else ""), ("Institution region", lambda u: u.institution.region if u.institution_id else ""), ("RUFORUM member", lambda u: "Yes" if (u.institution_id and u.institution.is_member) else "No"), # ── Demographics (from profile) ──────────────────────── ("Gender", _profile_display("gender")), ("Country of residence", _profile_attr("country")), ("City", _profile_attr("city")), ("Date of birth", lambda u: (u.profile.date_of_birth.strftime("%Y-%m-%d") if u.profile.date_of_birth else "") if hasattr(u, "profile") else ""), ("Age", lambda u: str(u.profile.age_years) if hasattr(u, "profile") and u.profile.age_years is not None else ""), ("Is youth (≤35)", lambda u: "Yes" if (hasattr(u, "profile") and u.profile.is_youth) else ("No" if hasattr(u, "profile") and u.profile.is_youth is False else "")), # ── Academic / professional ──────────────────────────── ("Highest degree", _profile_display("degree_level")), ("Areas of interest", _profile_attr("areas_of_interest")), ("Professional background", _profile_attr("professional_background")), ("Preferred language", _profile_attr("preferred_language")), ("Timezone", _profile_attr("timezone")), ) def get_queryset(self): qs = User.objects.select_related("institution", "profile").order_by("-date_joined") params = self.request.GET q = (params.get("q") or "").strip() if q: qs = qs.filter( Q(email__icontains=q) | Q(first_name__icontains=q) | Q(last_name__icontains=q) | Q(non_partner_institution_name__icontains=q) ) role = (params.get("role") or "").strip() if role: qs = qs.filter(role=role) institution_id = (params.get("institution") or "").strip() if institution_id: qs = qs.filter(institution_id=institution_id) status = (params.get("status") or "").strip() if status == "active": qs = qs.filter(is_active=True, pending_admin_review=False) elif status == "inactive": qs = qs.filter(is_active=False) elif status == "unverified": qs = qs.filter(email_verified=False, is_active=True) elif status == "pending": qs = qs.filter(pending_admin_review=True) return qs def get_context_data(self, **kwargs): ctx = super().get_context_data(**kwargs) ctx["filter_q"] = self.request.GET.get("q", "") ctx["filter_role"] = self.request.GET.get("role", "") ctx["filter_status"] = self.request.GET.get("status", "") ctx["filter_institution"] = self.request.GET.get("institution", "") ctx["roles"] = UserRole.choices ctx["institutions"] = Institution.objects.filter(is_active=True).order_by("name") ctx["pending_count"] = User.objects.filter(pending_admin_review=True).count() ctx["unverified_count"] = User.objects.filter( email_verified=False, is_active=True ).count() return ctx # ----------------------------------------------------------- pending queue class PendingReviewQueueView(StaffUserManageMixin, ListView): """List of accounts flagged for review — typically non-partner registrations.""" template_name = "authentication/manage/user_pending_queue.html" context_object_name = "users" paginate_by = 25 def get_queryset(self): return ( User.objects.filter(pending_admin_review=True) .select_related("institution") .order_by("-date_joined") ) class ApprovePendingUserView(StaffUserManageMixin, View): """Approve a non-partner registration. Optionally promote the typed institution name into a real ``Institution`` row so future learners from the same place can pick it from the dropdown.""" def post(self, request, pk): user = get_object_or_404(User, pk=pk, pending_admin_review=True) promote = request.POST.get("promote_institution") == "on" institution_id = (request.POST.get("institution_id") or "").strip() if institution_id: inst = get_object_or_404(Institution, pk=institution_id) user.institution = inst user.non_partner_institution_name = "" elif promote and user.non_partner_institution_name: inst, _ = Institution.objects.get_or_create( name=user.non_partner_institution_name.strip(), defaults={"is_member": False, "is_active": True}, ) user.institution = inst user.non_partner_institution_name = "" user.pending_admin_review = False user.save( update_fields=["pending_admin_review", "institution", "non_partner_institution_name"] ) audit_rep( actor=request.user, action=action_update(), target_model="CustomUser", object_id=user.pk, object_repr=user.email, changes={ "pending_admin_review": False, "institution_id": user.institution_id, "promoted": promote, }, request=request, ) messages.success(request, f"Approved {user.email}.") return redirect("accounts:manage_pending_queue") class RejectPendingUserView(StaffUserManageMixin, View): """Reject a pending registration: deactivate the account, capture the reason in the audit trail. We do NOT delete the row so the email address stays reserved (no impersonation by a follow-up signup).""" def post(self, request, pk): user = get_object_or_404(User, pk=pk, pending_admin_review=True) reason = (request.POST.get("reason") or "").strip() if not reason: messages.error(request, "Rejection reason is required.") return redirect("accounts:manage_pending_queue") user.is_active = False user.pending_admin_review = False user.save(update_fields=["is_active", "pending_admin_review"]) audit_rep( actor=request.user, action=action_update(), target_model="CustomUser", object_id=user.pk, object_repr=user.email, changes={"is_active": False, "pending_admin_review": False, "reason": reason}, request=request, ) messages.success(request, f"Rejected {user.email}.") return redirect("accounts:manage_pending_queue") # ------------------------------------------------------------------ edit class UserEditView(StaffUserManageMixin, View): """WS5.3 UI follow-up — staff form to change role / institution / names / phone / email_verified on an existing account. Password is intentionally NOT editable — users own their own credentials. Email is treated as immutable here too: changing the email is rare and cascades through audit, notifications, and the email_verified state, so we route that through Django admin or a future dedicated flow. """ template_name = "authentication/manage/user_edit.html" _select_multi_attrs = { "class": "min-h-32 w-full rounded-lg border border-ruforum-border bg-white px-3 py-2 text-sm text-ruforum-text shadow-sm focus:border-ruforum-primary focus:outline-none focus:ring-2 focus:ring-ruforum-primary/[0.18]" } def _form_class(self): from django import forms as djforms class _UserEditForm(djforms.ModelForm): user_permissions = djforms.ModelMultipleChoiceField( queryset=Permission.objects.select_related("content_type").order_by( "content_type__app_label", "codename" ), required=False, widget=djforms.SelectMultiple(attrs=UserEditView._select_multi_attrs), label="Extra permissions (direct on user)", help_text="In addition to permissions inherited from the role group below.", ) additional_roles = djforms.ModelMultipleChoiceField( queryset=Group.objects.exclude(pk__in=iilmp_role_groups()).order_by("name"), required=False, widget=djforms.SelectMultiple(attrs=UserEditView._select_multi_attrs), label="Additional roles", help_text=( "Custom Groups created in Access control → Roles. Their permissions " "layer on top of the Role above, which still drives most permission logic." ), ) class Meta: model = User fields = ( "first_name", "last_name", "phone", "role", "institution", "email_verified", "user_permissions", ) def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.fields["institution"].queryset = Institution.objects.filter( is_active=True ).order_by("name") self.fields["institution"].required = False # Searchable Select2 on the two single-choice access fields. self.fields["role"].widget.attrs.update( {"data-s2": True, "data-s2-placeholder": "Search roles…"} ) self.fields["institution"].widget.attrs.update( {"data-s2": True, "data-s2-placeholder": "Search institutions…"} ) if self.instance.pk: self.fields["additional_roles"].initial = self.instance.groups.exclude( pk__in=iilmp_role_groups() ) return _UserEditForm def get(self, request, pk): user = get_object_or_404(User, pk=pk) form = self._form_class()(instance=user) return render(request, self.template_name, {"target": user, "form": form}) def post(self, request, pk): user = get_object_or_404(User, pk=pk) form = self._form_class()(request.POST, instance=user) if not form.is_valid(): return render(request, self.template_name, {"target": user, "form": form}) # Snapshot pre-save values so the audit captures real diffs, not just # which form fields were touched. _non_diffable_fields = {"user_permissions", "additional_roles"} before = { f: getattr(user, f) for f in form.changed_data if f not in _non_diffable_fields } form.save() enum_group_ids = iilmp_role_groups().values_list("pk", flat=True) user.groups.remove(*user.groups.exclude(pk__in=enum_group_ids)) if form.cleaned_data["additional_roles"]: user.groups.add(*form.cleaned_data["additional_roles"]) after = {f: getattr(user, f) for f in before} audit_rep( actor=request.user, action=action_update(), target_model="CustomUser", object_id=user.pk, object_repr=user.email, changes={ f: { "old": str(before[f]) if before[f] is not None else None, "new": str(after[f]) if after[f] is not None else None, } for f in before }, request=request, ) # FRREP-UR010 notifications: tell the target user that their record # changed. ROLE_CHANGED carries the from/to enum; PROFILE_UPDATED # fires when material identity fields (name / phone / institution) # changed. `email_verified` is admin-side housekeeping; skip it. try: if "role" in form.changed_data: send_notification( user, f"Your account role was changed from {before.get('role') or '—'} to {after.get('role') or '—'}.", verb=Notification.Verb.REP_ROLE_CHANGED, email_context={ "from_role": str(before.get("role") or ""), "to_role": str(after.get("role") or ""), "changed_by": getattr(request.user, "email", ""), }, ) material_changes = {f for f in form.changed_data if f in {"first_name", "last_name", "phone", "institution"}} if material_changes: send_notification( user, "An administrator updated your profile details.", verb=Notification.Verb.REP_PROFILE_UPDATED, email_context={ "changed_fields": sorted(material_changes), "changed_by": getattr(request.user, "email", ""), }, ) except Exception: logger.exception("UserEditView: notification dispatch failed for user=%s", user.pk) messages.success(request, f"Updated {user.email}.") return redirect("accounts:manage_users") # --------------------------------------------------- activate / deactivate class ActivateUserView(StaffUserManageMixin, View): def post(self, request, pk): user = get_object_or_404(User, pk=pk) user.is_active = True user.save(update_fields=["is_active"]) audit_rep( actor=request.user, action=action_update(), target_model="CustomUser", object_id=user.pk, object_repr=user.email, changes={"is_active": True}, request=request, ) messages.success(request, f"Activated {user.email}.") return redirect(request.POST.get("next") or "accounts:manage_users") class DeactivateUserView(StaffUserManageMixin, View): def post(self, request, pk): user = get_object_or_404(User, pk=pk) if user.is_superuser: messages.error(request, "Cannot deactivate a superuser from this UI.") return redirect("accounts:manage_users") if user.pk == request.user.pk: messages.error(request, "You cannot deactivate your own account.") return redirect("accounts:manage_users") user.is_active = False user.save(update_fields=["is_active"]) audit_rep( actor=request.user, action=action_update(), target_model="CustomUser", object_id=user.pk, object_repr=user.email, changes={"is_active": False}, request=request, ) messages.success(request, f"Deactivated {user.email}.") return redirect(request.POST.get("next") or "accounts:manage_users") class DeleteUserView(StaffUserManageMixin, View): """FRREP-UR010 — delete a user account. Same self/superuser guards as :class:`DeactivateUserView`. The platform has many ``on_delete=PROTECT`` relations hanging off user activity (grant applications, awards, repository authorship, etc.), so a hard delete can legitimately fail — that must never surface as a 500. We attempt ``user.delete()`` and, if Django's collector reports protected related objects, fall back to deactivating the account instead and tell the operator why the row itself couldn't be removed. """ def post(self, request, pk): user = get_object_or_404(User, pk=pk) if user.is_superuser: messages.error(request, "Cannot delete a superuser from this UI.") return redirect("accounts:manage_users") if user.pk == request.user.pk: messages.error(request, "You cannot delete your own account.") return redirect("accounts:manage_users") email = user.email pk_value = user.pk try: user.delete() except ProtectedError: user.is_active = False user.save(update_fields=["is_active"]) audit_rep( actor=request.user, action=action_update(), target_model="CustomUser", object_id=pk_value, object_repr=email, changes={"is_active": False, "delete_blocked": "protected related records"}, request=request, ) messages.warning( request, f"{email} has linked records (applications, awards, submissions, etc.) and " "can't be permanently deleted — the account was deactivated instead.", ) return redirect("accounts:manage_users") audit_rep( actor=request.user, action=action_delete(), target_model="CustomUser", object_id=pk_value, object_repr=email, changes={"deleted": True}, request=request, ) messages.success(request, f"Deleted {email}.") return redirect("accounts:manage_users") # ------------------------------------------------------------ bulk import _REQUIRED_CSV_HEADERS = {"email", "first_name", "last_name"} _OPTIONAL_CSV_HEADERS = {"phone", "institution", "role"} def _parse_csv(file_obj) -> tuple[list[dict], list[str]]: """Validate headers and return (row_dicts, header_errors). Lines starting with ``#`` are treated as comments and skipped — this lets the downloadable template ship with example rows that don't accidentally create users if the operator forgets to delete them. """ decoded = io.TextIOWrapper(file_obj, encoding="utf-8-sig", newline="") reader = csv.DictReader(decoded) if not reader.fieldnames: return [], ["CSV is empty or missing a header row."] header_set = {h.strip().lower() for h in reader.fieldnames} missing = _REQUIRED_CSV_HEADERS - header_set if missing: return [], [f"Missing required column(s): {', '.join(sorted(missing))}."] rows = [] for row in reader: first_value = next(iter(row.values()), "") or "" if first_value.lstrip().startswith("#"): continue rows.append({k.strip().lower(): (v or "").strip() for k, v in row.items()}) return rows, [] def _resolve_institution(name: str) -> Institution | None: if not name: return None return Institution.objects.filter(name__iexact=name.strip()).first() def _send_invitation(user, request) -> None: """Send an activation invite. Reuses the existing one-time-token flow from RegisterForm (TimestampSigner + verify_email URL).""" try: from django.core.mail import send_mail from django.template.loader import render_to_string token = signer.sign(str(user.pk)) verify_url = request.build_absolute_uri( reverse("accounts:verify_email", kwargs={"token": token}) ) plain = ( f"Hi {user.first_name or user.email},\n\n" "An administrator created an IILMP account for you. Open this link to set your " "password and verify your email:\n\n" f"{verify_url}\n\n" "If you weren't expecting this, ignore the message." ) try: html = render_to_string( "notifications/emails/generic.html", { "notification": None, "recipient_name": user.get_full_name() or user.email.split("@")[0], "headline": "An IILMP account has been created for you", "body": plain, "action_url": verify_url, "action_label": "Verify and set password", }, ) except Exception: html = None send_mail( subject="Welcome to RUFORUM IILMP — verify your account", message=plain, from_email=settings.DEFAULT_FROM_EMAIL, recipient_list=[user.email], html_message=html, fail_silently=True, ) except Exception: logger.exception("user.bulk_import: invite email failed for %s", user.email) class BulkImportTemplateView(StaffUserManageMixin, View): """WS5.3 UI follow-up — serve a blank, properly-headered CSV template so operators don't have to author the header row by hand.""" def get(self, request): from django.http import HttpResponse body = ( "email,first_name,last_name,phone,institution,role\n" # Two example rows, commented-out so they don't import. Operators # delete these lines after using them as a guide. "# example.learner@x.com,Adaeze,Eze,+234801XXXXXXX,Makerere University,learner\n" "# example.instructor@x.com,Bola,Adesanya,,,instructor\n" ) resp = HttpResponse(body, content_type="text/csv") resp["Content-Disposition"] = ( 'attachment; filename="iilmp-user-import-template.csv"' ) return resp class BulkImportView(StaffUserManageMixin, View): """CSV bulk import of user accounts (FRREP-UR011). Required columns: ``email``, ``first_name``, ``last_name``. Optional: ``phone``, ``institution`` (name match against existing rows), ``role`` (UserRole enum value; defaults to LEARNER). Per-row outcomes are returned to the operator as a summary so they can fix the source spreadsheet and re-run. """ template_name = "authentication/manage/user_bulk_import.html" def get(self, request): return render(request, self.template_name, {}) def post(self, request): upload = request.FILES.get("csv") if not upload: messages.error(request, "Choose a CSV file to upload.") return render(request, self.template_name, {}) if not (upload.name.lower().endswith(".csv")): messages.error(request, "File must have a .csv extension.") return render(request, self.template_name, {}) rows, header_errors = _parse_csv(upload) if header_errors: return render( request, self.template_name, {"errors": header_errors}, ) created: list[str] = [] skipped: list[dict] = [] for line_no, row in enumerate(rows, start=2): # line 1 is header email = (row.get("email") or "").strip().lower() if not email: skipped.append({"line": line_no, "email": "", "reason": "Missing email"}) continue if User.objects.filter(email__iexact=email).exists(): skipped.append({"line": line_no, "email": email, "reason": "Email already exists"}) continue role_in = (row.get("role") or "").strip().lower() if role_in and role_in not in {c.value for c in UserRole}: skipped.append({ "line": line_no, "email": email, "reason": f"Unknown role: {role_in}", }) continue inst = _resolve_institution(row.get("institution") or "") try: user = User.objects.create_user( email=email, password=None, # require activation first_name=row.get("first_name") or "", last_name=row.get("last_name") or "", phone=row.get("phone") or "", institution=inst, role=role_in or UserRole.LEARNER.value, ) except Exception as exc: # noqa: BLE001 skipped.append({"line": line_no, "email": email, "reason": str(exc)[:200]}) continue created.append(email) _send_invitation(user, request) if created: audit_rep( actor=request.user, action=action_create(), target_model="CustomUser", object_id="bulk", object_repr=f"bulk import: {len(created)} created", changes={"created_count": len(created), "skipped_count": len(skipped)}, request=request, ) # Notify the operator who ran the bulk import (FRREP-UR011 summary). try: send_notification( request.user, f"Bulk import complete — {len(created)} created, {len(skipped)} skipped.", verb=Notification.Verb.REP_BULK_IMPORT_COMPLETE, email_context={ "created_count": len(created), "skipped_count": len(skipped), "total_rows": len(rows), }, ) except Exception: logger.exception("BulkImportView: notification dispatch failed") return render( request, self.template_name, { "report": { "created": created, "skipped": skipped, "total": len(rows), } }, ) # Public helper so other modules can reuse the eligibility check (e.g. the # top-nav menu can hide the link for users who can't manage users). def can_manage_users(user) -> bool: return _can_manage_users(user)