from django.db import models, transaction from django.utils import timezone from rest_framework import generics, status, viewsets from rest_framework.decorators import action from rest_framework.permissions import AllowAny, IsAuthenticated from apps.core.permissions import IsGymStaff, IsGymOwnerOrManager, get_gym_owner from rest_framework.response import Response from .models import ( AccessLog, Equipment, EquipmentCategory, GiftCard, GymDoorQR, MaintenanceLog, POSTransaction, Product, PurchaseOrder, Supplier, Visitor, ) from .serializers import ( AccessLogSerializer, EquipmentCategorySerializer, EquipmentSerializer, GiftCardSerializer, GymDoorQRSerializer, MaintenanceLogSerializer, MemberCheckInSerializer, POSTransactionSerializer, ProductSerializer, PurchaseOrderSerializer, SupplierSerializer, VisitorCheckInSerializer, VisitorSerializer, ) # --------------------------------------------------------------------------- # Visitor CRUD (gym owner only) # --------------------------------------------------------------------------- class VisitorViewSet(viewsets.ModelViewSet): serializer_class = VisitorSerializer permission_classes = [IsGymStaff] def get_queryset(self): user = self.request.user qs = Visitor.objects.filter(is_active=True) if not user.is_superuser: gym_owner = get_gym_owner(user) owner = gym_owner if (gym_owner and gym_owner != user) else user qs = qs.filter(registered_by__gym=owner) | qs.filter(registered_by=owner) qs = qs.distinct() return qs def perform_create(self, serializer): serializer.save(registered_by=self.request.user) # --------------------------------------------------------------------------- # Visitor lookup (public — used after scanning door QR) # --------------------------------------------------------------------------- class VisitorLookupView(generics.RetrieveAPIView): """Look up a visitor by their 6-char code. Public endpoint.""" serializer_class = VisitorSerializer permission_classes = [AllowAny] lookup_field = "visitor_code" lookup_url_kwarg = "code" def get_queryset(self): return Visitor.objects.filter(is_active=True) # --------------------------------------------------------------------------- # Visitor check-in (public) # --------------------------------------------------------------------------- class VisitorCheckInView(generics.CreateAPIView): """Public endpoint: visitor enters code + gym QR payload to check in/out.""" serializer_class = VisitorCheckInSerializer permission_classes = [AllowAny] def create(self, request, *args, **kwargs): serializer = self.get_serializer(data=request.data) serializer.is_valid(raise_exception=True) visitor_code = serializer.validated_data["visitor_code"] qr_payload = serializer.validated_data["qr_payload"] direction = serializer.validated_data["direction"] try: visitor = Visitor.objects.get(visitor_code=visitor_code, is_active=True) except Visitor.DoesNotExist: return Response( {"detail": "Invalid visitor code."}, status=status.HTTP_404_NOT_FOUND, ) try: door_qr = GymDoorQR.objects.get(qr_payload=qr_payload, is_active=True) except GymDoorQR.DoesNotExist: return Response( {"detail": "Invalid gym QR code."}, status=status.HTTP_400_BAD_REQUEST, ) log = AccessLog.objects.create( visitor=visitor, person_type=AccessLog.PersonType.VISITOR, direction=direction, method=AccessLog.Method.QR_CODE, timestamp=timezone.now(), granted=True, ) return Response( { "status": "success", "direction": direction, "visitor_name": visitor.name, "timestamp": log.timestamp.isoformat(), }, status=status.HTTP_201_CREATED, ) class PublicCheckInView(generics.GenericAPIView): """Unified public check-in: accepts visitor codes or guest pass codes.""" permission_classes = [AllowAny] def post(self, request, *args, **kwargs): code = request.data.get("code", "").strip().upper() qr_payload = request.data.get("qr_payload", "").strip() direction = request.data.get("direction", "entry") if not code: return Response({"detail": "Code is required."}, status=status.HTTP_400_BAD_REQUEST) if direction not in ("entry", "exit"): return Response({"detail": "Direction must be 'entry' or 'exit'."}, status=status.HTTP_400_BAD_REQUEST) # Validate door QR if provided if qr_payload: if not GymDoorQR.objects.filter(qr_payload=qr_payload, is_active=True).exists(): return Response({"detail": "Invalid gym QR code."}, status=status.HTTP_400_BAD_REQUEST) today = timezone.now().date() max_daily = 30 now = timezone.now() # Try visitor code try: visitor = Visitor.objects.get(visitor_code=code, is_active=True) v_entries = AccessLog.objects.filter( visitor=visitor, direction=AccessLog.Direction.ENTRY, timestamp__date=today, person_type=AccessLog.PersonType.VISITOR, ).count() v_exits = AccessLog.objects.filter( visitor=visitor, direction=AccessLog.Direction.EXIT, timestamp__date=today, person_type=AccessLog.PersonType.VISITOR, ).count() if direction == "entry" and v_entries > v_exits: return Response( {"detail": "Already checked in. Please check out first."}, status=status.HTTP_400_BAD_REQUEST, ) if direction == "exit" and v_entries <= v_exits: return Response( {"detail": "Not checked in. Please check in first."}, status=status.HTTP_400_BAD_REQUEST, ) log = AccessLog.objects.create( visitor=visitor, person_type=AccessLog.PersonType.VISITOR, direction=direction, method=AccessLog.Method.QR_CODE, timestamp=now, granted=True, ) duration_minutes = None if direction == "exit": latest_entry = AccessLog.objects.filter( visitor=visitor, direction=AccessLog.Direction.ENTRY, timestamp__date=today, person_type=AccessLog.PersonType.VISITOR, ).order_by("-timestamp").first() if latest_entry: duration_minutes = round( (now - latest_entry.timestamp).total_seconds() / 60, 1 ) return Response({ "status": "success", "person_name": visitor.name, "person_type": "visitor", "direction": direction, "timestamp": log.timestamp.isoformat(), "duration_minutes": duration_minutes, }, status=status.HTTP_201_CREATED) except Visitor.DoesNotExist: pass # Try guest pass code try: from apps.members.models import GuestPass guest_pass = GuestPass.objects.get(check_in_code=code) # ── Expiry enforcement ── if direction == "entry" and guest_pass.valid_until and guest_pass.valid_until < today: # Fetch available daily packages for renewal from apps.billing.models import GymPackage gym_owner = guest_pass.issued_by available_packages = [] if gym_owner: for pkg in GymPackage.objects.filter(gym=gym_owner, is_active=True): plans = pkg.plans if isinstance(pkg.plans, list) else [] for plan in plans: if plan.get("duration") == "daily": available_packages.append({ "package_id": str(pkg.id), "package_name": pkg.name, "duration": plan.get("duration"), "price": plan.get("price"), }) return Response({ "detail": "Guest pass has expired. Please renew to check in.", "error_code": "GUEST_PASS_EXPIRED", "guest_name": guest_pass.guest_name, "guest_phone": guest_pass.guest_phone, "valid_until": str(guest_pass.valid_until), "check_in_code": guest_pass.check_in_code, "available_packages": available_packages, }, status=status.HTTP_400_BAD_REQUEST) g_entries = AccessLog.objects.filter( guest_name=guest_pass.guest_name, direction=AccessLog.Direction.ENTRY, timestamp__date=today, person_type=AccessLog.PersonType.GUEST, ).count() g_exits = AccessLog.objects.filter( guest_name=guest_pass.guest_name, direction=AccessLog.Direction.EXIT, timestamp__date=today, person_type=AccessLog.PersonType.GUEST, ).count() if direction == "entry" and g_entries > g_exits: return Response( {"detail": "Already checked in. Please check out first."}, status=status.HTTP_400_BAD_REQUEST, ) if direction == "exit" and g_entries <= g_exits: return Response( {"detail": "Not checked in. Please check in first."}, status=status.HTTP_400_BAD_REQUEST, ) if direction == "entry": guest_pass.checked_in = True guest_pass.checked_in_at = now guest_pass.save(update_fields=["checked_in", "checked_in_at"]) log = AccessLog.objects.create( person_type=AccessLog.PersonType.GUEST, guest_name=guest_pass.guest_name, direction=direction, method=AccessLog.Method.QR_CODE, timestamp=now, granted=True, ) duration_minutes = None if direction == "exit": latest_entry = AccessLog.objects.filter( guest_name=guest_pass.guest_name, direction=AccessLog.Direction.ENTRY, timestamp__date=today, person_type=AccessLog.PersonType.GUEST, ).order_by("-timestamp").first() if latest_entry: duration_minutes = round( (now - latest_entry.timestamp).total_seconds() / 60, 1 ) return Response({ "status": "success", "person_name": guest_pass.guest_name, "person_type": "guest", "direction": direction, "timestamp": log.timestamp.isoformat(), "duration_minutes": duration_minutes, }, status=status.HTTP_201_CREATED) except GuestPass.DoesNotExist: pass # Try member number code try: from apps.members.models import MemberProfile from apps.scheduling.models import ClassBooking, ClassAttendanceLog, ClassSession member_profile = MemberProfile.objects.select_related("user").get( member_number=code, status="active" ) member_name = member_profile.user.get_full_name() or member_profile.user.email m_entries = AccessLog.objects.filter( member=member_profile, direction=AccessLog.Direction.ENTRY, timestamp__date=today, person_type=AccessLog.PersonType.MEMBER, ).count() m_exits = AccessLog.objects.filter( member=member_profile, direction=AccessLog.Direction.EXIT, timestamp__date=today, person_type=AccessLog.PersonType.MEMBER, ).count() if direction == "entry": if m_entries > m_exits: return Response( {"detail": "Already checked in. Please check out first."}, status=status.HTTP_400_BAD_REQUEST, ) # Find today's booked classes booked_qs = ClassBooking.objects.filter( member=member_profile, status=ClassBooking.Status.BOOKED, class_session__start_time__date=today, ).select_related("class_session", "class_session__class_type") class_session_ids = request.data.get("class_session_ids", []) with transaction.atomic(): first_session = None if class_session_ids: first_session = ClassSession.objects.filter( id__in=class_session_ids ).first() log = AccessLog.objects.create( member=member_profile, person_type=AccessLog.PersonType.MEMBER, direction=AccessLog.Direction.ENTRY, method=AccessLog.Method.QR_CODE, class_session=first_session, timestamp=now, granted=True, ) classes_checked_in = [] for session_id in class_session_ids: try: booking = ClassBooking.objects.get( class_session_id=session_id, member=member_profile, status=ClassBooking.Status.BOOKED, ) booking.status = ClassBooking.Status.ATTENDED booking.checked_in_at = now booking.save(update_fields=["status", "checked_in_at", "updated_at"]) ClassAttendanceLog.objects.create( class_session_id=session_id, member=member_profile, check_in_time=now, method=ClassAttendanceLog.Method.APP, ) session = ClassSession.objects.filter(id=session_id).first() if session: classes_checked_in.append(session.title) except ClassBooking.DoesNotExist: continue booked_classes = [ { "class_session_id": str(b.class_session_id), "title": b.class_session.title, "class_type_name": ( b.class_session.class_type.name if b.class_session.class_type else "" ), "start_time": b.class_session.start_time.isoformat(), "end_time": b.class_session.end_time.isoformat(), } for b in booked_qs ] return Response({ "status": "success", "person_name": member_name, "person_type": "member", "direction": "entry", "timestamp": log.timestamp.isoformat(), "classes_checked_in": classes_checked_in, "visit_type": "class" if classes_checked_in else "general", "booked_classes": booked_classes, }, status=status.HTTP_201_CREATED) else: # Exit — guardrail: must have checked in first if m_entries <= m_exits: return Response( {"detail": "Not checked in. Cannot check out without checking in first."}, status=status.HTTP_400_BAD_REQUEST, ) latest_entry = AccessLog.objects.filter( member=member_profile, direction=AccessLog.Direction.ENTRY, timestamp__date=today, person_type=AccessLog.PersonType.MEMBER, ).order_by("-timestamp").first() # Get attended classes for confirmation attended_qs = ClassBooking.objects.filter( member=member_profile, status=ClassBooking.Status.ATTENDED, class_session__start_time__date=today, ).select_related("class_session", "class_session__class_type") attended_classes = [ { "class_session_id": str(a.class_session_id), "title": a.class_session.title, "class_type_name": ( a.class_session.class_type.name if a.class_session.class_type else "" ), "start_time": a.class_session.start_time.isoformat(), "end_time": a.class_session.end_time.isoformat(), "checked_in_at": a.checked_in_at.isoformat() if a.checked_in_at else None, } for a in attended_qs ] with transaction.atomic(): log = AccessLog.objects.create( member=member_profile, person_type=AccessLog.PersonType.MEMBER, direction=AccessLog.Direction.EXIT, method=AccessLog.Method.QR_CODE, timestamp=now, granted=True, ) ClassAttendanceLog.objects.filter( member=member_profile, check_in_time__date=today, check_out_time__isnull=True, ).update(check_out_time=now) duration_minutes = None checked_in_at = None if latest_entry: delta = now - latest_entry.timestamp duration_minutes = round(delta.total_seconds() / 60, 1) checked_in_at = latest_entry.timestamp.isoformat() return Response({ "status": "success", "person_name": member_name, "person_type": "member", "direction": "exit", "timestamp": log.timestamp.isoformat(), "duration_minutes": duration_minutes, "checked_in_at": checked_in_at, "attended_classes": attended_classes, }, status=status.HTTP_201_CREATED) except MemberProfile.DoesNotExist: pass return Response( {"detail": "Invalid code. No matching visitor, guest pass, or member found."}, status=status.HTTP_404_NOT_FOUND, ) # --------------------------------------------------------------------------- # Member check-in (authenticated) # --------------------------------------------------------------------------- class MemberCheckInView(generics.CreateAPIView): """Authenticated member scans door QR to check in/out.""" serializer_class = MemberCheckInSerializer permission_classes = [IsAuthenticated] def create(self, request, *args, **kwargs): serializer = self.get_serializer(data=request.data) serializer.is_valid(raise_exception=True) qr_payload = serializer.validated_data["qr_payload"] direction = serializer.validated_data["direction"] class_session_ids = serializer.validated_data.get("class_session_ids", []) try: GymDoorQR.objects.get(qr_payload=qr_payload, is_active=True) except GymDoorQR.DoesNotExist: return Response( {"detail": "Invalid gym QR code."}, status=status.HTTP_400_BAD_REQUEST, ) member_profile = getattr(request.user, "member_profile", None) if not member_profile: return Response( {"detail": "Member profile not found."}, status=status.HTTP_400_BAD_REQUEST, ) now = timezone.now() today = now.date() member_name = request.user.get_full_name() or request.user.email if direction == "entry": # Anti-duplicate: check for open entry today (entry with no matching exit) open_entry = AccessLog.objects.filter( member=member_profile, person_type=AccessLog.PersonType.MEMBER, direction=AccessLog.Direction.ENTRY, timestamp__date=today, ).exclude( id__in=AccessLog.objects.filter( member=member_profile, person_type=AccessLog.PersonType.MEMBER, direction=AccessLog.Direction.EXIT, timestamp__date=today, ).values_list("id", flat=True) ) # Simpler approach: count entries vs exits today entry_count = AccessLog.objects.filter( member=member_profile, direction=AccessLog.Direction.ENTRY, timestamp__date=today, person_type=AccessLog.PersonType.MEMBER, ).count() exit_count = AccessLog.objects.filter( member=member_profile, direction=AccessLog.Direction.EXIT, timestamp__date=today, person_type=AccessLog.PersonType.MEMBER, ).count() if entry_count > exit_count: return Response( {"detail": "Already checked in. Please check out first."}, status=status.HTTP_400_BAD_REQUEST, ) with transaction.atomic(): from apps.scheduling.models import ClassBooking, ClassAttendanceLog, ClassSession first_session = None if class_session_ids: first_session = ClassSession.objects.filter( id__in=class_session_ids ).first() log = AccessLog.objects.create( member=member_profile, person_type=AccessLog.PersonType.MEMBER, direction=AccessLog.Direction.ENTRY, method=AccessLog.Method.QR_CODE, class_session=first_session, timestamp=now, granted=True, ) classes_checked_in = [] for session_id in class_session_ids: try: booking = ClassBooking.objects.get( class_session_id=session_id, member=member_profile, status=ClassBooking.Status.BOOKED, ) booking.status = ClassBooking.Status.ATTENDED booking.checked_in_at = now booking.save(update_fields=["status", "checked_in_at", "updated_at"]) ClassAttendanceLog.objects.create( class_session_id=session_id, member=member_profile, check_in_time=now, method=ClassAttendanceLog.Method.APP, ) session = ClassSession.objects.filter(id=session_id).first() if session: classes_checked_in.append(session.title) except ClassBooking.DoesNotExist: continue return Response({ "status": "success", "direction": "entry", "member_name": member_name, "timestamp": log.timestamp.isoformat(), "classes_checked_in": classes_checked_in, "visit_type": "class" if classes_checked_in else "general", }, status=status.HTTP_201_CREATED) else: # Exit flow entry_count = AccessLog.objects.filter( member=member_profile, direction=AccessLog.Direction.ENTRY, timestamp__date=today, person_type=AccessLog.PersonType.MEMBER, ).count() exit_count = AccessLog.objects.filter( member=member_profile, direction=AccessLog.Direction.EXIT, timestamp__date=today, person_type=AccessLog.PersonType.MEMBER, ).count() if entry_count <= exit_count: return Response( {"detail": "Not checked in. Please check in first."}, status=status.HTTP_400_BAD_REQUEST, ) # Find the latest open entry latest_entry = AccessLog.objects.filter( member=member_profile, direction=AccessLog.Direction.ENTRY, timestamp__date=today, person_type=AccessLog.PersonType.MEMBER, ).order_by("-timestamp").first() # Get attended classes for checkout confirmation from apps.scheduling.models import ClassAttendanceLog, ClassBooking attended_qs = ClassBooking.objects.filter( member=member_profile, status=ClassBooking.Status.ATTENDED, class_session__start_time__date=today, ).select_related("class_session", "class_session__class_type") attended_classes = [ { "class_session_id": str(a.class_session_id), "title": a.class_session.title, "class_type_name": ( a.class_session.class_type.name if a.class_session.class_type else "" ), "start_time": a.class_session.start_time.isoformat(), "end_time": a.class_session.end_time.isoformat(), "checked_in_at": a.checked_in_at.isoformat() if a.checked_in_at else None, } for a in attended_qs ] with transaction.atomic(): log = AccessLog.objects.create( member=member_profile, person_type=AccessLog.PersonType.MEMBER, direction=AccessLog.Direction.EXIT, method=AccessLog.Method.QR_CODE, timestamp=now, granted=True, ) # Close open attendance logs for today ClassAttendanceLog.objects.filter( member=member_profile, check_in_time__date=today, check_out_time__isnull=True, ).update(check_out_time=now) duration_minutes = None checked_in_at = None if latest_entry: delta = now - latest_entry.timestamp duration_minutes = round(delta.total_seconds() / 60, 1) checked_in_at = latest_entry.timestamp.isoformat() return Response({ "status": "success", "direction": "exit", "member_name": member_name, "timestamp": log.timestamp.isoformat(), "duration_minutes": duration_minutes, "checked_in_at": checked_in_at, "attended_classes": attended_classes, }, status=status.HTTP_201_CREATED) class MemberCheckInStatusView(generics.GenericAPIView): """GET: returns current check-in status for the authenticated member.""" permission_classes = [IsAuthenticated] def get(self, request, *args, **kwargs): member_profile = getattr(request.user, "member_profile", None) if not member_profile: return Response( {"detail": "Member profile not found."}, status=status.HTTP_400_BAD_REQUEST, ) now = timezone.now() today = now.date() entry_count = AccessLog.objects.filter( member=member_profile, direction=AccessLog.Direction.ENTRY, timestamp__date=today, person_type=AccessLog.PersonType.MEMBER, ).count() exit_count = AccessLog.objects.filter( member=member_profile, direction=AccessLog.Direction.EXIT, timestamp__date=today, person_type=AccessLog.PersonType.MEMBER, ).count() is_checked_in = entry_count > exit_count entry_log = None duration_minutes = None if is_checked_in: latest_entry = AccessLog.objects.filter( member=member_profile, direction=AccessLog.Direction.ENTRY, timestamp__date=today, person_type=AccessLog.PersonType.MEMBER, ).select_related("class_session").order_by("-timestamp").first() if latest_entry: delta = now - latest_entry.timestamp duration_minutes = round(delta.total_seconds() / 60, 1) entry_log = { "id": str(latest_entry.id), "timestamp": latest_entry.timestamp.isoformat(), "class_session_title": ( latest_entry.class_session.title if latest_entry.class_session else None ), } # Today's booked and attended classes (including recurring that run today) from apps.scheduling.models import ClassBooking import re as _re DAY_MAP_STATUS = {0: "MO", 1: "TU", 2: "WE", 3: "TH", 4: "FR", 5: "SA", 6: "SU"} def _session_runs_today(session, today_date): """Check if a class session runs on today_date.""" if not session.is_recurring or session.recurrence_type == "one_time": return session.start_time.date() == today_date rule = session.recurrence_rule or "" weekday = today_date.weekday() # 0=Mon .. 6=Sun day_code = DAY_MAP_STATUS.get(weekday, "") if "EXCLUDE=" in rule: exclude_match = _re.search(r"EXCLUDE=([A-Z,]+)", rule) if exclude_match: excludes = exclude_match.group(1).split(",") if "SAT" in excludes and weekday == 5: return False if "SUN" in excludes and weekday == 6: return False if session.recurrence_type == "daily": return True if session.recurrence_type == "weekly": byday_match = _re.search(r"BYDAY=([A-Z,]+)", rule) if byday_match: days = byday_match.group(1).split(",") return day_code in days return session.start_time.weekday() == weekday return False booked_qs = ClassBooking.objects.filter( member=member_profile, status=ClassBooking.Status.BOOKED, ).select_related("class_session", "class_session__class_type") todays_booked_classes = [] for b in booked_qs: if _session_runs_today(b.class_session, today): todays_booked_classes.append({ "class_session_id": str(b.class_session_id), "title": b.class_session.title, "class_type_name": ( b.class_session.class_type.name if b.class_session.class_type else "" ), "start_time": b.class_session.start_time.isoformat(), "end_time": b.class_session.end_time.isoformat(), }) attended_qs = ClassBooking.objects.filter( member=member_profile, status=ClassBooking.Status.ATTENDED, ).select_related("class_session", "class_session__class_type") todays_attended_classes = [] for a in attended_qs: if _session_runs_today(a.class_session, today): todays_attended_classes.append({ "class_session_id": str(a.class_session_id), "title": a.class_session.title, "class_type_name": ( a.class_session.class_type.name if a.class_session.class_type else "" ), "start_time": a.class_session.start_time.isoformat(), "end_time": a.class_session.end_time.isoformat(), "checked_in_at": a.checked_in_at.isoformat() if a.checked_in_at else None, }) return Response({ "is_checked_in": is_checked_in, "entry_log": entry_log, "duration_minutes": duration_minutes, "todays_booked_classes": todays_booked_classes, "todays_attended_classes": todays_attended_classes, }) class PublicMemberLookupView(generics.GenericAPIView): """Public endpoint: look up a member by member_number and return today's class info.""" permission_classes = [AllowAny] def get(self, request, code, *args, **kwargs): code = code.strip().upper() try: from apps.members.models import MemberProfile member_profile = MemberProfile.objects.select_related("user").get( member_number=code, status="active" ) except Exception: return Response( {"detail": "Member not found."}, status=status.HTTP_404_NOT_FOUND, ) now = timezone.now() today = now.date() member_name = member_profile.user.get_full_name() or member_profile.user.email # Check if currently checked in entry_count = AccessLog.objects.filter( member=member_profile, direction=AccessLog.Direction.ENTRY, timestamp__date=today, person_type=AccessLog.PersonType.MEMBER, ).count() exit_count = AccessLog.objects.filter( member=member_profile, direction=AccessLog.Direction.EXIT, timestamp__date=today, person_type=AccessLog.PersonType.MEMBER, ).count() is_checked_in = entry_count > exit_count duration_minutes = None checked_in_at = None if is_checked_in: latest_entry = AccessLog.objects.filter( member=member_profile, direction=AccessLog.Direction.ENTRY, timestamp__date=today, person_type=AccessLog.PersonType.MEMBER, ).order_by("-timestamp").first() if latest_entry: delta = now - latest_entry.timestamp duration_minutes = round(delta.total_seconds() / 60, 1) checked_in_at = latest_entry.timestamp.isoformat() # Today's booked and attended classes (including recurring that run today) from apps.scheduling.models import ClassBooking import re as _re2 _DAY_MAP2 = {0: "MO", 1: "TU", 2: "WE", 3: "TH", 4: "FR", 5: "SA", 6: "SU"} def _runs_today(session, today_date): if not session.is_recurring or session.recurrence_type == "one_time": return session.start_time.date() == today_date rule = session.recurrence_rule or "" weekday = today_date.weekday() day_code = _DAY_MAP2.get(weekday, "") if "EXCLUDE=" in rule: exc = _re2.search(r"EXCLUDE=([A-Z,]+)", rule) if exc: excludes = exc.group(1).split(",") if "SAT" in excludes and weekday == 5: return False if "SUN" in excludes and weekday == 6: return False if session.recurrence_type == "daily": return True if session.recurrence_type == "weekly": bd = _re2.search(r"BYDAY=([A-Z,]+)", rule) if bd: return day_code in bd.group(1).split(",") return session.start_time.weekday() == weekday return False booked_qs = ClassBooking.objects.filter( member=member_profile, status=ClassBooking.Status.BOOKED, ).select_related("class_session", "class_session__class_type") todays_booked_classes = [] for b in booked_qs: if _runs_today(b.class_session, today): todays_booked_classes.append({ "class_session_id": str(b.class_session_id), "title": b.class_session.title, "class_type_name": ( b.class_session.class_type.name if b.class_session.class_type else "" ), "start_time": b.class_session.start_time.isoformat(), "end_time": b.class_session.end_time.isoformat(), }) attended_qs = ClassBooking.objects.filter( member=member_profile, status=ClassBooking.Status.ATTENDED, ).select_related("class_session", "class_session__class_type") todays_attended_classes = [] for a in attended_qs: if _runs_today(a.class_session, today): todays_attended_classes.append({ "class_session_id": str(a.class_session_id), "title": a.class_session.title, "class_type_name": ( a.class_session.class_type.name if a.class_session.class_type else "" ), "start_time": a.class_session.start_time.isoformat(), "end_time": a.class_session.end_time.isoformat(), "checked_in_at": a.checked_in_at.isoformat() if a.checked_in_at else None, }) return Response({ "member_name": member_name, "member_number": member_profile.member_number, "is_checked_in": is_checked_in, "duration_minutes": duration_minutes, "checked_in_at": checked_in_at, "todays_booked_classes": todays_booked_classes, "todays_attended_classes": todays_attended_classes, }) # --------------------------------------------------------------------------- # Access logs (gym owner) # --------------------------------------------------------------------------- class AccessLogViewSet(viewsets.ReadOnlyModelViewSet): serializer_class = AccessLogSerializer permission_classes = [IsGymStaff] def get_queryset(self): user = self.request.user qs = AccessLog.objects.select_related("member__user", "visitor").all() if not user.is_superuser: gym_owner = get_gym_owner(user) owner = gym_owner if (gym_owner and gym_owner != user) else user from django.db.models import Q qs = qs.filter( Q(member__user__gym=owner) | Q(visitor__registered_by__gym=owner) | Q(visitor__registered_by=owner) ) person_type = self.request.query_params.get("person_type") if person_type: qs = qs.filter(person_type=person_type) date = self.request.query_params.get("date") if date: qs = qs.filter(timestamp__date=date) return qs @action(detail=False, methods=["post"], url_path="manual-checkin") def manual_checkin(self, request): """Check in by code — accepts visitor codes (6-char) or guest pass codes (GP-XXXXXXXX).""" code = request.data.get("code", "").strip().upper() direction = request.data.get("direction", "entry") if not code: return Response({"detail": "Code is required."}, status=status.HTTP_400_BAD_REQUEST) if direction not in ("entry", "exit"): return Response({"detail": "Direction must be 'entry' or 'exit'."}, status=status.HTTP_400_BAD_REQUEST) now = timezone.now() today = now.date() # Try visitor code first try: visitor = Visitor.objects.get(visitor_code=code, is_active=True) v_entries = AccessLog.objects.filter( visitor=visitor, direction=AccessLog.Direction.ENTRY, timestamp__date=today, person_type=AccessLog.PersonType.VISITOR, ).count() v_exits = AccessLog.objects.filter( visitor=visitor, direction=AccessLog.Direction.EXIT, timestamp__date=today, person_type=AccessLog.PersonType.VISITOR, ).count() if direction == "entry" and v_entries > v_exits: return Response( {"detail": "Already checked in. Please check out first."}, status=status.HTTP_400_BAD_REQUEST, ) if direction == "exit" and v_entries <= v_exits: return Response( {"detail": "Not checked in. Please check in first."}, status=status.HTTP_400_BAD_REQUEST, ) log = AccessLog.objects.create( visitor=visitor, person_type=AccessLog.PersonType.VISITOR, direction=direction, method=AccessLog.Method.MANUAL, timestamp=now, granted=True, ) duration_minutes = None if direction == "exit": latest_entry = AccessLog.objects.filter( visitor=visitor, direction=AccessLog.Direction.ENTRY, timestamp__date=today, person_type=AccessLog.PersonType.VISITOR, ).order_by("-timestamp").first() if latest_entry: duration_minutes = round( (now - latest_entry.timestamp).total_seconds() / 60, 1 ) return Response({ "status": "success", "person_name": visitor.name, "person_type": "visitor", "direction": direction, "timestamp": log.timestamp.isoformat(), "duration_minutes": duration_minutes, }, status=status.HTTP_201_CREATED) except Visitor.DoesNotExist: pass # Try guest pass code try: from apps.members.models import GuestPass guest_pass = GuestPass.objects.get(check_in_code=code) g_entries = AccessLog.objects.filter( guest_name=guest_pass.guest_name, direction=AccessLog.Direction.ENTRY, timestamp__date=today, person_type=AccessLog.PersonType.GUEST, ).count() g_exits = AccessLog.objects.filter( guest_name=guest_pass.guest_name, direction=AccessLog.Direction.EXIT, timestamp__date=today, person_type=AccessLog.PersonType.GUEST, ).count() if direction == "entry" and g_entries > g_exits: return Response( {"detail": "Already checked in. Please check out first."}, status=status.HTTP_400_BAD_REQUEST, ) if direction == "exit" and g_entries <= g_exits: return Response( {"detail": "Not checked in. Please check in first."}, status=status.HTTP_400_BAD_REQUEST, ) if direction == "entry": guest_pass.checked_in = True guest_pass.checked_in_at = now guest_pass.save(update_fields=["checked_in", "checked_in_at"]) log = AccessLog.objects.create( person_type=AccessLog.PersonType.GUEST, guest_name=guest_pass.guest_name, direction=direction, method=AccessLog.Method.MANUAL, timestamp=now, granted=True, ) duration_minutes = None if direction == "exit": latest_entry = AccessLog.objects.filter( guest_name=guest_pass.guest_name, direction=AccessLog.Direction.ENTRY, timestamp__date=today, person_type=AccessLog.PersonType.GUEST, ).order_by("-timestamp").first() if latest_entry: duration_minutes = round( (now - latest_entry.timestamp).total_seconds() / 60, 1 ) return Response({ "status": "success", "person_name": guest_pass.guest_name, "person_type": "guest", "direction": direction, "timestamp": log.timestamp.isoformat(), "duration_minutes": duration_minutes, }, status=status.HTTP_201_CREATED) except Exception: pass return Response({"detail": "Invalid code. No matching visitor or guest pass found."}, status=status.HTTP_404_NOT_FOUND) # --------------------------------------------------------------------------- # Gym Door QR (gym owner) # --------------------------------------------------------------------------- class GymDoorQRViewSet(viewsets.ModelViewSet): serializer_class = GymDoorQRSerializer permission_classes = [IsGymOwnerOrManager] def get_queryset(self): user = self.request.user qs = GymDoorQR.objects.filter(is_active=True) if not user.is_superuser: gym_owner = get_gym_owner(user) owner = gym_owner if (gym_owner and gym_owner != user) else user # Include QRs linked to owner's locations OR QRs with no location from django.db.models import Q qs = qs.filter(Q(location__gym=owner) | Q(location__isnull=True)) return qs def perform_create(self, serializer): """Auto-assign the first location belonging to this gym owner.""" from apps.core.models import Location user = self.request.user gym_owner = get_gym_owner(user) owner = gym_owner if (gym_owner and gym_owner != user) else user location = Location.objects.filter(gym=owner).first() serializer.save(location=location) # --------------------------------------------------------------------------- # Equipment views # --------------------------------------------------------------------------- class EquipmentCategoryViewSet(viewsets.ModelViewSet): serializer_class = EquipmentCategorySerializer permission_classes = [IsGymStaff] def get_queryset(self): user = self.request.user qs = EquipmentCategory.objects.all() if not user.is_superuser: gym_owner = get_gym_owner(user) owner = gym_owner if (gym_owner and gym_owner != user) else user qs = qs.filter(gym=owner) return qs def perform_create(self, serializer): gym_owner = get_gym_owner(self.request.user) serializer.save(gym=gym_owner) class EquipmentViewSet(viewsets.ModelViewSet): serializer_class = EquipmentSerializer permission_classes = [IsGymStaff] def get_queryset(self): user = self.request.user qs = Equipment.objects.select_related("category", "location").prefetch_related("maintenance_logs").all() if not user.is_superuser: gym_owner = get_gym_owner(user) owner = gym_owner if (gym_owner and gym_owner != user) else user qs = qs.filter(gym=owner) equip_status = self.request.query_params.get("status") category = self.request.query_params.get("category") if equip_status: qs = qs.filter(status=equip_status) if category: qs = qs.filter(category_id=category) return qs def perform_create(self, serializer): gym_owner = get_gym_owner(self.request.user) serializer.save(gym=gym_owner) @action(detail=True, methods=["post"]) def schedule_maintenance(self, request, pk=None): equipment = self.get_object() equipment.status = Equipment.Status.MAINTENANCE equipment.next_maintenance = request.data.get("date") equipment.save() return Response(EquipmentSerializer(equipment).data) class MaintenanceLogViewSet(viewsets.ModelViewSet): serializer_class = MaintenanceLogSerializer permission_classes = [IsGymStaff] def get_queryset(self): user = self.request.user qs = MaintenanceLog.objects.select_related("equipment", "performed_by").all() if not user.is_superuser: gym_owner = get_gym_owner(user) owner = gym_owner if (gym_owner and gym_owner != user) else user qs = qs.filter(equipment__gym=owner) equipment_id = self.request.query_params.get("equipment") if equipment_id: qs = qs.filter(equipment_id=equipment_id) return qs def perform_create(self, serializer): log = serializer.save(performed_by=self.request.user) equipment = log.equipment equipment.last_maintenance = log.date if log.next_scheduled: equipment.next_maintenance = log.next_scheduled equipment.save() # --------------------------------------------------------------------------- # Inventory / Supplier views # --------------------------------------------------------------------------- class ProductViewSet(viewsets.ModelViewSet): serializer_class = ProductSerializer permission_classes = [IsGymStaff] def get_queryset(self): user = self.request.user qs = Product.objects.all() if not user.is_superuser: gym_owner = get_gym_owner(user) owner = gym_owner if (gym_owner and gym_owner != user) else user qs = qs.filter(gym=owner) category = self.request.query_params.get("category") if category: qs = qs.filter(category=category) return qs def perform_create(self, serializer): gym_owner = get_gym_owner(self.request.user) serializer.save(gym=gym_owner) @action(detail=False, methods=["get"]) def low_stock(self, request): from django.db.models import F qs = Product.objects.filter(stock_quantity__lte=F("reorder_level"), is_active=True) if not request.user.is_superuser: gym_owner = get_gym_owner(request.user) owner = gym_owner if (gym_owner and gym_owner != request.user) else request.user qs = qs.filter(gym=owner) serializer = self.get_serializer(qs, many=True) return Response(serializer.data) class SupplierViewSet(viewsets.ModelViewSet): serializer_class = SupplierSerializer permission_classes = [IsGymStaff] def get_queryset(self): user = self.request.user qs = Supplier.objects.all() if not user.is_superuser: gym_owner = get_gym_owner(user) owner = gym_owner if (gym_owner and gym_owner != user) else user qs = qs.filter(gym=owner) return qs def perform_create(self, serializer): gym_owner = get_gym_owner(self.request.user) serializer.save(gym=gym_owner) class PurchaseOrderViewSet(viewsets.ModelViewSet): serializer_class = PurchaseOrderSerializer permission_classes = [IsGymOwnerOrManager] def get_queryset(self): user = self.request.user qs = PurchaseOrder.objects.select_related("supplier").prefetch_related("items__product").all() if not user.is_superuser: gym_owner = get_gym_owner(user) owner = gym_owner if (gym_owner and gym_owner != user) else user qs = qs.filter(supplier__gym=owner) return qs @action(detail=True, methods=["post"]) def receive(self, request, pk=None): with transaction.atomic(): order = PurchaseOrder.objects.select_for_update().get(pk=pk) if order.status == PurchaseOrder.Status.RECEIVED: return Response({"detail": "Order already received."}, status=status.HTTP_400_BAD_REQUEST) order.status = PurchaseOrder.Status.RECEIVED order.save() for item in order.items.select_for_update().all(): item.received_quantity = item.quantity item.save() Product.objects.filter(pk=item.product_id).update( stock_quantity=models.F("stock_quantity") + item.quantity ) order.refresh_from_db() return Response(PurchaseOrderSerializer(order).data) # --------------------------------------------------------------------------- # POS views # --------------------------------------------------------------------------- class POSTransactionViewSet(viewsets.ModelViewSet): serializer_class = POSTransactionSerializer permission_classes = [IsGymStaff] def get_queryset(self): user = self.request.user qs = POSTransaction.objects.select_related("cashier").prefetch_related("items").all() if not user.is_superuser: from django.db.models import Q gym_owner = get_gym_owner(user) owner = gym_owner if (gym_owner and gym_owner != user) else user qs = qs.filter(Q(cashier__gym=owner) | Q(cashier=owner)) date = self.request.query_params.get("date") payment_method = self.request.query_params.get("payment_method") if date: qs = qs.filter(created_at__date=date) if payment_method: qs = qs.filter(payment_method=payment_method) return qs def perform_create(self, serializer): serializer.save(cashier=self.request.user) class GiftCardViewSet(viewsets.ModelViewSet): serializer_class = GiftCardSerializer permission_classes = [IsGymStaff] def get_queryset(self): user = self.request.user qs = GiftCard.objects.all() if not user.is_superuser: from django.db.models import Q gym_owner = get_gym_owner(user) owner = gym_owner if (gym_owner and gym_owner != user) else user qs = qs.filter( Q(purchased_by__gym=owner) | Q(purchased_by=owner) ) return qs @action(detail=True, methods=["post"]) def redeem(self, request, pk=None): amount = request.data.get("amount", 0) try: amount = float(amount) except (ValueError, TypeError): return Response({"detail": "Invalid amount."}, status=status.HTTP_400_BAD_REQUEST) if amount <= 0: return Response({"detail": "Invalid redemption amount."}, status=status.HTTP_400_BAD_REQUEST) # Use select_for_update to prevent race conditions on balance with transaction.atomic(): card = GiftCard.objects.select_for_update().get(pk=pk) if amount > float(card.balance): return Response({"detail": "Insufficient balance."}, status=status.HTTP_400_BAD_REQUEST) card.balance = float(card.balance) - amount if card.balance <= 0: card.is_active = False card.save() return Response(GiftCardSerializer(card).data) # --------------------------------------------------------------------------- # Guest pass renewal (public — used from QR check-in page) # --------------------------------------------------------------------------- class GuestPassRenewView(generics.GenericAPIView): """Public endpoint: renew an expired guest pass with payment.""" permission_classes = [AllowAny] def post(self, request, *args, **kwargs): from datetime import timedelta from apps.members.models import GuestPass from apps.billing.models import GymPackage, Payment, PaymentGatewayConfig code = request.data.get("code", "").strip().upper() package_id = request.data.get("package_id") payment_method = request.data.get("payment_method", "cash") phone_number = request.data.get("phone_number", "") days = int(request.data.get("days", 1)) if not code: return Response({"detail": "Code is required."}, status=status.HTTP_400_BAD_REQUEST) try: guest_pass = GuestPass.objects.get(check_in_code=code) except GuestPass.DoesNotExist: return Response({"detail": "Invalid guest pass code."}, status=status.HTTP_404_NOT_FOUND) today = timezone.now().date() amount = 0 package = None if package_id: package = GymPackage.objects.filter(id=package_id).first() if package: plans = package.plans if isinstance(package.plans, list) else [] daily_plan = next((p for p in plans if p.get("duration") == "daily"), None) if daily_plan: amount = float(daily_plan.get("price", 0)) * days if payment_method == "mpesa": if not phone_number: return Response({"detail": "Phone number is required for M-Pesa."}, status=status.HTTP_400_BAD_REQUEST) gym_owner = guest_pass.issued_by config_obj = PaymentGatewayConfig.objects.filter(gateway="mpesa", gym=gym_owner).first() if not config_obj: return Response( {"detail": "M-Pesa is not configured for this gym."}, status=status.HTTP_400_BAD_REQUEST, ) from apps.billing.mpesa import MpesaClient, decrypt_config, normalize_phone from apps.billing.views import _build_mpesa_client client = _build_mpesa_client(config_obj) phone = normalize_phone(phone_number) try: result = client.stk_push( phone=phone, amount=int(amount), account_ref=f"GP-{code}", description="Guest pass renewal", ) except Exception as e: return Response( {"detail": f"STK Push failed: {str(e)}"}, status=status.HTTP_502_BAD_GATEWAY, ) checkout_request_id = result.get("CheckoutRequestID", "") gateway_response = dict(result) gateway_response["guest_pass_id"] = str(guest_pass.id) gateway_response["renewal_days"] = days payment = Payment.objects.create( member=None, amount=amount, currency="KES", payment_method="mpesa", status="pending", gateway_transaction_id=checkout_request_id, gateway_response=gateway_response, ) return Response({ "detail": "STK Push sent. Enter your M-Pesa PIN to confirm.", "payment_id": str(payment.id), "checkout_request_id": checkout_request_id, }, status=status.HTTP_201_CREATED) else: # Cash or Card — immediate renewal guest_pass.visit_date = today guest_pass.days = days guest_pass.checked_in = False guest_pass.checked_in_at = None guest_pass.payment_method = payment_method guest_pass.payment_status = "completed" if package: guest_pass.package = package guest_pass.amount_paid = amount guest_pass.save() return Response({ "status": "success", "detail": "Guest pass renewed successfully.", "valid_until": str(guest_pass.valid_until), }) class GuestPassPaymentStatusView(generics.GenericAPIView): """Public endpoint: poll guest pass M-Pesa payment status.""" permission_classes = [AllowAny] def get(self, request, checkout_id, *args, **kwargs): from apps.billing.models import Payment payment = Payment.objects.filter( gateway_transaction_id=checkout_id, payment_method="mpesa", ).first() if not payment: payment = Payment.objects.filter( payment_method="mpesa", gateway_response__Body__stkCallback__CheckoutRequestID=checkout_id, ).first() if not payment: return Response({"detail": "Payment not found."}, status=status.HTTP_404_NOT_FOUND) return Response({ "payment_id": str(payment.id), "status": payment.status, "amount": str(payment.amount), })