303 lines
13 KiB
Python
303 lines
13 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 datetime import time as datetime_time
|
||
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().strip('"\'')
|
||
exit_code = config('SECRET_POINTAGE_EXIT', default='').strip().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()
|
||
requested_time = request.data.get('time')
|
||
if requested_time:
|
||
try:
|
||
selected_time = datetime_time.fromisoformat(str(requested_time)[:5])
|
||
except ValueError:
|
||
return Response({'error': 'Heure invalide.'}, status=status.HTTP_400_BAD_REQUEST)
|
||
else:
|
||
selected_time = now.time().replace(second=0, microsecond=0)
|
||
resolution = request.data.get('resolution')
|
||
|
||
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()
|
||
open_exit = TimeEntry.objects.select_for_update().filter(
|
||
user=request.user,
|
||
date=now.date(),
|
||
start_time__isnull=True,
|
||
end_time__isnull=False,
|
||
).order_by('-end_time').first()
|
||
|
||
if action == 'enter':
|
||
if open_entry:
|
||
if resolution == 'cancel_entry':
|
||
open_entry.delete()
|
||
return Response({'action': action, 'message': 'Entrée annulée.'})
|
||
if resolution == 'complete_entry':
|
||
if selected_time <= open_entry.start_time:
|
||
return Response({'error': 'L’heure de sortie doit être après l’entrée.'}, status=status.HTTP_400_BAD_REQUEST)
|
||
open_entry.end_time = selected_time
|
||
open_entry.save(update_fields=['end_time', 'updated_at'])
|
||
return Response({'action': action, 'message': 'Sortie enregistrée pour l’entrée précédente.', 'entry': TimeEntrySerializer(open_entry).data})
|
||
return Response({'error': 'Une entrée est déjà ouverte.', 'conflict': 'open_entry', 'options': ['cancel_entry', 'complete_entry']}, status=status.HTTP_409_CONFLICT)
|
||
if open_exit:
|
||
if resolution == 'cancel_exit':
|
||
open_exit.delete()
|
||
return Response({'action': action, 'message': 'Sortie annulée.'})
|
||
if resolution == 'complete_exit':
|
||
open_exit.start_time = selected_time
|
||
open_exit.save(update_fields=['start_time', 'updated_at'])
|
||
return Response({'action': action, 'message': 'Entrée enregistrée pour la sortie précédente.', 'entry': TimeEntrySerializer(open_exit).data})
|
||
return Response({'error': 'Une sortie existe déjà.', 'conflict': 'open_exit', 'options': ['cancel_exit', 'complete_exit']}, status=status.HTTP_409_CONFLICT)
|
||
entry = TimeEntry.objects.create(user=request.user, date=now.date(), start_time=selected_time)
|
||
message = 'Entrée enregistrée.'
|
||
elif open_entry:
|
||
open_entry.end_time = selected_time
|
||
open_entry.save(update_fields=['end_time', 'updated_at'])
|
||
entry = open_entry
|
||
message = 'Sortie enregistrée.'
|
||
elif open_exit:
|
||
if resolution == 'cancel_exit':
|
||
open_exit.delete()
|
||
return Response({'action': action, 'message': 'Sortie annulée.'})
|
||
if resolution == 'complete_exit':
|
||
if selected_time >= open_exit.end_time:
|
||
return Response({'error': 'L’heure d’entrée doit être avant la sortie.'}, status=status.HTTP_400_BAD_REQUEST)
|
||
open_exit.start_time = selected_time
|
||
open_exit.save(update_fields=['start_time', 'updated_at'])
|
||
return Response({'action': action, 'message': 'Entrée enregistrée pour la sortie précédente.', 'entry': TimeEntrySerializer(open_exit).data})
|
||
return Response({'error': 'Une sortie existe déjà.', 'conflict': 'open_exit', 'options': ['cancel_exit', 'complete_exit']}, status=status.HTTP_409_CONFLICT)
|
||
else:
|
||
previous_open_entry = TimeEntry.objects.select_for_update().filter(
|
||
user=request.user,
|
||
start_time__isnull=False,
|
||
end_time__isnull=True,
|
||
).exclude(date=now.date()).order_by('-date', '-start_time').first()
|
||
if previous_open_entry:
|
||
last_date = previous_open_entry.date
|
||
previous_open_entry.end_time = datetime_time(23, 59, 59)
|
||
previous_open_entry.save(update_fields=['end_time', 'updated_at'])
|
||
|
||
current_date = last_date + timedelta(days=1)
|
||
while current_date < now.date():
|
||
TimeEntry.objects.create(
|
||
user=request.user,
|
||
date=current_date,
|
||
start_time=datetime_time.min,
|
||
end_time=datetime_time(23, 59, 59),
|
||
)
|
||
current_date += timedelta(days=1)
|
||
|
||
entry = TimeEntry.objects.create(user=request.user, date=now.date(), end_time=selected_time)
|
||
message = 'Sortie enregistrée sur plusieurs jours.'
|
||
else:
|
||
entry = TimeEntry.objects.create(user=request.user, date=now.date(), end_time=selected_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]
|
||
lookup_url_kwarg = 'entry_id'
|
||
|
||
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)
|