# -*- encoding: utf-8 -*- """ @File : DevicePushService.py @Time : 2022/11/23 11:40 @Author : stephen @Email : zhangdongming@asj6.wecom.work @Software: PyCharm """ import datetime import hashlib import json import logging import threading import time import boto3 import botocore import oss2 import requests from obs import ObsClient from AnsjerPush.Config.aiConfig import DEVICE_EVENT_TYPE, ALGORITHM_COMBO_TYPES from AnsjerPush.config import CONFIG_INFO, CONFIG_CN, MULTI_CHANNEL_TYPE_LIST, SYS_EVENT_TYPE_LIST, AWS_ACCESS_KEY_ID, \ AWS_SECRET_ACCESS_KEY, EVENT_DICT, EVENT_DICT_CN, CONFIG_TEST, HUAWEICLOUD_AK, HUAWEICLOUD_SK, \ HUAWEICLOUD_OBS_SERVER, HUAWEICLOUD_PUSH_BUKET, OSS_STS_ACCESS_KEY, OSS_STS_ACCESS_SECRET from AnsjerPush.config import XMPUSH_CONFIG, OPPOPUSH_CONFIG, XM_PUSH_CHANNEL_ID from Model.models import UidPushModel, SysMsgModel, DeviceSharePermission, DeviceChannelUserSet, \ DeviceChannelUserPermission, UidSetModel, Device_Info, UserAudioVideoPush, PushLog from Object.ETkObject import ETkObject from Object.OCIObjectStorage import OCIObjectStorage from Object.UidTokenObject import UidTokenObject from Object.utils import LocalDateTimeUtil from Service.CommonService import CommonService from Service.EquipmentInfoService import EquipmentInfoService, EQUIPMENT_INFO_DICT from Service.HuaweiPushService.HuaweiPushService import HuaweiPushObject from Service.PushService import PushObject from django.db import close_old_connections LOGGING = logging.getLogger('info') TIME_LOGGER = logging.getLogger('time') ERROR_INFO_LOGGER = logging.getLogger('error_info') class DevicePushService: @staticmethod def decode_uid(etk, uidToken): """ 解密UID,优先解密etk 否则判断uidToken """ # 解密获取uid if etk: eto = ETkObject(etk) uid = eto.uid else: uto = UidTokenObject(uidToken) uid = uto.UID return uid @staticmethod def judge_sys_msg(event_type): """ 判断是否属于系统消息 @param event_type: 事件类型 @return: bool """ if event_type in SYS_EVENT_TYPE_LIST: return True return False @staticmethod def get_s3_client(region): """ 根据地区获取S3 client @param region: 地区,1：国外, 2:国内 @return: aws_s3_client """ if int(region) == 1: aws_s3_client = boto3.client( 's3', aws_access_key_id=AWS_ACCESS_KEY_ID[1], aws_secret_access_key=AWS_SECRET_ACCESS_KEY[1], config=botocore.client.Config(signature_version='s3v4'), region_name='us-east-1' ) else: aws_s3_client = boto3.client( 's3', aws_access_key_id=AWS_ACCESS_KEY_ID[0], aws_secret_access_key=AWS_SECRET_ACCESS_KEY[0], config=botocore.client.Config(signature_version='s3v4'), region_name='cn-northwest-1' ) return aws_s3_client @classmethod def query_uid_push(cls, uid, event_type, button='1'): """ 查询uid_push和uid_set数据 @param uid: uid @param event_type: 事件类型 @param button: 按钮 @return: uid_push_qs """ if event_type not in [606, 607]: uid_push_qs = UidPushModel.objects.filter(uid_set__uid=uid, uid_set__detect_status=1) \ .exclude(token_val='0'). \ values('token_val', 'app_type', 'appBundleId', 'm_code', 'push_type', 'userID_id', 'userID__NickName', 'lang', 'm_code', 'tz', 'uid_set__nickname', 'uid_set__detect_interval', 'uid_set__detect_group', 'uid_set__channel', 'uid_set__ai_type', 'uid_set__device_type', 'uid_set__new_detect_interval', 'uid_set__msg_notify') else: # 一键通话只推主用户 device_info_qs = Device_Info.objects.filter(UID=uid).values('vodPrimaryUserID') primary_user_id = device_info_qs[0]['vodPrimaryUserID'] if event_type == 607: # 音视频通话根据用户按钮来推送 button_qs = UserAudioVideoPush.objects.filter(uid=uid).values('buttonUser1', 'buttonUser2') if button_qs.exists(): primary_user_id = button_qs[0]['buttonUser1'] if button == '1' else button_qs[0]['buttonUser2'] uid_push_qs = UidPushModel.objects.filter(uid_set__uid=uid, userID_id=primary_user_id) \ .exclude(token_val='0'). \ values('token_val', 'app_type', 'appBundleId', 'm_code', 'push_type', 'userID_id', 'userID__NickName', 'lang', 'm_code', 'tz', 'uid_set__nickname', 'uid_set__detect_interval', 'uid_set__detect_group', 'uid_set__channel', 'uid_set__ai_type', 'uid_set__device_type', 'uid_set__new_detect_interval', 'uid_set__msg_notify', 'jg_token_val') return uid_push_qs @classmethod def get_uid_push_by_uid(cls, uid): """ 根据uid获取设备推送（目前仅用在OZI定制客户） """ device_info_qs = Device_Info.objects.filter(UID=uid).values('vodPrimaryUserID') primary_user_id = device_info_qs[0]['vodPrimaryUserID'] uid_push_qs = UidPushModel.objects.filter(uid_set__uid=uid, userID_id=primary_user_id) \ .exclude(token_val='0'). \ values('token_val', 'app_type', 'appBundleId', 'm_code', 'push_type', 'userID_id', 'userID__NickName', 'lang', 'm_code', 'tz', 'uid_set__nickname', 'uid_set__detect_interval', 'uid_set__detect_group', 'uid_set__channel', 'uid_set__ai_type', 'uid_set__device_type', 'uid_set__new_detect_interval', 'uid_set__msg_notify', 'jg_token_val') return uid_push_qs @staticmethod def qs_to_list(qs): """ qs对象转存列表 @param qs: query set对象 @return: qs_list """ qs_list = [] for i in qs: qs_list.append(i) return qs_list @staticmethod def cache_push_detect_interval(redis_obj, name, detect_interval, new_detect_interval): """ 缓存设置推送消息的时间间隔 @param redis_obj: redis对象 @param name: redis key @param detect_interval: 原推送时间间隔 @param new_detect_interval: 新推送时间间隔 """ if CONFIG_INFO != CONFIG_CN: detect_interval = new_detect_interval if new_detect_interval > 0 else detect_interval detect_interval = 60 if detect_interval < 60 else detect_interval else: # 国内推送兼容问题，有值并且大于旧消息间隔则使用new_detect_interval detect_interval = new_detect_interval if new_detect_interval > detect_interval else detect_interval redis_obj.set_data(key=name, val=1, expire=detect_interval - 5) @classmethod def push_msg(cls, **params): """ 推送消息 @param params: 推送参数 @return: bool """ try: uid = params['uid'] params['event_tag'] = cls.get_event_tag(params['ai_type'], params['event_type'], params['detection']) is_app_push = True if params['event_type'] in [606, 607] else \ cls.is_send_app_push( params['event_type'], params['event_tag'], params['app_push_config'], params['app_push'], uid, params['channel']) # 低功耗产品推送，休眠702,低电量704提醒,1023 ozi, 并且detection=0，0标识单事件类型，1标识多事件类型 is_app_push = True if params['event_type'] in [702, 704, 1022, 1023] and params[ 'detection'] == 0 else is_app_push redis_obj = params['redis_obj'] # 推送 if is_app_push: msg_key = 'PUSH:MSG:IMAGE:{}:{}:{}'.format(params['uid'], params['channel'], params['n_time']) d_params = {'is_st': params['is_st'], 'storage_location': params['storage_location'], 'event_tag': params['event_tag'], 'event_type': params['event_type']} redis_obj.set_data(msg_key, json.dumps(d_params), 60) push_kwargs = params['push_kwargs'] for up in params['uid_set_push_list']: push_type = up['push_type'] lang = up['lang'] tz = up['tz'] if tz is None or tz == '': tz = 0 if params['event_type'] in [606, 607] and push_type in [5, 6]: push_kwargs['jg_token_val'] = up['jg_token_val'] else: if 'jg_token_val' in push_kwargs: push_kwargs.pop('jg_token_val') appBundleId = up['appBundleId'] token_val = up['token_val'] # 发送标题 msg_title = cls.get_msg_title(nickname=params['nickname']) # 发送内容 msg_text = cls.get_msg_text(channel=params['channel'], n_time=params['n_time'], lang=lang, tz=tz, event_type=params['event_type'], ai_type=params['ai_type'], device_type=params['device_type'], electricity=params['electricity'], dealings_type=params['dealings_type'], event_tag=params['event_tag'] ) # 补齐推送参数 push_kwargs['appBundleId'] = appBundleId push_kwargs['token_val'] = token_val push_kwargs['msg_title'] = msg_title push_kwargs['msg_text'] = msg_text params['push_kwargs'] = push_kwargs params['appBundleId'] = appBundleId params['token_val'] = token_val params['lang'] = lang params['tz'] = tz params['push_type'] = push_type params['redis_obj'] = redis_obj # GlobalThreadPool().submit(cls.send_app_msg_push, **params) push_thread = threading.Thread( target=cls.send_app_msg_push, kwargs=params ) push_thread.start() except Exception as e: ERROR_INFO_LOGGER.info( '推送消息线程异常,uid:{},error_line:{},error_msg:{}' .format(params['uid'], e.__traceback__.tb_lineno, repr(e))) @classmethod def save_msg_push(cls, **params): """ 保存推送数据和推送消息 @param params: 推送参数 @return: bool """ sys_msg_list = [] saved_user_id_list = [] uid = params['uid'] now_time = int(time.time()) redis_obj = params['redis_obj'] try: params['event_tag'] = cls.get_event_tag(params['ai_type'], params['event_type'], params['detection']) save_equipment_info = False equipment_info_key = EquipmentInfoService.randoms_choice_equipment_info_key() for up in params['uid_set_push_list']: lang = up['lang'] tz = up['tz'] if tz is None or tz == '': tz = 0 # 保存系统消息或推送消息数据 user_id = up['userID_id'] if user_id not in saved_user_id_list: # 防止同一用户重复写入数据 # 系统消息 if params['is_sys_msg']: sys_msg_text = cls.get_msg_text(channel=params['channel'], n_time=params['n_time'], lang=lang, tz=tz, is_sys=1, device_type=params['device_type'], event_type=params['event_type'], electricity=params['electricity']) sys_msg_list.append(SysMsgModel(userID_id=user_id, msg=sys_msg_text, addTime=now_time, updTime=now_time, uid=uid, eventType=params['event_type'])) # 保存推送消息 else: if not save_equipment_info: save_equipment_info = True params['userID_id'] = user_id answer_status = 1 if params['dealings_type'] == 1 else 0 equipment_info_kwargs = { 'device_user_id': params['userID_id'], 'event_time': params['n_time'], 'event_type': params['event_type'], 'device_uid': params['uid'], 'device_nick_name': params['nickname'], 'channel': params['channel'], 'alarm': 'Motion \tChannel:{}'.format(params['channel']), 'is_st': params['is_st'], 'add_time': int(time.time()), 'storage_location': params['storage_location'], 'event_tag': params['event_tag'], 'answer_status': answer_status } # 保存到redis列表 equipment_info_value = json.dumps(equipment_info_kwargs) redis_obj.rpush(equipment_info_key, equipment_info_value) saved_user_id_list.append(user_id) close_old_connections() # 写入系统消息 if sys_msg_list: SysMsgModel.objects.bulk_create(sys_msg_list) if save_equipment_info: equipment_info_list = [] equipment_info_model = EQUIPMENT_INFO_DICT[equipment_info_key] # 一键通话和视频通话需要实时写入数据 # 正式服通过定时任务批量写入数据 if params['event_type'] in [606, 607] or CONFIG_INFO == CONFIG_TEST: end = 0 # 缓存数据多于100条,批量保存前100条,否则保存全部 equipment_info_len = redis_obj.llen(equipment_info_key) end = 99 if equipment_info_len > 100 else equipment_info_len - 1 if end != 0: equipment_info_redis_list = redis_obj.lrange(equipment_info_key, 0, end) redis_obj.ltrim(equipment_info_key, end + 1, -1) for equipment_info in equipment_info_redis_list: equipment_info_data = eval(equipment_info) # 设备昵称存在表情,解码utf-8 if equipment_info_data.get('device_nick_name') is not None: equipment_info_data['device_nick_name'] = equipment_info_data['device_nick_name']. \ encode('UTF-8', 'ignore').decode('UTF-8') equipment_info_list.append(equipment_info_model(**equipment_info_data)) equipment_info_model.objects.bulk_create(equipment_info_list) return True except Exception as e: ERROR_INFO_LOGGER.info( '保存推送数据和推送消息线程异常,uid:{}, error_line:{}, error_msg:{}'. format(uid, e.__traceback__.tb_lineno, repr(e))) return False @classmethod def get_event_tag(cls, ai_type, event_type, detection=0): """ 获取事件标签 """ algorithm = False if ai_type > 0 and detection == 1: algorithm = True elif (ai_type == 7 and event_type <= 7) or (ai_type == 47 and event_type <= 47) or (detection == 1): algorithm = True if not algorithm: return ',' + str(event_type) + ',' event_res = DEVICE_EVENT_TYPE.get(event_type, 0) if event_res > 0: return ',' + str(event_res) + ',' event_type = cls.dec_to_bin(event_type) types = cls.get_combo_types(event_type) res = ','.join(types) + ',' return ',' + res @classmethod def get_combo_types(cls, event_type): """ 获取设备算法组合类型 51：移动侦测，52：传感器报警，53：影像遗失，54：PIR，55：门磁报警，56：外部发报，57：人型报警（提示：有人出现），58：车型，59：宠物，60：人脸，61：异响， 62：区域闯入，63：区域闯出，64：长时间无人检测，65：长时间无人检测，66：往来检测，67：哭声检测，68：手势检测,69:火焰检测 0：代表空字符，702：摄像头休眠，703：摄像头唤醒，704：电量过低 AWS AI识别 1：人形，2：车型，3：宠物，4：包裹。云端AI类型 """ try: types = [] event_type = str(event_type) len_type = len(event_type) for i in range(len_type): e_type = event_type[len_type - 1 - i] if e_type == '1': types.append(str(ALGORITHM_COMBO_TYPES[i])) return types except Exception as e: LOGGING.info('推送错误异常,errLine:{}, errMsg:{}'.format(e.__traceback__.tb_lineno, repr(e))) return event_type @staticmethod def dec_to_bin(num): """ 十进制转二进制 """ result = "" while num != 0: ret = num % 2 num //= 2 result = str(ret) + result return result @classmethod def send_app_msg_push(cls, **kwargs): """ 发送推送 @kwargs : @return push_result: bool """ try: push_type = kwargs['push_type'] push_kwargs = kwargs['push_kwargs'] push_result = False uid = kwargs['uid'] # is_st为1或3,且推送类型为apns,gcm,华为,异步推送图片 if (kwargs['is_st'] == 1 or kwargs['is_st'] == 3) and \ (push_type == 0 or push_type == 1 or push_type == 3): if kwargs['is_st'] == 1: key = '{}/{}/{}.jpeg'.format(kwargs['uid'], kwargs['channel'], kwargs['n_time']) else: key = '{}/{}/{}_0.jpeg'.format(kwargs['uid'], kwargs['channel'], kwargs['n_time']) # 开始异步推送图片 # thread_pool = GlobalThreadPool() # thread_pool.submit(cls.async_send_picture_push, ( # push_type, kwargs['aws_s3_client'], kwargs['bucket'], key, # kwargs['uid'], kwargs['appBundleId'], kwargs['token_val'], kwargs['event_type'], kwargs['n_time'], # push_kwargs['msg_title'], push_kwargs['msg_text'], kwargs['channel'], kwargs['storage_location'])) push_thread = threading.Thread(target=cls.async_send_picture_push, args=( push_type, kwargs['aws_s3_client'], kwargs['bucket'], key, kwargs['uid'], kwargs['appBundleId'], kwargs['token_val'], kwargs['event_type'], kwargs['n_time'], push_kwargs['msg_title'], push_kwargs['msg_text'], kwargs['channel'], kwargs['storage_location'], kwargs['redis_obj'])) push_thread.start() push_result = True # 不推图 else: if push_type in [0, 1, 2]: kwargs = { 'nickname': kwargs['uid'], 'app_bundle_id': kwargs['appBundleId'], 'token_val': kwargs['token_val'], 'n_time': kwargs['n_time'], 'event_type': kwargs['event_type'], 'msg_title': push_kwargs['msg_title'], 'msg_text': push_kwargs['msg_text'], 'uid': kwargs['uid'], 'channel': kwargs['channel'] } if push_type == 0: # ios apns push_result = PushObject.ios_apns_push(**kwargs) elif push_type == 1: # android gcm push_result = PushObject.android_fcm_push_v1(**kwargs) elif push_type == 2: # android jpush kwargs.pop('uid') push_result = PushObject.android_jpush(**kwargs) elif push_type == 3: huawei_push_object = HuaweiPushObject() huawei_push_object.send_push_notify_message(**push_kwargs) elif push_type == 4: # android xmpush if kwargs['event_type'] in [606, 607]: channel_id = 111934 else: channel_id = 104551 push_result = cls.do_xmpush(channel_id=channel_id, **push_kwargs) elif push_type == 5: # android vivopush push_result = PushObject.android_vivopush(**push_kwargs) elif push_type == 6: # android oppopush channel_id = 'DEVICE_REMINDER' push_result = cls.do_oppopush(channel_id=channel_id, **push_kwargs) elif push_type == 7: # android meizupush push_result = PushObject.android_meizupush(**push_kwargs) elif push_type == 8: # android honorpush push_result = PushObject.android_honorpush(**push_kwargs) if kwargs['event_type'] in [606, 607]: close_old_connections() # 写入日志表 PushLog.objects.create(uid=uid, event_type=kwargs['event_type'], created_time=int(time.time()), content=push_kwargs, push_result=push_result, push_type=push_type) return push_result except Exception as e: ERROR_INFO_LOGGER.info('发送推送线程异常uid:{},type:{},error_line:{},error_msg:{}' .format(kwargs['uid'], kwargs['push_type'], e.__traceback__.tb_lineno, repr(e))) return False @staticmethod def get_msg_title(nickname): """ 获取消息标题 """"" return nickname @staticmethod def get_event_type_text(lang, event_type, dealings_type): """ 事件类型文案键值查找 """ if lang == 'cn': if event_type in EVENT_DICT_CN: if isinstance(EVENT_DICT_CN[event_type], dict): msg_type = EVENT_DICT_CN[event_type][dealings_type] else: msg_type = EVENT_DICT_CN[event_type] else: msg_type = '未知事件类型 ' return msg_type else: if event_type in EVENT_DICT: if isinstance(EVENT_DICT[event_type], dict): msg_type = EVENT_DICT[event_type][dealings_type] else: msg_type = EVENT_DICT[event_type] else: msg_type = 'Unknown event type' return msg_type @staticmethod def get_msg_text(channel, n_time, lang, tz, event_type, electricity='', is_sys=0, dealings_type=0, ai_type=0, device_type=0, event_tag=''): """ 获取消息文本 @param: channel 通道号 @param: n_time 触发事件 @param: lang 语言 @param: tz 时区 @param: event_type 事件类型 @param: electricity 电量 @param: is_sys 是否系统消息 @param: dealings_type 往来类型 1 进 1 离开 @param: ai_type 设备本地AI只能算法 事件类型 @param: device_type 设备类型 @param: event_tag 设备算法事件标签 """ msg_type = '' event_type = int(event_type) device_type = int(device_type) event_list = [] if event_tag: event_list = [int(event) for event in event_tag.split(',') if event] events_to_remove = [702, 703, 704] for event in events_to_remove: if event in event_list: event_list.remove(event) if lang == 'cn': if event_type == 51: msg_type = '检测到画面变化' elif event_type == 52: msg_type = '传感器报警' elif event_type == 53: msg_type = '影像遗失' elif event_type == 54: msg_type = 'PIR' elif event_type == 55: msg_type = '门磁报警' elif event_type == 56: msg_type = '外部发报' elif event_type == 57: msg_type = '有人出现' elif event_type == 58: msg_type = '有车出现' elif event_type == 59: msg_type = '有宠物出现' elif event_type == 60: msg_type = '发现人脸' elif event_type == 61: msg_type = '有异响' elif event_type == 62: msg_type = '区域闯入' elif event_type == 63: msg_type = '区域闯出' elif event_type == 64: msg_type = '有人徘徊' elif event_type == 65: msg_type = '长时间无人出现' elif event_type == 1022: msg_type = '有人按下门铃' elif event_type == 1023: msg_type = '儿童保护模式开启' elif event_type == 704: msg_type = '电量低' elif event_type == 702: msg_type = '摄像头休眠' elif event_type == 703: msg_type = '摄像头唤醒' elif event_type in [606, 607]: msg_type = '有人呼叫，请点击查看' if event_type not in [606, 607, 1022] and ai_type > 0 and event_list: msg_type = ''.join([DevicePushService.get_event_type_text(lang, item, dealings_type) for item in event_list]) if is_sys: if device_type in MULTI_CHANNEL_TYPE_LIST: send_text = '{} 通道:{}'.format(msg_type, channel) else: send_text = msg_type else: if device_type in MULTI_CHANNEL_TYPE_LIST: send_text = '{} 通道:{}'.format(msg_type, channel) else: send_text = '{}'.format(msg_type) else: if event_type == 51: msg_type = 'Screen change detected' elif event_type == 52: msg_type = 'Sensor alarms' elif event_type == 53: msg_type = 'Lost images' elif event_type == 54: msg_type = 'PIR' elif event_type == 55: msg_type = 'Door magnetic alarm' elif event_type == 56: msg_type = 'External reporting' elif event_type == 57: msg_type = 'Person detected' elif event_type == 58: msg_type = 'Vehicle detected' elif event_type == 59: msg_type = 'Pet detected' elif event_type == 60: msg_type = 'Human face detected' elif event_type == 61: msg_type = 'Abnormal sound detected' elif event_type == 62: msg_type = 'Intrusion detected in the area' elif event_type == 63: msg_type = 'Area vacated' elif event_type == 64: msg_type = 'Loitering detected' elif event_type == 65: msg_type = 'No appearance for a long time' elif event_type == 1022: msg_type = 'Someone rang the doorbell' elif event_type == 1023: msg_type = 'Child protection mode is enabled' elif event_type == 704: msg_type = 'low battery' elif event_type == 702: msg_type = 'Camera sleep' elif event_type == 703: msg_type = 'Camera wake' elif event_type in [606, 607]: msg_type = 'Someone is calling, please click to view' if event_type not in [606, 607, 1022] and ai_type > 0 and event_list: msg_type = ''.join([DevicePushService.get_event_type_text(lang, item, dealings_type) for item in event_list]) if is_sys: if device_type in MULTI_CHANNEL_TYPE_LIST: send_text = '{} channel:{}'.format(msg_type, channel) else: send_text = msg_type else: if device_type in MULTI_CHANNEL_TYPE_LIST: send_text = '{} channel:{}'.format(msg_type, channel) else: send_text = '{}'.format(msg_type) return send_text @staticmethod def do_xmpush(channel_id, uid, channel, appBundleId, token_val, event_type, n_time, msg_title, msg_text): """ android 国内小米APP消息提醒推送 """ try: url = 'https://api.xmpush.xiaomi.com/v3/message/regid' app_secret = XMPUSH_CONFIG[appBundleId] # payload = {'alert': 'Motion', 'msg': '', 'sound': 'sound.aif', 'zpush': '1', # 'received_at': n_time, 'event_time': n_time, 'event_type': event_type, # 'uid': uid, 'channel': channel # } data = { 'title': msg_title, 'description': msg_text, 'payload': 'payload', 'restricted_package_name': appBundleId, 'registration_id': token_val, 'extra.channel_id': channel_id, 'extra.alert': 'Motion', 'extra.msg': '', 'extra.sound': 'sound.aif', 'extra.zpush': '1', 'extra.received_at': n_time, 'extra.event_time': n_time, 'extra.event_type': event_type, 'extra.uid': uid, 'extra.channel': channel, } if event_type in [606, 607]: data['extra.sound_uri'] = 'android.resource://com.ansjer.zccloud_ab/raw/phone_call' headers = { 'Authorization': 'key={}'.format(app_secret) } response = requests.post(url, data=data, headers=headers) if response.status_code == 200: return True ERROR_INFO_LOGGER.info('小米推送异常,uid:{},状态码{},时间:{},结果:{},' .format(uid, response.status_code, n_time, response.json())) return False except Exception as e: ERROR_INFO_LOGGER.info('小米推送异常,uid:{},时间:{},error_line:{},error_msg:{}'. format(uid, n_time, e.__traceback__.tb_lineno, repr(e))) return False @staticmethod def do_oppopush(channel_id, uid, channel, appBundleId, token_val, event_type, n_time, msg_title, msg_text, jg_token_val=''): """ android 国内oppo APP消息提醒推送 """ try: if event_type in [606, 607]: channel_id = XM_PUSH_CHANNEL_ID['push_to_talk'] app_key = OPPOPUSH_CONFIG[appBundleId]['Key'] master_secret = OPPOPUSH_CONFIG[appBundleId]['Secret'] url = 'https://api.push.oppomobile.com/' now_time = str(round(time.time() * 1000)) # 1、实例化一个sha256对象 sha256 = hashlib.sha256() # 2、调用update方法进行加密 sha256.update((app_key + now_time + master_secret).encode('utf-8')) # 3、调用hexdigest方法,获取加密结果 sign = sha256.hexdigest() # 获取auth_token get_token_url = url + 'server/v1/auth' post_data = { 'app_key': app_key, 'sign': sign, 'timestamp': now_time } headers = {'Content-Type': 'application/x-www-form-urlencoded'} response = requests.post(get_token_url, data=post_data, headers=headers) result = response.json() # 发送推送 push_url = url + 'server/v1/message/notification/unicast' extra_data = {'alert': 'Motion', 'msg': '', 'sound': 'sound.aif', 'zpush': '1', 'received_at': n_time, 'event_time': n_time, 'event_type': event_type, 'uid': uid, 'channel': channel} message = { "target_type": 2, "target_value": token_val, "notification": { "title": msg_title, "content": msg_text, 'channel_id': channel_id, 'action_parameters': extra_data, 'click_action_type': 4, 'click_action_activity': 'com.ansjer.zccloud_a.AJ_MainView.AJ_Home.AJMainActivity' } } push_data = { 'auth_token': result['data']['auth_token'], 'message': json.dumps(message) } response = requests.post(push_url, data=push_data, headers=headers) if response.status_code == 200: if event_type in [606, 607]: PushObject.jpush_transparent_transmission(msg_title, msg_text, appBundleId, jg_token_val, extra_data) return True return False except Exception as e: ERROR_INFO_LOGGER.info('oppo推送异常,uid:{},time:{},error_line:{},error_msg:{}'. format(uid, n_time, e.__traceback__.tb_lineno, repr(e))) return False @classmethod def async_send_picture_push(cls, push_type, aws_s3_client, bucket, key, uid, appBundleId, token_val, event_type, n_time, msg_title, msg_text, channel, storage_reg, redis_obj): """ 异步推送图片 """ try: if storage_reg == 1: # 阿里云 auth = oss2.Auth(OSS_STS_ACCESS_KEY, OSS_STS_ACCESS_SECRET) oss_img_bucket = oss2.Bucket(auth, 'oss-cn-shenzhen.aliyuncs.com', 'apg') image_url = oss_img_bucket.sign_url('GET', key, 300) elif storage_reg in [3, 4]: image_url = DevicePushService.oci_object_url(uid, redis_obj, storage_reg, bucket, key) elif storage_reg == 5: image_url = DevicePushService.create_obs_signed_url(key, 'GET') else: image_url = aws_s3_client.generate_presigned_url( 'get_object', Params={'Bucket': bucket, 'Key': key}, ExpiresIn=3600) push_result = False if push_type == 0: push_result = PushObject.ios_apns_push( uid, appBundleId, token_val, n_time, event_type, msg_title, msg_text, uid, channel, image_url) elif push_type == 1: push_result = PushObject.android_fcm_push_v1( uid, appBundleId, token_val, n_time, event_type, msg_title, msg_text, uid, channel, image_url) elif push_type == 3: huawei_push_object = HuaweiPushObject() push_result = huawei_push_object.send_push_notify_message( token_val=token_val, msg_title=msg_title, msg_text=msg_text, uid=uid, event_type=event_type, n_time=n_time, image_url=image_url, channel=channel) TIME_LOGGER.info('{}推送图片,push_type:{},推送结果:{}'.format(uid, push_type, push_result)) except Exception as e: ERROR_INFO_LOGGER.error( '异步推送图片异常,error_line:{},error_msg:{}'.format(e.__traceback__.tb_lineno, repr(e))) @staticmethod def oci_object_url(uid, redis_obj, storage_location, bucket, obj_name): """ 获取OCI对象存储URL 有效期5分钟 @param uid: uid @param redis_obj: 缓存客户端 @param storage_location: 存储区域 @param bucket: 存储桶 @param obj_name: 对象名称 @return: url """ try: uid_key = f'PUSH:PICTURE:OCI:URL:{uid}' oci_url = redis_obj.get_data(uid_key) if oci_url: return oci_url + obj_name oci = OCIObjectStorage('eur' if storage_location == 4 else 'us') prefix_name = f'{uid}/' time_expires = datetime.datetime.utcnow() + datetime.timedelta(minutes=60) result = oci.get_preauthenticated_request_url(bucket, 'ociPush', prefix_name, time_expires, 'AnyObjectRead') # 授权到指定uid文件夹 full_url = result.full_path if result else '' redis_obj.set_data(uid_key, full_url, 3580) return full_url + obj_name except Exception as e: LOGGING.error('oci查询消息列表异常error_line:{}, error_msg:{}'.format(e.__traceback__.tb_lineno, repr(e))) return '' @staticmethod def create_oci_req_url(storage_location, bucket, obj_name, oci=None): """ 创建oci预认证请求url """ if not oci: region = 'eur' if storage_location == 4 else 'us' oci = OCIObjectStorage(region) time_expires = datetime.datetime.utcnow() + datetime.timedelta(minutes=60) result = oci.create_ereauthenticated_request(bucket, 'ociPush', obj_name, time_expires) if not result: return '' return result.full_path + result.object_name @staticmethod def get_res_data(**kwargs): """ 获取响应数据 @return: res_data """ res_data = {'code': 0, 'msg': 'success'} is_st = kwargs['is_st'] storage_location = kwargs['storage_location'] if is_st == 0 or is_st == 2: res_data['msg'] = 'success 0 or 2' elif is_st == 1: key_name = '{}/{}/{}.jpeg'.format(kwargs['uid'], kwargs['channel'], kwargs['n_time']) params = {'Key': key_name} if kwargs['region'] == 2: # 2:国内 params['Bucket'] = 'push' else: # 1：国外 params['Bucket'] = 'foreignpush' # 根据存储区域返回链接 if storage_location in [3, 4]: # OCI img_url = DevicePushService.create_oci_req_url(storage_location, params['Bucket'], key_name) res_data['img_push'] = img_url res_data['msg'] = 'success 1' elif storage_location == 2: # AWS img_url = DevicePushService.generate_s3_url(kwargs['aws_s3_client'], params) res_data['img_push'] = img_url else: # 华为云 img_url = DevicePushService.create_obs_signed_url(key_name, 'PUT') res_data['img_push'] = img_url res_data['msg'] = 'success 1' elif is_st == 3: img_url_list = [] if kwargs['region'] == 2: # 2:国内 params = {'Bucket': 'push'} else: # 1：国外 params = {'Bucket': 'foreignpush'} oci_client = None if storage_location in [3, 4]: # 三张图的时候提前获取实例化OCI region = 'eur' if storage_location == 4 else 'us' oci_client = OCIObjectStorage(region) for i in range(kwargs['is_st']): key_name = '{}/{}/{}_{}.jpeg'.format(kwargs['uid'], kwargs['channel'], kwargs['n_time'], i) params['Key'] = key_name if storage_location in [3, 4]: # OCI img_url = DevicePushService.create_oci_req_url( storage_location, params['Bucket'], key_name, oci_client) elif storage_location == 2: # AWS img_url = DevicePushService.generate_s3_url(kwargs['aws_s3_client'], params) else: # 华为云 img_url = DevicePushService.create_obs_signed_url(key_name, 'PUT') img_url_list.append(img_url) res_data['img_url_list'] = img_url_list res_data['msg'] = 'success 3' return res_data @staticmethod def generate_s3_url(aws_s3_client, params): """ 获取S3对象URL """ response_url = aws_s3_client.generate_presigned_url( ClientMethod='put_object', Params=params, ExpiresIn=3600 ) return response_url @staticmethod def create_obs_signed_url(key_name, method): """ 生成对象存储预签名URL @param key_name: 对象名称 @param method: 方法 @return: 预签名URL """ obs_client = ObsClient( access_key_id=HUAWEICLOUD_AK, secret_access_key=HUAWEICLOUD_SK, server=HUAWEICLOUD_OBS_SERVER) res = obs_client.createSignedUrl( method=method, bucketName=HUAWEICLOUD_PUSH_BUKET, objectKey=key_name, expires=3600) return res.signedUrl @staticmethod def check_share_permission(user_id, channel, uid): """ 检查用户是否有权限接收设备报警推送 """ user_permission_qs = DeviceChannelUserSet.objects.filter(user_id=user_id, uid=uid) \ .values('id', 'channels') # 根据当前用户与uid查询是否设置过通道权限，不存在则不是分享设备 if not user_permission_qs.exists(): return True up_id = user_permission_qs[0]['id'] channels = user_permission_qs[0]['channels'] channels_list = [int(val) for val in channels.split(',')] # 当前uid是属于分享设备并且设置了权限 # 判断通道是否设置了权限，不存在则当前通道没有权限接受消息推送 if int(channel) not in channels_list: return False permission_qs = DeviceSharePermission.objects.filter(code='AlarmMessages').values('id') p_id = permission_qs[0]['id'] # 当前通道存在设置则查看是否有 消息推送权限 channel_permission_qs = DeviceChannelUserPermission.objects \ .filter(channel_user_id=up_id, permission_id=p_id) \ .values('permission_id', 'channel_user_id') if not channel_permission_qs.exists(): return False return True @classmethod def is_algorithm_type(cls, uid, event_type): """ 判断是否是算法类型 62、63、64、65、66不限制推送 """ uid_set_qs = UidSetModel.objects.filter(uid=uid).values('ai_type') if not uid_set_qs.exists(): return False if uid_set_qs[0]['ai_type'] == 0: return False event_types = [62, 63, 64, 65, 66] event_res = DEVICE_EVENT_TYPE.get(event_type, 0) if event_res in event_types: return True event_types2 = cls.get_combo_types(event_type) if not event_types2: return False c = [x for x in event_types if x in event_types2] return True if c else False @staticmethod def is_send_app_push(event_type, event_tag, app_push_config, msg_interval=None, uid=None, channel=1): """ 是否进行APP消息提醒 @return: True|False """ try: if not app_push_config: return True is_push = app_push_config['appPush'] if is_push != 1: # 1：进行APP提醒，其它则不执行APP提醒 return False if msg_interval: # 存在消息间隔数据缓存 不推送APP消息 return False if 'nvr' in app_push_config: push = DevicePushService.is_msg_push_nvr(event_type, int(channel), app_push_config) TIME_LOGGER.info(f'uid:{uid}NVR推送结果:{push}') return push all_day = app_push_config['pushTime']['allDay'] # 允许设备类型APP提醒列表 app_event_types = app_push_config['eventTypes']['device'] if all_day == 0: # 1：全天提醒，0：自定义时间提醒 push_time_config = app_push_config['pushTime'] # 计算当前时间是否在自定义消息提醒范围内 if not DevicePushService.is_push_notify_allowed_now(push_time_config): return False # APP接收提醒，判断识别类型是否勾选提醒 push_result = DevicePushService.is_type_push(event_type, event_tag, app_event_types) return push_result except Exception as e: LOGGING.info( '{}判断是否执行APP推送异常,errLine:{}, errMsg:{}'.format(uid, e.__traceback__.tb_lineno, repr(e))) return True @staticmethod def is_msg_push_nvr(event_type, channel, config): try: # 获取 NVR 配置中记录，默认为空列表 nvr_entries = config.get('nvr', []) # 构建通道键，例如 'channel1' 或 'channel2' channel_key = f'channel{channel}' if not nvr_entries: return True # 查找给定通道 matching_entry = next((entry for entry in nvr_entries if channel_key in entry), None) # 如果没有找到匹配项，返回 True，表示可以推送 if not matching_entry: return True # 获取与 channel_key 对应的 event_types event_types = matching_entry[channel_key] if channel_key in matching_entry else [] # 调用 is_type_push 方法检查事件类型是否可以推送 return DevicePushService.is_type_push(event_type, None, event_types) except Exception as e: # 记录异常信息并返回 False ERROR_INFO_LOGGER.error( 'NVRAPP推送异常, errLine: {}, errMsg: {}'.format(e.__traceback__.tb_lineno, repr(e))) return False @staticmethod def is_type_push(event_type, event_tag, app_event_types): # 检查事件标签和应用事件类型是否都存在 if event_tag and app_event_types: # 将事件标签按逗号分割成列表，并转换为整数类型 tag_list = [int(event) for event in event_tag.split(',') if event] # 判断是否有任一标签允许应用提醒 return any(item in app_event_types for item in tag_list) # 检查事件类型和用户所选事件类型是否都存在，并判断事件类型在用户所选事件类型列表中 return event_type and app_event_types and event_type in app_event_types @staticmethod def is_push_notify_allowed_now(push_time_config): """ 判断当前时间是否在允许APP推送提醒 """ now_time = int(time.time()) start_time = push_time_config['startTime'] end_time = push_time_config['endTime'] repeat = push_time_config['repeat'] tz = push_time_config['timeZone'] # 获取当前日期和周几 now_date, week = DevicePushService.get_now_date_and_week(now_time, tz) # 判断是否在重复日范围内 if not DevicePushService.is_repeated(week, repeat): return False # 计算当前日期在一天中的秒数 seconds = LocalDateTimeUtil.convert_time_to_seconds(now_date) # 判断是否在APP推送提醒范围内 return DevicePushService.is_in_effect(start_time, end_time, seconds) @staticmethod def is_in_effect(start, end, now_seconds): """ 判断是否在提醒时间范围内 @params: 开始时间秒 @params: 结束时间秒 @params: 当前时间秒 @return: 当前时间是在范围内返回True 否则False """ if start < end: return start <= now_seconds <= end else: return start <= now_seconds or now_seconds <= end @staticmethod def is_repeated(week_day, repeat_day): """ 判断是否重复日 @params: week_day 周几 @params: 重复日1-127 @return: 如果当前日期在重复日则返回True否则False """ # 判断对应位置上的值是否为 1 is_repeat = (repeat_day >> (week_day - 1)) & 1 == 1 return is_repeat @staticmethod def get_now_date_and_week(now_time, tz): now_data = CommonService.get_now_time_str(now_time, tz, 'cn') week = LocalDateTimeUtil.date_to_week(now_data) return now_data, week