ASJPush/Controller/ReportController.py

# @Author    : Rocky
# @File      : ReportController.py
# @Time      : 2025/5/12 10:42
import json
import time

from django.views import View

from Object.RedisObject import RedisObject
from Object.ResponseObject import ResponseObject
from Model.models import AbnormalEvent


class ReportView(View):

    def get(self, request, *args, **kwargs):
        request.encoding = 'utf-8'
        operation = kwargs.get('operation')
        return self.validation(request.GET, operation)

    def post(self, request, *args, **kwargs):
        request.encoding = 'utf-8'
        operation = kwargs.get('operation')
        return self.validation(request.POST, operation)

    def validation(self, request_dict, operation):
        response = ResponseObject()
        if operation == 'abnormalEvent':  # 上报异常事件
            return self.abnormal_event(request_dict, response)

    @staticmethod
    def abnormal_event(request_dict, response):
        uid = request_dict.get('uid')
        device_type = request_dict.get('device_type')
        version = request_dict.get('version')
        event_list = request_dict.get('event_list')

        if not all([uid, device_type, version, event_list]):
            return response.json(444)

        # 同一设备限制十分钟之内不能重复请求
        redis_obj = RedisObject()
        key = 'report_abnormal_event_{}'.format(uid)
        ttl = redis_obj.get_ttl(key)
        if ttl > 0:
            res = {
                'limit_time': ttl
            }
            return response.json(10, res)
        redis_obj.set_data(key=key, val=1, expire=600)

        try:
            device_type = int(device_type)
            now_time = int(time.time())
            event_list = json.loads(event_list)

            # 批量创建异常事件数据
            abnormal_event_list = []
            data = {
                'uid': uid,
                'device_type': device_type,
                'version': version,
                'created_time': now_time
            }
            for event in event_list:
                data['event_code'] = event['event_code']
                data['event_time'] = event['event_time']
                if event.get('content'):
                    data['content'] = event['content']
                abnormal_event = AbnormalEvent(**data)
                abnormal_event_list.append(abnormal_event)
            AbnormalEvent.objects.bulk_create(abnormal_event_list)
            return response.json(0)
        except Exception as e:
            return response.json(500, 'error_line:{}, error_msg:{}'.format(e.__traceback__.tb_lineno, repr(e)))