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.parts.admin import RevisionForm from .admin import InventoryAdmin, _adopt from . import document_services as services, document_queries as queries from .models import GoodsReceipt, GoodsReceiptLine, StockTransfer, StockTransferLine, SerializedStockUnit from .queries import authorized_locations class ReceiptForm(RevisionForm): class Meta: model = GoodsReceipt fields = ("destination", "external_reference", "note") class TransferForm(RevisionForm): class Meta: model = StockTransfer fields = ("source", "destination", "note") class ReceiptLineForm(forms.ModelForm): serial_identifiers = forms.CharField(required=False, widget=forms.Textarea(attrs={"rows": 3}), help_text="One identifier per line. Draft text does not create stock units.") class Meta: model = GoodsReceiptLine fields = ("spare_part", "quantity") def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) if not self.instance._state.adding: self.initial["serial_identifiers"] = "\n".join(row.identifier for row in self.instance.identifiers.all() if row.is_active) class TransferLineForm(forms.ModelForm): selected_units = forms.ModelMultipleChoiceField(queryset=SerializedStockUnit.objects.none(), required=False) class Meta: model = StockTransferLine fields = ("spare_part", "quantity") def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) if not self.instance._state.adding: self.initial["selected_units"] = [row.unit_id for row in self.instance.units.all() if row.is_active] class DocumentLineInline(admin.TabularInline): extra = 1 def has_add_permission(self, request, obj=None): return (obj is None or obj.status == "DRAFT") and super().has_add_permission(request, obj) def has_change_permission(self, request, obj=None): return (obj is None or obj.status == "DRAFT") and super().has_change_permission(request, obj) def has_delete_permission(self, request, obj=None): return (obj is None or obj.status == "DRAFT") and super().has_delete_permission(request, obj) class ReceiptLineInline(DocumentLineInline): model = GoodsReceiptLine form = ReceiptLineForm def get_queryset(self, request): return super().get_queryset(request).filter(is_active=True).select_related("spare_part", "movement").prefetch_related("identifiers") def get_fields(self, request, obj=None): return ("spare_part", "quantity", "serial_identifiers") if obj is None or obj.status == "DRAFT" else ("spare_part", "quantity", "recorded_identifiers", "movement") def get_readonly_fields(self, request, obj=None): return () if obj is None or obj.status == "DRAFT" else self.get_fields(request, obj) @admin.display(description="Identifiers") def recorded_identifiers(self, obj): return ", ".join(row.identifier for row in obj.identifiers.all() if row.is_active) class TransferLineInline(DocumentLineInline): model = StockTransferLine form = TransferLineForm def get_queryset(self, request): return super().get_queryset(request).filter(is_active=True).select_related("spare_part", "dispatch_movement", "receive_movement").prefetch_related("units__unit") def get_formset(self, request, obj=None, **kwargs): formset = super().get_formset(request, obj, **kwargs) locations = authorized_locations(actor=request.user, permission="inventory.transfer_stock").values("pk") if "selected_units" in formset.form.base_fields: formset.form.base_fields["selected_units"].queryset = SerializedStockUnit.objects.filter( current_location_id__in=locations, state="IN_STOCK").select_related("spare_part", "current_location") return formset def get_fields(self, request, obj=None): return ("spare_part", "quantity", "selected_units") if obj is None or obj.status == "DRAFT" else ( "spare_part", "quantity", "recorded_units", "dispatch_movement", "receive_movement") def get_readonly_fields(self, request, obj=None): return () if obj is None or obj.status == "DRAFT" else self.get_fields(request, obj) @admin.display(description="Units") def recorded_units(self, obj): return ", ".join(row.unit.identifier for row in obj.units.all() if row.is_active) class DocumentAdmin(InventoryAdmin): list_filter = ("status",) search_fields = ("number", "note") actions = ("cancel_selected",) def has_change_permission(self, request, obj=None): return (obj is None or obj.status == "DRAFT") and super().has_change_permission(request, obj) def get_readonly_fields(self, request, obj=None): editable = {"destination", "note", "external_reference"} if self.model is GoodsReceipt else {"source", "destination", "note"} if obj: editable -= {"source", "destination"} if obj.status != "DRAFT": editable.clear() return tuple(field.name for field in self.model._meta.concrete_fields if field.name not in editable) def get_form(self, request, obj=None, **kwargs): if obj is not None and obj.status != "DRAFT": return forms.modelform_factory(self.model, fields=[]) return super().get_form(request, obj, **kwargs) def formfield_for_foreignkey(self, db_field, request, **kwargs): if db_field.name in ("source", "destination"): permission = "inventory.receive_stock" if self.model is GoodsReceipt else "inventory.transfer_stock" kwargs["queryset"] = authorized_locations(actor=request.user, permission=permission).filter(is_active=True).exclude(location_type__in=["TRANSIT", "CUSTODY"]) return super().formfield_for_foreignkey(db_field, request, **kwargs) def _transition(self, request, queryset, action_name, operation, label, *, reason=False): rows = list(queryset[:2]) if len(rows) != 1: self.message_user(request, "Select one document to review and confirm.", messages.ERROR) return None row = rows[0] payload = [self.model._meta.label_lower, str(row.pk), row.updated_at.isoformat(), action_name, str(request.user.pk)] if request.POST.get("confirm_transition"): try: supplied = signing.loads(request.POST.get("transition_revision", ""), salt="inventory-document-transition") if supplied != payload: raise ValidationError("Document changed; review its current state before confirming.") argument = "receipt" if self.model is GoodsReceipt else "transfer" options = {"reason": request.POST.get("reason", "")} if reason else {} current = operation(actor=request.user, expected_revision=supplied[2], **{argument: row}, **options) self.log_change(request, current, label) self.message_user(request, f"{label}: {current.number}") except (ValidationError, IntegrityError, signing.BadSignature): self.message_user(request, "Transition rejected. Reload and review document state, stock and revision.", messages.ERROR) return None detail = (queries.goods_receipt_detail(actor=request.user, receipt=row) if self.model is GoodsReceipt else queries.stock_transfer_detail(actor=request.user, transfer=row)) context = {**self.admin_site.each_context(request), "opts": self.model._meta, "title": label, "document": detail, "lines": [line for line in detail.lines.all() if line.is_active], "action_name": action_name, "transition_revision": signing.dumps(payload, salt="inventory-document-transition"), "needs_reason": reason} return TemplateResponse(request, "admin/inventory/confirm_transition.html", context) @admin.action(description="Cancel draft document", permissions=["change"]) def cancel_selected(self, request, queryset): operation = services.cancel_goods_receipt if self.model is GoodsReceipt else services.cancel_stock_transfer return self._transition(request, queryset, "cancel_selected", operation, "Cancel draft document", reason=True) @admin.register(GoodsReceipt) class GoodsReceiptAdmin(DocumentAdmin): form = ReceiptForm list_display = ("number", "company", "destination", "status", "received_at", "received_by") inlines = (ReceiptLineInline,) actions = ("post_selected", "cancel_selected") def get_queryset(self, request): return queries.goods_receipts(actor=request.user) def save_model(self, request, obj, form, change): options = dict(actor=request.user, external_reference=form.cleaned_data["external_reference"], note=form.cleaned_data["note"]) current = (services.update_goods_receipt(receipt=obj, expected_revision=form.expected_revision, **options) if change else services.create_goods_receipt(destination=obj.destination, **options)) _adopt(obj, current) def save_formset(self, request, form, formset, change): lines = [dict(spare_part=item.cleaned_data["spare_part"], quantity=item.cleaned_data["quantity"], identifiers=[value for value in item.cleaned_data.get("serial_identifiers", "").splitlines() if value.strip()]) for item in formset.forms if item.cleaned_data and not item.cleaned_data.get("DELETE")] current = services.set_goods_receipt_lines(actor=request.user, receipt=form.instance, expected_revision=form.instance.updated_at.isoformat(), lines=lines) _adopt(form.instance, current) # No model-formset save: all writes have gone through the aggregate service. formset.new_objects, formset.changed_objects, formset.deleted_objects = [], [], [] @admin.action(description="Post receipt", permissions=["change"]) def post_selected(self, request, queryset): return self._transition(request, queryset, "post_selected", services.post_goods_receipt, "Post receipt") @admin.register(StockTransfer) class StockTransferAdmin(DocumentAdmin): form = TransferForm list_display = ("number", "company", "source", "destination", "status", "dispatched_at", "received_at") inlines = (TransferLineInline,) actions = ("dispatch_selected", "receive_selected", "cancel_selected") def get_queryset(self, request): return queries.stock_transfers(actor=request.user) def save_model(self, request, obj, form, change): options = dict(actor=request.user, note=form.cleaned_data["note"]) current = (services.update_stock_transfer(transfer=obj, expected_revision=form.expected_revision, **options) if change else services.create_stock_transfer(source=obj.source, destination=obj.destination, **options)) _adopt(obj, current) def save_formset(self, request, form, formset, change): lines = [dict(spare_part=item.cleaned_data["spare_part"], quantity=item.cleaned_data["quantity"], units=list(item.cleaned_data.get("selected_units", []))) for item in formset.forms if item.cleaned_data and not item.cleaned_data.get("DELETE")] current = services.set_stock_transfer_lines(actor=request.user, transfer=form.instance, expected_revision=form.instance.updated_at.isoformat(), lines=lines) _adopt(form.instance, current) formset.new_objects, formset.changed_objects, formset.deleted_objects = [], [], [] @admin.action(description="Dispatch transfer", permissions=["change"]) def dispatch_selected(self, request, queryset): return self._transition(request, queryset, "dispatch_selected", services.dispatch_stock_transfer, "Dispatch transfer") @admin.action(description="Receive transfer", permissions=["change"]) def receive_selected(self, request, queryset): return self._transition(request, queryset, "receive_selected", services.receive_stock_transfer, "Receive transfer")