ASJPush/Service/DevicePushService.py

# -*- encoding: utf-8 -*-
"""
@File    : DevicePushService.py
@Time    : 2022/11/23 11:40
@Author  : stephen
@Email   : zhangdongming@asj6.wecom.work
@Software: PyCharm
"""
import datetime
import hashlib
import json
import logging
import threading
import time

import boto3
import botocore
import oss2
import requests
from obs import ObsClient

from AnsjerPush.Config.aiConfig import DEVICE_EVENT_TYPE, ALGORITHM_COMBO_TYPES
from AnsjerPush.config import CONFIG_INFO, CONFIG_CN, MULTI_CHANNEL_TYPE_LIST, SYS_EVENT_TYPE_LIST, AWS_ACCESS_KEY_ID, \
    AWS_SECRET_ACCESS_KEY, EVENT_DICT, EVENT_DICT_CN, CONFIG_TEST, HUAWEICLOUD_AK, HUAWEICLOUD_SK, \
    HUAWEICLOUD_OBS_SERVER, HUAWEICLOUD_PUSH_BUKET, OSS_STS_ACCESS_KEY, OSS_STS_ACCESS_SECRET
from AnsjerPush.config import XMPUSH_CONFIG, OPPOPUSH_CONFIG, XM_PUSH_CHANNEL_ID, XM_PUSH_CHANNEL_DICT
from Model.models import UidPushModel, SysMsgModel, DeviceSharePermission, DeviceChannelUserSet, \
    DeviceChannelUserPermission, UidSetModel, Device_Info, UserAudioVideoPush, PushLog
from Object.ETkObject import ETkObject
from Object.OCIObjectStorage import OCIObjectStorage
from Object.UidTokenObject import UidTokenObject
from Object.enums.EventTypeEnum import EventTypeEnumObj
from Object.utils import LocalDateTimeUtil
from Service.CommonService import CommonService
from Service.EquipmentInfoService import EquipmentInfoService, EQUIPMENT_INFO_DICT
from Service.HuaweiPushService.HuaweiPushService import HuaweiPushObject
from Service.PushService import PushObject
from django.db import close_old_connections

LOGGING = logging.getLogger('info')
TIME_LOGGER = logging.getLogger('time')
ERROR_INFO_LOGGER = logging.getLogger('error_info')


