Produits Entr'ouvert » Chrono

    url(r'^statistics/$', views.statistics_list, name='api-statistics-list'),

    url(r'^statistics/bookings/$', views.bookings_statistics, name='api-statistics-bookings'),

from django.db.models import Count, Prefetch, Q

from django.db.models.functions import TruncDay

from chrono.agendas.models import (

    Agenda,

    Booking,

    BookingColor,

    Category,

    Desk,

    Event,

    MeetingType,

    TimePeriodException,

)

from chrono.interval import IntervalSet

class StatisticsList(APIView):

    permission_classes = (permissions.IsAuthenticated,)

    def get(self, request, *args, **kwargs):

        categories = Category.objects.all()

        category_options = [{'id': '_all', 'label': _('All')}] + [

            {'id': x.slug, 'label': x.label} for x in categories

        ]

        return Response(

            {

                'data': [

                    {

                        'name': _('Bookings Count'),

                        'url': request.build_absolute_uri(reverse('api-statistics-bookings')),

                        'id': 'bookings_count',

                        'filters': [

                            {

                                'id': 'time_interval',

                                'label': _('Interval'),

                                'options': [{'id': 'day', 'label': _('Day')}],

                                'required': True,

                                'default': 'month',

                            },

                            {

                                'id': 'category',

                                'label': _('Category'),

                                'options': category_options,

                                'required': False,

                                'default': '_all',

                            },

                        ],

                    }

                ]

            }

        )

statistics_list = StatisticsList.as_view()

class StatisticsFiltersSerializer(serializers.Serializer):

    time_interval = serializers.ChoiceField(choices=('day', _('Day')), default='day')

    start = serializers.DateTimeField(required=False, input_formats=['iso-8601', '%Y-%m-%d'])

    end = serializers.DateTimeField(required=False, input_formats=['iso-8601', '%Y-%m-%d'])

    category = serializers.SlugField(required=False, allow_blank=False, max_length=256)

class BookingsStatistics(APIView):

    permission_classes = (permissions.IsAuthenticated,)

    def get(self, request, *args, **kwargs):

        serializer = StatisticsFiltersSerializer(data=request.query_params)

        if not serializer.is_valid():

            raise APIError(

                _('invalid statistics filters'),

                err_class='invalid statistics filters',

                errors=serializer.errors,

                http_status=status.HTTP_400_BAD_REQUEST,

            )

        data = serializer.validated_data

        bookings = Booking.objects

        if 'start' in data:

            bookings = bookings.filter(creation_datetime__gte=data['start'])

        if 'end' in data:

            bookings = bookings.filter(creation_datetime__lte=data['end'])

        if 'category' in data and data['category'] != '_all':

            bookings = bookings.filter(event__agenda__category__slug=data['category'])

        bookings = bookings.annotate(day=TruncDay('creation_datetime'))

        bookings = bookings.values('day', 'user_was_present').annotate(total=Count('id')).order_by('day')

        bookings_by_day = collections.OrderedDict()

        for booking in bookings:

            totals_by_presence = bookings_by_day.setdefault(booking['day'], {})

            totals_by_presence[booking['user_was_present']] = booking['total']

        bookings_by_presence = {None: [], True: [], False: []}

        for bookings in bookings_by_day.values():

            for presence, data in bookings_by_presence.items():

                data.append(bookings.get(presence))

        labels = {None: _('Unknown'), True: _('Present'), False: _('Absent')}

        series = [{'label': labels[k], 'data': data} for k, data in bookings_by_presence.items() if any(data)]

        if len(series) == 1 and series[0]['label'] == _('Unknown'):

            series[0]['label'] = _('Bookings Count')

        return Response(

            {

                'data': {

                    'x_labels': [day.strftime('%Y-%m-%d') for day in bookings_by_day],

                    'series': series,

                },

                'err': 0,

            }

        )

bookings_statistics = BookingsStatistics.as_view()

def test_statistics_list(app, user):

    category_a = Category.objects.create(label='Category A')

    category_b = Category.objects.create(label='Category B')

    # unauthorized

    app.get('/api/statistics/', status=401)

    app.authorization = ('Basic', ('john.doe', 'password'))

    resp = app.get('/api/statistics/')

    category_filter = [x for x in resp.json['data'][0]['filters'] if x['id'] == 'category'][0]

    assert len(category_filter['options']) == 3

def test_statistics_bookings(app, user, freezer):

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

    event = Event.objects.create(start_datetime=now(), places=5, agenda=agenda)

    app.authorization = ('Basic', ('john.doe', 'password'))

    resp = app.get('/api/statistics/')

    url = [x for x in resp.json['data'] if x['id'] == 'bookings_count'][0]['url']

    resp = app.get(url)

    assert len(resp.json['data']['series']) == 0

    freezer.move_to('2020-10-10')

    for _ in range(10):

        Booking.objects.create(event=event)

    freezer.move_to('2020-10-15')

    Booking.objects.create(event=event)

    resp = app.get(url + '?time_interval=day')

    assert resp.json['data'] == {

        'x_labels': ['2020-10-10', '2020-10-15'],

        'series': [{'label': 'Bookings Count', 'data': [10, 1]}],

    }

    # period filter

    resp = app.get(url + '?start=2020-10-14&end=2020-10-16')

    assert resp.json['data'] == {

        'x_labels': ['2020-10-15'],

        'series': [{'label': 'Bookings Count', 'data': [1]}],

    }

    category = Category.objects.create(label='Category A', slug='category-a')

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

    event = Event.objects.create(start_datetime=now(), places=5, agenda=agenda)

    freezer.move_to('2020-10-25')

    Booking.objects.create(event=event)

    # category filter

    resp = app.get(url + '?category=category-a')

    assert resp.json['data'] == {

        'x_labels': ['2020-10-25'],

        'series': [{'label': 'Bookings Count', 'data': [1]}],

    }

    # invalid time_interval

    resp = app.get(url + '?time_interval=month', status=400)

    assert resp.json['err'] == 1

    assert 'time_interval' in resp.json['errors']

    # absence/presence

    for i in range(10):

        Booking.objects.create(event=event, user_was_present=bool(i % 2))

    freezer.move_to('2020-11-01')

    Booking.objects.create(event=event, user_was_present=True)

    resp = app.get(url)

    assert resp.json['data'] == {

        'x_labels': ['2020-10-10', '2020-10-15', '2020-10-25', '2020-11-01'],

        'series': [

            {'label': 'Unknown', 'data': [10, 1, 1, None]},

            {'label': 'Present', 'data': [None, None, 5, 1]},

            {'label': 'Absent', 'data': [None, None, 5, None]},

        ],

    }