K-Fêt - Statistics New base class - StatScale - create scale, given chunk size (timedelta), start and end times - get labels of - get start and end datetimes of chunks DayStatScale: Scale whose chunks interval is 1 day WeekStatScale: same with 1 week MonthStatScale: same with 1 month AccountStatOperationList: manifest of operations stats of an account - renamed from AccountStatLastAll - updated according to SingleResumeStat AccountStatOperation: - renamed from AccountStatLast - remove scale logic with use of StatScale objects - used scale is given by `scale` and `scale_args` GET params - add filter on operations types with `types` GET param AccountStatLast(Day,Week,Month) are deleted ("merged" in AccountStatOperation)
150 lines
4.2 KiB
Python
150 lines
4.2 KiB
Python
# -*- coding: utf-8 -*-
|
|
|
|
from dateutil.relativedelta import relativedelta
|
|
|
|
from django.utils import timezone
|
|
from django.db.models import Sum
|
|
|
|
KFET_WAKES_UP_AT = 7
|
|
|
|
|
|
def kfet_day(year, month, day, start_at=KFET_WAKES_UP_AT):
|
|
return timezone.datetime(year, month, day, hour=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.hour < start_at:
|
|
kfet_dt -= timezone.timedelta(days=1)
|
|
return kfet_dt
|
|
|
|
|
|
class StatScale(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 StatScale.__subclasses__():
|
|
if cls.name == name:
|
|
return cls
|
|
raise Exception('scale %s not found' % name)
|
|
|
|
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]
|
|
|
|
@classmethod
|
|
def get_chunk_start(cls, dt):
|
|
dt_kfet = to_kfet_day(dt)
|
|
start = dt_kfet - cls.offset_to_chunk_start(dt_kfet)
|
|
return start
|
|
|
|
|
|
class DayStatScale(StatScale):
|
|
name = 'day'
|
|
step = timezone.timedelta(days=1)
|
|
label_fmt = '%A'
|
|
|
|
@classmethod
|
|
def get_chunk_start(cls, dt):
|
|
return to_kfet_day(dt)
|
|
|
|
|
|
class WeekStatScale(StatScale):
|
|
name = 'week'
|
|
step = timezone.timedelta(days=7)
|
|
label_fmt = 'Semaine %W'
|
|
|
|
@classmethod
|
|
def offset_to_chunk_start(cls, dt):
|
|
return timezone.timedelta(days=dt.weekday())
|
|
|
|
|
|
class MonthStatScale(StatScale):
|
|
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 this_first_month_day():
|
|
now = timezone.now()
|
|
first_day = timezone.datetime(year=now.year,
|
|
month=now.month,
|
|
day=1,
|
|
hour=KFET_WAKES_UP_AT)
|
|
return first_day
|
|
|
|
|
|
def this_monday_morning():
|
|
now = timezone.now()
|
|
monday = now - timezone.timedelta(days=now.isoweekday()-1)
|
|
monday_morning = timezone.datetime(year=monday.year,
|
|
month=monday.month,
|
|
day=monday.day,
|
|
hour=KFET_WAKES_UP_AT)
|
|
return monday_morning
|
|
|
|
|
|
def this_morning():
|
|
now = timezone.now()
|
|
morning = timezone.datetime(year=now.year,
|
|
month=now.month,
|
|
day=now.day,
|
|
hour=KFET_WAKES_UP_AT)
|
|
return morning
|
|
|
|
|
|
# É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
|