99 lines
4.7 KiB
Python
99 lines
4.7 KiB
Python
from datetime import timedelta
|
|
import os
|
|
from django.utils import timezone
|
|
from rest_framework import status, viewsets
|
|
from rest_framework.exceptions import Throttled
|
|
from rest_framework.permissions import AllowAny, IsAdminUser
|
|
from rest_framework.response import Response
|
|
from rest_framework.views import APIView
|
|
from rest_framework_simplejwt.serializers import TokenObtainPairSerializer
|
|
from rest_framework_simplejwt.views import TokenObtainPairView
|
|
from .models import AvailabilityRequest, ScheduleEntry
|
|
from .notifications import send_ntfy_notification
|
|
from .serializers import *
|
|
|
|
def client_ip(request):
|
|
forwarded = request.META.get("HTTP_X_FORWARDED_FOR")
|
|
return forwarded.split(",")[0].strip() if forwarded else request.META.get("REMOTE_ADDR")
|
|
|
|
class AdminTokenSerializer(TokenObtainPairSerializer):
|
|
def validate(self, attrs):
|
|
data = super().validate(attrs)
|
|
if not self.user.is_staff:
|
|
from rest_framework.exceptions import AuthenticationFailed
|
|
raise AuthenticationFailed("Compte administrateur requis.")
|
|
data["user"] = {"username": self.user.username, "first_name": self.user.first_name, "last_name": self.user.last_name, "is_staff": True}
|
|
return data
|
|
|
|
class AdminTokenView(TokenObtainPairView):
|
|
serializer_class = AdminTokenSerializer
|
|
|
|
class PublicScheduleView(APIView):
|
|
permission_classes = [AllowAny]
|
|
def get(self, request):
|
|
qs = ScheduleEntry.objects.all()
|
|
start, end = request.query_params.get("start"), request.query_params.get("end")
|
|
if start and end:
|
|
qs = qs.filter(start_date__lte=end, end_date__gte=start)
|
|
return Response(ScheduleEntrySerializer(qs, many=True).data)
|
|
|
|
class PublicAvailabilityRequestCreateView(APIView):
|
|
permission_classes = [AllowAny]
|
|
def post(self, request):
|
|
ip = client_ip(request)
|
|
if ip and AvailabilityRequest.objects.filter(requester_ip=ip, requested_at__gte=timezone.now()-timedelta(seconds=45)).exists():
|
|
raise Throttled(detail="Une demande vient déjà d'être envoyée. Réessaie dans quelques secondes.")
|
|
s = AvailabilityRequestCreateSerializer(data=request.data)
|
|
s.is_valid(raise_exception=True)
|
|
item = s.save(requester_ip=ip)
|
|
site = os.getenv("PUBLIC_SITE_URL", "").rstrip("/")
|
|
msg = f"{item.guest_name} demande si tu es disponible maintenant."
|
|
if item.message:
|
|
msg += f"\n{item.message}"
|
|
send_ntfy_notification("Nouvelle demande de disponibilité", msg, f"{site}/admin/requests" if site else None)
|
|
return Response(AvailabilityRequestPublicSerializer(item).data, status=status.HTTP_201_CREATED)
|
|
|
|
class PublicAvailabilityRequestDetailView(APIView):
|
|
permission_classes = [AllowAny]
|
|
def get(self, request, token):
|
|
try:
|
|
item = AvailabilityRequest.objects.get(public_token=token)
|
|
except AvailabilityRequest.DoesNotExist:
|
|
return Response({"detail": "Demande introuvable."}, status=404)
|
|
return Response(AvailabilityRequestPublicSerializer(item).data)
|
|
|
|
class AdminScheduleViewSet(viewsets.ModelViewSet):
|
|
serializer_class = ScheduleEntrySerializer
|
|
permission_classes = [IsAdminUser]
|
|
def get_queryset(self):
|
|
qs = ScheduleEntry.objects.all()
|
|
start, end = self.request.query_params.get("start"), self.request.query_params.get("end")
|
|
return qs.filter(start_date__lte=end, end_date__gte=start) if start and end else qs
|
|
|
|
class AdminAvailabilityRequestViewSet(viewsets.ReadOnlyModelViewSet):
|
|
serializer_class = AvailabilityRequestAdminSerializer
|
|
permission_classes = [IsAdminUser]
|
|
def get_queryset(self):
|
|
qs = AvailabilityRequest.objects.all()
|
|
status_filter = self.request.query_params.get("status")
|
|
return qs.filter(status=status_filter) if status_filter else qs
|
|
def respond(self, request, pk=None):
|
|
item = self.get_object()
|
|
s = AvailabilityResponseSerializer(data=request.data)
|
|
s.is_valid(raise_exception=True)
|
|
new_status = s.validated_data["status"]
|
|
duration = s.validated_data.get("duration_minutes")
|
|
item.status = new_status
|
|
item.duration_minutes = duration if new_status == "accepted" else None
|
|
item.available_until = timezone.now()+timedelta(minutes=duration) if new_status == "accepted" and duration else None
|
|
item.response_message = s.validated_data.get("response_message", "")
|
|
item.responded_at = timezone.now()
|
|
item.save()
|
|
return Response(AvailabilityRequestAdminSerializer(item).data)
|
|
|
|
class AdminMeView(APIView):
|
|
permission_classes = [IsAdminUser]
|
|
def get(self, request):
|
|
u = request.user
|
|
return Response({"username": u.username, "first_name": u.first_name, "last_name": u.last_name, "is_staff": u.is_staff})
|