"""Scoped inventory presentation. Quantities and mutations stay in inventory APIs.""" from urllib.parse import urlencode from django import forms from django.contrib.auth.decorators import login_required from django.core.exceptions import PermissionDenied from django.core.paginator import Paginator from django.db.models import Exists, OuterRef, Q, Prefetch from django.http import Http404 from django.shortcuts import get_object_or_404, render from django.urls import reverse from django.utils import timezone from django.views.decorators.cache import never_cache from django.views.decorators.http import require_GET from apps.inventory import document_queries as documents, control_queries as control from apps.inventory import request_queries as requests, usage_queries as usage from apps.inventory.queries import authorized_locations from apps.inventory.models import StockLedgerEntry, PartsRequestLine, PartsDisposition, InventoryLocation from apps.organization.models import Company, ServiceCenter from . import queries KINDS = {"requests": "Parts requests", "receipts": "Goods receiving", "positions": "Inventory positions", "transfers": "Transfers", "counts": "Stock counts", "adjustments": "Posted adjustments", "history": "Inventory history", "issues": "Issue and custody", "returns": "Consumption and unused returns", "recoveries": "Defective recoveries"} def url(kind="overview", pk=None): return reverse("operations:inventory_record", args=[kind, pk]) if pk else reverse("operations:inventory_workspace", args=[kind]) def source(actor, kind): if kind == "requests": return requests.parts_requests(actor=actor).prefetch_related(Prefetch("lines", queryset=PartsRequestLine.objects.select_related("spare_part"))) if kind == "receipts": return documents.goods_receipts(actor=actor) if kind == "transfers": return documents.stock_transfers(actor=actor) if kind == "positions": return control.control_positions(actor=actor) if kind == "counts": return control.stock_counts(actor=actor) if kind == "adjustments": return control.stock_adjustments(actor=actor) if kind == "issues": return usage.parts_issues(actor=actor) if kind == "returns": return usage.parts_usage(actor=actor) if kind == "recoveries": return usage.defective_recoveries(actor=actor) if kind == "history": # Same location-side ledger boundary as frozen inventory reporting. A row # shows only its authorized location, never a hidden counterpart endpoint. return StockLedgerEntry.objects.filter(location_id__in=authorized_locations(actor=actor).values("pk")).select_related( "location", "spare_part", "movement__actor") raise Http404 def gate(actor): if not queries.capable(actor, "inventory.view_stock"): raise PermissionDenied def case_path(kind): return {"requests": "service_case_id", "issues": "service_case_id", "returns": "issue__service_case_id", "recoveries": "service_case_id"}.get(kind) class LocationChoice(forms.ModelChoiceField): def label_from_instance(self, row): center = str(row.service_center) if row.service_center_id else "Company location" return f"{row.company} / {center} / {row.code}: {row.name}" class Filters(forms.Form): search = forms.CharField(required=False, max_length=100, label="Part code or name") part = forms.UUIDField(required=False, label="Part ID") location = forms.UUIDField(required=False, label="Location ID") case = forms.UUIDField(required=False, label="Service job ID") state = forms.CharField(required=False, max_length=24, label="Exact recorded state / movement type") reference = forms.CharField(required=False, max_length=128) start = forms.DateField(required=False, widget=forms.DateInput(attrs={"type": "date"}), label="From local date") end = forms.DateField(required=False, widget=forms.DateInput(attrs={"type": "date"}), label="Through local date") def __init__(self, actor, kind, *args, **kwargs): super().__init__(*args, **kwargs) visible_locations = authorized_locations(actor=actor) self.fields["location"] = LocationChoice(queryset=visible_locations, required=False) if kind == "positions": self.fields["state"] = forms.ChoiceField(choices=[("", "All location types"), *InventoryLocation.Type.choices], required=False, label="Location type") self.fields["company"] = forms.ModelChoiceField(queryset=Company.objects.filter(pk__in=visible_locations.values("company_id")), required=False) self.fields["center"] = forms.ModelChoiceField(queryset=ServiceCenter.objects.filter(pk__in=visible_locations.values("service_center_id")).select_related("region"), required=False, label="Service center") for field in ("start", "end", "case", "reference"): self.fields.pop(field) elif kind in ("issues", "recoveries", "adjustments"): self.fields.pop("state") def clean(self): data = super().clean() if data.get("start") and data.get("end") and data["start"] > data["end"]: raise forms.ValidationError("End date must not precede start date.") return data def filtered(actor, kind, rows, data): if kind == "positions": if data.get("company"): rows = rows.filter(location__company=data["company"]) if data.get("center"): rows = rows.filter(location__service_center=data["center"]) if data.get("location"): location = data["location"] key = {"receipts": "destination_id", "requests": "lines__reservations__location_id", "issues": "custody_location_id", "returns": "issue__custody_location_id"}.get(kind, "location_id") rows = rows.filter(Q(source_id=location.pk) | Q(destination_id=location.pk)) if kind == "transfers" else rows.filter(**{key: location.pk}) if data.get("part") or data.get("search"): key = "lines__spare_part_id" if kind in ("requests", "receipts", "transfers") else "reservation__spare_part_id" if kind == "issues" else "issue__reservation__spare_part_id" if kind == "returns" else "replacement__issue__reservation__spare_part_id" if kind == "recoveries" else "spare_part_id" if data.get("part"): rows = rows.filter(**{key: data["part"]}) if data.get("search"): relation = key.removesuffix("_id") rows = rows.filter(Q(**{relation + "__part_code__icontains": data["search"]}) | Q(**{relation + "__name__icontains": data["search"]})) if data.get("case"): case = get_object_or_404(queries.cases(actor), pk=data["case"]) path = case_path(kind) if path: rows = rows.filter(**{path: case.pk}) elif kind == "history": # Only recorded issue/disposition evidence establishes a job link. from apps.inventory.models import PartsIssue movements = PartsDisposition.objects.filter(issue__service_case=case).values("movement_id") issues = PartsIssue.objects.filter(service_case=case).values("movement_id") rows = rows.filter(Q(movement_id__in=movements) | Q(movement_id__in=issues)) else: rows = rows.none() if data.get("state"): key = "movement__kind" if kind == "history" else "kind" if kind == "returns" else "location__location_type" if kind == "positions" else "status" rows = rows.none() if kind in ("issues", "recoveries", "adjustments") else rows.filter(**{key: data["state"].upper()}) if data.get("reference"): key = "movement__reference" if kind in ("history", "adjustments") else "number" if kind in ("receipts", "transfers") else None rows = rows.filter(**{key + "__icontains": data["reference"]}) if key else rows.none() timestamp = {"history": "movement__posted_at", "requests": "requested_at", "issues": "issued_at", "returns": "recorded_at", "recoveries": "recovered_at"}.get(kind, "created_at") for name, lookup in (("start", "gte"), ("end", "lte")): if data.get(name) and kind != "positions": rows = rows.filter(**{timestamp + "__date__" + lookup: data[name]}) return rows.distinct() def cards(actor, kind, rows): path = case_path(kind) if path: rows = rows.annotate(job_visible=Exists(queries.cases(actor).filter(pk=OuterRef(path)))) result = [] part_links = queries.capable(actor, "parts.view_sparepart") if kind == "positions" else False if kind == "positions": rows = rows.annotate(can_adjust=Exists(authorized_locations(actor=actor, permission="inventory.adjust_stock").filter(pk=OuterRef("location_id")))) for row in rows: links, facts = [], [] title = str(getattr(row, "number", row.pk)) if kind == "positions": title = row.spare_part.part_code facts = [("Location", row.location.code), ("Location type", row.location.get_location_type_display()), ("On hand", row.on_hand), ("Active reserved", row.reserved)] for label, params in (("Part positions", {"part": row.spare_part_id}), ("Location history", {"location": row.location_id, "part": row.spare_part_id})): links.append((label, url("positions" if label == "Part positions" else "history") + "?" + urlencode(params))) links.append(("Location", reverse("operations:detail", args=["inventory", row.location_id]))) if part_links and row.spare_part.is_active: links.append(("Part details", reverse("operations:detail", args=["parts", row.spare_part_id]))) if row.can_adjust and row.location.is_active and row.spare_part.is_active and row.location.location_type not in ("TRANSIT", "CUSTODY"): links.append(("Review stock adjustment", reverse("operations:inventory_adjust", args=[row.pk]))) elif kind == "history": title = row.movement.reference facts = [("Movement", row.movement_id), ("Posted", row.movement.posted_at), ("Type", row.movement.kind), ("Part", row.spare_part.part_code), ("Authorized location", row.location.code), ("Quantity delta", row.quantity_delta), ("Actor", row.movement.actor.get_username())] elif kind in ("receipts", "transfers"): facts = [("Status", row.status), ("Destination", row.destination.code), ("Created", row.created_at)] if kind == "receipts": facts += [("External reference", row.external_reference), ("Received", row.received_at), ("Receiving actor", row.received_by)] else: facts += [("Source", row.source.code), ("Dispatched", row.dispatched_at), ("Received", row.received_at)] elif kind == "requests": facts = [("Status", row.status), ("Requested", row.requested_at), ("Requester", row.requester.get_username())] facts += [(line.spare_part.part_code, line.quantity) for line in row.lines.all()] elif kind == "counts": facts = [("Part", row.spare_part.part_code), ("Location", row.location.code), ("Status", row.status), ("Expected at start", row.expected_quantity), ("Counted", row.counted_quantity), ("Variance", row.variance), ("Finished", row.finished_at)] elif kind == "adjustments": facts = [("Part", row.spare_part.part_code), ("Location", row.location.code), ("Posted adjustment", row.quantity_delta), ("Reason", row.get_reason_display()), ("Movement", row.movement_id), ("Posted", row.movement.posted_at)] elif kind == "issues": facts = [("Part", row.reservation.spare_part.part_code), ("Issued", row.issued_at), ("Issued quantity", row.reservation.quantity), ("Custody location", row.custody_location.code), ("Custodian", row.recipient.get_username()), ("Resolved quantity", row.resolved_quantity)] elif kind == "returns": facts = [("Part", row.issue.reservation.spare_part.part_code), ("Disposition", row.get_kind_display()), ("Quantity", row.quantity), ("Recorded", row.recorded_at)] else: facts = [("Component", row.component_description), ("Quantity", row.quantity), ("Location", row.location.code), ("Recovered", row.recovered_at)] if kind in ("requests", "receipts", "transfers", "counts"): links.append(("Review evidence", url(kind, row.pk))) if path and row.job_visible: case = row.issue.service_case if kind == "returns" else row.service_case links.append(("Open job " + case.job_number, reverse("operations:detail", args=["cases", case.pk]))) result.append(dict(title=title, facts=facts, links=links)) return result @never_cache @login_required(login_url="operations:login") @require_GET def workspace(request, kind="overview"): gate(request.user) links = [(label, url(key)) for key, label in KINDS.items()] if kind == "overview": today = timezone.localdate() metrics = [("Requests awaiting approval", source(request.user, "requests").filter(status="REQUESTED").count()), ("Approved requests", source(request.user, "requests").filter(status="APPROVED").count()), ("Active reservations", requests.stock_reservations(actor=request.user, active_only=True).count()), ("Receipts posted today", source(request.user, "receipts").filter(status="POSTED", received_at__date=today).count()), ("Draft / dispatched transfers", source(request.user, "transfers").filter(status__in=["DRAFT", "DISPATCHED"]).count()), ("Draft / active counts", source(request.user, "counts").filter(status__in=["DRAFT", "COUNTING"]).count())] recent = cards(request.user, "history", source(request.user, "history").order_by("-movement__posted_at", "pk")[:5]) return render(request, "operations/inventory.html", dict(title="Parts & inventory", links=links, metrics=metrics, records=recent, overview=True)) if kind not in KINDS: raise Http404 form = Filters(request.user, kind, request.GET) rows = source(request.user, kind) rows = filtered(request.user, kind, rows, form.cleaned_data) if form.is_valid() else rows.none() rows = rows.order_by("-movement__posted_at", "pk") if kind == "history" else rows.order_by("-created_at", "pk") page = Paginator(rows, 25).get_page(request.GET.get("page")) from .inventory_actions import create_links parameters = request.GET.copy() parameters.pop("page", None) return render(request, "operations/inventory.html", dict(title=KINDS[kind], links=links, form=form, records=cards(request.user, kind, page.object_list), page=page, retained=parameters.urlencode(), actions=create_links(request.user, kind)), status=200 if form.is_valid() else 400) @never_cache @login_required(login_url="operations:login") @require_GET def detail(request, kind, pk): gate(request.user) if kind not in ("requests", "receipts", "transfers", "counts"): raise Http404 row = get_object_or_404(source(request.user, kind), pk=pk) records = cards(request.user, kind, source(request.user, kind).filter(pk=pk)) children = [] if kind in ("receipts", "transfers"): row = documents.goods_receipt_detail(actor=request.user, receipt=row) if kind == "receipts" else documents.stock_transfer_detail(actor=request.user, transfer=row) for line in row.lines.all(): facts = [("Part", line.spare_part.part_code), ("Quantity", line.quantity), ("Active", line.is_active)] if kind == "receipts": facts += [("Movement", line.movement_id), ("Recorded identifiers", ", ".join(r.identifier for r in line.identifiers.all() if r.is_active))] else: facts += [("Dispatch movement", line.dispatch_movement_id), ("Receipt movement", line.receive_movement_id), ("Selected units", ", ".join(r.unit.identifier for r in line.units.all() if r.is_active))] children.append(dict(title="Document line", facts=facts)) elif kind == "requests": for reservation in requests.stock_reservations(actor=request.user).filter(line__request=row): children.append(dict(title="Reservation", facts=[("Part", reservation.spare_part.part_code), ("Location", reservation.location.code), ("Quantity", reservation.quantity), ("Status", reservation.status)])) if queries.cases(request.user).filter(pk=row.service_case_id).exists(): records[0]["links"].append(("Job parts evidence", reverse("operations:job_section", args=[row.service_case_id, "parts"]))) children += cards(request.user, "issues", usage.parts_issues(actor=request.user).filter(reservation__line__request=row)[:25]) children += cards(request.user, "returns", usage.parts_usage(actor=request.user).filter(issue__reservation__line__request=row)[:25]) elif kind == "counts": for link in row.units.select_related("unit").order_by("pk")[:100]: children.append(dict(title="Counted identity", facts=[("Identifier", link.unit.identifier), ("Expected", link.expected), ("Counted", link.counted)])) children += cards(request.user, "adjustments", control.stock_adjustments(actor=request.user).filter(count=row)) from .inventory_actions import record_links return render(request, "operations/inventory.html", dict(title=KINDS[kind] + " ยท " + str(getattr(row, "number", row.pk)), records=records + children, links=[("Back to " + KINDS[kind], url(kind))], actions=record_links(request.user, kind, row), detail_note="Issue and disposition previews show up to 25 rows each; count identities show up to 100. Use the dedicated workspaces or guarded Admin for longer histories."))