import pytz from functools import reduce # from collections import OrderedDict from datetime import datetime, time, timedelta from django.db.models import Max from celery import task from counter.models import DeviceCount from mosquito.models import MosqPostStatistic def max_count(x, y): if x.count > y.count: return x return y def shortest_date(date, date_list): distance_list = [(date - _date).days for _date in date_list] shortest_distance = min(filter(lambda x: x > 0, distance_list)) return shortest_distance def fill_date(ret=None, tz=None): date_list = [k for k in ret] min_date = min(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)] return ret def get_daily_statistic(enable_container=False): device_container = {} daily_ret = {} queryset = DeviceCount.objects.raw('select id, date(data_time) as date from device_count group by date') if queryset: tz = pytz.timezone("UTC") device_list = {item.device_id for item in DeviceCount.objects.distinct()} for item in device_list: device_container[item] = set() 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) 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( 'device_id').annotate(max_time=Max('data_time'))] if daily_queryset: daily_queryset = [reduce(max_count, entry) if len(entry) > 1 else entry[0] for entry in daily_queryset] if enable_container: for device_id in device_list: device_container[device_id] = device_container[device_id] | \ {(entry.count, entry.data_time) for entry in daily_queryset if entry.device_id == device_id} calc_result = sum(map(lambda x: int(x.count), daily_queryset)) date = q.date daily_ret[date] = calc_result return daily_ret, device_container def update_daily_statistic(): full_ret = None # 计算每天聚合值 daily_ret, device_container = get_daily_statistic(enable_container=True) for k in device_container: print(k, device_container[k]) # 计算没有记录的日期,值 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() is_first = True for d in _date_list: try: obj = MosqPostStatistic.objects.get(date=d) except MosqPostStatistic.DoesNotExist: obj = None if is_first: if obj: obj.total, obj.increment = full_ret[d], full_ret[d] obj.save() else: MosqPostStatistic.objects.create(date=d, total=full_ret[d], increment=full_ret[d]) else: increment = full_ret[d] - full_ret[d - timedelta(days=1)] if obj: obj.total, obj.increment = full_ret[d], increment obj.save() else: MosqPostStatistic.objects.create(date=d, total=full_ret[d], increment=increment) is_first = False return _date_list @task() def update_latest_statistic(): pass