ASJPush/Controller/DetectControllerV2.py

import json
import logging
import threading

from django.http import JsonResponse
from django.views.generic.base import View
from AnsjerPush.config import CONFIG_EUR, CONFIG_INFO, CONFIG_CN, CONFIG_US

from Object.RedisObject import RedisObject
from Service.DevicePushService import DevicePushService
from concurrent.futures import ThreadPoolExecutor

TIME_LOGGER = logging.getLogger('time')
# 创建一个全局的线程池实例
# 线程池最大数量
# 1、CPU密集型：操作内存处理的业务，一般线程数设置为：CPU核数 + 1 或者 CPU核数*2。核数为4的话一般设置5或8
# 2、IO密集型：文件操作，网络操作，数据库操作，一般线程设置为：CPU核数 / (1-0.9)，核数为4的话，一般设置 40
executor = ThreadPoolExecutor(max_workers=20)
if CONFIG_INFO == CONFIG_US or CONFIG_INFO == CONFIG_EUR:
    ThreadPoolExecutor(max_workers=80)
elif CONFIG_INFO == CONFIG_CN:
    ThreadPoolExecutor(max_workers=40)


# 移动侦测V2接口
class NotificationV2View(View):

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

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

    @staticmethod
    def validation(request_dict):
        """
        设备触发报警消息推送
        @param request_dict:uidToken 加密uid
        @param request_dict:etk 加密uid
        @param request_dict:channel 设备通道号
        @param request_dict:n_time 设备触发报警时间
        @param request_dict:event_type 设备事件类型
        @param request_dict:is_st 文件类型(0,2:无图, 1:单张图片, 3:三张图片)
        @param request_dict:region 文件存储区域（1：国外, 2:国内)
        @param request_dict:electricity 电量值
        @param request_dict:time_token 时间戳token
        @param request_dict:uid uid
        @param request_dict:dealings_type 往来检测 1：来，2：离开
        @param request_dict:detection 检测类型 0：普通，1：算法
        """
        uidToken = request_dict.get('uidToken', None)
        etk = request_dict.get('etk', None)
        channel = request_dict.get('channel', '1')
        n_time = request_dict.get('n_time', None)
        event_type = request_dict.get('event_type', None)
        is_st = request_dict.get('is_st', None)
        region = request_dict.get('region', None)
        electricity = request_dict.get('electricity', '')
        dealings_type = int(request_dict.get('dealingsType', 0))
        detection = int(request_dict.get('detection', 0))
        button = request_dict.get('button', '1')
        storage_location = request_dict.get('storage_location', None)
        uid = ""
        # 参数校验
        if not all([channel, n_time]):
            return JsonResponse(status=200, data={'code': 444, 'msg': 'param is wrong'})
        if not region or not is_st:
            return JsonResponse(status=200, data={'code': 404, 'msg': 'no region or is_st'})

        is_st = int(is_st)
        region = int(region)
        event_type = int(event_type)
        redis_obj = RedisObject()
        try:
            uid = DevicePushService.decode_uid(etk, uidToken)
            if len(uid) != 20 and len(uid) != 14:
                return JsonResponse(status=200, data={'code': 404, 'msg': 'wrong uid'})
            TIME_LOGGER.info('开始推送,uid:{},参数:{}'.format(uid, request_dict))
            # 判断是否为系统消息
            is_sys_msg = DevicePushService.judge_sys_msg(event_type)
            if is_sys_msg:
                push_interval = '{}_{}_{}_flag'.format(uid, channel, event_type)
            else:
                push_interval = '{}_{}_flag'.format(uid, channel)

            req_limiting = '{}_{}_{}_ptl'.format(uid, channel, event_type)
            cache_req_limiting = redis_obj.get_data(key=req_limiting)  # 获取请求限流缓存数据
            cache_app_push = redis_obj.get_data(key=push_interval)  # 获取APP推送消息时间间隔缓存数据
            if event_type not in [606, 607]:
                if cache_req_limiting:  # 限流存在则直接返回
                    return JsonResponse(status=200, data={'code': 0, 'msg': 'Push again in one minute'})
            redis_obj.set_data(key=req_limiting, val=1, expire=60)  # 当缓存不存在限流数据 重新设置一分钟请求一次

            # 查询uid_push和uid_set数据
            uid_push_qs = DevicePushService.query_uid_push(uid, event_type, button)
            if not uid_push_qs.exists():
                TIME_LOGGER.info('推送响应,uid:{},uid_push数据不存在!'.format(uid))
                return JsonResponse(status=200, data={'code': 176, 'msg': 'no uid_push data'})

            ai_type = uid_push_qs.first()['uid_set__ai_type']
            device_type = uid_push_qs.first()['uid_set__device_type']

            # uid_push_qs转存列表
            uid_set_push_list = DevicePushService.qs_to_list(uid_push_qs)
            nickname = uid_set_push_list[0]['uid_set__nickname']
            nickname = uid if not nickname else nickname

            # APP消息提醒推送间隔
            detect_interval = uid_set_push_list[0]['uid_set__detect_interval']
            if event_type not in [606, 607]:
                if not cache_app_push:
                    # 缓存APP提醒推送间隔 默认1分钟提醒一次
                    DevicePushService.cache_push_detect_interval(redis_obj, push_interval, detect_interval,
                                                                 uid_set_push_list[0]['uid_set__new_detect_interval'])
            else:
                cache_app_push = ''

            bucket = ''
            aws_s3_client = ''
            # 推图,初始化s3 client
            if is_st == 1 or is_st == 3:
                aws_s3_client = DevicePushService.get_s3_client(region=region)
                bucket = 'foreignpush' if region == 1 else 'push'

            # 推送相关参数
            push_kwargs = {
                'uid': uid,
                'channel': channel,
                'event_type': event_type,
                'n_time': n_time,
            }
            # 对象存储区域, 测试/国内: 旧:aws(2),新:华为云(5),设备传参, 美洲: oci凤凰城(3), 欧洲: oci伦敦(4)
            if CONFIG_INFO == CONFIG_US:
                storage_location = 3
            elif CONFIG_INFO == CONFIG_EUR:
                storage_location = 4
            else:
                if storage_location is not None:
                    storage_location = int(storage_location)
                else:
                    storage_location = 2

            params = {'nickname': nickname, 'uid': uid, 'push_kwargs': push_kwargs, 'is_st': is_st, 'region': region,
                      'is_sys_msg': is_sys_msg, 'channel': channel, 'event_type': event_type, 'n_time': n_time,
                      'electricity': electricity, 'bucket': bucket, 'aws_s3_client': aws_s3_client,
                      'app_push': cache_app_push, 'storage_location': storage_location, 'ai_type': ai_type, 'device_type': device_type,
                      'dealings_type': dealings_type, 'detection': detection,
                      'app_push_config': uid_set_push_list[0]['uid_set__msg_notify'],
                      'uid_set_push_list': uid_set_push_list}

            # 使用全局的线程池提交推送任务
            executor.submit(push_and_save_data, **params)

            # 异步推送消息和保存数据
            # push_thread = threading.Thread(
            #     target=push_and_save_data,
            #     kwargs=params)
            # push_thread.start()

            # 视频通话不返回图片链接
            if event_type == 607:
                TIME_LOGGER.info('推送响应,uid:{},n_time:{},事件类型:{}'.format(uid, n_time, event_type))
                return JsonResponse(status=200, data={'code': 0, 'msg': 'success'})

            # 获取S3对象上传链接
            kwargs = {
                'is_st': is_st,
                'uid': uid,
                'channel': channel,
                'n_time': n_time,
                'region': region,
                'aws_s3_client': aws_s3_client,
                'storage_location': storage_location
            }
            res_data = DevicePushService.get_res_data(**kwargs)

            TIME_LOGGER.info('推送响应,uid:{},n_time:{},事件类型:{},响应:{}'.format(
                uid, n_time, event_type, json.dumps(res_data)))
            return JsonResponse(status=200, data=res_data)
        except Exception as e:
            TIME_LOGGER.info('V2推送接口异常uid:{},etk:{},error_line:{},error_msg:{}'.
                             format(uid, etk, e.__traceback__.tb_lineno, repr(e)))
            data = {
                'error_line': e.__traceback__.tb_lineno,
                'error_msg': repr(e)
            }
            return JsonResponse(status=200, data=json.dumps(data), safe=False)


def push_and_save_data(**params):
    uid = params['uid']
    TIME_LOGGER.info('{}开始异步存表和推送'.format(uid))
    # 异步推送消息
    push_thread = threading.Thread(
        target=DevicePushService.push_msg,
        kwargs=params)
    push_thread.start()
    # 保存推送数据
    result = DevicePushService.save_msg_push(**params)
    TIME_LOGGER.info('{}存表结果:{}'.format(uid, result))