from django import forms from django.contrib import admin, messages from django.core import signing from django.core.exceptions import ValidationError from django.db import IntegrityError from django.template.response import TemplateResponse from apps.access.authorization import authorized_queryset from apps.organization.models import ServiceCenter from apps.service.models import ServiceCase from apps.parts.admin import RevisionForm from .admin import InventoryAdmin, _adopt from . import request_services as services, request_queries as queries from .models import PartsRequest, PartsRequestLine, StockReservation, SerializedStockUnit from .queries import authorized_locations, USABLE_TYPES class RequestForm(RevisionForm): class Meta: model = PartsRequest fields = ("service_case", "note") class RequestLineInline(admin.TabularInline): model = PartsRequestLine fields = ("spare_part", "quantity") extra = 1 def has_change_permission(self, request, obj=None): return False def has_delete_permission(self, request, obj=None): return False def has_add_permission(self, request, obj=None): return obj is None and super().has_add_permission(request, obj) def get_readonly_fields(self, request, obj=None): return self.fields if obj else () class HistoricalWorkflowAdmin(InventoryAdmin): def has_change_permission(self, request, obj=None): return obj is None and super().has_change_permission(request, obj) def get_form(self, request, obj=None, **kwargs): return forms.modelform_factory(self.model, fields=[]) if obj else super().get_form(request, obj, **kwargs) def get_readonly_fields(self, request, obj=None): editable = {"service_case", "note"} if self.model is PartsRequest else {"line", "location", "quantity"} return tuple(field.name for field in self.model._meta.concrete_fields if obj or field.name not in editable) def transition(self, request, queryset, action, operation, reason=False): rows = list(queryset[:2]) if len(rows) != 1: self.message_user(request, "Select one record to review.", messages.ERROR) return None row = rows[0] payload = [self.model._meta.label_lower, str(row.pk), row.updated_at.isoformat(), action, str(request.user.pk)] if request.POST.get("confirm_transition"): try: if signing.loads(request.POST.get("transition_revision", ""), salt="inventory-job-transition") != payload: raise ValidationError("The record changed; review it again.") options = {"request" if self.model is PartsRequest else "reservation": row} if reason: options["reason"] = request.POST.get("reason", "") result = operation(actor=request.user, expected_revision=payload[2], **options) self.log_change(request, result, action) self.message_user(request, "Operation completed.") except (ValidationError, IntegrityError, signing.BadSignature): self.message_user(request, "Operation rejected. Review the current state and revision.", messages.ERROR) return None if self.model is PartsRequest: row = queries.parts_request_detail(actor=request.user, request=row) return TemplateResponse(request, "admin/inventory/confirm_job_transition.html", { **self.admin_site.each_context(request), "opts": self.model._meta, "title": action.replace("_", " ").title(), "record": row, "action_name": action, "needs_reason": reason, "transition_revision": signing.dumps(payload, salt="inventory-job-transition")}) @admin.register(PartsRequest) class PartsRequestAdmin(HistoricalWorkflowAdmin): form = RequestForm inlines = (RequestLineInline,) list_display = ("service_case", "requester", "requested_at", "status") list_filter = ("status",) actions = ("approve_selected", "reject_selected", "cancel_selected") def get_queryset(self, request): return queries.parts_requests(actor=request.user) def formfield_for_foreignkey(self, db_field, request, **kwargs): if db_field.name == "service_case": centers = authorized_queryset(user=request.user, permission="inventory.request_parts", queryset=ServiceCenter.objects.all()) kwargs["queryset"] = ServiceCase.objects.filter(service_center_id__in=centers.values("pk"), status__in=["DIAGNOSED", "REPAIRING"]).select_related("service_center") return super().formfield_for_foreignkey(db_field, request, **kwargs) def save_model(self, request, obj, form, change): if change: raise ValidationError("Request evidence is immutable.") # Aggregate creation waits for the validated line formset below. def save_formset(self, request, form, formset, change): lines = [dict(spare_part=item.cleaned_data["spare_part"], quantity=item.cleaned_data["quantity"]) for item in formset.forms if item.cleaned_data and not item.cleaned_data.get("DELETE")] row = services.create_parts_request(actor=request.user, service_case=form.instance.service_case, note=form.instance.note, lines=lines) _adopt(form.instance, row) formset.new_objects, formset.changed_objects, formset.deleted_objects = [], [], [] @admin.action(description="Approve requested parts", permissions=["change"]) def approve_selected(self, request, queryset): return self.transition(request, queryset, "approve_selected", services.approve_parts_request) @admin.action(description="Reject requested parts", permissions=["change"]) def reject_selected(self, request, queryset): return self.transition(request, queryset, "reject_selected", services.reject_parts_request, True) @admin.action(description="Cancel parts request", permissions=["change"]) def cancel_selected(self, request, queryset): return self.transition(request, queryset, "cancel_selected", services.cancel_parts_request, True) class ReservationForm(forms.ModelForm): selected_units = forms.ModelMultipleChoiceField(queryset=SerializedStockUnit.objects.none(), required=False) request_revisions = forms.CharField(widget=forms.HiddenInput) class Meta: model = StockReservation fields = ("line", "location", "quantity") def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) if not self.is_bound: revisions = {str(line.pk): line.request.updated_at.isoformat() for line in self.fields["line"].queryset} self.initial["request_revisions"] = signing.dumps(revisions, salt="inventory-reserve-request") def clean(self): cleaned = super().clean() try: revisions = signing.loads(cleaned.get("request_revisions", ""), salt="inventory-reserve-request") line = cleaned.get("line") self.expected_revision = revisions[str(line.pk)] if line else None except (signing.BadSignature, KeyError, TypeError): raise ValidationError("Request selection changed; reload the form.") return cleaned @admin.register(StockReservation) class StockReservationAdmin(HistoricalWorkflowAdmin): form = ReservationForm list_display = ("service_case", "spare_part", "location", "quantity", "status", "reserved_at") list_filter = ("status",) actions = ("release_selected",) def get_queryset(self, request): return queries.stock_reservations(actor=request.user) def formfield_for_foreignkey(self, db_field, request, **kwargs): if db_field.name == "line": kwargs["queryset"] = PartsRequestLine.objects.filter(request_id__in=queries.parts_requests(actor=request.user, permission="inventory.reserve_parts").filter(status="APPROVED").values("pk")).select_related("request", "spare_part") if db_field.name == "location": kwargs["queryset"] = authorized_locations(actor=request.user, permission="inventory.reserve_parts").filter(is_active=True, location_type__in=USABLE_TYPES) return super().formfield_for_foreignkey(db_field, request, **kwargs) def get_form(self, request, obj=None, **kwargs): form = super().get_form(request, obj, **kwargs) if "selected_units" in form.base_fields: form.base_fields["selected_units"].queryset = SerializedStockUnit.objects.filter(state="IN_STOCK", current_location_id__in=authorized_locations(actor=request.user, permission="inventory.reserve_parts").values("pk")).select_related("spare_part", "current_location") return form def save_model(self, request, obj, form, change): if change: raise ValidationError("Reservation history is immutable.") row = services.reserve_parts(actor=request.user, line=obj.line, location=obj.location, quantity=obj.quantity, units=list(form.cleaned_data["selected_units"]), expected_request_revision=form.expected_revision) _adopt(obj, row) @admin.action(description="Release reservation", permissions=["change"]) def release_selected(self, request, queryset): return self.transition(request, queryset, "release_selected", services.release_reservation, True)