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 = []
            for event in event_list:
                event_code = event['event_code']
                event_time = event['event_time']
                abnormal_event = AbnormalEvent(
                    uid=uid, device_type=device_type, version=version, event_code=event_code, event_time=event_time,
                    created_time=now_time
                )
                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)))