# improved mosq task
This commit is contained in:
parent
43a55c96a6
commit
626cac02e7
|
@ -20,53 +20,60 @@ def shortest_date(date, date_list):
|
||||||
return shortest_distance
|
return shortest_distance
|
||||||
|
|
||||||
|
|
||||||
@task()
|
def fill_date(ret=None, tz=None):
|
||||||
def update_daily_statistic():
|
date_list = [k for k in ret]
|
||||||
# calc_ret = {}
|
min_date = min(date_list)
|
||||||
# queryset = DeviceCount.objects.raw(
|
max_date = datetime.now(tz).date()
|
||||||
# 'select id, device_id, max(data_time) as max_date '
|
days = (max_date - min_date).days
|
||||||
# 'from device_count group by device_id, date(data_time) order by max_date')
|
for d in range(1, days + 1):
|
||||||
#
|
date_ahead = min_date + timedelta(days=d)
|
||||||
# if queryset:
|
if date_ahead not in ret:
|
||||||
# ret = []
|
distance = shortest_date(date_ahead, date_list)
|
||||||
# for q in queryset:
|
ret[date_ahead] = ret[date_ahead - timedelta(days=distance)]
|
||||||
# entry = DeviceCount.objects.filter(device_id=q.device_id, data_time=q.max_date)
|
return ret
|
||||||
# entry = reduce(max_count, entry) if len(entry) > 1 else entry[0]
|
|
||||||
# ret.append((entry.data_time.date(), int(entry.count)))
|
|
||||||
# if ret:
|
def get_daily_statistic(SPLIT_BY_DEVICE=False):
|
||||||
# for item in ret:
|
|
||||||
# key, value = item[0], item[1]
|
|
||||||
# try:
|
|
||||||
# calc_ret[key] += value
|
|
||||||
# except KeyError:
|
|
||||||
# calc_ret[key] = value
|
|
||||||
queryset = DeviceCount.objects.raw('select id, date(data_time) as date from device_count group by date')
|
queryset = DeviceCount.objects.raw('select id, date(data_time) as date from device_count group by date')
|
||||||
if queryset:
|
if queryset:
|
||||||
|
calc_result = {}
|
||||||
tz = pytz.timezone("UTC")
|
tz = pytz.timezone("UTC")
|
||||||
|
device_list = {item.device_id for item in DeviceCount.objects.distinct()}
|
||||||
|
for item in device_list:
|
||||||
|
calc_result[item] = []
|
||||||
ret = {}
|
ret = {}
|
||||||
for q in queryset:
|
for q in queryset:
|
||||||
|
# 按照 device_id 分组查询最新时间 max_time,然后通过id + max_time 获取实例,用来得到count
|
||||||
midnight = tz.localize(datetime.combine(q.date, time(23, 59)), is_dst=None)
|
midnight = tz.localize(datetime.combine(q.date, time(23, 59)), is_dst=None)
|
||||||
daily_queryset = [DeviceCount.objects.filter(device_id=x['device_id'], data_time=x['max_time'])
|
daily_queryset = [DeviceCount.objects.filter(device_id=x['device_id'], data_time=x['max_time'])
|
||||||
for x in DeviceCount.objects.filter(data_time__lte=midnight).values(
|
for x in DeviceCount.objects.filter(data_time__lte=midnight).values(
|
||||||
'device_id').annotate(max_time=Max('data_time'))]
|
'device_id').annotate(max_time=Max('data_time'))]
|
||||||
if daily_queryset:
|
if daily_queryset:
|
||||||
daily_queryset = [reduce(max_count, entry) if len(entry) > 1 else entry[0] for entry in daily_queryset]
|
daily_queryset = [reduce(max_count, entry) if len(entry) > 1 else entry[0] for entry in daily_queryset]
|
||||||
calc_result = sum(map(lambda x: int(x.count), daily_queryset))
|
if SPLIT_BY_DEVICE:
|
||||||
|
for device_id in device_list:
|
||||||
|
calc_result[device_id] += [entry for entry in daily_queryset if entry.device_id == device_id]
|
||||||
|
else:
|
||||||
|
calc_result = sum(map(lambda x: int(x.count), daily_queryset))
|
||||||
date = q.date
|
date = q.date
|
||||||
ret[date] = calc_result
|
ret[date] = calc_result
|
||||||
date_list = [k for k in ret]
|
|
||||||
min_date = min(date_list)
|
|
||||||
# max_date = max(date_list)
|
|
||||||
max_date = datetime.now(tz).date()
|
|
||||||
days = (max_date - min_date).days
|
|
||||||
for d in range(1, days+1):
|
|
||||||
date_ahead = min_date + timedelta(days=d)
|
|
||||||
if date_ahead not in ret:
|
|
||||||
distance = shortest_date(date_ahead, date_list)
|
|
||||||
ret[date_ahead] = ret[date_ahead - timedelta(days=distance)]
|
|
||||||
|
|
||||||
# Update statistic to MosqPostStatistic
|
return ret
|
||||||
_date_list = [k for k in ret]
|
return None
|
||||||
|
|
||||||
|
@task()
|
||||||
|
def update_daily_statistic():
|
||||||
|
full_ret = None
|
||||||
|
# 计算每天聚合值
|
||||||
|
daily_ret = get_daily_statistic()
|
||||||
|
|
||||||
|
# 计算没有记录的日期,值
|
||||||
|
if daily_ret:
|
||||||
|
full_ret = fill_date(daily_ret, tz=pytz.timezone("UTC"))
|
||||||
|
|
||||||
|
# 写入数据库
|
||||||
|
if full_ret:
|
||||||
|
_date_list = [k for k in full_ret]
|
||||||
_date_list.sort()
|
_date_list.sort()
|
||||||
is_first = True
|
is_first = True
|
||||||
for d in _date_list:
|
for d in _date_list:
|
||||||
|
@ -76,17 +83,17 @@ def update_daily_statistic():
|
||||||
obj = None
|
obj = None
|
||||||
if is_first:
|
if is_first:
|
||||||
if obj:
|
if obj:
|
||||||
obj.total, obj.increment = ret[d], ret[d]
|
obj.total, obj.increment = full_ret[d], full_ret[d]
|
||||||
obj.save()
|
obj.save()
|
||||||
else:
|
else:
|
||||||
MosqPostStatistic.objects.create(date=d, total=ret[d], increment=ret[d])
|
MosqPostStatistic.objects.create(date=d, total=full_ret[d], increment=full_ret[d])
|
||||||
else:
|
else:
|
||||||
increment = ret[d] - ret[d-timedelta(days=1)]
|
increment = full_ret[d] - full_ret[d-timedelta(days=1)]
|
||||||
if obj:
|
if obj:
|
||||||
obj.total, obj.increment = ret[d], increment
|
obj.total, obj.increment = full_ret[d], increment
|
||||||
obj.save()
|
obj.save()
|
||||||
else:
|
else:
|
||||||
MosqPostStatistic.objects.create(date=d, total=ret[d], increment=increment)
|
MosqPostStatistic.objects.create(date=d, total=full_ret[d], increment=increment)
|
||||||
is_first = False
|
is_first = False
|
||||||
return _date_list
|
return _date_list
|
||||||
|
|
||||||
|
|
Loading…
Reference in New Issue