class DevicePushService:

    @staticmethod
    def decode_uid(etk, uidToken):
        """
        解密UID,优先解密etk 否则判断uidToken
        """
        # 解密获取uid
        if etk:
            eto = ETkObject(etk)
            uid = eto.uid
        else:
            uto = UidTokenObject(uidToken)
            uid = uto.UID
        return uid

    @staticmethod
    def judge_sys_msg(event_type):
        """
        判断是否属于系统消息
        @param event_type: 事件类型
        @return: bool
        """
        if event_type in SYS_EVENT_TYPE_LIST:
            return True
        return False

    @staticmethod
    def get_s3_client(region):
        """
        根据地区获取S3 client
        @param region: 地区,1：国外, 2:国内
        @return: aws_s3_client
        """
        if int(region) == 1:
            aws_s3_client = boto3.client(
                's3',
                aws_access_key_id=AWS_ACCESS_KEY_ID[1],
                aws_secret_access_key=AWS_SECRET_ACCESS_KEY[1],
                config=botocore.client.Config(signature_version='s3v4'),
                region_name='us-east-1'
            )
        else:
            aws_s3_client = boto3.client(
                's3',
                aws_access_key_id=AWS_ACCESS_KEY_ID[0],
                aws_secret_access_key=AWS_SECRET_ACCESS_KEY[0],
                config=botocore.client.Config(signature_version='s3v4'),
                region_name='cn-northwest-1'
            )
        return aws_s3_client

    @classmethod
    def query_uid_push(cls, uid, event_type, button='1'):
        """
        查询uid_push和uid_set数据
        @param uid: uid
        @param event_type: 事件类型
        @param button: 按钮
        @return: uid_push_qs
        """
        if event_type not in [606, 607]:
            uid_push_qs = UidPushModel.objects.filter(uid_set__uid=uid, uid_set__detect_status=1) \
                .exclude(token_val='0'). \
                values('token_val', 'app_type', 'appBundleId', 'm_code', 'push_type', 'userID_id', 'userID__NickName',
                       'lang', 'm_code', 'tz', 'uid_set__nickname', 'uid_set__detect_interval', 'uid_set__detect_group',
                       'uid_set__channel', 'uid_set__ai_type', 'uid_set__device_type', 'uid_set__new_detect_interval',
                       'uid_set__msg_notify')
        else:
            # 一键通话只推主用户
            device_info_qs = Device_Info.objects.filter(UID=uid).values('vodPrimaryUserID')
            primary_user_id = device_info_qs[0]['vodPrimaryUserID']
            if event_type == 607:  # 音视频通话根据用户按钮来推送
                button_qs = UserAudioVideoPush.objects.filter(uid=uid).values('buttonUser1', 'buttonUser2')
                if button_qs.exists():
                    primary_user_id = button_qs[0]['buttonUser1'] if button == '1' else button_qs[0]['buttonUser2']
            uid_push_qs = UidPushModel.objects.filter(uid_set__uid=uid, userID_id=primary_user_id) \
                .exclude(token_val='0'). \
                values('token_val', 'app_type', 'appBundleId', 'm_code', 'push_type', 'userID_id', 'userID__NickName',
                       'lang', 'm_code', 'tz', 'uid_set__nickname', 'uid_set__detect_interval', 'uid_set__detect_group',
                       'uid_set__channel', 'uid_set__ai_type', 'uid_set__device_type', 'uid_set__new_detect_interval',
                       'uid_set__msg_notify', 'jg_token_val')
        return uid_push_qs

    @classmethod
    def get_uid_push_by_uid(cls, uid):
        """
        根据uid获取设备推送（目前仅用在OZI定制客户）
        """
        device_info_qs = Device_Info.objects.filter(UID=uid).values('vodPrimaryUserID')
        primary_user_id = device_info_qs[0]['vodPrimaryUserID']
        uid_push_qs = UidPushModel.objects.filter(uid_set__uid=uid, userID_id=primary_user_id) \
            .exclude(token_val='0'). \
            values('token_val', 'app_type', 'appBundleId', 'm_code', 'push_type', 'userID_id', 'userID__NickName',
                   'lang', 'm_code', 'tz', 'uid_set__nickname', 'uid_set__detect_interval', 'uid_set__detect_group',
                   'uid_set__channel', 'uid_set__ai_type', 'uid_set__device_type', 'uid_set__new_detect_interval',
                   'uid_set__msg_notify', 'jg_token_val')
        return uid_push_qs

    @staticmethod
    def qs_to_list(qs):
        """
        qs对象转存列表
        @param qs: query set对象
        @return: qs_list
        """
        qs_list = []
        for i in qs:
            qs_list.append(i)
        return qs_list

    @staticmethod
    def cache_push_detect_interval(redis_obj, name, detect_interval, new_detect_interval):
        """
        缓存设置推送消息的时间间隔
        @param redis_obj: redis对象
        @param name: redis key
        @param detect_interval: 原推送时间间隔
        @param new_detect_interval: 新推送时间间隔
        """
        if CONFIG_INFO != CONFIG_CN:
            detect_interval = new_detect_interval if new_detect_interval > 0 else detect_interval
            detect_interval = 60 if detect_interval < 60 else detect_interval
        else:  # 国内推送兼容问题，有值并且大于旧消息间隔则使用new_detect_interval
            detect_interval = new_detect_interval if new_detect_interval > detect_interval else detect_interval
        redis_obj.set_data(key=name, val=1, expire=detect_interval - 5)

    @classmethod
    def push_msg(cls, **params):
        """
        推送消息
        @param params: 推送参数
        @return: bool
        """
        try:
            uid = params['uid']
            params['event_tag'] = cls.get_event_tag(params['ai_type'], params['event_type'], params['detection'])
            is_app_push = True if params['event_type'] in [606, 607] else \
                cls.is_send_app_push(
                    params['event_type'], params['event_tag'], params['app_push_config'], params['app_push'],
                    uid, params['channel'])

            # 低功耗产品推送，休眠702,低电量704提醒,1023 ozi, 并且detection=0，0标识单事件类型，1标识多事件类型
            is_app_push = True if params['event_type'] in [702, 704, 1022, 1023] and params[
                'detection'] == 0 else is_app_push
            redis_obj = params['redis_obj']
            # 推送
            if is_app_push:
                msg_key = 'PUSH:MSG:IMAGE:{}:{}:{}'.format(params['uid'], params['channel'], params['n_time'])
                d_params = {'is_st': params['is_st'], 'storage_location': params['storage_location'],
                            'event_tag': params['event_tag'], 'event_type': params['event_type']}
                redis_obj.set_data(msg_key, json.dumps(d_params), 60)

                push_kwargs = params['push_kwargs']
                for up in params['uid_set_push_list']:
                    push_type = up['push_type']
                    lang = up['lang']
                    tz = up['tz']
                    if tz is None or tz == '':
                        tz = 0
                    if params['event_type'] in [606, 607] and push_type in [5, 6]:
                        push_kwargs['jg_token_val'] = up['jg_token_val']
                    else:
                        if 'jg_token_val' in push_kwargs:
                            push_kwargs.pop('jg_token_val')

                    appBundleId = up['appBundleId']
                    token_val = up['token_val']
                    # 发送标题
                    msg_title = cls.get_msg_title(nickname=params['nickname'])
                    # 发送内容
                    msg_text = cls.get_msg_text(channel=params['channel'], n_time=params['n_time'], lang=lang, tz=tz,
                                                event_type=params['event_type'], ai_type=params['ai_type'],
                                                device_type=params['device_type'], electricity=params['electricity'],
                                                dealings_type=params['dealings_type'], event_tag=params['event_tag']
                                                )

                    # 补齐推送参数
                    push_kwargs['appBundleId'] = appBundleId
                    push_kwargs['token_val'] = token_val
                    push_kwargs['msg_title'] = msg_title
                    push_kwargs['msg_text'] = msg_text
                    params['push_kwargs'] = push_kwargs
                    params['appBundleId'] = appBundleId
                    params['token_val'] = token_val
                    params['lang'] = lang
                    params['tz'] = tz
                    params['push_type'] = push_type
                    params['redis_obj'] = redis_obj

                    # GlobalThreadPool().submit(cls.send_app_msg_push, **params)
                    push_thread = threading.Thread(
                        target=cls.send_app_msg_push,
                        kwargs=params
                    )
                    push_thread.start()
        except Exception as e:
            ERROR_INFO_LOGGER.info(
                '推送消息线程异常,uid:{},error_line:{},error_msg:{}'
                .format(params['uid'], e.__traceback__.tb_lineno, repr(e)))

    @classmethod
    def save_msg_push(cls, **params):
        """
        保存推送数据和推送消息
        @param params: 推送参数
        @return: bool
        """
        sys_msg_list = []
        saved_user_id_list = []
        uid = params['uid']
        now_time = int(time.time())
        redis_obj = params['redis_obj']
        try:
            params['event_tag'] = cls.get_event_tag(params['ai_type'], params['event_type'], params['detection'])

            save_equipment_info = False
            equipment_info_key = EquipmentInfoService.randoms_choice_equipment_info_key()
            for up in params['uid_set_push_list']:
                lang = up['lang']
                tz = up['tz']
                if tz is None or tz == '':
                    tz = 0

                # 保存系统消息或推送消息数据
                user_id = up['userID_id']
                if user_id not in saved_user_id_list:  # 防止同一用户重复写入数据
                    # 系统消息
                    if params['is_sys_msg']:
                        sys_msg_text = cls.get_msg_text(channel=params['channel'], n_time=params['n_time'], lang=lang,
                                                        tz=tz, is_sys=1, device_type=params['device_type'],
                                                        event_type=params['event_type'],
                                                        electricity=params['electricity'])
                        sys_msg_list.append(SysMsgModel(userID_id=user_id, msg=sys_msg_text, addTime=now_time,
                                                        updTime=now_time, uid=uid, eventType=params['event_type']))
                    # 保存推送消息
                    else:
                        if not save_equipment_info:
                            save_equipment_info = True
                        params['userID_id'] = user_id
                        answer_status = 1 if params['dealings_type'] == 1 else 0
                        equipment_info_kwargs = {
                            'device_user_id': params['userID_id'],
                            'event_time': params['n_time'],
                            'event_type': params['event_type'],
                            'device_uid': params['uid'],
                            'device_nick_name': params['nickname'],
                            'channel': params['channel'],
                            'alarm': 'Motion \tChannel:{}'.format(params['channel']),
                            'is_st': params['is_st'],
                            'add_time': int(time.time()),
                            'storage_location': params['storage_location'],
                            'event_tag': params['event_tag'],
                            'answer_status': answer_status
                        }
                        # 保存到redis列表
                        equipment_info_value = json.dumps(equipment_info_kwargs)
                        redis_obj.rpush(equipment_info_key, equipment_info_value)
                    saved_user_id_list.append(user_id)
            close_old_connections()
            # 写入系统消息
            if sys_msg_list:
                SysMsgModel.objects.bulk_create(sys_msg_list)
            if save_equipment_info:
                equipment_info_list = []
                equipment_info_model = EQUIPMENT_INFO_DICT[equipment_info_key]

                # 一键通话和视频通话需要实时写入数据
                # 正式服通过定时任务批量写入数据
                if params['event_type'] in [606, 607] or CONFIG_INFO == CONFIG_TEST:
                    end = 0
                    # 缓存数据多于100条,批量保存前100条,否则保存全部
                    equipment_info_len = redis_obj.llen(equipment_info_key)
                    end = 99 if equipment_info_len > 100 else equipment_info_len - 1

                    if end != 0:
                        equipment_info_redis_list = redis_obj.lrange(equipment_info_key, 0, end)
                        redis_obj.ltrim(equipment_info_key, end + 1, -1)

                        for equipment_info in equipment_info_redis_list:
                            equipment_info_data = eval(equipment_info)
                            # 设备昵称存在表情,解码utf-8
                            if equipment_info_data.get('device_nick_name') is not None:
                                equipment_info_data['device_nick_name'] = equipment_info_data['device_nick_name']. \
                                    encode('UTF-8', 'ignore').decode('UTF-8')
                            equipment_info_list.append(equipment_info_model(**equipment_info_data))
                        equipment_info_model.objects.bulk_create(equipment_info_list)

            return True
        except Exception as e:
            ERROR_INFO_LOGGER.info(
                '保存推送数据和推送消息线程异常,uid:{}, error_line:{}, error_msg:{}'.
                format(uid, e.__traceback__.tb_lineno, repr(e)))
            return False

    @classmethod
    def get_event_tag(cls, ai_type, event_type, detection=0):
        """
        获取事件标签
        """
        algorithm = False
        if ai_type > 0 and detection == 1:
            algorithm = True
        elif (ai_type == 7 and event_type <= 7) or (ai_type == 47 and event_type <= 47) or (detection == 1):
            algorithm = True
        if not algorithm:
            return ',' + str(event_type) + ','
        event_res = DEVICE_EVENT_TYPE.get(event_type, 0)
        if event_res > 0:
            return ',' + str(event_res) + ','
        event_type = cls.dec_to_bin(event_type)
        types = cls.get_combo_types(event_type)
        res = ','.join(types) + ','
        return ',' + res

    @classmethod
    def get_combo_types(cls, event_type):
        """
        获取设备算法组合类型
        51：移动侦测，52：传感器报警，53：影像遗失，54：PIR，55：门磁报警，56：外部发报，57：人型报警（提示：有人出现），58：车型，59：宠物，60：人脸，61：异响，
        62：区域闯入，63：区域闯出，64：长时间无人检测，65：长时间无人检测，66：往来检测，67：哭声检测，68：手势检测,69:火焰检测
        0：代表空字符，702：摄像头休眠，703：摄像头唤醒，704：电量过低
        AWS AI识别 1：人形，2：车型，3：宠物，4：包裹。云端AI类型
        """
        try:
            types = []
            event_type = str(event_type)
            len_type = len(event_type)
            for i in range(len_type):
                e_type = event_type[len_type - 1 - i]
                if e_type == '1':
                    types.append(str(ALGORITHM_COMBO_TYPES[i]))
            return types
        except Exception as e:
            LOGGING.info('推送错误异常,errLine:{}, errMsg:{}'.format(e.__traceback__.tb_lineno, repr(e)))
            return event_type

    @staticmethod
    def dec_to_bin(num):
        """
        十进制转二进制
        """
        result = ""
        while num != 0:
            ret = num % 2
            num //= 2
            result = str(ret) + result
        return result

    @classmethod
    def send_app_msg_push(cls, **kwargs):
        """
        发送推送
        @kwargs :
        @return push_result: bool
        """
        try:
            push_type = kwargs['push_type']
            push_kwargs = kwargs['push_kwargs']
            push_result = False
            uid = kwargs['uid']
            # is_st为1或3,且推送类型为apns,gcm,华为,异步推送图片
            if (kwargs['is_st'] == 1 or kwargs['is_st'] == 3) and \
                    (push_type == 0 or push_type == 1 or push_type == 3):
                if kwargs['is_st'] == 1:
                    key = '{}/{}/{}.jpeg'.format(kwargs['uid'], kwargs['channel'], kwargs['n_time'])
                else:
                    key = '{}/{}/{}_0.jpeg'.format(kwargs['uid'], kwargs['channel'], kwargs['n_time'])
                # 开始异步推送图片
                # thread_pool = GlobalThreadPool()
                # thread_pool.submit(cls.async_send_picture_push, (
                #     push_type, kwargs['aws_s3_client'], kwargs['bucket'], key,
                #     kwargs['uid'], kwargs['appBundleId'], kwargs['token_val'], kwargs['event_type'], kwargs['n_time'],
                #     push_kwargs['msg_title'], push_kwargs['msg_text'], kwargs['channel'], kwargs['storage_location']))

                push_thread = threading.Thread(target=cls.async_send_picture_push, args=(
                    push_type, kwargs['aws_s3_client'], kwargs['bucket'], key,
                    kwargs['uid'], kwargs['appBundleId'], kwargs['token_val'], kwargs['event_type'], kwargs['n_time'],
                    push_kwargs['msg_title'], push_kwargs['msg_text'], kwargs['channel'], kwargs['storage_location'],
                    kwargs['redis_obj']))
                push_thread.start()

                push_result = True

            # 不推图
            else:
                if push_type in [0, 1, 2]:
                    kwargs = {
                        'nickname': kwargs['uid'],
                        'app_bundle_id': kwargs['appBundleId'],
                        'token_val': kwargs['token_val'],
                        'n_time': kwargs['n_time'],
                        'event_type': kwargs['event_type'],
                        'msg_title': push_kwargs['msg_title'],
                        'msg_text': push_kwargs['msg_text'],
                        'uid': kwargs['uid'],
                        'channel': kwargs['channel']
                    }

                    if push_type == 0:  # ios apns
                        push_result = PushObject.ios_apns_push(**kwargs)
                    elif push_type == 1:  # android gcm
                        push_result = PushObject.android_fcm_push_v1(**kwargs)
                    elif push_type == 2:  # android jpush
                        kwargs.pop('uid')
                        push_result = PushObject.android_jpush(**kwargs)

                elif push_type == 3:
                    huawei_push_object = HuaweiPushObject(appBundleId=kwargs["appBundleId"])
                    huawei_push_object.send_push_notify_message(**push_kwargs)
                elif push_type == 4:  # android xmpush
                    if kwargs['event_type'] in EventTypeEnumObj.DATA_PUSH_EVENT_TYPE_LIST.value:
                        push_channel = 'push_to_talk'
                    else:
                        push_channel = 'device_reminder'
                    channel_id = XM_PUSH_CHANNEL_DICT[kwargs['appBundleId']][push_channel]
                    push_result = cls.do_xmpush(channel_id=channel_id, **push_kwargs)
                elif push_type == 5:  # android vivopush
                    push_result = PushObject.android_vivopush(**push_kwargs)
                elif push_type == 6:  # android oppopush
                    channel_id = 'DEVICE_REMINDER'
                    push_result = cls.do_oppopush(channel_id=channel_id, **push_kwargs)
                elif push_type == 7:  # android meizupush
                    push_result = PushObject.android_meizupush(**push_kwargs)
                elif push_type == 8:  # android honorpush
                    push_result = PushObject.android_honorpush(**push_kwargs)

            if kwargs['event_type'] in [606, 607]:
                close_old_connections()
                # 写入日志表
                PushLog.objects.create(uid=uid, event_type=kwargs['event_type'], created_time=int(time.time()),
                                       content=push_kwargs, push_result=push_result, push_type=push_type)
            return push_result
        except Exception as e:
            ERROR_INFO_LOGGER.info('发送推送线程异常uid:{},type:{},error_line:{},error_msg:{}'
                                   .format(kwargs['uid'], kwargs['push_type'], e.__traceback__.tb_lineno, repr(e)))
            return False

    @staticmethod
    def get_msg_title(nickname):
        """
        获取消息标题
        """""
        return nickname

    @staticmethod
    def get_event_type_text(lang, event_type, dealings_type):
        """
        事件类型文案键值查找
        """
        if lang == 'cn':
            if event_type in EVENT_DICT_CN:
                if isinstance(EVENT_DICT_CN[event_type], dict):
                    msg_type = EVENT_DICT_CN[event_type][dealings_type]
                else:
                    msg_type = EVENT_DICT_CN[event_type]
            else:
                msg_type = '未知事件类型 '
            return msg_type
        else:
            if event_type in EVENT_DICT:
                if isinstance(EVENT_DICT[event_type], dict):
                    msg_type = EVENT_DICT[event_type][dealings_type]
                else:
                    msg_type = EVENT_DICT[event_type]
            else:
                msg_type = 'Unknown event type'
            return msg_type

    @staticmethod
    def get_msg_text(channel, n_time, lang, tz, event_type, electricity='', is_sys=0, dealings_type=0, ai_type=0,
                     device_type=0, event_tag=''):
        """
        获取消息文本
        @param: channel 通道号
        @param: n_time 触发事件
        @param: lang 语言
        @param: tz 时区
        @param: event_type 事件类型
        @param: electricity 电量
        @param: is_sys 是否系统消息
        @param: dealings_type 往来类型 1 进 1 离开
        @param: ai_type 设备本地AI只能算法 事件类型
        @param: device_type 设备类型
        @param: event_tag 设备算法事件标签
        """
        msg_type = ''
        event_type = int(event_type)
        device_type = int(device_type)
        event_list = []
        if event_tag:
            event_list = [int(event) for event in event_tag.split(',') if event]

            events_to_remove = [702, 703, 704]
            for event in events_to_remove:
                if event in event_list:
                    event_list.remove(event)
        if lang == 'cn':
            if event_type == 51:
                msg_type = '检测到画面变化'
            elif event_type == 52:
                msg_type = '传感器报警'
            elif event_type == 53:
                msg_type = '影像遗失'
            elif event_type == 54:
                msg_type = 'PIR'
            elif event_type == 55:
                msg_type = '门磁报警'
            elif event_type == 56:
                msg_type = '外部发报'
            elif event_type == 57:
                msg_type = '有人出现'
            elif event_type == 58:
                msg_type = '有车出现'
            elif event_type == 59:
                msg_type = '有宠物出现'
            elif event_type == 60:
                msg_type = '发现人脸'
            elif event_type == 61:
                msg_type = '有异响'
            elif event_type == 62:
                msg_type = '区域闯入'
            elif event_type == 63:
                msg_type = '区域闯出'
            elif event_type == 64:
                msg_type = '有人徘徊'
            elif event_type == 65:
                msg_type = '长时间无人出现'
            elif event_type == 1022:
                msg_type = '有人按下门铃'
            elif event_type == 1023:
                msg_type = '儿童保护模式开启'

            elif event_type == 704:
                msg_type = '电量低'
            elif event_type == 702:
                msg_type = '摄像头休眠'
            elif event_type == 703:
                msg_type = '摄像头唤醒'
            elif event_type in [606, 607]:
                msg_type = '有人呼叫，请点击查看'
            if event_type not in [606, 607, 1022] and ai_type > 0 and event_list:
                msg_type = ''.join([DevicePushService.get_event_type_text(lang, item, dealings_type)
                                    for item in event_list])
            if is_sys:
                if device_type in MULTI_CHANNEL_TYPE_LIST:
                    send_text = '{} 通道:{}'.format(msg_type, channel)
                else:
                    send_text = msg_type
            else:
                if device_type in MULTI_CHANNEL_TYPE_LIST:
                    send_text = '{} 通道:{}'.format(msg_type, channel)
                else:
                    send_text = '{}'.format(msg_type)
        else:
            if event_type == 51:
                msg_type = 'Screen change detected'
            elif event_type == 52:
                msg_type = 'Sensor alarms'
            elif event_type == 53:
                msg_type = 'Lost images'
            elif event_type == 54:
                msg_type = 'PIR'
            elif event_type == 55:
                msg_type = 'Door magnetic alarm'
            elif event_type == 56:
                msg_type = 'External reporting'
            elif event_type == 57:
                msg_type = 'Person detected'
            elif event_type == 58:
                msg_type = 'Vehicle detected'
            elif event_type == 59:
                msg_type = 'Pet detected'
            elif event_type == 60:
                msg_type = 'Human face detected'
            elif event_type == 61:
                msg_type = 'Abnormal sound detected'
            elif event_type == 62:
                msg_type = 'Intrusion detected in the area'
            elif event_type == 63:
                msg_type = 'Area vacated'
            elif event_type == 64:
                msg_type = 'Loitering detected'
            elif event_type == 65:
                msg_type = 'No appearance for a long time'
            elif event_type == 1022:
                msg_type = 'Someone rang the doorbell'
            elif event_type == 1023:
                msg_type = 'Child protection mode is enabled'

            elif event_type == 704:
                msg_type = 'low battery'
            elif event_type == 702:
                msg_type = 'Camera sleep'
            elif event_type == 703:
                msg_type = 'Camera wake'
            elif event_type in [606, 607]:
                msg_type = 'Someone is calling, please click to view'
            if event_type not in [606, 607, 1022] and ai_type > 0 and event_list:
                msg_type = ''.join([DevicePushService.get_event_type_text(lang, item, dealings_type)
                                    for item in event_list])
            if is_sys:
                if device_type in MULTI_CHANNEL_TYPE_LIST:
                    send_text = '{} channel:{}'.format(msg_type, channel)
                else:
                    send_text = msg_type
            else:
                if device_type in MULTI_CHANNEL_TYPE_LIST:
                    send_text = '{} channel:{}'.format(msg_type, channel)
                else:
                    send_text = '{}'.format(msg_type)
        return send_text

    @staticmethod
    def do_xmpush(channel_id, uid, channel, appBundleId, token_val, event_type, n_time,
                  msg_title, msg_text):
        """
        android 国内小米APP消息提醒推送
        """
        try:
            url = 'https://api.xmpush.xiaomi.com/v3/message/regid'
            app_secret = XMPUSH_CONFIG[appBundleId]
            # payload = {'alert': 'Motion', 'msg': '', 'sound': 'sound.aif', 'zpush': '1',
            #            'received_at': n_time, 'event_time': n_time, 'event_type': event_type,
            #            'uid': uid, 'channel': channel
            #            }
            data = {
                'title': msg_title,
                'description': msg_text,
                'payload': 'payload',
                'restricted_package_name': appBundleId,
                'registration_id': token_val,
                'extra.channel_id': channel_id,
                'extra.alert': 'Motion',
                'extra.msg': '',
                'extra.sound': 'sound.aif',
                'extra.zpush': '1',
                'extra.received_at': n_time,
                'extra.event_time': n_time,
                'extra.event_type': event_type,
                'extra.uid': uid,
                'extra.channel': channel,
            }
            if event_type in [606, 607]:
                data['extra.sound_uri'] = 'android.resource://com.ansjer.zccloud_ab/raw/phone_call'
            headers = {
                'Authorization': 'key={}'.format(app_secret)
            }
            response = requests.post(url, data=data, headers=headers)
            if response.status_code == 200:
                return True
            ERROR_INFO_LOGGER.info('小米推送异常,uid:{},状态码{},时间:{},结果:{},'
                                   .format(uid, response.status_code, n_time, response.json()))
            return False
        except Exception as e:
            ERROR_INFO_LOGGER.info('小米推送异常,uid:{},时间:{},error_line:{},error_msg:{}'.
                                   format(uid, n_time, e.__traceback__.tb_lineno, repr(e)))
            return False

    @staticmethod
    def do_oppopush(channel_id, uid, channel, appBundleId, token_val, event_type, n_time,
                    msg_title, msg_text, jg_token_val=''):
        """
        android 国内oppo APP消息提醒推送
        """
        try:
            if event_type in [606, 607]:
                channel_id = XM_PUSH_CHANNEL_ID['push_to_talk']
            app_key = OPPOPUSH_CONFIG[appBundleId]['Key']
            master_secret = OPPOPUSH_CONFIG[appBundleId]['Secret']
            url = 'https://api.push.oppomobile.com/'
            now_time = str(round(time.time() * 1000))
            # 1、实例化一个sha256对象
            sha256 = hashlib.sha256()
            # 2、调用update方法进行加密
            sha256.update((app_key + now_time + master_secret).encode('utf-8'))
            # 3、调用hexdigest方法,获取加密结果
            sign = sha256.hexdigest()
            # 获取auth_token
            get_token_url = url + 'server/v1/auth'
            post_data = {
                'app_key': app_key,
                'sign': sign,
                'timestamp': now_time
            }
            headers = {'Content-Type': 'application/x-www-form-urlencoded'}
            response = requests.post(get_token_url, data=post_data, headers=headers)
            result = response.json()
            # 发送推送
            push_url = url + 'server/v1/message/notification/unicast'
            extra_data = {'alert': 'Motion', 'msg': '', 'sound': 'sound.aif', 'zpush': '1',
                          'received_at': n_time, 'event_time': n_time, 'event_type': event_type,
                          'uid': uid, 'channel': channel}
            message = {
                "target_type": 2,
                "target_value": token_val,
                "notification": {
                    "title": msg_title,
                    "content": msg_text,
                    'channel_id': channel_id,
                    'action_parameters': extra_data,
                    'click_action_type': 4,
                    'click_action_activity': OPPOPUSH_CONFIG[appBundleId]['click_action_activity']
                }
            }
            push_data = {
                'auth_token': result['data']['auth_token'],
                'message': json.dumps(message)
            }

            response = requests.post(push_url, data=push_data, headers=headers)
            if response.status_code == 200:
                if event_type in [606, 607]:
                    PushObject.jpush_transparent_transmission(msg_title, msg_text, appBundleId, jg_token_val,
                                                              extra_data)
                return True
            return False
        except Exception as e:
            ERROR_INFO_LOGGER.info('oppo推送异常,uid:{},time:{},error_line:{},error_msg:{}'.
                                   format(uid, n_time, e.__traceback__.tb_lineno, repr(e)))
            return False

    @classmethod
    def async_send_picture_push(cls, push_type, aws_s3_client, bucket, key, uid, appBundleId, token_val,
                                event_type, n_time, msg_title, msg_text, channel, storage_reg, redis_obj):
        """
        异步推送图片
        """
        try:
            if storage_reg == 1:
                # 阿里云
                auth = oss2.Auth(OSS_STS_ACCESS_KEY, OSS_STS_ACCESS_SECRET)
                oss_img_bucket = oss2.Bucket(auth, 'oss-cn-shenzhen.aliyuncs.com', 'apg')
                image_url = oss_img_bucket.sign_url('GET', key, 300)
            elif storage_reg in [3, 4]:
                image_url = DevicePushService.oci_object_url(uid, redis_obj, storage_reg, bucket, key)
            elif storage_reg == 5:
                image_url = DevicePushService.create_obs_signed_url(key, 'GET')
            else:
                image_url = aws_s3_client.generate_presigned_url(
                    'get_object', Params={'Bucket': bucket, 'Key': key}, ExpiresIn=3600)
            push_result = False
            if push_type == 0:
                push_result = PushObject.ios_apns_push(
                    uid, appBundleId, token_val, n_time, event_type, msg_title, msg_text, uid, channel, image_url)
            elif push_type == 1:
                push_result = PushObject.android_fcm_push_v1(
                    uid, appBundleId, token_val, n_time, event_type, msg_title, msg_text, uid, channel, image_url)
            elif push_type == 3:
                huawei_push_object = HuaweiPushObject(appBundleId=appBundleId)
                push_result = huawei_push_object.send_push_notify_message(
                    token_val=token_val, msg_title=msg_title, msg_text=msg_text, uid=uid, event_type=event_type,
                    n_time=n_time, image_url=image_url, channel=channel)

            TIME_LOGGER.info('{}推送图片,push_type:{},推送结果:{}'.format(uid, push_type, push_result))
        except Exception as e:
            ERROR_INFO_LOGGER.error(
                '异步推送图片异常,error_line:{},error_msg:{}'.format(e.__traceback__.tb_lineno, repr(e)))

    @staticmethod
    def oci_object_url(uid, redis_obj, storage_location, bucket, obj_name):
        """
        获取OCI对象存储URL 有效期5分钟
        @param uid: uid
        @param redis_obj: 缓存客户端
        @param storage_location: 存储区域
        @param bucket: 存储桶
        @param obj_name: 对象名称
        @return: url
        """
        try:
            uid_key = f'PUSH:PICTURE:OCI:URL:{uid}'
            oci_url = redis_obj.get_data(uid_key)
            if oci_url:
                return oci_url + obj_name
            oci = OCIObjectStorage('eur' if storage_location == 4 else 'us')
            prefix_name = f'{uid}/'
            time_expires = datetime.datetime.utcnow() + datetime.timedelta(minutes=60)
            result = oci.get_preauthenticated_request_url(bucket, 'ociPush', prefix_name, time_expires,
                                                          'AnyObjectRead')  # 授权到指定uid文件夹
            full_url = result.full_path if result else ''
            redis_obj.set_data(uid_key, full_url, 3580)
            return full_url + obj_name
        except Exception as e:
            LOGGING.error('oci查询消息列表异常error_line:{}, error_msg:{}'.format(e.__traceback__.tb_lineno, repr(e)))
            return ''

    @staticmethod
    def create_oci_req_url(storage_location, bucket, obj_name, oci=None):
        """
        创建oci预认证请求url
        """
        if not oci:
            region = 'eur' if storage_location == 4 else 'us'
            oci = OCIObjectStorage(region)
        time_expires = datetime.datetime.utcnow() + datetime.timedelta(minutes=60)
        result = oci.create_ereauthenticated_request(bucket, 'ociPush', obj_name, time_expires)
        if not result:
            return ''
        return result.full_path + result.object_name

    @staticmethod
    def get_res_data(**kwargs):
        """
        获取响应数据
        @return: res_data
        """
        res_data = {'code': 0, 'msg': 'success'}
        is_st = kwargs['is_st']
        storage_location = kwargs['storage_location']
        if is_st == 0 or is_st == 2:
            res_data['msg'] = 'success 0 or 2'
        elif is_st == 1:
            key_name = '{}/{}/{}.jpeg'.format(kwargs['uid'], kwargs['channel'], kwargs['n_time'])
            params = {'Key': key_name}
            if kwargs['region'] == 2:  # 2:国内
                params['Bucket'] = 'push'
            else:  # 1：国外
                params['Bucket'] = 'foreignpush'
            # 根据存储区域返回链接
            if storage_location in [3, 4]:
                # OCI
                img_url = DevicePushService.create_oci_req_url(storage_location, params['Bucket'], key_name)
                res_data['img_push'] = img_url
                res_data['msg'] = 'success 1'
            elif storage_location == 2:
                # AWS
                img_url = DevicePushService.generate_s3_url(kwargs['aws_s3_client'], params)
                res_data['img_push'] = img_url
            else:
                # 华为云
                img_url = DevicePushService.create_obs_signed_url(key_name, 'PUT')
                res_data['img_push'] = img_url
                res_data['msg'] = 'success 1'
        elif is_st == 3:
            img_url_list = []
            if kwargs['region'] == 2:  # 2:国内
                params = {'Bucket': 'push'}
            else:  # 1：国外
                params = {'Bucket': 'foreignpush'}
            oci_client = None
            if storage_location in [3, 4]:  # 三张图的时候提前获取实例化OCI
                region = 'eur' if storage_location == 4 else 'us'
                oci_client = OCIObjectStorage(region)
            for i in range(kwargs['is_st']):
                key_name = '{}/{}/{}_{}.jpeg'.format(kwargs['uid'], kwargs['channel'], kwargs['n_time'], i)
                params['Key'] = key_name
                if storage_location in [3, 4]:
                    # OCI
                    img_url = DevicePushService.create_oci_req_url(
                        storage_location, params['Bucket'], key_name, oci_client)
                elif storage_location == 2:
                    # AWS
                    img_url = DevicePushService.generate_s3_url(kwargs['aws_s3_client'], params)
                else:
                    # 华为云
                    img_url = DevicePushService.create_obs_signed_url(key_name, 'PUT')
                img_url_list.append(img_url)
            res_data['img_url_list'] = img_url_list
            res_data['msg'] = 'success 3'
        return res_data

    @staticmethod
    def generate_s3_url(aws_s3_client, params):
        """
        获取S3对象URL
        """
        response_url = aws_s3_client.generate_presigned_url(
            ClientMethod='put_object',
            Params=params,
            ExpiresIn=3600
        )
        return response_url

    @staticmethod
    def create_obs_signed_url(key_name, method):
        """
        生成对象存储预签名URL
        @param key_name: 对象名称
        @param method: 方法
        @return: 预签名URL
        """
        obs_client = ObsClient(
            access_key_id=HUAWEICLOUD_AK, secret_access_key=HUAWEICLOUD_SK, server=HUAWEICLOUD_OBS_SERVER)
        res = obs_client.createSignedUrl(
            method=method, bucketName=HUAWEICLOUD_PUSH_BUKET, objectKey=key_name, expires=3600)
        return res.signedUrl

    @staticmethod
    def check_share_permission(user_id, channel, uid):
        """
        检查用户是否有权限接收设备报警推送
        """
        user_permission_qs = DeviceChannelUserSet.objects.filter(user_id=user_id, uid=uid) \
            .values('id', 'channels')
        # 根据当前用户与uid查询是否设置过通道权限，不存在则不是分享设备
        if not user_permission_qs.exists():
            return True
        up_id = user_permission_qs[0]['id']
        channels = user_permission_qs[0]['channels']
        channels_list = [int(val) for val in channels.split(',')]
        # 当前uid是属于分享设备并且设置了权限
        # 判断通道是否设置了权限，不存在则当前通道没有权限接受消息推送
        if int(channel) not in channels_list:
            return False
        permission_qs = DeviceSharePermission.objects.filter(code='AlarmMessages').values('id')
        p_id = permission_qs[0]['id']

        # 当前通道存在设置则查看是否有 消息推送权限
        channel_permission_qs = DeviceChannelUserPermission.objects \
            .filter(channel_user_id=up_id, permission_id=p_id) \
            .values('permission_id', 'channel_user_id')
        if not channel_permission_qs.exists():
            return False
        return True

    @classmethod
    def is_algorithm_type(cls, uid, event_type):
        """
        判断是否是算法类型 62、63、64、65、66不限制推送
        """
        uid_set_qs = UidSetModel.objects.filter(uid=uid).values('ai_type')
        if not uid_set_qs.exists():
            return False
        if uid_set_qs[0]['ai_type'] == 0:
            return False
        event_types = [62, 63, 64, 65, 66]
        event_res = DEVICE_EVENT_TYPE.get(event_type, 0)
        if event_res in event_types:
            return True
        event_types2 = cls.get_combo_types(event_type)
        if not event_types2:
            return False
        c = [x for x in event_types if x in event_types2]
        return True if c else False

    @staticmethod
    def is_send_app_push(event_type, event_tag, app_push_config, msg_interval=None, uid=None, channel=1):
        """
        是否进行APP消息提醒
        @return: True|False
        """
        try:
            if not app_push_config:
                return True

            is_push = app_push_config['appPush']
            if is_push != 1:  # 1：进行APP提醒，其它则不执行APP提醒
                return False
            if msg_interval:  # 存在消息间隔数据缓存 不推送APP消息
                return False

            if 'nvr' in app_push_config:
                push = DevicePushService.is_msg_push_nvr(event_type, int(channel), app_push_config)
                TIME_LOGGER.info(f'uid:{uid}NVR推送结果:{push}')
                return push

            all_day = app_push_config['pushTime']['allDay']
            # 允许设备类型APP提醒列表
            app_event_types = app_push_config['eventTypes']['device']

            if all_day == 0:  # 1：全天提醒，0：自定义时间提醒
                push_time_config = app_push_config['pushTime']
                # 计算当前时间是否在自定义消息提醒范围内
                if not DevicePushService.is_push_notify_allowed_now(push_time_config):
                    return False
            # APP接收提醒，判断识别类型是否勾选提醒
            push_result = DevicePushService.is_type_push(event_type, event_tag, app_event_types)
            return push_result
        except Exception as e:
            LOGGING.info(
                '{}判断是否执行APP推送异常,errLine:{}, errMsg:{}'.format(uid, e.__traceback__.tb_lineno, repr(e)))
            return True

    @staticmethod
    def is_msg_push_nvr(event_type, channel, config):
        try:
            # 获取 NVR 配置中记录，默认为空列表
            nvr_entries = config.get('nvr', [])
            # 构建通道键，例如 'channel1' 或 'channel2'
            channel_key = f'channel{channel}'

            if not nvr_entries:
                return True

            # 查找给定通道
            matching_entry = next((entry for entry in nvr_entries if channel_key in entry), None)

            # 如果没有找到匹配项，返回 True，表示可以推送
            if not matching_entry:
                return True

                # 获取与 channel_key 对应的 event_types
            event_types = matching_entry[channel_key] if channel_key in matching_entry else []

            # 调用 is_type_push 方法检查事件类型是否可以推送
            return DevicePushService.is_type_push(event_type, None, event_types)

        except Exception as e:
            # 记录异常信息并返回 False
            ERROR_INFO_LOGGER.error(
                'NVRAPP推送异常, errLine: {}, errMsg: {}'.format(e.__traceback__.tb_lineno, repr(e)))
            return False

    @staticmethod
    def is_type_push(event_type, event_tag, app_event_types):
        # 检查事件标签和应用事件类型是否都存在
        if event_tag and app_event_types:
            # 将事件标签按逗号分割成列表，并转换为整数类型
            tag_list = [int(event) for event in event_tag.split(',') if event]
            # 判断是否有任一标签允许应用提醒
            return any(item in app_event_types for item in tag_list)

        # 检查事件类型和用户所选事件类型是否都存在，并判断事件类型在用户所选事件类型列表中
        return event_type and app_event_types and event_type in app_event_types

    @staticmethod
    def is_push_notify_allowed_now(push_time_config):
        """
        判断当前时间是否在允许APP推送提醒
        """
        now_time = int(time.time())
        start_time = push_time_config['startTime']
        end_time = push_time_config['endTime']
        repeat = push_time_config['repeat']
        tz = push_time_config['timeZone']

        # 获取当前日期和周几
        now_date, week = DevicePushService.get_now_date_and_week(now_time, tz)

        # 判断是否在重复日范围内
        if not DevicePushService.is_repeated(week, repeat):
            return False

        # 计算当前日期在一天中的秒数
        seconds = LocalDateTimeUtil.convert_time_to_seconds(now_date)

        # 判断是否在APP推送提醒范围内
        return DevicePushService.is_in_effect(start_time, end_time, seconds)

    @staticmethod
    def is_in_effect(start, end, now_seconds):
        """
        判断是否在提醒时间范围内
        @params: 开始时间秒
        @params: 结束时间秒
        @params: 当前时间秒
        @return: 当前时间是在范围内返回True 否则False
        """
        if start < end:
            return start <= now_seconds <= end
        else:
            return start <= now_seconds or now_seconds <= end

    @staticmethod
    def is_repeated(week_day, repeat_day):
        """
        判断是否重复日
        @params: week_day 周几
        @params: 重复日1-127
        @return: 如果当前日期在重复日则返回True否则False
        """
        # 判断对应位置上的值是否为 1
        is_repeat = (repeat_day >> (week_day - 1)) & 1 == 1
        return is_repeat

    @staticmethod
    def get_now_date_and_week(now_time, tz):
        now_data = CommonService.get_now_time_str(now_time, tz, 'cn')
        week = LocalDateTimeUtil.date_to_week(now_data)
        return now_data, week