from __future__ import annotations import json from datetime import date from django.contrib import messages from django.contrib.auth.decorators import login_required from django.contrib.auth import get_user_model from django.core.exceptions import ValidationError from django.core.paginator import Paginator from django.db import IntegrityError, models, transaction from django.db.models import Count, Prefetch, Q from django.db.models.functions import TruncMonth from django.http import FileResponse, HttpRequest, HttpResponse, JsonResponse, QueryDict from django.shortcuts import get_object_or_404, redirect, render from django.urls import reverse from django.utils import timezone from django.views.decorators.http import require_GET, require_POST from .cpu_catalog import CPU_FAMILY_CHOICES, CPU_VENDOR_CHOICES, suggest_cpus from .device_types import COMPUTER_FIELDS, HARDWARE_FIELDS, applicable_fields, clear_inapplicable from .forms import ChoiceValueForm, DeviceForm, EmployeeForm, ExcelUploadForm, FieldDefinitionForm, InspectionForm, QuickAddForm, RepairForm, StorageRecipientForm, UserManagementForm from .import_export import analyze_workbook, csv_response, import_workbook, ods_response, suggested_mapping, xlsx_response from .models import AuditLog, AllInOneSpecification, ChoiceValue, ComputerSpecification, DesktopSpecification, Device, DeviceTask, Employee, FieldDefinition, ImportJob, InspectionRecord, LaptopSpecification, NumberSequence, RepairRecord, UserPreference, UserProfile from .permissions import can_edit, is_admin, role_required from .query import COMPUTER_RELATIONS, SPECIFICATION_RELATIONS, device_queryset from .services import ( FIELD_LABELS, add_audit, check_device_version, create_device_batch, display_device_value, display_value, revert_audit_change, serialize_audit_value, update_device_field, validate_custom_value, ) from .workflow import TRANSITIONS, advance_device, assign_storage_recipient, complete_task, create_task, record_repair, save_inspection GERMAN_MONTHS = ("", "Jan", "Feb", "Mär", "Apr", "Mai", "Jun", "Jul", "Aug", "Sep", "Okt", "Nov", "Dez") CHART_COLORS = ("#0066ff", "#17a673", "#f59e0b", "#8b5cf6", "#ef4444", "#06b6d4", "#64748b", "#ec4899") def _month_start(value: date, offset: int = 0) -> date: month_index = value.year * 12 + value.month - 1 + offset return date(month_index // 12, month_index % 12 + 1, 1) @login_required def overview(request: HttpRequest) -> HttpResponse: try: period_months = int(request.GET.get("months", 3)) except (TypeError, ValueError): period_months = 3 if period_months not in {3, 6, 12}: period_months = 3 today = timezone.localdate() period_start = _month_start(today, -(period_months - 1)) previous_start = _month_start(period_start, -period_months) active_devices = Device.objects.filter(archived_at__isnull=True) arrivals = Device.objects.filter(date__gte=period_start, date__lte=today) previous_arrivals = Device.objects.filter(date__gte=previous_start, date__lt=period_start) active_count = active_devices.count() period_count = arrivals.count() previous_count = previous_arrivals.count() difference = period_count - previous_count period_laptops = arrivals.filter(device_type__label__iexact="Laptop").count() period_finished = arrivals.filter(status__label__iexact="Fertig").count() archived_count = Device.objects.filter(archived_at__isnull=False).count() monthly_start = _month_start(today, -11) monthly_rows = ( Device.objects.filter(date__gte=monthly_start, date__lte=today) .annotate(month=TruncMonth("date")) .values("month") .annotate(count=Count("id")) .order_by("month") ) monthly_counts = {row["month"].date() if hasattr(row["month"], "date") else row["month"]: row["count"] for row in monthly_rows} month_series = [] for offset in range(12): month = _month_start(monthly_start, offset) month_series.append({"label": f"{GERMAN_MONTHS[month.month]} {str(month.year)[2:]}", "count": monthly_counts.get(month, 0)}) maximum_month = max((item["count"] for item in month_series), default=0) chart_maximum = max(maximum_month, 1) for index, item in enumerate(month_series): item["x"] = f"{40 + index * (820 / 11):.1f}" item["y"] = f"{205 - item['count'] / chart_maximum * 170:.1f}" line_points = " ".join(f'{item["x"]},{item["y"]}' for item in month_series) area_points = f"40,205 {line_points} 860,205" type_rows = list( arrivals.values("device_type__label") .annotate(count=Count("id")) .order_by("-count", "device_type__label") ) type_maximum = max((row["count"] for row in type_rows), default=1) for row in type_rows: row["label"] = row.pop("device_type__label") or "Ohne Typ" row["percentage"] = round(row["count"] / type_maximum * 100) status_rows = list( active_devices.values("status__label", "status__color") .annotate(count=Count("id")) .order_by("-count", "status__label") ) status_total = sum(row["count"] for row in status_rows) gradient_parts = [] gradient_cursor = 0.0 for index, row in enumerate(status_rows): row["label"] = row.pop("status__label") or "Ohne Status" row["color"] = row.pop("status__color") or CHART_COLORS[index % len(CHART_COLORS)] row["percentage"] = round(row["count"] / status_total * 100) if status_total else 0 next_cursor = gradient_cursor + (row["count"] / status_total * 100 if status_total else 0) gradient_parts.append(f'{row["color"]} {gradient_cursor:.2f}% {next_cursor:.2f}%') gradient_cursor = next_cursor status_gradient = ", ".join(gradient_parts) if gradient_parts else "#e5e7eb 0% 100%" recipient_rows = list( arrivals.exclude(recipient__isnull=True) .values("recipient__label") .annotate(count=Count("id")) .order_by("-count", "recipient__label")[:7] ) recipient_maximum = max((row["count"] for row in recipient_rows), default=1) for row in recipient_rows: row["label"] = row.pop("recipient__label") row["percentage"] = round(row["count"] / recipient_maximum * 100) cpu_present = Q() ram_present = Q() storage_present = Q() for relation in COMPUTER_RELATIONS: cpu_present |= Q(**{f"{relation}__cpu__isnull": False}) & ~Q(**{f"{relation}__cpu": ""}) ram_present |= Q(**{f"{relation}__ram_gb__isnull": False}) storage_present |= Q(**{f"{relation}__ssd_gb__isnull": False}) | Q(**{f"{relation}__hdd_gb__isnull": False}) completeness_counts = { "model": active_devices.exclude(model="").count(), "cpu": active_devices.filter(cpu_present).count(), "ram": active_devices.filter(ram_present).count(), "storage": active_devices.filter(storage_present).count(), "employee": active_devices.filter(employee__isnull=False).count(), } completeness = [] for key, label in (("model", "Modell"), ("cpu", "Prozessor"), ("ram", "Arbeitsspeicher"), ("storage", "Datenspeicher"), ("employee", "Bearbeiter")): percentage = round(completeness_counts[key] / active_count * 100) if active_count else 0 completeness.append({"label": label, "percentage": percentage}) return render(request, "inventory/overview.html", { "period_months": period_months, "period_start": period_start, "today": today, "active_count": active_count, "period_count": period_count, "period_laptops": period_laptops, "period_finished": period_finished, "archived_count": archived_count, "difference": difference, "month_series": month_series, "maximum_month": maximum_month, "half_month": round(maximum_month / 2), "line_points": line_points, "area_points": area_points, "type_rows": type_rows, "status_rows": status_rows, "status_total": status_total, "status_gradient": status_gradient, "recipient_rows": recipient_rows, "completeness": completeness, }) def _list_context(request: HttpRequest, archived: bool = False) -> dict: queryset = device_queryset(request.GET, archived=archived) selected_types = request.GET.getlist("type") selected_type = ChoiceValue.objects.filter( pk=selected_types[0], kind=ChoiceValue.Kind.DEVICE_TYPE, ).first() if len(selected_types) == 1 and selected_types[0].isascii() and selected_types[0].isdigit() and len(selected_types[0]) <= 18 else None preference, _ = UserPreference.objects.get_or_create(user=request.user) try: page_size = int(request.GET.get("page_size") or preference.page_size) except ValueError: page_size = 50 page_size = page_size if page_size in {25, 50, 100} else 50 page = Paginator(queryset, page_size).get_page(request.GET.get("page")) params = request.GET.copy() params.pop("page", None) navigation_params = request.GET.copy() for key in ("partial", "exact", "open"): navigation_params.pop(key, None) control_keys = {"q", "exact", "page", "page_size", "sort", "dir", "open", "partial"} has_filters = any(key not in control_keys and value for key in request.GET for value in request.GET.getlist(key)) active_filter_count = sum( 1 for key in request.GET if key not in control_keys and any(value for value in request.GET.getlist(key)) ) return { "page_obj": page, "page_size": page_size, "query_string": params.urlencode(), "archived": archived, "can_edit_devices": can_edit(request.user), "statuses": ChoiceValue.objects.filter(kind=ChoiceValue.Kind.STATUS, active=True), "stages": Device.Stage.choices, "types": ChoiceValue.objects.filter(kind=ChoiceValue.Kind.DEVICE_TYPE, active=True), "visible_hardware_fields": applicable_fields(selected_type) if selected_type else COMPUTER_FIELDS, "recipients": ChoiceValue.objects.filter(kind=ChoiceValue.Kind.RECIPIENT, active=True), "employees": Employee.objects.filter(active=True), "cpu_vendors": CPU_VENDOR_CHOICES, "cpu_families": CPU_FAMILY_CHOICES, "custom_columns": FieldDefinition.objects.filter(active=True, show_in_list=True), "custom_filters": FieldDefinition.objects.filter(active=True, filterable=True).prefetch_related("options"), "total_count": page.paginator.count, "has_filters": has_filters, "active_filter_count": active_filter_count, "preference": preference, "navigation_query": navigation_params.urlencode(), } @login_required def device_list(request: HttpRequest) -> HttpResponse: context = _list_context(request) if request.headers.get("HX-Request") == "true" or request.GET.get("partial") == "1": return render(request, "inventory/_device_table.html", context) return render(request, "inventory/device_list.html", context) @require_POST @role_required(UserProfile.Role.ADMIN, UserProfile.Role.EDITOR) def bulk_update_devices(request: HttpRequest) -> HttpResponse: try: selected_ids = {int(value) for value in request.POST.getlist("device_ids") if value} except (TypeError, ValueError): selected_ids = set() status_value = request.POST.get("status", "").strip() recipient_value = request.POST.get("recipient", "").strip() return_query = QueryDict(request.POST.get("return_query", "")[:8000]).urlencode() return_url = reverse("inventory:device-list") if return_query: return_url = f"{return_url}?{return_query}" if not selected_ids: messages.error(request, "Bitte mindestens ein Gerät auswählen.") return redirect(return_url) if not status_value and not recipient_value: messages.error(request, "Bitte einen Status oder einen Empfänger auswählen.") return redirect(return_url) devices = list(Device.objects.filter(pk__in=selected_ids, archived_at__isnull=True).order_by("pk")) if len(devices) != len(selected_ids): messages.error(request, "Mindestens ein ausgewähltes Gerät ist nicht mehr verfügbar.") return redirect(return_url) changed_devices = 0 try: with transaction.atomic(): for device in devices: changed = False if status_value and str(device.status_id) != status_value: device = update_device_field(device, "status", status_value, request.user) changed = True if recipient_value: raw_recipient = "" if recipient_value == "__clear__" else recipient_value current_recipient = "" if device.recipient_id is None else str(device.recipient_id) if current_recipient != raw_recipient: update_device_field(device, "recipient", raw_recipient, request.user) changed = True changed_devices += int(changed) except ValidationError as exc: messages.error(request, exc.messages[0]) return redirect(return_url) if changed_devices: messages.success(request, f"{changed_devices} Geräte wurden gemeinsam aktualisiert.") else: messages.info(request, "Die ausgewählten Geräte hatten bereits diese Werte.") return redirect(return_url) def _device_navigation(request: HttpRequest, device: Device) -> dict: params = request.GET.copy() for key in ("drawer", "partial", "exact", "open"): params.pop(key, None) ordered_ids = list(device_queryset(params).values_list("pk", flat=True)) try: position = ordered_ids.index(device.pk) except ValueError: position = -1 previous_id = ordered_ids[position - 1] if position > 0 else None next_id = ordered_ids[position + 1] if position >= 0 and position + 1 < len(ordered_ids) else None query = params.urlencode() list_url = reverse("inventory:device-list") if query: list_url = f"{list_url}?{query}" return { "previous_device_id": previous_id, "next_device_id": next_id, "navigation_query": query, "list_url": list_url, "navigation_position": position + 1 if position >= 0 else None, "navigation_total": len(ordered_ids), } def _hardware_type_map() -> dict[str, list[str]]: choices = ChoiceValue.objects.filter(kind=ChoiceValue.Kind.DEVICE_TYPE) return {"": sorted(COMPUTER_FIELDS), **{ str(choice.pk): sorted(applicable_fields(choice)) for choice in choices }} @role_required(UserProfile.Role.ADMIN, UserProfile.Role.EDITOR) def quick_add(request: HttpRequest) -> HttpResponse: initial_status = ChoiceValue.objects.filter(kind=ChoiceValue.Kind.STATUS, label="Eingang", active=True).first() form = QuickAddForm( request.POST or None, initial={"number": NumberSequence.suggested_device_number(), "quantity": 1, "status": initial_status, "date": timezone.localdate()}, ) if request.method == "POST" and form.is_valid(): try: devices = create_device_batch( first_number=form.cleaned_data["number"], quantity=form.cleaned_data["quantity"], shelf=form.cleaned_data["shelf"], status=form.cleaned_data["status"], device_date=form.cleaned_data["date"], device_type=form.cleaned_data["device_type"], employee=form.cleaned_data["employee"], donor=form.cleaned_data["donor"], user=request.user, model=form.cleaned_data["model"], hardware_values=form.selected_hardware_values(), ) except (IntegrityError, ValidationError) as exc: form.add_error("number", "Diese Gerätenummer wurde inzwischen vergeben. Bitte wählen Sie eine andere Nummer.") else: if len(devices) == 1: messages.success(request, f"Gerät #{devices[0].number} wurde angelegt. Sie können direkt das nächste Gerät erfassen.") else: messages.success(request, f"{len(devices)} Geräte wurden als #{devices[0].number} bis #{devices[-1].number} angelegt.") return redirect("inventory:quick-add") return render(request, "inventory/quick_add.html", { "form": form, "type_field_map": _hardware_type_map(), "all_hardware_fields": HARDWARE_FIELDS, }) @login_required @transaction.atomic def device_detail(request: HttpRequest, pk: int) -> HttpResponse: devices = Device.objects.select_related("status", "recipient", "location", "device_type", "employee", *SPECIFICATION_RELATIONS) if request.method == "POST": devices = devices.select_for_update(of=("self",)) device = get_object_or_404(devices, pk=pk) navigation = _device_navigation(request, device) definitions = FieldDefinition.objects.filter(active=True).prefetch_related("options") if request.method == "POST": if not can_edit(request.user): return JsonResponse({"error": "Keine Berechtigung."}, status=403) expected = request.POST.get("expected_updated_at") revision = request.POST.get("expected_revision") try: if not expected and not revision: raise RuntimeError("conflict") check_device_version(device, expected, revision) except RuntimeError: messages.error(request, "Das Gerät wurde zwischenzeitlich geändert. Bitte neu laden.") target = reverse("inventory:device-detail", args=[device.pk]) if navigation["navigation_query"]: target = f'{target}?{navigation["navigation_query"]}' return redirect(target) old_values = {name: getattr(device, name) for name in FIELD_LABELS} old_custom = dict(device.custom_values) form = DeviceForm(request.POST, instance=device) if form.is_valid(): with transaction.atomic(): device = form.save(commit=False) clear_inapplicable(device) custom_values = dict(device.custom_values) try: for definition in definitions: key = f"custom_{definition.key}" raw = request.POST.getlist(key) if definition.field_type == FieldDefinition.FieldType.MULTI_SELECT else request.POST.get(key) custom_values[definition.key] = validate_custom_value(definition, raw) except (ValidationError, ValueError) as exc: form.add_error(None, exc) else: device.custom_values = custom_values device.full_clean() device.save() for name, label in FIELD_LABELS.items(): old = old_values[name] new = getattr(device, name) old_data = serialize_audit_value(name, old) new_data = serialize_audit_value(name, new) if old_data != new_data: add_audit( request.user, device, AuditLog.Action.UPDATE, label, display_device_value(name, old), display_device_value(name, new), field_key=name, old_data=old_data, new_data=new_data, ) for definition in definitions: old = old_custom.get(definition.key) new = custom_values.get(definition.key) if old != new: add_audit( request.user, device, AuditLog.Action.UPDATE, definition.name, old, new, field_key=f"custom:{definition.key}", old_data=old, new_data=new, ) messages.success(request, "Änderungen gespeichert.") target = reverse("inventory:device-detail", args=[device.pk]) if navigation["navigation_query"]: target = f'{target}?{navigation["navigation_query"]}' return redirect(target) else: form = DeviceForm(instance=device) if not can_edit(request.user): for field in form.fields.values(): field.disabled = True type_field_map = _hardware_type_map() selected_type_id = str(form["device_type"].value() or "") context = { "device": device, "form": form, "definitions": definitions, "can_edit_devices": can_edit(request.user), "hardware_fields": type_field_map.get(selected_type_id, applicable_fields(device.device_type)), "all_hardware_fields": HARDWARE_FIELDS, "type_field_map": type_field_map, "next_stages": [(value, label) for value, label in Device.Stage.choices if value in TRANSITIONS[device.stage]], "tasks": device.tasks.select_related("assigned_to", "completed_by"), "task_assignees": get_user_model().objects.filter(is_active=True).filter( Q(is_superuser=True) | Q(profile__role__in=[UserProfile.Role.ADMIN, UserProfile.Role.EDITOR]), ).order_by("username"), "workflow_events": device.workflow_events.select_related("user")[:10], "inspections": device.inspections.select_related("recorded_by")[:5], "repairs": device.repairs.select_related("recorded_by")[:5], "ram_suggestions": ChoiceValue.objects.filter(kind=ChoiceValue.Kind.RAM, active=True), "ssd_suggestions": ChoiceValue.objects.filter(kind=ChoiceValue.Kind.SSD, active=True), "hdd_suggestions": ChoiceValue.objects.filter(kind=ChoiceValue.Kind.HDD, active=True), "display_suggestions": ChoiceValue.objects.filter(kind=ChoiceValue.Kind.DISPLAY, active=True), **navigation, } return render(request, "inventory/device_detail.html", context) @require_POST @role_required(UserProfile.Role.ADMIN, UserProfile.Role.EDITOR) def device_advance(request: HttpRequest, pk: int) -> HttpResponse: try: advance_device(pk, request.POST.get("stage", ""), user=request.user, note=request.POST.get("note", "").strip(), expected_revision=request.POST.get("expected_revision")) messages.success(request, "Arbeitsschritt aktualisiert.") except (ValidationError, Device.DoesNotExist) as exc: messages.error(request, exc.messages[0] if isinstance(exc, ValidationError) else "Gerät nicht gefunden.") return workflow_action_redirect(request, pk) @require_POST @role_required(UserProfile.Role.ADMIN, UserProfile.Role.EDITOR) def device_task_add(request: HttpRequest, pk: int) -> HttpResponse: from django.utils.dateparse import parse_date raw_due = request.POST.get("due_date", "").strip() due_date = parse_date(raw_due) if raw_due else None if raw_due and due_date is None: messages.error(request, "Ungültiges Fälligkeitsdatum.") return workflow_action_redirect(request, pk) assigned_to = None if raw_assignee := request.POST.get("assigned_to", "").strip(): if not raw_assignee.isascii() or not raw_assignee.isdecimal(): messages.error(request, "Ungültige zuständige Person.") return workflow_action_redirect(request, pk) assigned_to = get_user_model().objects.filter(pk=raw_assignee, is_active=True).filter( Q(is_superuser=True) | Q(profile__role__in=[UserProfile.Role.ADMIN, UserProfile.Role.EDITOR]), ).first() if assigned_to is None: messages.error(request, "Ungültige zuständige Person.") return workflow_action_redirect(request, pk) try: create_task(pk, title=request.POST.get("title", ""), description=request.POST.get("description", ""), due_date=due_date, assigned_to=assigned_to, user=request.user) messages.success(request, "Aufgabe angelegt.") except (ValidationError, Device.DoesNotExist) as exc: messages.error(request, exc.messages[0] if isinstance(exc, ValidationError) else "Gerät nicht gefunden.") return workflow_action_redirect(request, pk) @require_POST @role_required(UserProfile.Role.ADMIN, UserProfile.Role.EDITOR) def device_task_complete(request: HttpRequest, pk: int, task_id: int) -> HttpResponse: try: complete_task(pk, task_id, user=request.user) messages.success(request, "Aufgabe erledigt.") except (ValidationError, Device.DoesNotExist, DeviceTask.DoesNotExist) as exc: messages.error(request, exc.messages[0] if isinstance(exc, ValidationError) else "Aufgabe nicht gefunden.") return workflow_action_redirect(request, pk) @require_POST @role_required(UserProfile.Role.ADMIN, UserProfile.Role.EDITOR) def device_inspection_save(request: HttpRequest, pk: int) -> HttpResponse: device = get_object_or_404(Device, pk=pk, archived_at__isnull=True) form = InspectionForm(request.POST, device=device) if form.is_valid(): try: save_inspection(pk, form.cleaned_data, user=request.user, expected_revision=request.POST.get("expected_revision")) messages.success(request, "Inspektion gespeichert.") except (ValidationError, Device.DoesNotExist) as exc: messages.error(request, exc.messages[0] if isinstance(exc, ValidationError) else "Gerät nicht gefunden.") else: messages.error(request, form.errors.as_text()) return workflow_action_redirect(request, pk) @require_POST @role_required(UserProfile.Role.ADMIN, UserProfile.Role.EDITOR) def device_repair_add(request: HttpRequest, pk: int) -> HttpResponse: form = RepairForm(request.POST) if form.is_valid(): try: record_repair(pk, form.cleaned_data["work_done"], user=request.user, expected_revision=request.POST.get("expected_revision")) messages.success(request, "Reparatur dokumentiert.") except (ValidationError, Device.DoesNotExist) as exc: messages.error(request, exc.messages[0] if isinstance(exc, ValidationError) else "Gerät nicht gefunden.") else: messages.error(request, form.errors.as_text()) return workflow_action_redirect(request, pk) @require_POST @role_required(UserProfile.Role.ADMIN, UserProfile.Role.EDITOR) def device_storage_assign(request: HttpRequest, pk: int) -> HttpResponse: form = StorageRecipientForm(request.POST) if form.is_valid(): try: assign_storage_recipient(pk, form.cleaned_data["recipient"], user=request.user, expected_revision=request.POST.get("expected_revision")) messages.success(request, "Empfänger gespeichert.") except (ValidationError, Device.DoesNotExist) as exc: messages.error(request, exc.messages[0] if isinstance(exc, ValidationError) else "Gerät nicht gefunden.") else: messages.error(request, form.errors.as_text()) return workflow_action_redirect(request, pk) def workflow_action_redirect(request: HttpRequest, pk: int) -> HttpResponse: if request.POST.get("return_to") == "workflow": return redirect("inventory:workflow-board") return redirect("inventory:device-detail", pk=pk) @login_required def workflow_board(request: HttpRequest) -> HttpResponse: open_tasks = Prefetch( "tasks", queryset=DeviceTask.objects.filter(state=DeviceTask.State.OPEN).order_by("due_date", "created_at"), to_attr="open_tasks", ) columns = [] for stage, label in Device.Stage.choices: queryset = Device.objects.filter(stage=stage, archived_at__isnull=True).select_related( "device_type", "status", "employee", "recipient", *SPECIFICATION_RELATIONS, ) devices = list(queryset.order_by("-updated_at", "-number").prefetch_related( open_tasks, Prefetch("inspections", queryset=InspectionRecord.objects.select_related("recorded_by").order_by("-created_at"), to_attr="inspection_history"), Prefetch("repairs", queryset=RepairRecord.objects.select_related("recorded_by").order_by("-created_at"), to_attr="repair_history"), )[:30]) for device in devices: device.open_task_count = len(device.open_tasks) device.next_stages = [(value, stage_label) for value, stage_label in Device.Stage.choices if value in TRANSITIONS[stage]] if stage == Device.Stage.INSPECTION: device.stage_form = InspectionForm(device=device) elif stage == Device.Stage.REPAIR: device.stage_form = RepairForm() elif stage == Device.Stage.STORAGE: device.stage_form = StorageRecipientForm(initial={"recipient": device.recipient.label if device.recipient else ""}) columns.append({ "stage": stage, "label": label, "total": queryset.count(), "devices": devices, }) return render(request, "inventory/workflow_board.html", { "columns": columns, "can_edit_devices": can_edit(request.user), }) @login_required def task_list(request: HttpRequest) -> HttpResponse: state = request.GET.get("state", "OPEN") if state not in DeviceTask.State.values: state = DeviceTask.State.OPEN tasks = DeviceTask.objects.select_related("device", "assigned_to", "created_by").filter( state=state, device__archived_at__isnull=True, ).order_by("due_date", "-created_at") return render(request, "inventory/task_list.html", { "tasks": tasks, "state": state, "can_edit_devices": can_edit(request.user), }) @require_POST @role_required(UserProfile.Role.ADMIN, UserProfile.Role.EDITOR) def inline_update(request: HttpRequest, pk: int) -> JsonResponse: device = get_object_or_404(Device, pk=pk, archived_at__isnull=True) try: payload = json.loads(request.body) if not isinstance(payload, dict): raise ValidationError("Ein JSON-Objekt ist erforderlich.") if not payload.get("expected_updated_at") and payload.get("expected_revision") is None: raise ValidationError("Die Datensatzversion fehlt. Bitte neu laden.") updated = update_device_field(device, payload.get("field", ""), payload.get("value"), request.user, payload.get("expected_updated_at"), expected_revision=payload.get("expected_revision")) field = payload["field"] value = getattr(updated, field) display = display_device_value(field, value) return JsonResponse({ "ok": True, "display": display, "value": serialize_audit_value(field, value), "color": value.color if field == "status" and value else "", "updated_at": updated.updated_at.isoformat(), "revision": updated.revision, }) except RuntimeError: return JsonResponse({"error": "Das Gerät wurde inzwischen geändert. Bitte die Ansicht neu laden."}, status=409) except (ValidationError, ValueError, KeyError, json.JSONDecodeError) as exc: message = exc.messages[0] if isinstance(exc, ValidationError) else str(exc) return JsonResponse({"error": message}, status=400) @require_POST @role_required(UserProfile.Role.ADMIN, UserProfile.Role.EDITOR) @transaction.atomic def archive_device(request: HttpRequest, pk: int) -> HttpResponse: device = get_object_or_404(Device.objects.select_for_update(), pk=pk, archived_at__isnull=True) device.archived_at = timezone.now() device.archived_by = request.user device.save(update_fields=["archived_at", "archived_by", "updated_at"]) add_audit(request.user, device, AuditLog.Action.ARCHIVE) messages.success(request, f"Gerät #{device.number} wurde archiviert.") return redirect("inventory:device-list") @role_required(UserProfile.Role.ADMIN) def archive_list(request: HttpRequest) -> HttpResponse: return render(request, "inventory/archive_list.html", _list_context(request, archived=True)) @require_POST @role_required(UserProfile.Role.ADMIN) @transaction.atomic def restore_device(request: HttpRequest, pk: int) -> HttpResponse: device = get_object_or_404(Device.objects.select_for_update(), pk=pk, archived_at__isnull=False) device.archived_at = None device.archived_by = None device.save(update_fields=["archived_at", "archived_by", "updated_at"]) add_audit(request.user, device, AuditLog.Action.RESTORE) messages.success(request, f"Gerät #{device.number} wurde wiederhergestellt.") return redirect("inventory:archive-list") @require_GET @login_required def autocomplete(request: HttpRequest, field: str) -> JsonResponse: term = request.GET.get("q", "").strip() results: list[dict] = [] kind_map = {"status": ChoiceValue.Kind.STATUS, "recipient": ChoiceValue.Kind.RECIPIENT, "location": ChoiceValue.Kind.LOCATION, "device_type": ChoiceValue.Kind.DEVICE_TYPE} if field in kind_map: rows = ChoiceValue.objects.filter(kind=kind_map[field], active=True, label__icontains=term).order_by("sort_order", "label") results = [{"id": row.pk, "label": row.label} for row in rows] elif field == "employee": rows = Employee.objects.filter(active=True).filter(models.Q(code__icontains=term) | models.Q(name__icontains=term)).order_by("code") results = [{"id": row.pk, "label": f"{row.code} – {row.name}"} for row in rows] elif field == "cpu": existing = (cpu for model in (LaptopSpecification, DesktopSpecification, AllInOneSpecification, ComputerSpecification) for cpu in model.objects.exclude(cpu="").values_list("cpu", flat=True).distinct()) results = [{"id": value, "label": value} for value in suggest_cpus(term, existing)] return JsonResponse({"results": results}) @login_required def audit_list(request: HttpRequest) -> HttpResponse: logs = AuditLog.objects.select_related("user", "device", "reverted_by") query = request.GET.get("q", "").strip() if query: logs = logs.filter(models.Q(field__icontains=query) | models.Q(old_value__icontains=query) | models.Q(new_value__icontains=query)) page = Paginator(logs, 50).get_page(request.GET.get("page")) return render(request, "inventory/audit_list.html", {"page_obj": page, "can_revert_audits": can_edit(request.user)}) @require_POST @role_required(UserProfile.Role.ADMIN, UserProfile.Role.EDITOR) def audit_revert(request: HttpRequest, pk: int) -> HttpResponse: log = get_object_or_404(AuditLog, pk=pk) try: revert_audit_change(log, request.user) except ValidationError as exc: messages.error(request, exc.messages[0]) else: messages.success(request, f"Die Änderung an Gerät #{log.device.number} wurde zurückgesetzt.") return redirect("inventory:audit-list") @role_required(UserProfile.Role.ADMIN) def configuration(request: HttpRequest) -> HttpResponse: choices = ChoiceValue.objects.all() definitions = FieldDefinition.objects.prefetch_related("options") employees = Employee.objects.all() return render(request, "inventory/configuration.html", { "choices": choices, "definitions": definitions, "employees": employees, "choice_kinds": ChoiceValue.Kind.choices, }) @role_required(UserProfile.Role.ADMIN) def employee_edit(request: HttpRequest, pk: int | None = None) -> HttpResponse: instance = get_object_or_404(Employee, pk=pk) if pk else None form = EmployeeForm(request.POST or None, instance=instance) if request.method == "POST" and form.is_valid(): employee = form.save() messages.success(request, f"Bearbeiter {employee.name} wurde gespeichert.") return redirect("inventory:configuration") return render(request, "inventory/simple_form.html", { "form": form, "title": "Bearbeiter bearbeiten" if instance else "Bearbeiter hinzufügen", "cancel_url": reverse("inventory:configuration"), }) @require_POST @role_required(UserProfile.Role.ADMIN) def employee_remove(request: HttpRequest, pk: int) -> HttpResponse: employee = get_object_or_404(Employee, pk=pk) name = employee.name if employee.devices.exists(): employee.active = False employee.save(update_fields=["active", "updated_at"]) messages.success(request, f"Bearbeiter {name} wurde aus der Auswahl entfernt. Bestehende Gerätezuordnungen bleiben erhalten.") else: employee.delete() messages.success(request, f"Bearbeiter {name} wurde entfernt.") return redirect("inventory:configuration") @role_required(UserProfile.Role.ADMIN) def user_list(request: HttpRequest) -> HttpResponse: from django.contrib.auth.models import User users = User.objects.select_related("profile").order_by("username") return render(request, "inventory/user_list.html", {"users": users}) @role_required(UserProfile.Role.ADMIN) def user_edit(request: HttpRequest, pk: int | None = None) -> HttpResponse: from django.contrib.auth.models import User instance = get_object_or_404(User, pk=pk) if pk else None form = UserManagementForm(request.POST or None, instance=instance) if request.method == "POST" and form.is_valid(): user = form.save() messages.success(request, f"Konto {user.username} gespeichert.") return redirect("inventory:user-list") return render(request, "inventory/simple_form.html", {"form": form, "title": "Benutzer bearbeiten" if instance else "Benutzer anlegen", "cancel_url": reverse("inventory:user-list")}) @role_required(UserProfile.Role.ADMIN) def choice_edit(request: HttpRequest, pk: int | None = None) -> HttpResponse: instance = get_object_or_404(ChoiceValue, pk=pk) if pk else None form = ChoiceValueForm(request.POST or None, instance=instance) if request.method == "POST" and form.is_valid(): form.save() messages.success(request, "Auswahlwert gespeichert.") return redirect("inventory:configuration") return render(request, "inventory/simple_form.html", {"form": form, "title": "Auswahlwert bearbeiten" if instance else "Auswahlwert anlegen"}) @role_required(UserProfile.Role.ADMIN) def field_edit(request: HttpRequest, pk: int | None = None) -> HttpResponse: instance = get_object_or_404(FieldDefinition, pk=pk) if pk else None form = FieldDefinitionForm(request.POST or None, instance=instance) if request.method == "POST" and form.is_valid(): form.save() messages.success(request, "Benutzerdefiniertes Feld gespeichert.") return redirect("inventory:configuration") return render(request, "inventory/simple_form.html", {"form": form, "title": "Feld bearbeiten" if instance else "Feld anlegen"}) @role_required(UserProfile.Role.ADMIN) def import_excel(request: HttpRequest) -> HttpResponse: form = ExcelUploadForm(request.POST or None, request.FILES or None) if request.method == "POST" and form.is_valid(): upload = form.cleaned_data["file"] try: headers, preview = analyze_workbook(upload) except Exception as exc: form.add_error("file", f"Datei konnte nicht gelesen werden: {exc}") else: upload.seek(0) job = ImportJob.objects.create(user=request.user, file=upload, original_name=upload.name, headers=headers, preview_rows=preview) return redirect("inventory:import-map", pk=job.pk) return render(request, "inventory/import_upload.html", {"form": form}) @role_required(UserProfile.Role.ADMIN) def import_map(request: HttpRequest, pk: int) -> HttpResponse: job = get_object_or_404(ImportJob, pk=pk, user=request.user, status=ImportJob.Status.UPLOADED) suggestions = suggested_mapping(job.headers) fields = [("", "Nicht importieren")] + [(key, label) for key, label in [ ("number", "Nr."), ("shelf", "Regal"), ("status", "Status"), ("date", "Datum"), ("recipient", "Empfänger"), ("location", "Ort"), ("ticket", "Ticket"), ("device_type", "Typ"), ("price", "Preis"), ("battery", "Akku"), ("wifi", "WLAN"), ("lan", "LAN"), ("display_inches", "Zoll"), ("ram_gb", "RAM"), ("ssd_gb", "SSD"), ("hdd_gb", "HDD"), ("cpu", "CPU"), ("hwid", "HWID"), ("ports", "Anschlüsse"), ("resolution", "Auflösung"), ("print_technology", "Druckverfahren"), ("duplex", "Duplex"), ("color_printing", "Farbdruck"), ("model", "Modell"), ("employee", "Bearbeiter"), ("donor", "Spender"), ("comment", "Kommentar")]] if request.method == "POST": mapping = {index: request.POST.get(f"column_{index}", "") for index in range(len(job.headers))} try: with job.file.open("rb") as source: result = import_workbook(source, mapping, request.user) except Exception as exc: job.status = ImportJob.Status.FAILED job.result = {"error": str(exc)} job.save(update_fields=["status", "result", "updated_at"]) messages.error(request, f"Import abgebrochen: {exc}") else: job.status = ImportJob.Status.COMPLETED job.result = result job.save(update_fields=["status", "result", "updated_at"]) messages.success(request, f"Import abgeschlossen: {result['created']} Geräte angelegt.") return redirect("inventory:device-list") return render(request, "inventory/import_map.html", {"job": job, "suggestions": suggestions, "fields": fields}) @login_required def export_devices(request: HttpRequest, file_format: str) -> HttpResponse: queryset = device_queryset(request.GET) if request.GET.get("scope") == "filtered" else device_queryset(type("Params", (), {"get": lambda self, key, default=None: default, "getlist": lambda self, key: []})()) if file_format == "csv": return csv_response(queryset) if file_format == "xlsx": return xlsx_response(queryset) if file_format == "ods": return ods_response(queryset) return JsonResponse({"error": "Unbekanntes Exportformat."}, status=400) @require_POST @login_required def save_preferences(request: HttpRequest) -> JsonResponse: try: payload = json.loads(request.body) if not isinstance(payload, dict): raise ValueError allowed_columns = {"number", *FIELD_LABELS} | {f"custom_{key}" for key in FieldDefinition.objects.values_list("key", flat=True)} if set(payload) - {"page_size", "visible_columns", "column_order", "column_widths"}: raise ValueError if "page_size" in payload and (type(payload["page_size"]) is not int or payload["page_size"] not in {25, 50, 100}): raise ValueError for key in ("visible_columns", "column_order"): if key in payload: value = payload[key] if not isinstance(value, list) or len(value) > len(allowed_columns) or any(not isinstance(item, str) or item not in allowed_columns for item in value): raise ValueError if len(set(value)) != len(value): raise ValueError if "column_widths" in payload: widths = payload["column_widths"] if not isinstance(widths, dict) or any(key not in allowed_columns or type(value) is not int or not 40 <= value <= 2000 for key, value in widths.items()): raise ValueError preference, _ = UserPreference.objects.get_or_create(user=request.user) if payload.get("page_size") in {25, 50, 100}: preference.page_size = payload["page_size"] for key in ["visible_columns", "column_order", "column_widths"]: if key in payload: setattr(preference, key, payload[key]) preference.save() return JsonResponse({"ok": True}) except (ValueError, json.JSONDecodeError): return JsonResponse({"error": "Ungültige Einstellungen."}, status=400) def health(request: HttpRequest) -> JsonResponse: try: Device.objects.values("pk").first() return JsonResponse({"status": "ok"}) except Exception: return JsonResponse({"status": "unavailable"}, status=503)