from rest_framework import viewsets, status, mixins from rest_framework.decorators import action from rest_framework.response import Response from rest_framework.views import APIView from rest_framework.permissions import IsAuthenticated, AllowAny from .models import (ComPortReading, User, Report, ScaleProtocol, License, StationSlot, generate_doc_number) from .serializers import ( ComPortReadingSerializer, UserSerializer, UserDetailSerializer, ChangePasswordSerializer, VehicleSerializer, NomenclatureSerializer, NomenclatureEntrySerializer, ReportSerializer, ExtraCombinationSerializer, ScaleProtocolSerializer, ) from django.utils import timezone from vehicles.models import Vehicle, ExtraCombination, combination_hash from nomenclatures.models import Nomenclature, NomenclatureEntry from scalesapp.sse import sse_broadcast_update from . import licensing class IsAdminUser(IsAuthenticated): def has_permission(self, request, view): return bool(request.user and request.user.is_authenticated and request.user.is_admin) class UserViewSet(viewsets.ModelViewSet): """ API endpoint for user management. list: Get all users create: Create a new user retrieve: Get a specific user update: Update a user destroy: Soft-delete a user (sets is_active=False) me: Get current authenticated user change_password: Change password for current user """ queryset = User.objects.all() serializer_class = UserSerializer filterset_fields = ['role', 'is_admin', 'is_active'] ordering = ['username'] def get_permissions(self): if self.action in ('me', 'change_password'): return [IsAuthenticated()] return [IsAuthenticated(), IsAdminUser()] def destroy(self, request, *args, **kwargs): instance = self.get_object() instance.is_active = False instance.save() return Response(status=status.HTTP_204_NO_CONTENT) @action(detail=False, methods=['get']) def me(self, request): """Get current authenticated user details""" serializer = UserDetailSerializer(request.user) return Response(serializer.data) @action(detail=False, methods=['post'], url_path='change-password', url_name='change_password') def change_password(self, request): """Change password for current user""" serializer = ChangePasswordSerializer( data=request.data, context={'request': request} ) if serializer.is_valid(): serializer.save() return Response({ 'detail': 'Password updated successfully' }, status=status.HTTP_200_OK) return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST) class ComPortReadingViewSet(viewsets.ModelViewSet): """ API endpoint for COM port readings. list: Get all readings with optional filtering create: Create a new reading (allows unauthenticated for serial bridge) retrieve: Get a specific reading destroy: Delete a reading latest: Get the latest reading by_port: Get readings for a specific port """ queryset = ComPortReading.objects.all() serializer_class = ComPortReadingSerializer filterset_fields = ['port'] ordering = ['-timestamp'] def get_permissions(self): """Allow unauthenticated POST for serial bridge""" if self.action == 'create': return [AllowAny()] return [IsAuthenticated()] @action(detail=False, methods=['get']) def latest(self, request): """Get the latest reading""" reading = ComPortReading.objects.first() if reading: serializer = self.get_serializer(reading) return Response(serializer.data) return Response({'detail': 'No readings yet'}, status=status.HTTP_404_NOT_FOUND) @action(detail=False, methods=['get']) def by_port(self, request): """Get readings for a specific port""" port = request.query_params.get('port', None) if not port: return Response( {'detail': 'port parameter is required'}, status=status.HTTP_400_BAD_REQUEST ) readings = ComPortReading.objects.filter(port=port) serializer = self.get_serializer(readings, many=True) return Response(serializer.data) def get_client_ip(self, request): """Get client IP address""" x_forwarded_for = request.META.get('HTTP_X_FORWARDED_FOR') if x_forwarded_for: ip = x_forwarded_for.split(',')[0] else: ip = request.META.get('REMOTE_ADDR') return ip def perform_create(self, serializer): """Save the reading with client IP""" serializer.save(source_ip=self.get_client_ip(self.request)) def _maybe_finalize(vehicle, user): """If both tare and gross are set and net hasn't been calculated yet, compute net and assign doc_number.""" if vehicle.tare is not None and vehicle.gross is not None and vehicle.net is None: vehicle.net = vehicle.gross - vehicle.tare vehicle.net_date = timezone.now() vehicle.net_user = user vehicle.doc_number = generate_doc_number(timezone.now().year) vehicle.save() def _assign_weighing(vehicle, value, user, manual=False, note=None): """Record a weighing. Assigns by weight: lighter becomes tare, heavier becomes gross. The first weighing is parked in tare; the second may push it to gross if the new reading is lighter, since trucks may arrive either empty (inbound) or loaded (outbound). """ now = timezone.now() def apply(prefix): setattr(vehicle, prefix, value) setattr(vehicle, f'{prefix}_date', now) setattr(vehicle, f'{prefix}_user', user) setattr(vehicle, f'{prefix}_manual', manual) setattr(vehicle, f'{prefix}_manual_user', user if manual else None) setattr(vehicle, f'{prefix}_manual_date', now if manual else None) setattr(vehicle, f'{prefix}_manual_note', note if manual else None) def move(src_prefix, dst_prefix): """Move a recorded weighing to the other slot, carrying all its provenance.""" for suffix in ('', '_date', '_user', '_manual', '_manual_user', '_manual_date', '_manual_note'): setattr(vehicle, f'{dst_prefix}{suffix}', getattr(vehicle, f'{src_prefix}{suffix}')) if vehicle.tare is not None and vehicle.gross is not None: raise ValueError('Vehicle already has both weights.') if vehicle.tare is None and vehicle.gross is None: apply('tare') elif vehicle.gross is None: # tare holds the first weighing - keep the lighter one as tare if value >= vehicle.tare: apply('gross') else: move('tare', 'gross') apply('tare') else: # gross set but tare empty (only reachable via direct API use) if value <= vehicle.gross: apply('tare') else: move('gross', 'tare') apply('gross') vehicle.save() _maybe_finalize(vehicle, user) def _record_extra_combination(extra_data): """Record the lookup-only projection of a vehicle's extra data as a reusable combination.""" if not extra_data: return lookup_codes = set( Nomenclature.objects.filter(applies_to='vehicle', kind=Nomenclature.LOOKUP) .values_list('code', flat=True) ) combo = {k: v for k, v in extra_data.items() if k in lookup_codes and v is not None} if not combo: return existing, created = ExtraCombination.objects.get_or_create( data_hash=combination_hash(combo), defaults={'data': combo} ) if not created: existing.use_count += 1 existing.last_used = timezone.now() existing.save() action = 'created' if created else 'updated' sse_broadcast_update('extra_combination', action, ExtraCombinationSerializer(existing).data) class VehicleViewSet(viewsets.ModelViewSet): """ API endpoint for vehicle management. list: Get all vehicles with pagination create: Create a new vehicle with optional extra data retrieve: Get a specific vehicle by ID update: Full update of vehicle and extra data partial_update: Partial update of vehicle destroy: Delete a vehicle (cascades to VehicleExtra) by_number: Get vehicle by vehicle_number measure: Record a weighing (app assigns tare vs gross by weight) """ queryset = Vehicle.objects.select_related('extra', 'tare_user', 'gross_user').all() serializer_class = VehicleSerializer permission_classes = [IsAuthenticated] filterset_fields = ['vehicle_number'] search_fields = ['vehicle_number'] ordering = ['vehicle_number'] def perform_create(self, serializer): instance = serializer.save() data = VehicleSerializer(instance).data sse_broadcast_update('vehicle', 'created', data) _record_extra_combination(instance.extra.data if hasattr(instance, 'extra') else None) def perform_update(self, serializer): instance = serializer.save() data = VehicleSerializer(instance).data sse_broadcast_update('vehicle', 'updated', data) _record_extra_combination(instance.extra.data if hasattr(instance, 'extra') else None) def perform_destroy(self, instance): data = VehicleSerializer(instance).data instance.delete() sse_broadcast_update('vehicle', 'deleted', data) @action(detail=False, methods=['get'], url_path='by-number') def by_number(self, request): """Get vehicle by vehicle_number query parameter""" vehicle_number = request.query_params.get('vehicle_number', None) if not vehicle_number: return Response( {'detail': 'vehicle_number parameter is required'}, status=status.HTTP_400_BAD_REQUEST ) try: vehicle = Vehicle.objects.get(vehicle_number=vehicle_number) serializer = self.get_serializer(vehicle) return Response(serializer.data) except Vehicle.DoesNotExist: return Response( {'detail': 'Vehicle not found'}, status=status.HTTP_404_NOT_FOUND ) @action(detail=True, methods=['post'], url_path='measure') def measure(self, request, pk=None): """Record a weighing. The app assigns tare vs gross by weight.""" if not request.user.can_measure: return Response({'detail': 'You do not have permission to measure.'}, status=status.HTTP_403_FORBIDDEN) # Licence enforcement. The UI also disables the button, but this is the # actual rule - the button state is only a convenience. station_key = (request.data.get('station_key') or '').strip() if not _station_allowed(station_key): limit = licensing.effective_station_limit() return Response( {'detail': f'This station is not licensed to measure ' f'(limit: {limit} stations).'}, status=status.HTTP_403_FORBIDDEN) manual = bool(request.data.get('manual', False)) if manual and not request.user.can_manually_measure: return Response({'detail': 'You do not have permission to measure manually.'}, status=status.HTTP_403_FORBIDDEN) note = (request.data.get('note') or '').strip() if manual and not note: return Response({'detail': 'A note is required for a manual measurement.'}, status=status.HTTP_400_BAD_REQUEST) value = request.data.get('value') if value is None: return Response({'detail': 'value is required'}, status=status.HTTP_400_BAD_REQUEST) try: value = int(value) except (ValueError, TypeError): return Response({'detail': 'value must be a number'}, status=status.HTTP_400_BAD_REQUEST) vehicle = self.get_object() try: _assign_weighing(vehicle, value, request.user, manual=manual, note=note or None) except ValueError as exc: return Response({'detail': str(exc)}, status=status.HTTP_400_BAD_REQUEST) serializer = self.get_serializer(vehicle) sse_broadcast_update('vehicle', 'updated', serializer.data) return Response(serializer.data) class ExtraCombinationViewSet(mixins.ListModelMixin, mixins.RetrieveModelMixin, mixins.DestroyModelMixin, viewsets.GenericViewSet): """ API endpoint for previously-used combinations of vehicle extra (lookup) data. list: Get all combinations, most recently used first retrieve: Get a specific combination destroy: Delete a combination the user no longer needs Rows are created/bumped automatically when a vehicle's extra data is saved. """ queryset = ExtraCombination.objects.all() serializer_class = ExtraCombinationSerializer permission_classes = [IsAuthenticated] ordering = ['-last_used'] def perform_destroy(self, instance): data = ExtraCombinationSerializer(instance).data instance.delete() sse_broadcast_update('extra_combination', 'deleted', data) def _live_slots(): return StationSlot.objects.filter(last_seen__gte=licensing.live_slot_cutoff()) def _station_allowed(station_key): """Whether this station may measure. A station already holding a live slot is always allowed - including when the licence is full - so a brief disconnection does not lock a working station out behind its own expiring slot. """ if not station_key: return False if _live_slots().filter(station_key=station_key).exists(): return True return _live_slots().count() < licensing.effective_station_limit() class LicenseView(APIView): """Licence status, and installing a new licence key.""" permission_classes = [IsAuthenticated] def get(self, request): status_info = licensing.license_status() slots = _live_slots() payload = { **status_info, 'stations_used': slots.count(), 'components': licensing.machine_components(), } if request.user.is_admin: payload['slots'] = [ {'station_key': s.station_key, 'com_port': s.com_port, 'bridge_version': s.bridge_version, 'last_seen': s.last_seen} for s in slots ] return Response(payload) def post(self, request): if not request.user.is_admin: return Response({'detail': 'Only administrators can install a licence.'}, status=status.HTTP_403_FORBIDDEN) blob = (request.data.get('key_blob') or '').strip() if not blob: return Response({'detail': 'key_blob is required'}, status=status.HTTP_400_BAD_REQUEST) # Validate before storing, so a bad paste cannot displace a working licence. valid, payload, reason = licensing.verify_license(blob) if not valid: return Response({'detail': reason}, status=status.HTTP_400_BAD_REQUEST) record = License.objects.first() or License() record.key_blob = blob record.grace_started = None record.save() return Response(licensing.license_status()) class LicenseHeartbeatView(APIView): """Stations report in here so the server can count them. The bridge deliberately does not talk to Django - the browser, which already talks to both, reports which bridge it is using. That keeps the bridge pure transport with no startup dependency on the server. """ permission_classes = [IsAuthenticated] def post(self, request): station_key = (request.data.get('station_key') or '').strip() if not station_key: return Response({'detail': 'station_key is required'}, status=status.HTTP_400_BAD_REQUEST) limit = licensing.effective_station_limit() existing = StationSlot.objects.filter(station_key=station_key).first() live = _live_slots() if existing is None and live.count() >= limit: return Response({ 'allowed': False, 'stations_used': live.count(), 'stations_licensed': limit, 'detail': f'Licence limit reached ({limit} stations)', }, status=status.HTTP_403_FORBIDDEN) slot, _ = StationSlot.objects.update_or_create( station_key=station_key, defaults={ 'bridge_version': request.data.get('bridge_version', '') or '', 'com_port': request.data.get('com_port', '') or '', }, ) # update_or_create does not touch auto_now when nothing changed slot.save(update_fields=['last_seen']) return Response({ 'allowed': True, 'stations_used': _live_slots().count(), 'stations_licensed': limit, }) class ScaleProtocolViewSet(viewsets.ModelViewSet): """ API endpoint for scale protocol definitions. The frontend caches these and uses them to parse raw serial lines, so adding a scale type needs no change to the serial bridge running on station PCs. Built-in protocols cannot be deleted. """ queryset = ScaleProtocol.objects.all() serializer_class = ScaleProtocolSerializer permission_classes = [IsAuthenticated] search_fields = ['name'] ordering = ['name'] def perform_create(self, serializer): instance = serializer.save() sse_broadcast_update('scale_protocol', 'created', ScaleProtocolSerializer(instance).data) def perform_update(self, serializer): instance = serializer.save() sse_broadcast_update('scale_protocol', 'updated', ScaleProtocolSerializer(instance).data) def destroy(self, request, *args, **kwargs): instance = self.get_object() if instance.is_builtin: return Response({'detail': 'Built-in protocols cannot be deleted.'}, status=status.HTTP_400_BAD_REQUEST) data = ScaleProtocolSerializer(instance).data instance.delete() sse_broadcast_update('scale_protocol', 'deleted', data) return Response(status=status.HTTP_204_NO_CONTENT) class NomenclatureViewSet(viewsets.ModelViewSet): """ API endpoint for nomenclature management. list: Get all nomenclatures with pagination create: Create a new nomenclature with fields retrieve: Get a specific nomenclature by ID update: Full update of nomenclature and fields partial_update: Partial update of nomenclature destroy: Delete a nomenclature (cascades to fields and entries) by_code: Get nomenclature by code by_applies_to: Filter nomenclatures by applies_to type entries: List or create entries for a specific nomenclature """ queryset = Nomenclature.objects.prefetch_related('fields').all() serializer_class = NomenclatureSerializer permission_classes = [IsAuthenticated] filterset_fields = ['applies_to', 'code', 'kind'] search_fields = ['code', 'name'] ordering = ['sort_order', 'code'] def perform_create(self, serializer): instance = serializer.save() sse_broadcast_update('nomenclature', 'insert', NomenclatureSerializer(instance).data) def perform_update(self, serializer): instance = serializer.save() sse_broadcast_update('nomenclature', 'update', NomenclatureSerializer(instance).data) def perform_destroy(self, instance): data = NomenclatureSerializer(instance).data instance.delete() sse_broadcast_update('nomenclature', 'delete', data) @action(detail=False, methods=['get'], url_path='by-code') def by_code(self, request): """Get nomenclature by code query parameter""" code = request.query_params.get('code', None) if not code: return Response( {'detail': 'code parameter is required'}, status=status.HTTP_400_BAD_REQUEST ) try: nomenclature = Nomenclature.objects.get(code=code) serializer = self.get_serializer(nomenclature) return Response(serializer.data) except Nomenclature.DoesNotExist: return Response( {'detail': 'Nomenclature not found'}, status=status.HTTP_404_NOT_FOUND ) @action(detail=False, methods=['get'], url_path='by-applies-to') def by_applies_to(self, request): """Get nomenclatures filtered by applies_to query parameter""" applies_to = request.query_params.get('applies_to', None) if not applies_to: return Response( {'detail': 'applies_to parameter is required'}, status=status.HTTP_400_BAD_REQUEST ) if applies_to not in ['vehicle', 'container']: return Response( {'detail': 'applies_to must be either vehicle or container'}, status=status.HTTP_400_BAD_REQUEST ) nomenclatures = Nomenclature.objects.filter(applies_to=applies_to) page = self.paginate_queryset(nomenclatures) if page is not None: serializer = self.get_serializer(page, many=True) return self.get_paginated_response(serializer.data) serializer = self.get_serializer(nomenclatures, many=True) return Response(serializer.data) @action(detail=True, methods=['get', 'post'], url_path='entries') def entries(self, request, pk=None): """List or create entries for a specific nomenclature.""" nomenclature = self.get_object() if request.method == 'GET': entries = nomenclature.entries.all() is_active = request.query_params.get('is_active') if is_active is not None: entries = entries.filter(is_active=is_active.lower() == 'true') serializer = NomenclatureEntrySerializer(entries, many=True) return Response(serializer.data) serializer = NomenclatureEntrySerializer( data=request.data, context={'nomenclature': nomenclature}, ) serializer.is_valid(raise_exception=True) serializer.save(nomenclature=nomenclature) entry_data = {**serializer.data, 'nomenclature_code': nomenclature.code} sse_broadcast_update('nomenclature_entry', 'insert', entry_data) return Response(serializer.data, status=status.HTTP_201_CREATED) class NomenclatureEntryViewSet(viewsets.ModelViewSet): """ API endpoint for individual nomenclature entry management. For list/create use the nested endpoint: /api/nomenclatures/{id}/entries/ This viewset handles retrieve/update/delete of individual entries. """ queryset = NomenclatureEntry.objects.select_related('nomenclature').all() serializer_class = NomenclatureEntrySerializer permission_classes = [IsAuthenticated] def get_serializer_context(self): context = super().get_serializer_context() if self.kwargs.get('pk'): try: entry = NomenclatureEntry.objects.select_related('nomenclature').get(pk=self.kwargs['pk']) context['nomenclature'] = entry.nomenclature except NomenclatureEntry.DoesNotExist: pass return context def perform_update(self, serializer): instance = serializer.save() entry_data = {**NomenclatureEntrySerializer(instance).data, 'nomenclature_code': instance.nomenclature.code} sse_broadcast_update('nomenclature_entry', 'update', entry_data) def perform_destroy(self, instance): entry_data = {**NomenclatureEntrySerializer(instance).data, 'nomenclature_code': instance.nomenclature.code} instance.delete() sse_broadcast_update('nomenclature_entry', 'delete', entry_data) class ReportViewSet(viewsets.ModelViewSet): """ API endpoint for report template management. list: Get all reports create: Create a new report retrieve: Get a specific report by ID update: Update a report destroy: Delete a report """ queryset = Report.objects.select_related('created_by').all() serializer_class = ReportSerializer permission_classes = [IsAuthenticated] filterset_fields = ['name'] search_fields = ['name'] ordering = ['-updated_at'] def perform_create(self, serializer): serializer.save(created_by=self.request.user) def perform_update(self, serializer): serializer.save()