233 lines
8.9 KiB
Python
233 lines
8.9 KiB
Python
from rest_framework import generics, status
|
||
from rest_framework.response import Response
|
||
from rest_framework.permissions import IsAuthenticated
|
||
from rest_framework.views import APIView
|
||
from django.contrib.auth.models import User
|
||
from django.db import transaction
|
||
from django.utils import timezone
|
||
from datetime import datetime, date, timedelta
|
||
from decouple import config
|
||
from .models import TimeEntry
|
||
from .serializers import (
|
||
TimeEntrySerializer,
|
||
TimeEntryCreateSerializer,
|
||
TimeEntryUpdateSerializer,
|
||
WeeklySummarySerializer
|
||
)
|
||
|
||
class TimeEntryListView(generics.ListAPIView):
|
||
serializer_class = TimeEntrySerializer
|
||
permission_classes = [IsAuthenticated]
|
||
|
||
def get_queryset(self):
|
||
user = self.request.user
|
||
# Staff members can view all entries, regular users only their own
|
||
if user.is_staff or user.is_superuser:
|
||
return TimeEntry.objects.all()
|
||
return TimeEntry.objects.filter(user=user)
|
||
|
||
class TimeEntryCreateView(generics.CreateAPIView):
|
||
serializer_class = TimeEntryCreateSerializer
|
||
permission_classes = [IsAuthenticated]
|
||
|
||
def perform_create(self, serializer):
|
||
serializer.save(user=self.request.user)
|
||
|
||
class TimeEntryScanView(APIView):
|
||
permission_classes = [IsAuthenticated]
|
||
|
||
def post(self, request):
|
||
code = str(request.data.get('code') or '').strip().strip('"\'')
|
||
if '=' in code:
|
||
code = code.split('=', 1)[1].strip().strip('"\'')
|
||
enter_code = config('SECRET_POINTAGE_ENTER', default='').strip()
|
||
exit_code = config('SECRET_POINTAGE_EXIT', default='').strip()
|
||
if code == enter_code:
|
||
action = 'enter'
|
||
elif code == exit_code:
|
||
action = 'exit'
|
||
else:
|
||
return Response({'error': 'QR code de pointage invalide.'}, status=status.HTTP_400_BAD_REQUEST)
|
||
|
||
now = timezone.localtime()
|
||
with transaction.atomic():
|
||
open_entry = TimeEntry.objects.select_for_update().filter(
|
||
user=request.user,
|
||
date=now.date(),
|
||
start_time__isnull=False,
|
||
end_time__isnull=True,
|
||
).order_by('-start_time').first()
|
||
|
||
if action == 'enter':
|
||
if open_entry:
|
||
return Response({'error': 'Une entrée est déjà ouverte pour aujourd’hui.'}, status=status.HTTP_400_BAD_REQUEST)
|
||
entry = TimeEntry.objects.create(user=request.user, date=now.date(), start_time=now.time())
|
||
message = 'Entrée enregistrée.'
|
||
elif open_entry:
|
||
open_entry.end_time = now.time()
|
||
open_entry.save(update_fields=['end_time', 'updated_at'])
|
||
entry = open_entry
|
||
message = 'Sortie enregistrée.'
|
||
else:
|
||
entry = TimeEntry.objects.create(user=request.user, date=now.date(), end_time=now.time())
|
||
message = 'Sortie enregistrée sans entrée correspondante.'
|
||
|
||
return Response({'action': action, 'message': message, 'entry': TimeEntrySerializer(entry).data})
|
||
|
||
class TimeEntryDetailView(generics.RetrieveUpdateDestroyAPIView):
|
||
serializer_class = TimeEntrySerializer
|
||
permission_classes = [IsAuthenticated]
|
||
|
||
def get_queryset(self):
|
||
user = self.request.user
|
||
# Staff members can view all entries, regular users only their own
|
||
if user.is_staff or user.is_superuser:
|
||
return TimeEntry.objects.all()
|
||
return TimeEntry.objects.filter(user=user)
|
||
|
||
class TimeEntryUpdateView(APIView):
|
||
permission_classes = [IsAuthenticated]
|
||
|
||
def put(self, request, entry_id):
|
||
try:
|
||
entry = TimeEntry.objects.get(id=entry_id)
|
||
# Check if user is authorized to update this entry
|
||
if not (request.user.is_staff or request.user.is_superuser or entry.user == request.user):
|
||
return Response(
|
||
{'error': 'Accès refusé'},
|
||
status=status.HTTP_403_FORBIDDEN
|
||
)
|
||
|
||
serializer = TimeEntryUpdateSerializer(entry, data=request.data)
|
||
if serializer.is_valid():
|
||
serializer.save()
|
||
return Response(serializer.data)
|
||
return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
|
||
except TimeEntry.DoesNotExist:
|
||
return Response(
|
||
{'error': 'Pointage non trouvé'},
|
||
status=status.HTTP_404_NOT_FOUND
|
||
)
|
||
|
||
class WeeklySummaryView(APIView):
|
||
permission_classes = [IsAuthenticated]
|
||
|
||
def get(self, request, user_id=None):
|
||
# If user_id is provided and user is staff, get that user's summary
|
||
# Otherwise, get current user's summary
|
||
if user_id and (request.user.is_staff or request.user.is_superuser):
|
||
try:
|
||
user = User.objects.get(id=user_id)
|
||
except User.DoesNotExist:
|
||
return Response(
|
||
{'error': 'Utilisateur non trouvé'},
|
||
status=status.HTTP_404_NOT_FOUND
|
||
)
|
||
else:
|
||
user = request.user
|
||
|
||
# Get current week (Monday to Sunday)
|
||
today = timezone.now().date()
|
||
start_of_week = today - timedelta(days=today.weekday())
|
||
end_of_week = start_of_week + timedelta(days=6)
|
||
|
||
# Get all time entries for the current week for this user
|
||
entries = TimeEntry.objects.filter(
|
||
user=user,
|
||
date__gte=start_of_week,
|
||
date__lte=end_of_week
|
||
).order_by('date')
|
||
|
||
# Group entries by day
|
||
days_data = []
|
||
for i in range(7):
|
||
current_day = start_of_week + timedelta(days=i)
|
||
day_entries = entries.filter(date=current_day)
|
||
|
||
if day_entries.exists():
|
||
# Get the first entry for this day (assuming one entry per day)
|
||
entry = day_entries.first()
|
||
days_data.append({
|
||
'date': entry.date.isoformat(),
|
||
'day': entry.date.strftime('%A'),
|
||
'start_time': str(entry.start_time) if entry.start_time else None,
|
||
'end_time': str(entry.end_time) if entry.end_time else None,
|
||
'duration': entry.duration
|
||
})
|
||
else:
|
||
days_data.append({
|
||
'date': current_day.isoformat(),
|
||
'day': current_day.strftime('%A'),
|
||
'start_time': None,
|
||
'end_time': None,
|
||
'duration': 0
|
||
})
|
||
|
||
total_hours = sum(entry.duration or 0 for entry in entries)
|
||
|
||
summary = {
|
||
'user': user.username,
|
||
'username': user.username,
|
||
'total_hours': round(total_hours, 2),
|
||
'days': days_data
|
||
}
|
||
|
||
serializer = WeeklySummarySerializer(summary)
|
||
return Response(serializer.data)
|
||
|
||
class UserWeeklySummaryView(APIView):
|
||
permission_classes = [IsAuthenticated]
|
||
|
||
def get(self, request):
|
||
# Get current user's weekly summary
|
||
user = request.user
|
||
|
||
# Get current week (Monday to Sunday)
|
||
today = timezone.now().date()
|
||
start_of_week = today - timedelta(days=today.weekday())
|
||
end_of_week = start_of_week + timedelta(days=6)
|
||
|
||
# Get all time entries for the current week for this user
|
||
entries = TimeEntry.objects.filter(
|
||
user=user,
|
||
date__gte=start_of_week,
|
||
date__lte=end_of_week
|
||
).order_by('date')
|
||
|
||
# Group entries by day
|
||
days_data = []
|
||
for i in range(7):
|
||
current_day = start_of_week + timedelta(days=i)
|
||
day_entries = entries.filter(date=current_day)
|
||
|
||
if day_entries.exists():
|
||
# Get the first entry for this day (assuming one entry per day)
|
||
entry = day_entries.first()
|
||
days_data.append({
|
||
'date': entry.date.isoformat(),
|
||
'day': entry.date.strftime('%A'),
|
||
'start_time': str(entry.start_time) if entry.start_time else None,
|
||
'end_time': str(entry.end_time) if entry.end_time else None,
|
||
'duration': entry.duration
|
||
})
|
||
else:
|
||
days_data.append({
|
||
'date': current_day.isoformat(),
|
||
'day': current_day.strftime('%A'),
|
||
'start_time': None,
|
||
'end_time': None,
|
||
'duration': 0
|
||
})
|
||
|
||
total_hours = sum(entry.duration or 0 for entry in entries)
|
||
|
||
summary = {
|
||
'user': user.username,
|
||
'username': user.username,
|
||
'total_hours': round(total_hours, 2),
|
||
'days': days_data
|
||
}
|
||
|
||
serializer = WeeklySummarySerializer(summary)
|
||
return Response(serializer.data)
|