kpsul/kfet/statistic.py

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

from datetime import date, datetime, time, timedelta

from dateutil.relativedelta import relativedelta
from dateutil.parser import parse as dateutil_parse

from django.utils import timezone
from django.db.models import Sum

KFET_WAKES_UP_AT = time(7, 0)


def kfet_day(year, month, day, start_at=KFET_WAKES_UP_AT):
    """datetime wrapper with time offset."""
    return datetime.combine(date(year, month, day), start_at)


def to_kfet_day(dt, start_at=KFET_WAKES_UP_AT):
    kfet_dt = kfet_day(year=dt.year, month=dt.month, day=dt.day)
    if dt.time() < start_at:
        kfet_dt -= timedelta(days=1)
    return kfet_dt


class Scale(object):
    name = None
    step = None

    def __init__(self, n_steps=0, begin=None, end=None,
                 last=False, std_chunk=True):
        self.std_chunk = std_chunk
        if last:
            end = timezone.now()

        if begin is not None and n_steps != 0:
            self.begin = self.get_from(begin)
            self.end = self.do_step(self.begin, n_steps=n_steps)
        elif end is not None and n_steps != 0:
            self.end = self.get_from(end)
            self.begin = self.do_step(self.end, n_steps=-n_steps)
        elif begin is not None and end is not None:
            self.begin = self.get_from(begin)
            self.end = self.get_from(end)
        else:
            raise Exception('Two of these args must be specified: '
                            'n_steps, begin, end; '
                            'or use last and n_steps')

        self.datetimes = self.get_datetimes()

    @staticmethod
    def by_name(name):
        for cls in Scale.__subclasses__():
            if cls.name == name:
                return cls
        return None

    def get_from(self, dt):
        return self.std_chunk and self.get_chunk_start(dt) or dt

    def __getitem__(self, i):
        return self.datetimes[i], self.datetimes[i+1]

    def __len__(self):
        return len(self.datetimes) - 1

    def do_step(self, dt, n_steps=1):
        return dt + self.step * n_steps

    def get_datetimes(self):
        datetimes = [self.begin]
        tmp = self.begin
        while tmp <= self.end:
            tmp = self.do_step(tmp)
            datetimes.append(tmp)
        return datetimes

    def get_labels(self, label_fmt=None):
        if label_fmt is None:
            label_fmt = self.label_fmt
        return [begin.strftime(label_fmt) for begin, end in self]


class DayScale(Scale):
    name = 'day'
    step = timedelta(days=1)
    label_fmt = '%A'

    @classmethod
    def get_chunk_start(cls, dt):
        return to_kfet_day(dt)


class WeekScale(Scale):
    name = 'week'
    step = timedelta(days=7)
    label_fmt = 'Semaine %W'

    @classmethod
    def get_chunk_start(cls, dt):
        dt_kfet = to_kfet_day(dt)
        offset = timedelta(days=dt_kfet.weekday())
        return dt_kfet - offset


class MonthScale(Scale):
    name = 'month'
    step = relativedelta(months=1)
    label_fmt = '%B'

    @classmethod
    def get_chunk_start(cls, dt):
        return to_kfet_day(dt).replace(day=1)


def stat_manifest(scales_def=None, scale_args=None, scale_prefix=None,
                  **other_url_params):
    if scale_prefix is None:
        scale_prefix = 'scale_'
    if scales_def is None:
        scales_def = []
    if scale_args is None:
        scale_args = {}
    manifest = []
    for label, cls in scales_def:
        url_params = {scale_prefix+'name': cls.name}
        url_params.update({scale_prefix+key: value
                           for key, value in scale_args.items()})
        url_params.update(other_url_params)
        manifest.append(dict(
            label=label,
            url_params=url_params,
        ))
    return manifest


def last_stats_manifest(scales_def=None, scale_args=None, scale_prefix=None,
                        **url_params):
    scales_def = [
        ('Derniers mois', MonthScale, ),
        ('Dernières semaines', WeekScale, ),
        ('Derniers jours', DayScale, ),
    ]
    if scale_args is None:
        scale_args = {}
    scale_args.update(dict(
        last=True,
        n_steps=7,
    ))
    return stat_manifest(scales_def=scales_def, scale_args=scale_args,
                         scale_prefix=scale_prefix, **url_params)


# Étant donné un queryset d'operations
# rend la somme des article_nb
def tot_ventes(queryset):
    res = queryset.aggregate(Sum('article_nb'))['article_nb__sum']
    return res and res or 0


class ScaleMixin(object):
    scale_args_prefix = 'scale_'

    def get_scale_args(self, params=None, prefix=None):
        """Retrieve scale args from params.

        Should search the same args of Scale constructor.

        Args:
            params (dict, optional): Scale args are searched in this.
                Default to GET params of request.
            prefix (str, optional): Appended at the begin of scale args names.
                Default to `self.scale_args_prefix`.

        """
        if params is None:
            params = self.request.GET
        if prefix is None:
            prefix = self.scale_args_prefix

        scale_args = {}

        name = params.get(prefix+'name', None)
        if name is not None:
            scale_args['name'] = name

        n_steps = params.get(prefix+'n_steps', None)
        if n_steps is not None:
            scale_args['n_steps'] = int(n_steps)

        begin = params.get(prefix+'begin', None)
        if begin is not None:
            scale_args['begin'] = dateutil_parse(begin)

        end = params.get(prefix+'send', None)
        if end is not None:
            scale_args['end'] = dateutil_parse(end)

        last = params.get(prefix+'last', None)
        if last is not None:
            scale_args['last'] = (
                last in ['true', 'True', '1'] and True or False)

        return scale_args

    def get_context_data(self, *args, **kwargs):
        context = super().get_context_data(*args, **kwargs)

        scale_args = self.get_scale_args()
        scale_name = scale_args.pop('name', None)
        scale_cls = Scale.by_name(scale_name)

        if scale_cls is None:
            scale = self.get_default_scale()
        else:
            scale = scale_cls(**scale_args)

        self.scale = scale
        context['labels'] = scale.get_labels()
        return context

    def get_default_scale(self):
        return DayScale(n_steps=7, last=True)

    def chunkify_qs(self, qs, scale, field=None):
        if field is None:
            field = 'at'
        begin_f = '{}__gte'.format(field)
        end_f = '{}__lte'.format(field)
        return [
            qs.filter(**{begin_f: begin, end_f: end})
            for begin, end in scale
        ]