Produits Entr'ouvert » Chrono

# chrono - agendas system

# Copyright (C) 2021  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/>.

from django.core.management.base import BaseCommand

from django.utils import timezone

from chrono.agendas.models import Agenda

class Command(BaseCommand):

    help = 'Update event recurrences to reflect exceptions'

    def handle(self, **options):

        agendas = Agenda.objects.filter(kind='events', event__recurrence_rule__isnull=False).distinct()

        for agenda in agendas:

            agenda.update_event_recurrences()

# -*- coding: utf-8 -*-

# Generated by Django 1.11.18 on 2021-02-18 14:41

from __future__ import unicode_literals

from django.db import migrations, models

import django.db.models.deletion

class Migration(migrations.Migration):

    dependencies = [

        ('agendas', '0077_auto_20210218_1533'),

    ]

    operations = [

        migrations.CreateModel(

            name='RecurrenceExceptionsReport',

            fields=[

                (

                    'id',

                    models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID'),

                ),

                (

                    'agenda',

                    models.OneToOneField(

                        on_delete=django.db.models.deletion.CASCADE,

                        related_name='recurrence_exceptions_report',

                        to='agendas.Agenda',

                    ),

                ),

                ('events', models.ManyToManyField(to='agendas.Event')),

            ],

        ),

    ]

from django.db.models import Count, Q, Case, When, Max

    @transaction.atomic

    def update_event_recurrences(self):

        recurring_events = self.event_set.filter(recurrence_rule__isnull=False)

        recurrences = self.event_set.filter(primary_event__isnull=False)

        # remove recurrences

        datetimes = []

        min_start = localtime(now())

        max_start = recurrences.aggregate(dt=Max('start_datetime'))['dt']

        if not max_start:

            return

        exceptions = self.get_recurrence_exceptions(min_start, max_start)

        for event in recurring_events:

            events = event.get_recurrences(min_start, max_start, exceptions=exceptions)

            datetimes.extend([event.start_datetime for event in events])

        events = recurrences.filter(start_datetime__gt=min_start).exclude(start_datetime__in=datetimes)

        events.filter(Q(booking__isnull=True) | Q(booking__cancellation_datetime__isnull=False)).delete()

        # report events that weren't deleted because they have bookings

        report, _ = RecurrenceExceptionsReport.objects.get_or_create(agenda=self)

        report.events.set(events)

        # add recurrences

        excluded_datetimes = [make_naive(dt) for dt in recurrences.values_list('start_datetime', flat=True)]

        Event.create_events_recurrences(

            recurring_events.filter(recurrence_end_date__isnull=False), excluded_datetimes

        )

        Event.create_events_recurrences([self], excluded_datetimes)

    @classmethod

    def create_events_recurrences(cls, events, excluded_datetimes=None):

        recurrences = []

        for event in events:

            max_datetime = datetime.datetime.combine(event.recurrence_end_date, datetime.time(0, 0))

            recurrences.extend(

                event.get_recurrences(localtime(now()), make_aware(max_datetime), excluded_datetimes)

            )

class RecurrenceExceptionsReport(models.Model):

    agenda = models.OneToOneField(

        Agenda, related_name='recurrence_exceptions_report', on_delete=models.CASCADE

    )

    events = models.ManyToManyField(Event)

{% if object.recurrence_exceptions_report.events.exists %}

<div class="warningnotice">

<p>{% trans "The following events exist despite exceptions because they have active bookings:" %}</p>

<ul>

{% for event in object.recurrence_exceptions_report.events.all %}

<li><a href="{{ event.get_absolute_view_url }}">{{ event }}{% if event.label %} - {{ event.start_datetime|date:"DATETIME_FORMAT" }}{% endif %}</a></li>

{% endfor %}

</ul>

<p>{% trans "You can cancel them manually for this warning to go away, or wait until they are passed." %}</p>

</div>

{% endif %}

cron = -5 -1 -1 -1 -1 /usr/bin/chrono-manage tenant_command update_event_recurrences --all-tenants -v0

