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 copy
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.MessageConfig import EVENT_CONFIGS, DEFAULT_TEXTS, MSG_CONFIG
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, JPUSH_UID_LIST, \
    DATA_PUSH_EVENT_TYPE_LIST, PRIMARY_USERS_PUSH_EVENT_TYPE_LIST
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.GlobalThreadPoolObject import GlobalThreadPool
from Object.OCIObjectStorage import OCIObjectStorage
from Object.RedisObject import RedisObject
from Object.UidTokenObject import UidTokenObject
from Object.enums.ConstantEnum import ConstantEnum
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

from Service.VSeesHuaweiPushService.VseesHuaweiPushObject import VseesHuaweiPushObject

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 PRIMARY_USERS_PUSH_EVENT_TYPE_LIST:
            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:
            # 一键通话,视频通话,按下门铃,SD卡异常 只推主用户
            device_info_qs = Device_Info.objects.filter(UID=uid).values('vodPrimaryUserID')
            if not device_info_qs.exists():
                return None
            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')
        if not device_info_qs.exists():
            return None
        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 PRIMARY_USERS_PUSH_EVENT_TYPE_LIST \
                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']
                push_token_list = []
                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 DATA_PUSH_EVENT_TYPE_LIST and push_type in [5, 6]:
                        process_token = up['jg_token_val']
                        push_kwargs['jg_token_val'] = up['jg_token_val']
                    else:
                        process_token = up['token_val']
                        if 'jg_token_val' in push_kwargs:
                            push_kwargs.pop('jg_token_val')

                    appBundleId = up['appBundleId']
                    token_val = up['token_val']
                    if process_token in push_token_list:
                        LOGGING.info("uid:{}, 重复token_val{},".format(uid, token_val))
                        continue
                    # 发送标题
                    msg_title = cls.get_msg_title(nickname=params['nickname'])
                    # 发送内容
                    msg_text = cls.get_msg_text_v2(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'],
                                                   redis_obj=params['redis_obj'])

                    # 补齐推送参数
                    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

                    # 使用 copy.deepcopy 保证每个任务的 params 是独立的
                    push_params = copy.copy(params)  # 浅拷贝字典结构
                    push_params['push_kwargs'] = copy.deepcopy(params['push_kwargs'])  # 独立修改的字段再深拷贝

                    try:
                        GlobalThreadPool().submit(cls.send_app_msg_push, **push_params)
                    except RuntimeError as e:
                        ERROR_INFO_LOGGER.error(
                            '推送消息线程池异常uid:{} 线程池已满，触发降级，使用原始线程启动. error_line:{} error_msg:{}'.format(
                                uid, e.__traceback__.tb_lineno, repr(e)
                            )
                        )
                        push_thread = threading.Thread(
                            target=cls.send_app_msg_push,
                            kwargs=push_params
                        )
                        push_thread.start()
                    # 过滤相同的token_val
                    push_token_list.append(process_token)
        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'], is_save_msg=1)
                        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'],
                                                        channel=params['channel']))
                    # 保存推送消息
                    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 DATA_PUSH_EVENT_TYPE_LIST 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
                        uid = kwargs.pop('uid')
                        redis_obj = RedisObject()
                        jpush_uid_list = redis_obj.lrange(JPUSH_UID_LIST, 0, -1)
                        jpush_uid_list = [str(i, 'utf-8') for i in jpush_uid_list]
                        if uid in jpush_uid_list:
                            push_result = PushObject.jpush(**kwargs)
                elif push_type == 3:
                    if kwargs["appBundleId"] == ConstantEnum.ZOSI_APP_BUNDLE_ID.value:
                        huawei_push_object = HuaweiPushObject()
                        huawei_push_object.send_push_notify_message(**push_kwargs)
                    elif kwargs["appBundleId"] == ConstantEnum.VSEES_APP_BUNDLE_ID.value:
                        vsees_huawei_push_object = VseesHuaweiPushObject()
                        vsees_huawei_push_object.send_push_notify_message(**push_kwargs)
                elif push_type == 4:  # android xmpush
                    if kwargs['event_type'] in DATA_PUSH_EVENT_TYPE_LIST:
                        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 DATA_PUSH_EVENT_TYPE_LIST:
                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='', is_save_msg=0):
        """
        获取消息文本
        @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 设备算法事件标签
        @parm: is_save_msg 保存消息
        """
        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 = SYS_EVENT_TYPE_LIST
            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 == 702:
                msg_type = '摄像头休眠'
            elif event_type == 703:
                msg_type = '摄像头唤醒'
            elif event_type == 704:
                msg_type = '电量低'
            elif event_type == 706:
                msg_type = 'SD卡异常，请重新插拔或格式化SD卡。'
                if is_save_msg:
                    msg_type = 'SD卡异常，请重新插拔或格式化SD卡。如问题仍未解决，请联系在线客服。'

            elif event_type in DATA_PUSH_EVENT_TYPE_LIST:
                msg_type = '有人呼叫，请点击查看'
            if event_type not in DATA_PUSH_EVENT_TYPE_LIST 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 == 702:
                msg_type = 'Camera sleep'
            elif event_type == 703:
                msg_type = 'Camera wake'
            elif event_type == 704:
                msg_type = 'low battery'
            elif event_type == 706:
                msg_type = 'SD card is abnormal. Please re-insert or format the SD card.'
                if is_save_msg:
                    msg_type = 'SD card is abnormal, please re-insert or format the SD card. ' \
                               'If the problem is still not solved, please contact online customer service.'

            elif event_type in DATA_PUSH_EVENT_TYPE_LIST:
                msg_type = 'Someone is calling, please click to view'
            elif event_type not in DATA_PUSH_EVENT_TYPE_LIST 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 get_msg_text_v2(channel, n_time, lang, tz, event_type, electricity='', is_sys=0, dealings_type=0, ai_type=0,
                        device_type=0, event_tag='', is_save_msg=0, redis_obj=None):
        """
        获取消息文案V2
        """
        msg_type = ''
        try:
            event_type = int(event_type)
            device_type = int(device_type)
            event_list = []

            # 保持原有event_tag处理逻辑（完全不变）
            if event_tag:
                event_list = [int(event) for event in event_tag.split(',') if event]
                events_to_remove = SYS_EVENT_TYPE_LIST
                for event in events_to_remove:
                    if event in event_list:
                        event_list.remove(event)

            if event_type == 1022:
                msg_type = MSG_CONFIG[1022].get(lang, '')
            elif event_type == 1023:
                msg_type = MSG_CONFIG[1023].get(lang, '')

            elif event_type == 702:
                msg_type = MSG_CONFIG[702].get(lang, '')
            elif event_type == 703:
                msg_type = MSG_CONFIG[703].get(lang, '')
            elif event_type == 704:
                msg_type = MSG_CONFIG[704].get(lang, '')
            elif event_type == 706:
                if is_save_msg:
                    msg_type = MSG_CONFIG[706]['save'].get(lang, '')
                else:
                    msg_type = MSG_CONFIG[706]['default'].get(lang, '')

            elif event_type in DATA_PUSH_EVENT_TYPE_LIST:
                msg_type = MSG_CONFIG['data_push'].get(lang, '')

            # 组合事件处理文案
            if event_type not in DATA_PUSH_EVENT_TYPE_LIST and ai_type > 0 and event_list:
                msg_type = ''.join(
                    DevicePushService.get_event_type_text_v2(lang, item, dealings_type, redis_obj)
                    for item in event_list
                )
            elif not msg_type and event_type not in DATA_PUSH_EVENT_TYPE_LIST:
                # --- 新增代码：高频事件拦截 ---
                cache_key_tuple = (lang, event_type)
                if cache_key_tuple in DevicePushService._HIGH_FREQ_CACHE:
                    msg_type = DevicePushService._HIGH_FREQ_CACHE[cache_key_tuple]
                else:
                    msg_type = DevicePushService.get_event_type_text_v2(lang, event_type, None, redis_obj)

            if is_sys:
                if device_type in MULTI_CHANNEL_TYPE_LIST:
                    send_text = f'{msg_type} 通道:{channel}' if lang == 'cn' else f'{msg_type} channel:{channel}'
                else:
                    send_text = msg_type
            else:
                if device_type in MULTI_CHANNEL_TYPE_LIST:  # NVR多通道文案
                    channel_dict = {
                        'cn': '通道',  # 简体中文
                        'en': 'Channel',  # 英语
                        'es': 'Canal',  # 西班牙语
                        'fr': 'Canal',  # 法语
                        'de': 'Kanal ',  # 德语
                        'cn_tw': '通道',  # 繁体中文
                        'pt': 'Canal',  # 葡萄牙语
                        'ru': 'Канал',  # 俄语
                        'ja': 'チャンネル',  # 日语
                        'it': 'Canale',  # 意大利语
                        'pl': 'Kanał',  # 波兰语
                        'nl': 'Kanaal',  # 荷兰语
                        'ko': '채널',  # 韩语
                        'ar': 'قناة',  # 阿拉伯语
                        'vi': 'kênh',  # 越南语
                        'th': 'ช่องเสียง'  # 泰语
                    }
                    prefix = channel_dict.get(lang, channel_dict['en'])
                    send_text = f'{msg_type} {prefix}:{channel}'
                else:
                    send_text = f'{msg_type}'
            return send_text
        except Exception as e:
            ERROR_INFO_LOGGER.error('文案异常time:{},event_tag:{},error行数:{},内容:{}'
                                    .format(n_time, event_tag, e.__traceback__.tb_lineno, repr(e)))
            return msg_type

    @staticmethod
    def get_event_type_text_v2(lang, event_type, dealings_type, redis_obj):
        """ 改造后：优先读缓存，无缓存时读配置并写入 """
        msg = ''
        try:
            # 生成缓存键（示例：event_msg:en:66:1）
            cache_key = f"event_msg:{lang}:{event_type}"
            field = str(dealings_type) if dealings_type else 'default'

            # 1. 先尝试从 Redis 读取
            cached_msg = redis_obj.hget_data(cache_key, field)
            if cached_msg:
                return cached_msg

            # 2. 缓存未命中，从原配置读取
            config = EVENT_CONFIGS.get(lang, EVENT_CONFIGS['en'])
            default_msg = DEFAULT_TEXTS.get(lang, DEFAULT_TEXTS['en'])
            event_data = config.get(event_type)

            # 处理子类型
            if isinstance(event_data, dict):
                msg = event_data.get(dealings_type, default_msg)
            else:
                msg = event_data if event_data is not None else default_msg

            redis_obj.hset_data(cache_key, field, msg, 3600 * 24 * 7)
            return msg
        except Exception as e:
            ERROR_INFO_LOGGER.error('redis写入文案异常lang:{},event_type:{},error行数:{},内容:{}'
                                    .format(lang, event_type, e.__traceback__.tb_lineno, repr(e)))
            return msg


    @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]
            # 跳转类型
            jump_type = CommonService.get_jump_type(event_type)
            data = {
                'title': msg_title,
                'description': msg_text,
                'payload': 'payload',
                'restricted_package_name': appBundleId,
                'registration_id': token_val,
                'extra.channel_id': channel_id,
                'extra.alert': msg_text,
                '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,
                'extra.jump_type': jump_type
            }
            if event_type in DATA_PUSH_EVENT_TYPE_LIST:
                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)
            LOGGING.info('uid:{},时间:{}小米推送参数：{}'.format(uid, n_time, data))
            if response.status_code == 200:
                LOGGING.info('uid:{},时间:{}小米推送结果：{}'.format(uid, n_time, response.json()))
                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 DATA_PUSH_EVENT_TYPE_LIST:
                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'
            # 跳转类型
            jump_type = CommonService.get_jump_type(event_type)
            extra_data = {'alert': msg_text, 'msg': '', 'sound': 'sound.aif', 'zpush': '1',
                          'received_at': n_time, 'event_time': n_time, 'event_type': event_type,
                          'uid': uid, 'channel': channel, 'jump_type': jump_type}
            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 DATA_PUSH_EVENT_TYPE_LIST:
                    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:
                if appBundleId == ConstantEnum.ZOSI_APP_BUNDLE_ID.value:
                    huawei_push_object = HuaweiPushObject()
                    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)
                elif appBundleId == ConstantEnum.VSEES_APP_BUNDLE_ID.value:
                    vsees_huawei_push_object = VseesHuaweiPushObject()
                    push_result = vsees_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

    # 新增内存缓存专用字典（线程安全，无需锁）
    _HIGH_FREQ_CACHE = {
        # 格式: (lang, event_type, dealings_type) : msg
        ('cn', 51): '检测到画面变化',
        ('cn', 57): '有人出现',
        ('en', 51): 'Detected scene change',
        ('en', 57): 'Person detected',
        ('es', 51): 'Cambio de pantalla detectado',
        ('es', 57): 'Persona detectada',
        ('fr', 51): "Changement d'écran détecté",
        ('fr', 57): "Présence d'une personne",
        ('de', 51): "Bildschirmänderung erkannt",
        ('de', 57): "Jemand anwesend",
        ('cn_tw', 51): "檢測到畫面變化",
        ('cn_tw', 57): "有人出現",
        ('pt', 51): "Alteração do ecrã detectada",
        ('pt', 57): "Pessoa detectada",
        ('ru', 51): "Обнаружено изменение сцены",
        ('ru', 57): "Обнаружен человек",
        ('ja', 51): "画面変化を検知",
        ('ja', 57): "人を検知",
        ('it', 51): "Rilevato Cambio di Schermata",
        ('it', 57): "Presenza Umana",
        ('pl', 51): "Wykryto zmianę obrazu",
        ('pl', 57): "Wykryto osobę",
        ('nl', 51): "Schermverandering gedetecteerd",
        ('nl', 57): "Iemand aanwezig",
        ('ko', 51): "화면 변화 감지됨",
        ('ko', 57): "사람 출현",
        ('ar', 51): "تم اكتشاف تغيير في الصورة",
        ('ar', 57): "وجود شخص",
        ('vi', 51): "Phát hiện thay đổi hình ảnh",
        ('vi', 57): "Phát hiện người xuất hiện ",
        ('th', 51): "ตรวจพบการเปลี่ยนแปลงภาพ",
        ('th', 57): "ตรวจพบคนปรากฏตัว",
    }