86 lines
3.6 KiB
Python
86 lines
3.6 KiB
Python
"""Règles communes aux saisies manuelles, aux groupes et aux scans."""
|
||
from datetime import datetime, time, timedelta
|
||
from uuid import uuid4
|
||
|
||
from django.contrib.auth import get_user_model
|
||
from django.db import transaction
|
||
from django.db.models import Q
|
||
from rest_framework.exceptions import ValidationError
|
||
|
||
from .models import TimeEntry
|
||
|
||
|
||
def validate_period(data):
|
||
if data.get('start_time') is None:
|
||
raise ValidationError({'start_time': "L'heure d'entrée est requise."})
|
||
data['end_date'] = data.get('end_date') or data['start_date']
|
||
start = datetime.combine(data['start_date'], data['start_time'])
|
||
end = None
|
||
if data['end_date'] < data['start_date']:
|
||
raise ValidationError({'end_date': 'La sortie doit suivre l’entrée.'})
|
||
if data.get('end_time') is not None:
|
||
end = datetime.combine(data['end_date'], data['end_time'])
|
||
if end <= start:
|
||
raise ValidationError({'end_time': 'La sortie doit suivre l’entrée.'})
|
||
elif data['end_date'] != data['start_date']:
|
||
raise ValidationError({'end_time': 'Une sortie est requise pour un pointage multi-jours.'})
|
||
return start, end
|
||
|
||
|
||
def lock_user(user):
|
||
# Verrou stable, même lorsqu'il n'existe encore aucun pointage.
|
||
get_user_model().objects.select_for_update().get(pk=user.pk)
|
||
|
||
|
||
def validate_overlap(user, data, excluded_ids=()):
|
||
start, end = validate_period(data)
|
||
entries = TimeEntry.objects.filter(user=user).exclude(pk__in=excluded_ids)
|
||
entries = entries.filter(Q(end_date__gte=start.date()) | Q(end_time__isnull=True))
|
||
if end is not None:
|
||
entries = entries.filter(start_date__lte=end.date())
|
||
for entry in entries:
|
||
if entry.start_time is None:
|
||
raise ValidationError({'detail': 'Un ancien pointage sans entrée doit être corrigé.'})
|
||
existing_start = datetime.combine(entry.start_date, entry.start_time)
|
||
existing_end = (
|
||
datetime.combine(entry.end_date or entry.start_date, entry.end_time)
|
||
if entry.end_time is not None else None
|
||
)
|
||
if (end is None or existing_start < end) and (existing_end is None or start < existing_end):
|
||
raise ValidationError({'detail': 'Ce pointage chevauche un pointage existant ou une entrée ouverte.'})
|
||
|
||
|
||
@transaction.atomic
|
||
def save_period(user, data, entries=()):
|
||
"""Enregistre un pointage logique entier, sans laisser de groupe partiel."""
|
||
lock_user(user)
|
||
data = dict(data)
|
||
validate_overlap(user, data, [entry.pk for entry in entries])
|
||
group_id = None
|
||
if data['start_date'] != data['end_date']:
|
||
group_id = next((entry.group_id for entry in entries if entry.group_id), None) or uuid4()
|
||
|
||
first = entries[0] if entries else None
|
||
if entries:
|
||
TimeEntry.objects.filter(pk__in=[entry.pk for entry in entries[1:]]).delete()
|
||
current_date = data['start_date']
|
||
created = []
|
||
while current_date <= data['end_date']:
|
||
values = dict(
|
||
user=user,
|
||
group_id=group_id,
|
||
start_date=current_date,
|
||
end_date=current_date,
|
||
start_time=data['start_time'] if current_date == data['start_date'] else time.min,
|
||
end_time=data.get('end_time') if current_date == data['end_date'] else time(23, 59, 59),
|
||
)
|
||
if first is not None and not created:
|
||
for key, value in values.items():
|
||
setattr(first, key, value)
|
||
first.save()
|
||
created.append(first)
|
||
else:
|
||
created.append(TimeEntry.objects.create(**values))
|
||
current_date += timedelta(days=1)
|
||
return created[0]
|