from __future__ import annotations from typing import Any, Dict, List from django import forms from django.contrib.auth.models import Group, Permission from django.core.paginator import Paginator from django.db.models import Q from django.http import HttpRequest, HttpResponse, JsonResponse from django.shortcuts import redirect from django.views import View from django.views.generic.detail import SingleObjectMixin from django.views.generic.edit import FormMixin from django.views.generic.list import MultipleObjectMixin from inertia import render from labs.models import ( CertificateIssuance, CertificateRecipient, CreditTransaction, User, UserCreditDetail, UserCreditSummary, UserEvent, ) from .base import PermissionRequiredAdminMixin class UserEditForm(forms.ModelForm): """Form for editing user details.""" # Submitted from the Inertia admin UI via FormData. We keep these fields # explicitly `required=False` to match checkbox semantics (unchecked => absent). is_staff = forms.BooleanField(required=False) is_active = forms.BooleanField(required=False) groups = forms.ModelMultipleChoiceField( queryset=Group.objects.all(), required=False ) user_permissions = forms.ModelMultipleChoiceField( queryset=Permission.objects.all(), required=False ) class Meta: model = User fields = [ "name", "first_name", "last_name", "email", "is_staff", "is_active", "groups", "user_permissions", ] class UserListView(PermissionRequiredAdminMixin, MultipleObjectMixin, View): """View for listing users with search and pagination.""" model = User permission_required = "labs.view_user" paginate_by = 10 page_kwarg = "page" ordering = "-date_joined" search_fields = ["email", "name", "first_name", "last_name"] serialize_fields = [ "id", "email", "name", "organization", "is_staff", "is_active", "date_joined", ] component = "users/Index" def get_queryset(self): queryset = self.model._default_manager.all() if self.ordering: queryset = queryset.order_by(self.ordering) query = self.request.GET.get("q", "").strip() if query: conditions = Q() for field in self.search_fields: conditions |= Q(**{f"{field}__icontains": query}) queryset = queryset.filter(conditions) is_staff = self.request.GET.get("is_staff") if is_staff in ("true", "false"): queryset = queryset.filter(is_staff=is_staff == "true") is_active = self.request.GET.get("is_active") if is_active in ("true", "false"): queryset = queryset.filter(is_active=is_active == "true") return queryset def serialize_user(self, user: User) -> Dict[str, Any]: return { "id": user.id, "email": user.email, "name": user.name, "organization": user.organization, "is_staff": user.is_staff, "is_active": user.is_active, "date_joined": user.date_joined.isoformat(), } def get(self, request: HttpRequest) -> HttpResponse: self.object_list = self.get_queryset() paginate_by = self.get_paginate_by(self.object_list) if paginate_by: paginator, page, object_list, is_paginated = self.paginate_queryset( self.object_list, paginate_by ) else: paginator = None page = None object_list = self.object_list is_paginated = False users: List[Dict[str, Any]] = [self.serialize_user(u) for u in object_list] pagination: Dict[str, Any] | None = None if paginator and page: pagination = { "page": page.number, "per_page": paginator.per_page, "total": paginator.count, "num_pages": paginator.num_pages, "is_paginated": is_paginated, } return render( request, self.component, { "users": users, "filters": { "q": request.GET.get("q", "").strip(), "is_staff": request.GET.get("is_staff"), "is_active": request.GET.get("is_active"), }, "pagination": pagination, }, ) class UserDetailView(PermissionRequiredAdminMixin, FormMixin, SingleObjectMixin, View): """Detail/Edit view for a single user.""" model = User permission_required = "labs.view_user" form_class = UserEditForm component = "users/Detail" def serialize_user(self, user: User) -> Dict[str, Any]: return { "id": user.id, "email": user.email, "name": user.name, "first_name": user.first_name, "last_name": user.last_name, "organization": user.organization, "is_staff": user.is_staff, "is_active": user.is_active, "date_joined": user.date_joined.isoformat(), "groups": list(user.groups.values_list("id", flat=True)), "user_permissions": list( user.user_permissions.values_list("id", flat=True) ), } def get_available_groups(self) -> List[Dict[str, Any]]: return [{"id": g.id, "name": g.name} for g in Group.objects.all()] def get_available_permissions(self) -> List[Dict[str, Any]]: perms = Permission.objects.select_related("content_type").all() return [ { "id": p.id, "name": f"{p.content_type.app_label} | {p.content_type.model} | {p.name}", } for p in perms ] def get_form_kwargs(self) -> Dict[str, Any]: kwargs = super().get_form_kwargs() kwargs["instance"] = self.object return kwargs def get(self, request: HttpRequest, pk: int) -> HttpResponse: self.object = self.get_object() tab = request.GET.get("tab", "info") context = { "user": self.serialize_user(self.object), "tab": tab, } if tab == "info": context["available_groups"] = self.get_available_groups() context["available_permissions"] = self.get_available_permissions() if tab == "credits": context.update(self.get_credits_context(request)) elif tab == "events": context.update(self.get_events_context(request)) elif tab == "certificates": context.update(self.get_certificates_context(request)) return render(request, self.component, context) def get_credits_context(self, request: HttpRequest) -> Dict[str, Any]: # Summary summary, _ = UserCreditSummary.objects.get_or_create(user=self.object) # Free credits free_credits = UserCreditDetail.objects.filter( user=self.object, credit_type=UserCreditDetail.CreditType.FREE ).order_by("-created_at") # Transactions transactions = CreditTransaction.objects.filter(user=self.object).order_by( "-created_at" ) # Date filter date_from = request.GET.get("date_from") date_to = request.GET.get("date_to") if date_from: transactions = transactions.filter(created_at__gte=date_from) if date_to: transactions = transactions.filter(created_at__lte=date_to) # Pagination for transactions paginator = Paginator(transactions, 10) page_number = request.GET.get("page", 1) page_obj = paginator.get_page(page_number) return { "credit_summary": { "total": summary.paid_credits + summary.free_credits, "paid": summary.paid_credits, "free": summary.free_credits, }, "free_credits": [ { "id": fc.id, "amount": fc.amount, "initial_amount": fc.initial_amount, "expires_at": fc.expires_at.isoformat() if fc.expires_at else None, "remark": fc.remark, "created_at": fc.created_at.isoformat(), } for fc in free_credits ], "transactions": { "data": [ { "id": t.id, "type": t.get_transaction_type_display(), "paid_credits": t.paid_credits, "free_credits": t.free_credits, "remark": t.remark, "created_at": t.created_at.isoformat(), } for t in page_obj ], "pagination": { "page": page_obj.number, "per_page": paginator.per_page, "total": paginator.count, "num_pages": paginator.num_pages, }, }, "filters": { "date_from": date_from, "date_to": date_to, }, } def get_events_context(self, request: HttpRequest) -> Dict[str, Any]: events = UserEvent.objects.filter(user=self.object).order_by("-created_at") # Filters date_from = request.GET.get("date_from") date_to = request.GET.get("date_to") source = request.GET.get("source") if date_from: events = events.filter(created_at__gte=date_from) if date_to: events = events.filter(created_at__lte=date_to) if source: events = events.filter(event_source=source) paginator = Paginator(events, 20) page_number = request.GET.get("page", 1) page_obj = paginator.get_page(page_number) return { "events": { "data": [ { "id": e.id, "source": e.get_event_source_display(), "name": e.event_name, "body": e.event_body, "created_at": e.created_at.isoformat(), } for e in page_obj ], "pagination": { "page": page_obj.number, "per_page": paginator.per_page, "total": paginator.count, "num_pages": paginator.num_pages, }, }, "filters": { "date_from": date_from, "date_to": date_to, "source": source, }, } def get_certificates_context(self, request: HttpRequest) -> Dict[str, Any]: certificates = [] try: recipient = CertificateRecipient.objects.get(user=self.object) issuances = ( CertificateIssuance.objects.filter(recipient=recipient) .select_related("cert_type") .order_by("-issued_at") ) certificates = [ { "id": c.id, "number": c.cert_number, "type": c.cert_type.name, "issued_at": c.issued_at.isoformat(), "url": c.get_file_public_url() if c.file_object_key else None, "file_status": "ready" if c.file_object_key else "pending", } for c in issuances ] except CertificateRecipient.DoesNotExist: pass return {"certificates": certificates} def post(self, request: HttpRequest, pk: int) -> HttpResponse: if not request.user.has_perm("labs.change_user"): return redirect(f"/admin/users/{pk}/") self.object = self.get_object() form = self.get_form() if form.is_valid(): return self.form_valid(form) return self.form_invalid(form) def form_valid(self, form: forms.ModelForm) -> HttpResponse: form.save() self.request.session["flash_success"] = "User updated successfully" return redirect(f"/admin/users/{self.object.id}/") def form_invalid(self, form: forms.ModelForm) -> HttpResponse: return render( self.request, self.component, { "user": self.serialize_user(self.object), "errors": form.errors, "form": form.data, }, ) class UserJsonSearchView(PermissionRequiredAdminMixin, View): """JSON API to search users for autocomplete.""" permission_required = "labs.view_user" def get(self, request: HttpRequest) -> HttpResponse: query = request.GET.get("q", "").strip() if not query: return JsonResponse({"results": []}) users = User.objects.filter( Q(email__icontains=query) | Q(name__icontains=query) )[:20] return JsonResponse( { "results": [ {"value": u.id, "label": f"{u.name} ({u.email})"} for u in users ] } )