# -*- 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.MessageConfig import EVENT_CONFIGS, DEFAULT_TEXTS, CHANNEL_MAP, 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 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.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 [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'] 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 [606, 607] 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'] ) # 补齐推送参数 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() # 过滤相同的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']) 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 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 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 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): """ 获取消息文案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) # 简体中文、英文、西班牙语、法语、德语、繁体中文、葡萄牙语、俄语、日语、意大利语、波兰语、韩语、阿拉伯语 SUPPORTED_LANGS = {'cn', 'en', 'es', 'fr', 'de', 'cn_tw', 'pt', 'ru', 'ja', 'it', 'pl', 'nl', 'ko', 'ar'} MIN_EVENT_TYPE, MAX_EVENT_TYPE = 51, 65 # 根据语言获取对应单个事件文案 if lang in SUPPORTED_LANGS and MIN_EVENT_TYPE <= event_type <= MAX_EVENT_TYPE: msg_type = MSG_CONFIG[event_type].get(lang, '') 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 == 705: if is_save_msg: msg_type = MSG_CONFIG[705]['save'].get(lang, '') else: msg_type = MSG_CONFIG[705]['default'].get(lang, '') elif event_type in EventTypeEnumObj.DATA_PUSH_EVENT_TYPE_LIST.value: msg_type = MSG_CONFIG['data_push'].get(lang, '') # 组合事件处理文案 if event_type not in EventTypeEnumObj.DATA_PUSH_EVENT_TYPE_LIST.value and ai_type > 0 and event_list: msg_type = ''.join( [DevicePushService.get_event_type_text_v2(lang, item, dealings_type) for item in event_list]) 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多通道文案 prefix = CHANNEL_MAP.get(lang, CHANNEL_MAP['en']) send_text = f'{msg_type} {prefix}:{channel}' else: send_text = f'{msg_type}' return send_text except Exception as e: ERROR_INFO_LOGGER.info('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): """统一处理多语言事件类型文案""" # 获取对应语言的配置（默认用英文） 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): return event_data.get(dealings_type, default_msg) elif event_data is not None: return event_data else: return default_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] # 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) 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 [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: 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