ants-hub/src/ants_hub/api/views/ants.py

# ANTS-Hub - Copyright (C) Entr'ouvert

import datetime
import functools
import secrets

from django.conf import settings
from django.http import JsonResponse

from ants_hub.data.models import Lieu, Plage, RendezVous, TypeDeRdv


def authenticate(func):
    @functools.wraps(func)
    def wrapper(request, *args, **kwargs):
        header = request.headers.get('X-Hub-Rdv-Auth-Token', '')
        if not header:
            return JsonResponse("Missing X-HUB-RDV-AUTH-TOKEN header", status=401, safe=False)
        setting = settings.ANTS_HUB_X_HUB_RDV_AUTH_TOKEN
        if not setting:
            return JsonResponse("X-HUB-RDV-AUTH-TOKEN header invalid", status=401, safe=False)
        if not secrets.compare_digest(header, setting):
            return JsonResponse("X-HUB-RDV-AUTH-TOKEN header invalid", status=401, safe=False)
        return func(request, *args, **kwargs)

    return wrapper


@authenticate
def get_managed_meeting_points(request):
    lieux = []
    for lieu in Lieu.objects.select_related('collectivite'):
        d = {
            'id': str(lieu.id),
            'name': lieu.nom,
            'longitude': lieu.longitude,
            'latitude': lieu.latitude,
            'public_entry_address': lieu.numero_rue,
            'zip_code': lieu.code_postal,
            'city_name': lieu.ville,
        }
        website = lieu.url or lieu.collectivite.url
        if website:
            d['website'] = website
        city_logo = lieu.logo_url or lieu.collectivite.logo_url
        if city_logo:
            d['city_logo'] = city_logo
        lieux.append(d)
    return JsonResponse(lieux, safe=False)


def available_time_slots_parse_qs(request):
    meeting_point_ids = request.GET.getlist('meeting_point_ids')
    if not meeting_point_ids:
        raise ValueError('missing meeting_point_ids')
    try:
        meeting_point_ids = map(int, map(str.strip, meeting_point_ids))
    except ValueError:
        raise ValueError('invalid meeting_point_ids')
    start_date = request.GET.get('start_date', '').strip()
    if not start_date:
        raise ValueError('missing start_date')
    try:
        start_date = datetime.datetime.strptime(start_date, '%Y-%m-%d').date()
    except ValueError:
        raise ValueError('invalid start_date')
    if start_date < datetime.date.today():
        start_date = datetime.date.today()
    end_date = request.GET.get('end_date', '').strip()
    if not end_date:
        raise ValueError('missing end_date')
    try:
        end_date = datetime.datetime.strptime(end_date, '%Y-%m-%d').date()
    except ValueError:
        raise ValueError('invalid end_date')
    if end_date < start_date:
        raise ValueError('end_date must be >= start_date')
    end_date = min(end_date, start_date + datetime.timedelta(days=180))
    reason = request.GET.get('reason', 'CNI').strip()
    if reason:
        try:
            reason = TypeDeRdv.from_label(reason)
        except ValueError:
            reason = TypeDeRdv.CNI
    try:
        documents_number = int(request.GET.get('documents_number', '').strip())
    except (ValueError, TypeError):
        documents_number = 1
    return meeting_point_ids, start_date, end_date, reason, documents_number


@authenticate
def available_time_slots(request):
    try:
        meeting_point_ids, start_date, end_date, reason, documents_number = available_time_slots_parse_qs(
            request
        )
    except ValueError as e:
        return JsonResponse(
            {
                'detail': [
                    {
                        'loc': 'query-string',
                        'msg': str(e),
                        'type': 'bad request',
                    }
                ]
            },
            status=422,
        )
    types_de_rdv = [reason]
    if reason in (TypeDeRdv.CNI, TypeDeRdv.PASSPORT):
        types_de_rdv.append(TypeDeRdv.CNI_PASSPORT)
    elif reason == TypeDeRdv.CNI_PASSPORT:
        types_de_rdv.append(TypeDeRdv.CNI)
        types_de_rdv.append(TypeDeRdv.PASSPORT)

    rdv_by_lieu = {}
    qs = Plage.objects.filter(lieu__id__in=meeting_point_ids)
    qs = qs.filter(type_de_rdv__in=types_de_rdv)
    qs = qs.filter(date__gte=start_date, date__lte=end_date)
    qs = qs.filter(persones__gte=documents_number)
    qs = qs.select_related('lieu', 'lieu__collectivite')
    for plage in qs.order_by('lieu', 'date'):
        for time_slot in plage.available_time_slots():
            time_slot['callback_url'] = request.build_absolute_uri(time_slot['callback_url'])
            rdv_by_lieu.setdefault(str(plage.lieu_id), []).append(time_slot)
    return JsonResponse(rdv_by_lieu)


@authenticate
def search_application_ids(request):
    application_ids = map(str.strip, request.GET.getlist('application_ids'))
    application_ids = filter(None, application_ids)
    application_ids = filter(lambda x: len(x) < 64, application_ids)
    if not application_ids:
        return JsonResponse(
            {
                'detail': [
                    {
                        'loc': 'query-string',
                        'msg': 'missing application_ids',
                        'type': 'bad request',
                    }
                ]
            },
            status=422,
        )
    rdv_by_identifiant_predemande = {}
    qs = RendezVous.objects.filter(identifiant_predemande__in=application_ids)
    qs = qs.select_related('lieu', 'lieu__collectivite')
    for rdv in qs:
        rdv_by_identifiant_predemande.setdefault(rdv.identifiant_predemande, []).append(
            {
                'meeting_point': rdv.lieu.nom,
                'datetime': rdv.date.isoformat(),
                'management_url': request.build_absolute_uri(rdv.make_gestion_url()),
                'annulation_url': request.build_absolute_uri(rdv.make_annulation_url()),
            }
        )
    return JsonResponse(rdv_by_identifiant_predemande)