356 lines
13 KiB
Python
356 lines
13 KiB
Python
# chrono - agendas system
|
|
# Copyright (C) 2023 Entr'ouvert
|
|
#
|
|
# This program is free software: you can redistribute it and/or modify it
|
|
# under the terms of the GNU Affero General Public License as published
|
|
# by the Free Software Foundation, either version 3 of the License, or
|
|
# (at your option) any later version.
|
|
#
|
|
# This program is distributed in the hope that it will be useful,
|
|
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
# GNU Affero General Public License for more details.
|
|
#
|
|
# You should have received a copy of the GNU Affero General Public License
|
|
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
|
|
|
import collections
|
|
import datetime
|
|
|
|
from django import forms
|
|
from django.conf import settings
|
|
from django.db import models, transaction
|
|
from django.urls import reverse
|
|
from django.utils.translation import gettext_lazy as _
|
|
|
|
from chrono.agendas.models import Agenda, Booking
|
|
from chrono.utils.timezone import localtime, now
|
|
|
|
from .hub import push_rendez_vous_disponibles
|
|
|
|
# fields without max_length problems
|
|
|
|
|
|
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 CityManager(models.Manager):
|
|
def get_by_natural_key(self, name):
|
|
return self.get(name=name)
|
|
|
|
|
|
def get_portal_url():
|
|
template_vars = getattr(settings, 'TEMPLATE_VARS', {})
|
|
return template_vars.get('portal_url', '')
|
|
|
|
|
|
class City(models.Model):
|
|
name = CharField(verbose_name=_('Name'), unique=True)
|
|
url = URLField(verbose_name=_('Portal URL'), default=get_portal_url, blank=True)
|
|
logo_url = URLField(verbose_name=_('Logo URL'), blank=True)
|
|
meeting_url = URLField(
|
|
verbose_name=_('Booking URL'), help_text=_('URL of the web form to make a booking.'), blank=True
|
|
)
|
|
management_url = URLField(
|
|
verbose_name=_('Booking management URL'),
|
|
help_text=_('Generic URL to find and manage an existing booking.'),
|
|
blank=True,
|
|
)
|
|
cancel_url = URLField(
|
|
verbose_name=_('Booking cancellation URL'),
|
|
help_text=_('Generic URL to find and cancel an existing booking.'),
|
|
blank=True,
|
|
)
|
|
created = models.DateTimeField(verbose_name=_('Created'), auto_now_add=True)
|
|
last_update = models.DateTimeField(verbose_name=_('Last update'), auto_now=True)
|
|
full_sync = models.BooleanField(
|
|
verbose_name=_('Full synchronization next time'), default=False, editable=False
|
|
)
|
|
last_sync = models.DateTimeField(verbose_name=_('Last synchronization'), null=True, editable=False)
|
|
|
|
objects = CityManager()
|
|
|
|
def __str__(self):
|
|
return f'{self.name}'
|
|
|
|
def natural_key(self):
|
|
return (self.name,)
|
|
|
|
def get_absolute_url(self):
|
|
return reverse('chrono-manager-ants-hub')
|
|
|
|
def details(self):
|
|
details = []
|
|
for key in ['url', 'logo_url', 'management_url', 'cancel_url', 'full_sync']:
|
|
value = getattr(self, key, None)
|
|
if value:
|
|
verbose_name = type(self)._meta.get_field(key).verbose_name
|
|
details.append((verbose_name, value))
|
|
return details
|
|
|
|
@classmethod
|
|
@transaction.atomic(savepoint=False)
|
|
def push(cls):
|
|
reference = now()
|
|
# prevent concurrent pushes with locks
|
|
cities = list(cls.objects.select_for_update())
|
|
push_rendez_vous_disponibles(
|
|
{
|
|
'collectivites': [city.export_to_push() for city in cities],
|
|
}
|
|
)
|
|
City.objects.update(last_sync=reference)
|
|
Place.objects.update(last_sync=reference)
|
|
|
|
def export_to_push(self):
|
|
payload = {
|
|
'full': True,
|
|
'id': str(self.pk),
|
|
'nom': self.name,
|
|
'url': self.url,
|
|
'logo_url': self.logo_url,
|
|
'rdv_url': self.meeting_url,
|
|
'gestion_url': self.management_url,
|
|
'annulation_url': self.cancel_url,
|
|
'lieux': [place.export_to_push() for place in self.places.all()],
|
|
}
|
|
return payload
|
|
|
|
class Meta:
|
|
verbose_name = _('City')
|
|
verbose_name_plural = _('Cities')
|
|
|
|
|
|
class PlaceManager(models.Manager):
|
|
def get_by_natural_key(self, name, *args):
|
|
return self.get(name=name, city=City.objects.get_by_natural_key(*args))
|
|
|
|
|
|
class Place(models.Model):
|
|
city = models.ForeignKey(
|
|
verbose_name=_('City'), to=City, related_name='places', on_delete=models.CASCADE, editable=False
|
|
)
|
|
name = CharField(verbose_name=_('Name'))
|
|
address = CharField(verbose_name=_('Address'))
|
|
zipcode = CharField(verbose_name=_('Code postal'))
|
|
city_name = CharField(verbose_name=_('City name'))
|
|
longitude = models.FloatField(verbose_name=_('Longitude'), default=2.476)
|
|
latitude = models.FloatField(verbose_name=_('Latitude'), default=46.596)
|
|
url = URLField(verbose_name=_('Portal URL'), blank=True)
|
|
logo_url = URLField(verbose_name=_('Logo URL'), blank=True)
|
|
meeting_url = URLField(
|
|
verbose_name=_('Booking URL'), help_text=_('URL of the web form to make a booking.'), blank=True
|
|
)
|
|
management_url = URLField(
|
|
verbose_name=_('Booking management URL'),
|
|
help_text=_('Generic URL to find and manage an existing booking.'),
|
|
blank=True,
|
|
)
|
|
cancel_url = URLField(
|
|
verbose_name=_('Booking cancellation URL'),
|
|
help_text=_('Generic URL to find and cancel an existing booking.'),
|
|
blank=True,
|
|
)
|
|
created = models.DateTimeField(verbose_name=_('Created'), auto_now_add=True)
|
|
last_update = models.DateTimeField(verbose_name=_('Last update'), auto_now=True)
|
|
full_sync = models.BooleanField(verbose_name=_('Full synchronization'), default=False)
|
|
last_sync = models.DateTimeField(verbose_name=_('Last synchronization'), null=True, editable=False)
|
|
|
|
objects = PlaceManager()
|
|
|
|
def __str__(self):
|
|
return f'{self.name}'
|
|
|
|
def natural_key(self):
|
|
return (self.name,) + self.city.natural_key()
|
|
|
|
natural_key.dependencies = ['ants_hub.city']
|
|
|
|
def get_absolute_url(self):
|
|
return reverse('chrono-manager-ants-hub-place', kwargs={'city_pk': self.city_id, 'pk': self.pk})
|
|
|
|
def url_details(self):
|
|
details = []
|
|
for key in ['url', 'logo_url', 'management_url', 'cancel_url', 'full_sync']:
|
|
value = getattr(self, key, None)
|
|
if value:
|
|
verbose_name = type(self)._meta.get_field(key).verbose_name
|
|
details.append((verbose_name, value))
|
|
return details
|
|
|
|
def export_to_push(self):
|
|
payload = {
|
|
'full': True,
|
|
'id': str(self.pk),
|
|
'nom': self.name,
|
|
'numero_rue': self.address,
|
|
'code_postal': self.zipcode,
|
|
'ville': self.city_name,
|
|
'longitude': self.longitude,
|
|
'latitude': self.latitude,
|
|
'url': self.url,
|
|
'rdv_url': self.meeting_url,
|
|
'gestion_url': self.management_url,
|
|
'annulation_url': self.cancel_url,
|
|
'plages': list(self.iter_open_dates()),
|
|
'rdvs': list(self.iter_predemandes()),
|
|
}
|
|
return payload
|
|
|
|
def iter_open_dates(self):
|
|
for place_agenda in self.agendas.all():
|
|
yield from place_agenda.iter_open_dates()
|
|
|
|
def iter_predemandes(self):
|
|
agendas = Agenda.objects.filter(ants_place__place=self)
|
|
agendas |= Agenda.objects.filter(virtual_agendas__ants_place__place=self)
|
|
agendas = set(agendas)
|
|
bookings = (
|
|
Booking.objects.filter(
|
|
event__desk__agenda__in=agendas,
|
|
event__start_datetime__gt=now(),
|
|
extra_data__ants_identifiant_predemande__isnull=False,
|
|
)
|
|
.values_list(
|
|
'extra_data__ants_identifiant_predemande', 'event__start_datetime', 'cancellation_datetime'
|
|
)
|
|
.order_by('event__state_datetime')
|
|
)
|
|
for identifiant_predemande, start_datetime, cancellation_datetime in bookings:
|
|
if not isinstance(identifiant_predemande, str):
|
|
continue
|
|
rdv = {
|
|
'id': identifiant_predemande,
|
|
'date': start_datetime.isoformat(),
|
|
}
|
|
if cancellation_datetime is not None:
|
|
rdv['annule'] = True
|
|
yield rdv
|
|
|
|
class Meta:
|
|
verbose_name = _('place')
|
|
verbose_name_plural = _('places')
|
|
unique_together = [
|
|
('city', 'name'),
|
|
]
|
|
|
|
|
|
class ANTSMeetingType(models.IntegerChoices):
|
|
CNI = 1, _('CNI')
|
|
PASSPORT = 2, _('Passport')
|
|
CNI_PASSPORT = 3, _('CNI and passport')
|
|
|
|
@property
|
|
def ants_name(self):
|
|
return super().name.replace('_', '-')
|
|
|
|
|
|
class ANTSPersonsNumber(models.IntegerChoices):
|
|
ONE = 1, _('1 person')
|
|
TWO = 2, _('2 persons')
|
|
THREE = 3, _('3 persons')
|
|
FOUR = 4, _('4 persons')
|
|
|
|
|
|
class PlaceAgenda(models.Model):
|
|
place = models.ForeignKey(
|
|
verbose_name=_('Place'),
|
|
to=Place,
|
|
on_delete=models.CASCADE,
|
|
related_name='agendas',
|
|
)
|
|
agenda = models.ForeignKey(
|
|
verbose_name=_('Agenda'),
|
|
to='agendas.Agenda',
|
|
on_delete=models.CASCADE,
|
|
related_name='+',
|
|
related_query_name='ants_place',
|
|
)
|
|
setting = models.JSONField(verbose_name=_('Setting'), default=dict, blank=True)
|
|
|
|
def set_meeting_type_setting(self, meeting_type, key, value):
|
|
assert key in ['ants_meeting_type', 'ants_persons_number']
|
|
meeting_types = self.setting.setdefault('meeting-types', {})
|
|
value = map(int, value)
|
|
if key == 'ants_meeting_type':
|
|
value = list(map(ANTSMeetingType, value))
|
|
else:
|
|
value = list(map(ANTSPersonsNumber, value))
|
|
meeting_types.setdefault(str(meeting_type.slug), {})[key] = value
|
|
|
|
def get_meeting_type_setting(self, meeting_type, key):
|
|
assert key in ['ants_meeting_type', 'ants_persons_number']
|
|
meeting_types = self.setting.setdefault('meeting-types', {})
|
|
value = meeting_types.get(str(meeting_type.slug), {}).get(key, [])
|
|
value = map(int, value)
|
|
if key == 'ants_meeting_type':
|
|
value = list(map(ANTSMeetingType, value))
|
|
else:
|
|
value = list(map(ANTSPersonsNumber, value))
|
|
return value
|
|
|
|
def iter_ants_meeting_types_and_persons(self):
|
|
d = collections.defaultdict(set)
|
|
for meeting_type in self.agenda.iter_meetingtypes():
|
|
for ants_meeting_type in self.get_meeting_type_setting(meeting_type, 'ants_meeting_type'):
|
|
for ants_persons_number in self.get_meeting_type_setting(meeting_type, 'ants_persons_number'):
|
|
meeting_type.id = meeting_type.slug
|
|
d[(meeting_type, ants_persons_number)].add(ants_meeting_type)
|
|
for (meeting_type, ants_persons_number), ants_meeting_types in d.items():
|
|
yield meeting_type, ants_persons_number, ants_meeting_types
|
|
|
|
@property
|
|
def ants_properties(self):
|
|
rdv = set()
|
|
persons = set()
|
|
for meeting_type in self.agenda.meetingtype_set.all():
|
|
rdv.update(self.get_meeting_type_setting(meeting_type, 'ants_meeting_type'))
|
|
persons.update(self.get_meeting_type_setting(meeting_type, 'ants_persons_number'))
|
|
rdv = sorted(list(rdv))
|
|
persons = sorted(list(persons))
|
|
return [x.label for x in rdv] + [x.label for x in persons]
|
|
|
|
def __str__(self):
|
|
return str(self.agenda)
|
|
|
|
def get_absolute_url(self):
|
|
return self.place.get_absolute_url()
|
|
|
|
def iter_open_dates(self):
|
|
settings = list(self.iter_ants_meeting_types_and_persons())
|
|
if not settings:
|
|
return
|
|
intervals = self.agenda.get_free_time(end_datetime=now() + datetime.timedelta(days=6 * 30))
|
|
for start, end in intervals:
|
|
for meeting_type, ants_persons_number, ants_meeting_types in settings:
|
|
duration = datetime.timedelta(minutes=meeting_type.duration)
|
|
if end - start < duration:
|
|
continue
|
|
yield {
|
|
'date': localtime(start).date().isoformat(),
|
|
# use naive local time representation
|
|
'heure_debut': localtime(start).time().replace(tzinfo=None).isoformat(),
|
|
'heure_fin': localtime(end).time().replace(tzinfo=None).isoformat(),
|
|
'duree': meeting_type.duration,
|
|
'personnes': int(ants_persons_number),
|
|
'types_rdv': [x.ants_name for x in ants_meeting_types],
|
|
}
|
|
|
|
class Meta:
|
|
unique_together = [
|
|
('place', 'agenda'),
|
|
]
|