def test_recurring_events_exceptions_update_recurrences(freezer):

    freezer.move_to('2021-05-01 12:00')

    agenda = Agenda.objects.create(label='Agenda', kind='events')

    desk = Desk.objects.get(slug='_exceptions_holder', agenda=agenda)

    daily_event = Event.objects.create(

        agenda=agenda,

        start_datetime=now(),

        repeat='daily',

        places=5,

        recurrence_end_date=datetime.date(year=2021, month=5, day=8),

    )

    weekly_event = Event.objects.create(

        agenda=agenda,

        start_datetime=now(),

        repeat='weekly',

        places=5,

        recurrence_end_date=datetime.date(year=2021, month=6, day=1),

    )

    Event.create_events_recurrences([daily_event, weekly_event])

    daily_event_no_end_date = Event.objects.create(

        agenda=agenda,

        start_datetime=now() + datetime.timedelta(hours=2),

        repeat='daily',

        places=5,

    )

    daily_event_no_end_date.refresh_from_db()

    # create one recurrence on 07/05

    daily_event_no_end_date.get_or_create_event_recurrence(now() + datetime.timedelta(days=6, hours=2))

    assert Event.objects.filter(primary_event=daily_event).count() == 7

    assert Event.objects.filter(primary_event=weekly_event).count() == 5

    assert Event.objects.filter(primary_event=daily_event_no_end_date).count() == 1

    time_period_exception = TimePeriodException.objects.create(

        desk=desk,

        start_datetime=datetime.date(year=2021, month=5, day=5),

        end_datetime=datetime.date(year=2021, month=5, day=10),

    )

    agenda.update_event_recurrences()

    assert Event.objects.filter(primary_event=daily_event).count() == 4

    assert Event.objects.filter(primary_event=weekly_event).count() == 4

    assert Event.objects.filter(primary_event=daily_event_no_end_date).count() == 0

    time_period_exception.delete()

    agenda.update_event_recurrences()

    assert Event.objects.filter(primary_event=daily_event).count() == 7

    assert Event.objects.filter(primary_event=weekly_event).count() == 5

    assert Event.objects.filter(primary_event=daily_event_no_end_date).count() == 0

    event = daily_event_no_end_date.get_or_create_event_recurrence(

        now() + datetime.timedelta(days=6, hours=2)

    )

    booking = Booking.objects.create(event=event)

    time_period_exception.save()

    agenda.update_event_recurrences()

    assert Booking.objects.count() == 1

    assert Event.objects.filter(primary_event=daily_event_no_end_date).count() == 1

    assert agenda.recurrence_exceptions_report.events.get() == event

def test_recurring_events_exceptions_report(settings, app, admin_user, freezer):

    freezer.move_to('2021-07-01 12:10')

    agenda = Agenda.objects.create(label='Foo bar', kind='events')

    event = Event.objects.create(

        start_datetime=now(),

        places=10,

        repeat='daily',

        recurrence_end_date=now() + datetime.timedelta(days=30),

        agenda=agenda,

    )

    event.create_all_recurrences()

    app = login(app)

    resp = app.get('/manage/agendas/%s/%s/%s/' % (agenda.pk, 2021, 7))

    assert len(resp.pyquery.find('.event-info')) == 30

    time_period_exception = TimePeriodException.objects.create(

        desk=agenda.desk_set.get(),

        start_datetime=datetime.date(year=2021, month=7, day=5),

        end_datetime=datetime.date(year=2021, month=7, day=10),

    )

    call_command('update_event_recurrences')

    resp = app.get('/manage/agendas/%s/%s/%s/' % (agenda.pk, 2021, 7))

    assert len(resp.pyquery.find('.event-info')) == 25

    resp = app.get('/manage/agendas/%s/settings' % agenda.id)

    assert not 'warningnotice' in resp.text

    event = Event.objects.get(start_datetime__day=11)

    booking = Booking.objects.create(event=event)

    time_period_exception.end_datetime = datetime.date(year=2021, month=7, day=12)

    time_period_exception.save()

    call_command('update_event_recurrences')

    resp = app.get('/manage/agendas/%s/%s/%s/' % (agenda.pk, 2021, 7))

    assert len(resp.pyquery.find('.event-info')) == 24

    resp = app.get('/manage/agendas/%s/settings' % agenda.id)

    assert 'warningnotice' in resp.text

    assert 'July 11, 2021, 2:10 p.m.' in resp.text

    booking.cancel()

    call_command('update_event_recurrences')

    resp = app.get('/manage/agendas/%s/%s/%s/' % (agenda.pk, 2021, 7))

    assert len(resp.pyquery.find('.event-info')) == 23

    resp = app.get('/manage/agendas/%s/settings' % agenda.id)

    assert not 'warningnotice' in resp.text