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)