# -*- encoding: utf-8 -*- """ @File : DeviceReportController.py @Time : 2025/4/3 16:20 @Author : stephen @Email : zhangdongming@asj6.wecom.work @Software: PyCharm """ import time from datetime import timedelta, datetime import pytz from django.db.models import Sum, Value from django.db.models.functions import Coalesce from django.http import QueryDict from django.views import View from Ansjer.config import LOGGER from Model.models import DeviceDailyReport from Object.ResponseObject import ResponseObject from Object.TokenObject import TokenObject class DeviceReportView(View): def get(self, request, *args, **kwargs): request.encoding = 'utf-8' operation = kwargs.get('operation') return self.validation(request.GET, request, operation) def post(self, request, *args, **kwargs): request.encoding = 'utf-8' operation = kwargs.get('operation') return self.validation(request.POST, request, operation) def delete(self, request, *args, **kwargs): request.encoding = 'utf-8' operation = kwargs.get('operation') delete = QueryDict(request.body) if not delete: delete = request.GET return self.validation(delete, request, operation) def put(self, request, *args, **kwargs): request.encoding = 'utf-8' operation = kwargs.get('operation') put = QueryDict(request.body) return self.validation(put, request, operation) def validation(self, request_dict, request, operation): response = ResponseObject('cn') tko = TokenObject(request.META.get('HTTP_AUTHORIZATION')) if tko.code != 0: return response.json(tko.code) response.lang = tko.lang userID = tko.userID if operation == 'getBatteryCapacityList': # 获取电池电量列表 return self.get_battery_capacity_list(userID, request, request_dict, response) elif operation == 'getWakeSleepData': return self.get_wake_sleep_data(userID, request_dict, response) return response.json(414) @classmethod def get_wake_sleep_data(cls, userID, request_dict, response): """ 获取设备睡眠唤醒统计数据 :param userID: 响应对象 :param request_dict: 参数对象 :param response: 响应对象 :return: 包含7天汇总统计和30天明细数据的JSON响应 """ try: # region 时间参数计算 uid = request_dict.get('device_id', None) if not uid: return response.json(444) channel = int(request_dict.get('channel', 1)) current_time = int(time.time()) SECONDS_PER_DAY = 24 * 60 * 60 # 单日秒数常量 # 计算时间范围（秒级时间戳） seven_days_ago = current_time - 7 * SECONDS_PER_DAY # 7天前 thirty_days_ago = current_time - 30 * SECONDS_PER_DAY # 30天前 # endregion # region 7天汇总统计 (使用Coalesce避免空值) seven_days_stats = DeviceDailyReport.objects.filter( device_id=uid, channel=channel, report_time__range=(seven_days_ago, current_time) # 时间范围查询 ).aggregate( total_human_detection=Coalesce(Sum('human_detection'), Value(0)), total_working_hours=Coalesce(Sum('working_hours'), Value(0)), total_wake_sleep=Coalesce(Sum('wake_sleep'), Value(0)) ) # endregion # region 30天明细数据（按时间倒序） thirty_days_details = DeviceDailyReport.objects.filter( device_id=uid, channel=channel, report_time__gte=thirty_days_ago, report_time__lte=current_time ).order_by('-report_time').values_list( 'human_detection', 'working_hours', 'wake_sleep', 'battery_level', 'report_time', named=True # 使用命名元组方便访问 ) # endregion # 组合返回数据结构 result_data = { **seven_days_stats, 'detail_report': list(thirty_days_details) # 转换查询集为列表 } return response.json(0, result_data) except Exception as e: error_msg = f"查询异常: {str(e)},userID: {userID}" error_line = e.__traceback__.tb_lineno LOGGER.error(f"{error_msg} 行号: {error_line}") # 返回安全空数据 return response.json(500, { 'total_human_detection': 0, 'total_working_hours': 0, 'total_wake_sleep': 0, 'detail_report': []}) @classmethod def get_battery_capacity_list(cls, userID, request, request_dict, response): try: # 参数校验 device_id = request_dict.get('device_id') if not device_id: return response.json(444, "设备ID不能为空") tz_offset = float(request_dict.get('tz', 0)) if not (-12 <= tz_offset <= 14): raise ValueError("时区偏移超出范围") tz = pytz.FixedOffset(int(tz_offset * 60)) # 计算日期范围 query_days = int(request_dict.get('days', 7)) client_now = datetime.now(tz) # 数据日期范围 = [今天 - query_days, 昨天] end_date = (client_now - timedelta(days=1)).date() # 昨天 start_date = end_date - timedelta(days=query_days - 1) # 起始日期 # 转换查询范围到UTC时间戳 start_utc = int(tz.localize( datetime.combine(start_date, datetime.min.time()) ).astimezone(pytz.utc).timestamp()) end_utc = int(tz.localize( datetime.combine(end_date + timedelta(days=1), datetime.min.time()) ).astimezone(pytz.utc).timestamp()) # 查询数据库 records = DeviceDailyReport.objects.filter( device_id=device_id, type=1, report_time__gte=start_utc, report_time__lt=end_utc ).order_by('report_time') # 构建日期-电量映射（直接使用上报日期） date_battery_map = {} for record in records: # 关键变更：直接使用上报日期作为数据日期 record_date = datetime.fromtimestamp(record.report_time, tz).date() # 只记录查询范围内的数据 if start_date <= record_date <= end_date: date_battery_map[record_date.isoformat()] = record.battery_level # 生成完整日期序列 report_data = [] for day_offset in range(query_days): current_date = start_date + timedelta(days=day_offset) report_data.append({ "index": current_date.day - 1, # 日期下标从0开始 "battery": date_battery_map.get(current_date.isoformat(), 0), "time": current_date.strftime("%Y-%m-%d") }) return response.json(0, report_data) except Exception as e: LOGGER.error(f'查询设备电量上报列表异常: {e}') return response.json(0, {})