244 lines
11 KiB
Python
244 lines
11 KiB
Python
from datetime import timedelta
|
||
|
||
from decouple import config
|
||
from django.contrib.auth import get_user_model
|
||
from django.db import transaction
|
||
from django.shortcuts import get_object_or_404
|
||
from django.utils import timezone
|
||
from rest_framework import generics, status
|
||
from rest_framework.exceptions import PermissionDenied, ValidationError
|
||
from rest_framework.permissions import IsAuthenticated
|
||
from rest_framework.response import Response
|
||
from rest_framework.views import APIView
|
||
|
||
from .models import TimeEntry
|
||
from .serializers import (
|
||
TimeEntryCreateSerializer, TimeEntrySerializer,
|
||
TimeEntryUpdateSerializer, WeeklySummarySerializer,
|
||
)
|
||
from .services import lock_user, save_period
|
||
|
||
User = get_user_model()
|
||
|
||
|
||
def visible_entries(user):
|
||
entries = TimeEntry.objects.select_related('user')
|
||
return entries if user.is_staff or user.is_superuser else entries.filter(user=user)
|
||
|
||
|
||
class TimeEntryListView(generics.ListAPIView):
|
||
serializer_class = TimeEntrySerializer
|
||
permission_classes = [IsAuthenticated]
|
||
|
||
def get_queryset(self):
|
||
return visible_entries(self.request.user)
|
||
|
||
|
||
class TimeEntryCreateView(generics.CreateAPIView):
|
||
serializer_class = TimeEntryCreateSerializer
|
||
permission_classes = [IsAuthenticated]
|
||
|
||
def create(self, request, *args, **kwargs):
|
||
serializer = self.get_serializer(data=request.data)
|
||
serializer.is_valid(raise_exception=True)
|
||
user = serializer.validated_data.get('user', request.user)
|
||
if user != request.user and not (request.user.is_staff or request.user.is_superuser):
|
||
raise PermissionDenied('Vous ne pouvez pas créer un pointage pour un autre utilisateur.')
|
||
entry = serializer.save(user=user)
|
||
return Response(TimeEntrySerializer(entry).data, status=status.HTTP_201_CREATED)
|
||
|
||
|
||
class TimeEntryDetailView(generics.RetrieveUpdateDestroyAPIView):
|
||
serializer_class = TimeEntrySerializer
|
||
permission_classes = [IsAuthenticated]
|
||
lookup_url_kwarg = 'entry_id'
|
||
|
||
def get_queryset(self):
|
||
return visible_entries(self.request.user)
|
||
|
||
@transaction.atomic
|
||
def update(self, request, *args, **kwargs):
|
||
entry = self.get_object()
|
||
lock_user(entry.user)
|
||
entry = self.get_queryset().select_for_update().get(pk=entry.pk)
|
||
if entry.group_id:
|
||
raise ValidationError({'detail': 'Modifiez le pointage entier avec la route groups.'})
|
||
serializer = TimeEntryUpdateSerializer(
|
||
entry, data=request.data, partial=True,
|
||
)
|
||
serializer.is_valid(raise_exception=True)
|
||
result = serializer.save()
|
||
return Response(TimeEntrySerializer(result).data)
|
||
|
||
@transaction.atomic
|
||
def destroy(self, request, *args, **kwargs):
|
||
entry = self.get_object()
|
||
lock_user(entry.user)
|
||
entry = self.get_queryset().select_for_update().get(pk=entry.pk)
|
||
if entry.group_id:
|
||
raise ValidationError({'detail': 'Supprimez le pointage entier avec la route groups.'})
|
||
entry.delete()
|
||
return Response(status=status.HTTP_204_NO_CONTENT)
|
||
|
||
|
||
class TimeEntryUpdateView(TimeEntryDetailView):
|
||
# Alias conservé pour les anciens clients, avec les mêmes validations.
|
||
http_method_names = ['put', 'patch', 'options']
|
||
|
||
|
||
class TimeEntryGroupView(APIView):
|
||
permission_classes = [IsAuthenticated]
|
||
|
||
def get_entries(self, request, group_id):
|
||
queryset = visible_entries(request.user).filter(group_id=group_id)
|
||
first = queryset.order_by('start_date', 'start_time').first()
|
||
if first is None:
|
||
from rest_framework.exceptions import NotFound
|
||
raise NotFound()
|
||
lock_user(first.user)
|
||
entries = list(queryset.filter(user=first.user).select_for_update().order_by('start_date', 'start_time'))
|
||
if not entries:
|
||
from rest_framework.exceptions import NotFound
|
||
raise NotFound()
|
||
return entries
|
||
|
||
@transaction.atomic
|
||
def patch(self, request, group_id):
|
||
entries = self.get_entries(request, group_id)
|
||
serializer = TimeEntryUpdateSerializer(
|
||
entries[0], data=request.data, partial=True,
|
||
context={'entries': entries},
|
||
)
|
||
serializer.is_valid(raise_exception=True)
|
||
result = serializer.save()
|
||
return Response(TimeEntrySerializer(result).data)
|
||
|
||
put = patch
|
||
|
||
@transaction.atomic
|
||
def delete(self, request, group_id):
|
||
entries = self.get_entries(request, group_id)
|
||
TimeEntry.objects.filter(pk__in=[entry.pk for entry in entries]).delete()
|
||
return Response(status=status.HTTP_204_NO_CONTENT)
|
||
|
||
|
||
class TimeEntryScanView(APIView):
|
||
permission_classes = [IsAuthenticated]
|
||
|
||
def parse_time(self, value, default):
|
||
if value is None:
|
||
return default
|
||
from rest_framework import serializers
|
||
return serializers.TimeField(input_formats=['%H:%M', '%H:%M:%S']).run_validation(value)
|
||
|
||
def conflict(self, kind, resolution, message, selected_time, today, entry=None):
|
||
return Response({
|
||
'error': message, 'message': message, 'conflict': kind,
|
||
'resolution': resolution,
|
||
'missing_time': 'end_time' if kind == 'open_entry' else 'start_time',
|
||
'scan_time': selected_time.strftime('%H:%M'),
|
||
'scan_date': today.isoformat(),
|
||
'entry_id': entry.pk if entry else None,
|
||
}, status=status.HTTP_409_CONFLICT)
|
||
|
||
@transaction.atomic
|
||
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 not enter_code or not exit_code or enter_code == exit_code:
|
||
return Response({'error': 'Les codes de pointage ne sont pas configurés.'}, status=503)
|
||
if code not in (enter_code, exit_code):
|
||
raise ValidationError({'error': 'QR code de pointage invalide.'})
|
||
action = 'enter' if code == enter_code else 'exit'
|
||
now = timezone.localtime()
|
||
today = now.date()
|
||
selected_time = self.parse_time(request.data.get('time'), now.time().replace(second=0, microsecond=0))
|
||
resolution = request.data.get('resolution')
|
||
if resolution not in (None, 'complete_exit', 'complete_missing_entry'):
|
||
raise ValidationError({'error': 'Résolution de conflit inconnue.'})
|
||
if resolution and request.data.get('scan_date') != today.isoformat():
|
||
raise ValidationError({'error': 'Le jour a changé. Scannez à nouveau le code.'})
|
||
|
||
lock_user(request.user)
|
||
open_entries = list(TimeEntry.objects.select_for_update().filter(
|
||
user=request.user, end_time__isnull=True,
|
||
).order_by('start_date', 'start_time'))
|
||
if len(open_entries) > 1:
|
||
raise ValidationError({'error': 'Plusieurs entrées sont ouvertes. Corrigez les pointages avant de scanner.'})
|
||
open_entry = open_entries[0] if open_entries else None
|
||
|
||
if resolution == 'complete_missing_entry':
|
||
if action != 'exit' or open_entry:
|
||
raise ValidationError({'error': 'La situation a changé. Scannez à nouveau le code.'})
|
||
scan_time = self.parse_time(request.data.get('scan_time'), None)
|
||
if scan_time is None:
|
||
raise ValidationError({'scan_time': 'L’heure de sortie scannée est requise.'})
|
||
entry = save_period(request.user, dict(
|
||
start_date=today, end_date=today,
|
||
start_time=selected_time, end_time=scan_time,
|
||
))
|
||
action = 'exit'
|
||
elif resolution == 'complete_exit' or action == 'exit':
|
||
if not open_entry:
|
||
if resolution:
|
||
raise ValidationError({'error': 'Aucune entrée ouverte à compléter.'})
|
||
return self.conflict('missing_entry', 'complete_missing_entry',
|
||
'Aucune entrée ouverte. Renseignez l’heure d’entrée.', selected_time, today)
|
||
if resolution and str(request.data.get('entry_id')) != str(open_entry.pk):
|
||
raise ValidationError({'error': 'Le pointage a changé. Scannez à nouveau le code.'})
|
||
entry = save_period(request.user, dict(
|
||
start_date=open_entry.start_date, start_time=open_entry.start_time,
|
||
end_date=today, end_time=selected_time,
|
||
), [open_entry])
|
||
action = 'exit'
|
||
else:
|
||
if open_entry:
|
||
return self.conflict('open_entry', 'complete_exit',
|
||
'Une entrée est déjà ouverte. Voulez-vous enregistrer une sortie ?',
|
||
selected_time, today, open_entry)
|
||
entry = save_period(request.user, dict(
|
||
start_date=today, end_date=today, start_time=selected_time, end_time=None,
|
||
))
|
||
return Response({
|
||
'action': action,
|
||
'message': 'Entrée enregistrée.' if action == 'enter' else 'Sortie enregistrée.',
|
||
'entry': TimeEntrySerializer(entry).data,
|
||
})
|
||
|
||
|
||
def build_weekly_summary(user):
|
||
today = timezone.localdate()
|
||
start_of_week = today - timedelta(days=today.weekday())
|
||
entries = list(TimeEntry.objects.filter(
|
||
user=user, start_date__range=(start_of_week, start_of_week + timedelta(days=6)),
|
||
).order_by('start_date', 'start_time'))
|
||
days_data = []
|
||
for offset in range(7):
|
||
day = start_of_week + timedelta(days=offset)
|
||
day_entries = [entry for entry in entries if entry.start_date == day]
|
||
completed = [entry for entry in day_entries if entry.end_time is not None]
|
||
days_data.append({
|
||
'date': day.isoformat(), 'day': day.strftime('%A'),
|
||
'start_time': str(day_entries[0].start_time) if day_entries else None,
|
||
'end_time': str(max(entry.end_time for entry in completed)) if completed and len(completed) == len(day_entries) else None,
|
||
'duration': sum(entry.duration or 0 for entry in day_entries),
|
||
'entries_count': len(day_entries),
|
||
})
|
||
return dict(user=user.username, username=user.username,
|
||
total_hours=round(sum(entry.duration or 0 for entry in entries), 2), entries=days_data)
|
||
|
||
|
||
class WeeklySummaryView(APIView):
|
||
permission_classes = [IsAuthenticated]
|
||
|
||
def get(self, request, user_id=None):
|
||
user = request.user
|
||
if user_id is not None and user_id != user.pk:
|
||
if not (user.is_staff or user.is_superuser):
|
||
raise PermissionDenied('Vous ne pouvez pas consulter ce résumé.')
|
||
user = get_object_or_404(User, pk=user_id)
|
||
return Response(WeeklySummarySerializer(build_weekly_summary(user)).data)
|