500 lines
16 KiB
Python
500 lines
16 KiB
Python
# ANTS-Hub - Copyright (C) Entr'ouvert
|
|
|
|
import datetime
|
|
import hashlib
|
|
import re
|
|
import secrets
|
|
import typing
|
|
import uuid
|
|
|
|
from django import forms
|
|
from django.core.exceptions import ValidationError
|
|
from django.core.validators import MaxValueValidator, MinValueValidator
|
|
from django.db import models
|
|
from django.urls import reverse
|
|
from django.utils.text import slugify
|
|
from django.utils.timezone import is_naive
|
|
|
|
from ants_hub.timezone import localtime, make_aware
|
|
|
|
|
|
class CharField(models.TextField):
|
|
'''TextField using forms.TextInput as widget'''
|
|
|
|
def formfield(self, **kwargs):
|
|
defaults = {'widget': forms.TextInput}
|
|
defaults.update(**kwargs)
|
|
return super().formfield(**defaults)
|
|
|
|
|
|
class URLField(models.URLField):
|
|
'''URLField using a TEXT column for storage'''
|
|
|
|
def get_internal_type(self):
|
|
return 'TextField'
|
|
|
|
|
|
class Horaire(typing.NamedTuple):
|
|
start: datetime.time
|
|
end: datetime.time
|
|
|
|
def __str__(self):
|
|
return f'{self.start.isoformat()}-{self.end.isoformat()}'
|
|
|
|
|
|
class HoraireList(list):
|
|
def __str__(self):
|
|
return '\n'.join(str(h) for h in self)
|
|
|
|
def append(self, value):
|
|
assert isinstance(value, Horaire)
|
|
if self and value.start < self[-1].end:
|
|
raise ValueError
|
|
super().append(value)
|
|
|
|
@classmethod
|
|
def from_str(cls, value):
|
|
try:
|
|
terms = map(str.strip, re.split('[\n,]', value.strip()))
|
|
hl = cls()
|
|
for term in terms:
|
|
start, end = map(str.strip, term.split('-', 1))
|
|
start = datetime.time.fromisoformat(start)
|
|
end = datetime.time.fromisoformat(end)
|
|
if end < start:
|
|
raise ValueError
|
|
hl.append(Horaire(start, end))
|
|
return hl
|
|
except (ValueError, TypeError):
|
|
raise ValueError
|
|
|
|
|
|
class HoraireFormField(forms.CharField):
|
|
widget = forms.Textarea
|
|
|
|
def to_python(self, value):
|
|
if value in self.empty_values:
|
|
return None
|
|
if isinstance(value, list):
|
|
return HoraireList(x if isinstance(x, Horaire) else Horaire(*x) for x in value)
|
|
try:
|
|
return HoraireList.from_str(value)
|
|
except ValueError:
|
|
raise ValidationError('Horaires invalides, ex.: 12:00-13:00,14:00-15:00')
|
|
|
|
|
|
class HoraireField(models.TextField):
|
|
def formfield(self, **kwargs):
|
|
defaults = {'form_class': HoraireFormField}
|
|
defaults.update(**kwargs)
|
|
return super().formfield(**defaults)
|
|
|
|
def get_prep_value(self, value):
|
|
return str(value)
|
|
|
|
def from_db_value(self, value, expression, connection):
|
|
return HoraireList.from_str(value)
|
|
|
|
def to_python(self, value):
|
|
if isinstance(value, HoraireList):
|
|
return value
|
|
|
|
if value is None:
|
|
return value
|
|
|
|
if isinstance(value, list):
|
|
return HoraireList(Horaire(x) for x in value)
|
|
|
|
try:
|
|
return HoraireList.from_str(value)
|
|
except ValueError:
|
|
raise ValidationError('Horaires invalides, ex.: 12:00-13:00,14:00-15:00')
|
|
|
|
|
|
class TypeDeRdv(models.IntegerChoices):
|
|
CNI = 1, 'CNI'
|
|
PASSPORT = 2, 'Passeport'
|
|
CNI_PASSPORT = 3, 'CNI et passeport'
|
|
|
|
@property
|
|
def ants_name(self):
|
|
return super().name.replace('_', '-')
|
|
|
|
@classmethod
|
|
def from_ants_name(cls, name):
|
|
return cls[name.replace('-', '_')]
|
|
|
|
|
|
class TypeDeRdvField(models.SmallIntegerField):
|
|
def __init__(self, **kwargs):
|
|
kwargs['choices'] = TypeDeRdv.choices
|
|
super().__init__(**kwargs)
|
|
|
|
|
|
def generate_apikey():
|
|
return uuid.uuid4().hex
|
|
|
|
|
|
def make_uuid_from_apikey(apikey):
|
|
return uuid.UUID(bytes=hashlib.sha256(apikey.encode()).digest()[:16])
|
|
|
|
|
|
class RaccordementManager(models.Manager):
|
|
def get_by_natural_key(self, name):
|
|
return self.get(name=name)
|
|
|
|
def get_by_apikey(self, apikey):
|
|
'''Use an hashed version of the apikey for lookup, to prevent timing attacks.'''
|
|
apikey = apikey.strip()
|
|
guid = make_uuid_from_apikey(apikey)
|
|
try:
|
|
raccordement = self.get(apikey_digest=guid)
|
|
except self.model.DoesNotExist:
|
|
return None
|
|
if secrets.compare_digest(raccordement.apikey, apikey):
|
|
return raccordement
|
|
return None
|
|
|
|
|
|
class Raccordement(models.Model):
|
|
uuid = models.UUIDField(verbose_name='UUID', default=uuid.uuid4, primary_key=True, editable=False)
|
|
name = CharField(verbose_name='Nom', unique=True)
|
|
apikey = CharField(
|
|
verbose_name='API key',
|
|
default=generate_apikey,
|
|
unique=True,
|
|
help_text='Écrire "NEW" pour en générer une nouvelle.',
|
|
)
|
|
apikey_digest = models.UUIDField(verbose_name='Condensat de l\'API key', db_index=True, editable=False)
|
|
notes = models.TextField(verbose_name="Notes", blank=True)
|
|
created = models.DateTimeField(verbose_name='Création', auto_now_add=True)
|
|
last_update = models.DateTimeField(verbose_name='Dernière mise à jour', auto_now=True)
|
|
|
|
objects = RaccordementManager()
|
|
|
|
def __str__(self):
|
|
return f'{self.name}'
|
|
|
|
def natural_key(self):
|
|
return (self.name,)
|
|
|
|
def clean(self):
|
|
if self.apikey.strip() == 'NEW':
|
|
self.apikey = generate_apikey()
|
|
self.apikey = self.apikey.strip()
|
|
if len(self.apikey) < 32:
|
|
raise ValidationError('API key doit faire au moins 32 caractères')
|
|
|
|
def save(self, *args, **kwargs):
|
|
self.apikey_digest = make_uuid_from_apikey(self.apikey)
|
|
return super().save(*args, **kwargs)
|
|
|
|
def lock(self):
|
|
type(self).objects.filter(pk=self.pk).select_for_update(nowait=True).get()
|
|
|
|
@property
|
|
def raccordement_url(self):
|
|
class URL:
|
|
@classmethod
|
|
def render(cls, context):
|
|
url = context['request'].build_absolute_uri('/api/chrono/')
|
|
url = url.replace('://', f'://{self.apikey}:@')
|
|
return f'<a href="{url}">{url}</a>'
|
|
|
|
return URL
|
|
|
|
class Meta:
|
|
verbose_name = 'raccordement'
|
|
verbose_name_plural = 'raccordements'
|
|
ordering = ('name',)
|
|
db_table = 'ants_hub_raccordement'
|
|
|
|
|
|
class CollectiviteManager(models.Manager):
|
|
def get_by_natural_key(self, source_id, *args):
|
|
return self.get(source_id=source_id, raccordement=Raccordement.objects.get_by_natural_key(*args))
|
|
|
|
|
|
class Collectivite(models.Model):
|
|
raccordement = models.ForeignKey(
|
|
verbose_name='Raccordement', to=Raccordement, related_name='collectivites', on_delete=models.PROTECT
|
|
)
|
|
nom = CharField(verbose_name='Nom')
|
|
source_id = CharField(verbose_name='Identifiant de la collectivité à la source')
|
|
url = URLField(verbose_name='URL du portail')
|
|
logo_url = URLField(verbose_name='URL du logo', blank=True)
|
|
rdv_url = URLField(verbose_name='URL de prise de rendez-vous ', blank=True)
|
|
gestion_url = URLField(verbose_name='URL de gestion des rendez-vous', blank=True)
|
|
annulation_url = URLField(verbose_name='URL d\'annulation des rendez-vous', blank=True)
|
|
created = models.DateTimeField(verbose_name='Création', auto_now_add=True)
|
|
last_update = models.DateTimeField(verbose_name='Dernière mise à jour', auto_now=True)
|
|
|
|
objects = CollectiviteManager()
|
|
|
|
def __str__(self):
|
|
return f'{self.nom}'
|
|
|
|
def natural_key(self):
|
|
return (self.source_id,) + self.raccordement.natural_key()
|
|
|
|
natural_key.dependencies = ['data.raccordement']
|
|
|
|
@property
|
|
def slug(self):
|
|
return slugify(self.nom)
|
|
|
|
class Meta:
|
|
verbose_name = 'collectivité'
|
|
verbose_name_plural = 'collectivités'
|
|
unique_together = [
|
|
['raccordement', 'source_id'],
|
|
['raccordement', 'nom'],
|
|
]
|
|
db_table = 'ants_hub_collectivite'
|
|
|
|
|
|
class LieuManager(models.Manager):
|
|
def get_by_natural_key(self, source_id, *args):
|
|
return self.get(source_id=source_id, collectivite=Collectivite.objects.get_by_natural_key(*args))
|
|
|
|
|
|
class Lieu(models.Model):
|
|
collectivite = models.ForeignKey(
|
|
verbose_name='Collectivité', to=Collectivite, related_name='lieux', on_delete=models.PROTECT
|
|
)
|
|
nom = CharField(verbose_name='Nom')
|
|
source_id = CharField(verbose_name='Identifiant du lieu à la source')
|
|
numero_rue = CharField(verbose_name='Numéro rue')
|
|
code_postal = CharField(verbose_name='Code postal')
|
|
ville = CharField(verbose_name='Ville')
|
|
longitude = models.FloatField(verbose_name='Longitude')
|
|
latitude = models.FloatField(verbose_name='Latitude')
|
|
url = URLField(verbose_name='URL du portail', blank=True)
|
|
logo_url = URLField(verbose_name='URL du logo', blank=True)
|
|
rdv_url = URLField(verbose_name='URL de prise de rendez-vous ', blank=True)
|
|
gestion_url = URLField(verbose_name='URL de gestion des rendez-vous', blank=True)
|
|
annulation_url = URLField(verbose_name='URL d\'annulation des rendez-vous', blank=True)
|
|
created = models.DateTimeField(verbose_name='Création', auto_now_add=True)
|
|
last_update = models.DateTimeField(verbose_name='Dernière mise à jour', auto_now=True)
|
|
|
|
objects = LieuManager()
|
|
|
|
def __str__(self):
|
|
return f'{self.nom} / {self.numero_rue} / {self.ville}'
|
|
|
|
def natural_key(self):
|
|
return (self.source_id,) + self.collectivite.natural_key()
|
|
|
|
natural_key.dependencies = ['data.collectivite']
|
|
|
|
@property
|
|
def slug(self):
|
|
return slugify(self.nom)
|
|
|
|
def get_rdv_url(self):
|
|
return self.rdv_url or self.collectivite.rdv_url or self.url or self.collectivite.url
|
|
|
|
class Meta:
|
|
verbose_name = 'lieu'
|
|
verbose_name_plural = 'lieux'
|
|
unique_together = [
|
|
('collectivite', 'source_id'),
|
|
('collectivite', 'nom'),
|
|
]
|
|
db_table = 'ants_hub_lieu'
|
|
|
|
|
|
class PlageManager(models.Manager):
|
|
def get_by_natural_key(self, date, type_de_rdv, *args):
|
|
return self.get(date=date, type_de_rdv=type_de_rdv, lieu=Lieu.objects.get_by_natural_key(*args))
|
|
|
|
|
|
def make_available_time_slots(date, horaires, duree, **kwargs):
|
|
for horaire in horaires:
|
|
start = datetime.datetime.combine(date, horaire.start)
|
|
if is_naive(start):
|
|
start = make_aware(start)
|
|
end = datetime.datetime.combine(date, horaire.end)
|
|
if is_naive(end):
|
|
end = make_aware(end)
|
|
start = localtime(start)
|
|
end = localtime(end)
|
|
while start < end:
|
|
yield start, make_rdv_url(date=start, **kwargs)
|
|
start += datetime.timedelta(minutes=duree)
|
|
|
|
|
|
template_make_rdv_url = None
|
|
|
|
|
|
def make_rdv_url(collectivite_pk, collectivite_slug, lieu_pk, lieu_slug, date):
|
|
return f'rdv/{collectivite_slug}-{collectivite_pk}/{lieu_slug}-{lieu_pk}/{date.isoformat()}/'
|
|
|
|
|
|
class Plage(models.Model):
|
|
lieu = models.ForeignKey(verbose_name='Lieu', to=Lieu, related_name='plages', on_delete=models.CASCADE)
|
|
date = models.DateField(verbose_name='Date')
|
|
horaires = HoraireField(verbose_name='Horaires')
|
|
duree = models.SmallIntegerField(verbose_name='Durée')
|
|
type_de_rdv = TypeDeRdvField(verbose_name='Type de rendez-vous')
|
|
personnes = models.SmallIntegerField(
|
|
verbose_name='Nombre de personnes maximum',
|
|
default=1,
|
|
validators=[MinValueValidator(1), MaxValueValidator(4)],
|
|
)
|
|
created = models.DateTimeField(verbose_name='Création', auto_now_add=True)
|
|
last_update = models.DateTimeField(verbose_name='Dernière mise à jour', auto_now=True)
|
|
|
|
objects = PlageManager()
|
|
|
|
def __str__(self):
|
|
return f'{self.date.isoformat()} / {self.horaires}'
|
|
|
|
def natural_key(self):
|
|
return (self.date, self.type_de_rdv) + self.lieu.natural_key()
|
|
|
|
natural_key.dependencies = ['data.lieu']
|
|
|
|
def _make_rdv_url(self, datetime):
|
|
return make_rdv_url(
|
|
collectivite_pk=self.lieu.collectivite.pk,
|
|
collectivite_slug=self.lieu.collectivite.slug,
|
|
lieu_pk=self.lieu.pk,
|
|
lieu_slug=self.lieu.slug,
|
|
date=datetime,
|
|
)
|
|
|
|
def available_time_slots(self):
|
|
return make_available_time_slots(
|
|
date=self.date,
|
|
horaires=self.horaires,
|
|
duree=self.duree,
|
|
collectivite_pk=self.lieu.collectivite.pk,
|
|
collectivite_slug=self.lieu.collectivite.slug,
|
|
lieu_pk=self.lieu.pk,
|
|
lieu_slug=self.lieu.slug,
|
|
)
|
|
|
|
class Meta:
|
|
verbose_name = 'plage'
|
|
verbose_name_plural = 'plages'
|
|
unique_together = [
|
|
['type_de_rdv', 'date', 'lieu', 'personnes'],
|
|
]
|
|
db_table = 'ants_hub_plage'
|
|
|
|
|
|
class RendezVousManager(models.Manager):
|
|
def get_by_natural_key(self, identifiant_predemande, *args):
|
|
return self.get(
|
|
identifiant_predemande=identifiant_predemande, lieu=Lieu.objects.get_by_natural_key(*args)
|
|
)
|
|
|
|
|
|
class RendezVous(models.Model):
|
|
uuid = models.UUIDField(verbose_name='UUID', default=uuid.uuid4, primary_key=True, editable=False)
|
|
lieu = models.ForeignKey(verbose_name='Lieu', to=Lieu, related_name='rdvs', on_delete=models.CASCADE)
|
|
identifiant_predemande = CharField(verbose_name='Identifiant de prédemande', db_index=True)
|
|
date = models.DateTimeField(verbose_name='Date')
|
|
gestion_url = URLField(verbose_name='URL de gestion des rendez-vous', blank=True)
|
|
annulation_url = URLField(verbose_name='URL d\'annulation des rendez-vous', blank=True)
|
|
created = models.DateTimeField(verbose_name='Création', auto_now_add=True)
|
|
last_update = models.DateTimeField(verbose_name='Dernière mise à jour', auto_now=True)
|
|
|
|
objects = RendezVousManager()
|
|
|
|
def __str__(self):
|
|
return f'{self.date} / {self.identifiant_predemande}'
|
|
|
|
def natural_key(self):
|
|
return (self.identifiant_predemande,) + self.lieu.natural_key()
|
|
|
|
natural_key.dependencies = ['data.lieu']
|
|
|
|
def make_gestion_url(self):
|
|
return reverse(
|
|
'gestion-redirect',
|
|
kwargs={
|
|
'collectivite_pk': self.lieu.collectivite.pk,
|
|
'collectivite_slug': self.lieu.collectivite.slug,
|
|
'lieu_pk': self.lieu.pk,
|
|
'lieu_slug': self.lieu.slug,
|
|
'date': localtime(self.date),
|
|
'rdv_pk': self.pk,
|
|
},
|
|
)
|
|
|
|
def get_gestion_url(self):
|
|
return (
|
|
self.gestion_url
|
|
or self.lieu.gestion_url
|
|
or self.lieu.collectivite.gestion_url
|
|
or self.lieu.rdv_url
|
|
or self.lieu.collectivite.rdv_url
|
|
or self.lieu.url
|
|
or self.lieu.collectivite.url
|
|
)
|
|
|
|
def make_annulation_url(self):
|
|
return reverse(
|
|
'annulation-redirect',
|
|
kwargs={
|
|
'collectivite_pk': self.lieu.collectivite.pk,
|
|
'collectivite_slug': self.lieu.collectivite.slug,
|
|
'lieu_pk': self.lieu.pk,
|
|
'lieu_slug': self.lieu.slug,
|
|
'date': localtime(self.date),
|
|
'rdv_pk': self.pk,
|
|
},
|
|
)
|
|
|
|
def get_annulation_url(self):
|
|
return (
|
|
self.annulation_url
|
|
or self.lieu.annulation_url
|
|
or self.lieu.collectivite.annulation_url
|
|
or self.get_gestion_url()
|
|
)
|
|
|
|
class Meta:
|
|
verbose_name = 'rendez-vous'
|
|
verbose_name_plural = 'rendez-vous'
|
|
db_table = 'ants_hub_rendez_vous'
|
|
|
|
|
|
class Config(models.Model):
|
|
REQUEST_FROM_ANTS_AUTH_TOKEN = 'REQUEST_FROM_ANTS_AUTH_TOKEN'
|
|
REQUEST_TO_ANTS_AUTH_TOKEN = 'REQUEST_TO_ANTS_AUTH_TOKEN'
|
|
REQUEST_TO_ANTS_BASE_URL = 'REQUEST_TO_ANTS_BASE_URL'
|
|
|
|
KEYS = [
|
|
(REQUEST_FROM_ANTS_AUTH_TOKEN, 'Token d\'authentification pour les appels en provenance de l\'ANTS'),
|
|
(REQUEST_TO_ANTS_AUTH_TOKEN, 'Token d\'authentification pour les appels en direction de l\'ANTS'),
|
|
(REQUEST_TO_ANTS_BASE_URL, 'URL de base des web-services de l\'API Rendez-vous de l\'ANTS'),
|
|
]
|
|
|
|
key = models.CharField(verbose_name='Clé', choices=KEYS, max_length=64, primary_key=True)
|
|
value = models.TextField(verbose_name='Valeur', blank=True)
|
|
|
|
def __str__(self):
|
|
return self.get_key_display()
|
|
|
|
@classmethod
|
|
def get(cls, key, default=None):
|
|
try:
|
|
return cls.objects.get(key=key).value
|
|
except cls.DoesNotExist:
|
|
return default
|
|
|
|
@classmethod
|
|
def set(cls, key, value):
|
|
if not any(x == key for x, y in cls.KEYS):
|
|
raise ValueError(f'unknown key {key}')
|
|
cls.objects.update_or_create(key=key, defaults={'value': value})
|
|
|
|
class Meta:
|
|
verbose_name = 'Configuration'
|
|
verbose_name_plural = 'Configurations'
|
|
db_table = 'ants_hub_configuration'
|
|
ordering = ['key']
|