# -*- encoding: utf-8 -*- """ @File : DevicePushService.py @Time : 2022/11/23 11:40 @Author : stephen @Email : zhangdongming@asj6.wecom.work @Software: PyCharm """ import hashlib import json import logging import os import threading import time import apns2 import boto3 import botocore import jpush as jpush import requests from pyfcm import FCMNotification from AnsjerPush.Config.aiConfig import DEVICE_EVENT_TYPE 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 from AnsjerPush.config import JPUSH_CONFIG, FCM_CONFIG, APNS_CONFIG, BASE_DIR, APNS_MODE, XMPUSH_CONFIG, OPPOPUSH_CONFIG from Model.models import UidPushModel, SysMsgModel, DeviceSharePermission, DeviceChannelUserSet, \ DeviceChannelUserPermission, UidSetModel, Device_Info from Object.ETkObject import ETkObject from Object.UidTokenObject import UidTokenObject from Object.utils import LocalDateTimeUtil from Service.CommonService import CommonService from Service.EquipmentInfoService import EquipmentInfoService from Service.HuaweiPushService.HuaweiPushService import HuaweiPushObject from Service.PushService import PushObject LOGGING = logging.getLogger('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 LOGGING.info('消息推送-当前UID：{}'.format(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): """ 查询uid_push和uid_set数据 @param uid: uid @param event_type: 事件类型 @return: uid_push_qs """ if event_type != 606: uid_push_qs = UidPushModel.objects.filter(uid_set__uid=uid, uid_set__detect_status=1). \ 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'] uid_push_qs = UidPushModel.objects.filter(uid_set__uid=uid, userID_id=primary_user_id). \ 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 redis_obj.set_data(key=name, val=1, expire=detect_interval - 5) @classmethod def save_msg_push(cls, uid_set_push_list, **params): """ 推送消息,返回推送数据列表 @param uid_set_push_list: redis对象 @param params: 推送参数 @return: dict """ LOGGING.info('uid_set_push_list:{}'.format(uid_set_push_list)) new_device_info_list = [] sys_msg_list = [] userID_ids = [] kwag_args = params['kwag_args'] code_data = {'do_apns_code': '', 'do_fcm_code': '', 'do_jpush_code': ''} local_date_time = '' # push_permission = True 多通道权限限制接收 try: params['event_tag'] = cls.get_event_tag(params['ai_type'], params['event_type'], params['detection']) is_app_push = True if params['event_tag'] == 606 else \ cls.is_send_app_push(params['event_type'], params['event_tag'], params['app_push_config']) for up in uid_set_push_list: appBundleId = up['appBundleId'] token_val = up['token_val'] lang = up['lang'] tz = up['tz'] if tz is None or tz == '': tz = 0 # 发送标题 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'] ) kwag_args['appBundleId'] = appBundleId kwag_args['token_val'] = token_val kwag_args['msg_title'] = msg_title kwag_args['msg_text'] = msg_text if params['event_type'] == 606 and up['push_type'] in [5, 6]: kwag_args['jg_token_val'] = up['jg_token_val'] local_date_time = CommonService.get_now_time_str(n_time=params['n_time'], tz=tz, lang='cn') local_date_time = local_date_time[0:10] # 以下是存库 userID_id = up["userID_id"] if userID_id not in userID_ids: now_time = int(time.time()) 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=userID_id, msg=sys_msg_text, addTime=now_time, updTime=now_time, uid=params['uid'], eventType=params['event_type'])) else: params['userID_id'] = userID_id # push_permission = DevicePushService.check_share_permission(userID_id, # params['channel'],params['uid']) new_device_info_list.append(cls.created_device_vo(local_date_time, **params)) userID_ids.append(userID_id) params['appBundleId'] = appBundleId params['token_val'] = token_val params['lang'] = lang params['tz'] = tz params['kwag_args'] = kwag_args code_data = cls.send_app_msg_push(up['push_type'], **params) if is_app_push else code_data return {'code_date': code_data, 'new_device_info_list': new_device_info_list, 'sys_msg_list': sys_msg_list, 'local_date_time': local_date_time} except Exception as e: LOGGING.info('推送消息或存表异常: errLine:{}, errMsg:{}'.format(e.__traceback__.tb_lineno, repr(e))) @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：手势检测 0：代表空字符，702：摄像头休眠，703：摄像头唤醒，704：电量过低 AWS AI识别 1：人形，2：车型，3：宠物，4：包裹。云端AI类型 """ try: types = [] combo_types = [51, 57, 58, 60, 59, 61, 62, 63, 64, 65, 66, 67, 68] 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(combo_types[i])) LOGGING.info('算法对照打印：{}'.format(combo_types)) return types except Exception as e: print('推送错误异常,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, push_type, **param): """ 发送app消息推送 """ try: kwargs = param['kwag_args'] result = {'do_apns_code': '', 'do_fcm_code': '', 'do_jpush_code': '', 'do_xmpush_code': '', 'do_vivopush_code': '', 'do_meizupush_code': '', 'do_oppopush_code': ''} # 判断是否进行APP消息推送，如app_push不为空，则不进行推送 if not param['app_push']: LOGGING.info('APP准备推送:{}, {}'.format(param['uid'], param)) # 推送显示图片 if (param['is_st'] == 1 or param['is_st'] == 3) and \ (push_type == 0 or push_type == 1 or push_type == 3): if param['is_st'] == 1: key = '{}/{}/{}.jpeg'.format(param['uid'], param['channel'], param['n_time']) else: key = '{}/{}/{}_0.jpeg'.format(param['uid'], param['channel'], param['n_time']) push_thread = threading.Thread(target=cls.async_send_picture_push, args=( push_type, param['aws_s3_client'], param['bucket'], key, param['uid'], param['appBundleId'], param['token_val'], param['event_type'], param['n_time'], param['kwag_args']['msg_title'], param['kwag_args']['msg_text'], param['channel'])) push_thread.start() else: if push_type == 0: # ios apns result['do_apns_code'] = cls.do_apns(**kwargs) elif push_type == 1: # android gcm result['do_fcm_code'] = cls.do_fcm(**kwargs) elif push_type == 2: # android jpush result['do_jpush_code'] = cls.do_jpush(**kwargs) elif push_type == 3: huawei_push_object = HuaweiPushObject() huawei_push_object.send_push_notify_message(**kwargs) elif push_type == 4: # android xmpush channel_id = 104551 result['do_xmpush_code'] = cls.do_xmpush(channel_id=channel_id, **kwargs) elif push_type == 5: # android vivopush result['do_vivopush_code'] = PushObject.android_vivopush(**kwargs) elif push_type == 6: # android oppopush channel_id = 'DEVICE_REMINDER' result['do_oppopush_code'] = cls.do_oppopush(channel_id=channel_id, **kwargs) elif push_type == 7: # android meizupush result['do_meizupush_code'] = PushObject.android_meizupush(**kwargs) return result except Exception as e: LOGGING.info('异常详情,errLine:{}, errMsg:{}'.format(e.__traceback__.tb_lineno, repr(e))) return None @classmethod def save_sys_msg(cls, is_sys_msg, local_date_time, sys_msg_list, new_device_info_list): """ 保存系统消息&设备推送消息存库 """ if is_sys_msg: SysMsgModel.objects.bulk_create(sys_msg_list) else: # new 分表批量存储 设备信息 if new_device_info_list and len(new_device_info_list) > 0: # 根据日期获得星期几 week = LocalDateTimeUtil.date_to_week(local_date_time) EquipmentInfoService.equipment_info_bulk_create(week, new_device_info_list) return True @classmethod def created_device_vo(cls, local_date_time, **params): """ 获取设备推送表对象 """ return EquipmentInfoService.get_equipment_info_obj( local_date_time, 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:{channel}'.format(channel=params['channel']), is_st=params['is_st'], receive_time=params['n_time'], add_time=int(time.time()), storage_location=params['storage_location'], border_coords='', event_tag=params['event_tag'], answer_status=True if params['dealings_type'] == 1 else 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] if lang == 'cn': if event_type == 51: 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 == 704: msg_type = '剩余电量 ' + electricity elif event_type == 702: msg_type = '摄像头休眠' elif event_type == 703: msg_type = '摄像头唤醒' elif event_type == 606: msg_type = '有人呼叫，请点击查看' elif 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 == 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 == 704: msg_type = 'Battery remaining ' + electricity elif event_type == 702: msg_type = 'Camera sleep' elif event_type == 703: msg_type = 'Camera wake' elif event_type == 606: msg_type = 'Someone is calling, please click to view' elif 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_jpush(uid, channel, appBundleId, token_val, event_type, n_time, msg_title, msg_text): """ android 国内极光APP消息提醒推送 """ app_key = JPUSH_CONFIG[appBundleId]['Key'] master_secret = JPUSH_CONFIG[appBundleId]['Secret'] _jpush = jpush.JPush(app_key, master_secret) push = _jpush.create_push() push.audience = jpush.registration_id(token_val) push_data = {"alert": "Motion ", "event_time": n_time, "event_type": event_type, "msg": "", "received_at": n_time, "sound": "sound.aif", "uid": uid, "zpush": "1", "channel": channel} android = jpush.android(alert=msg_text, priority=1, style=1, alert_type=7, big_text=msg_text, title=msg_title, extras=push_data) push.notification = jpush.notification(android=android) push.platform = jpush.all_ res = push.send() print(res) return res.status_code @staticmethod def do_fcm(uid, channel, appBundleId, token_val, event_type, n_time, msg_title, msg_text): """ android 谷歌APP消息提醒推送 """ try: serverKey = FCM_CONFIG[appBundleId] except Exception as e: LOGGING.info('异常详情,errLine:{}, errMsg:{}'.format(e.__traceback__.tb_lineno, repr(e))) return 'serverKey abnormal' push_service = FCMNotification(api_key=serverKey) data = {"alert": "Motion ", "event_time": n_time, "event_type": event_type, "msg": "", "received_at": n_time, "sound": "sound.aif", "uid": uid, "zpush": "1", "channel": channel } result = push_service.notify_single_device(registration_id=token_val, message_title=msg_title, message_body=msg_text, data_message=data, click_action='android.intent.action.VIEW', extra_kwargs={ 'default_vibrate_timings': True, 'default_sound': True, 'default_light_settings': True, }, ) return result @staticmethod def do_apns(uid, channel, appBundleId, token_val, event_type, n_time, msg_title, msg_text): """ ios 消息提醒推送 """ LOGGING.info("进来do_apns函数了") LOGGING.info(token_val) LOGGING.info(APNS_MODE) LOGGING.info(os.path.join(BASE_DIR, APNS_CONFIG[appBundleId]['pem_path'])) try: cli = apns2.APNSClient( mode=APNS_MODE, client_cert=os.path.join(BASE_DIR, APNS_CONFIG[appBundleId]['pem_path'])) push_data = {"alert": "Motion ", "event_time": n_time, "event_type": event_type, "msg": "", "received_at": n_time, "sound": "", "uid": uid, "zpush": "1", "channel": channel} alert = apns2.PayloadAlert(body=msg_text, title=msg_title) payload = apns2.Payload(alert=alert, custom=push_data, sound="default") # return uid, channel, appBundleId, str(token_val), event_type, n_time, msg_title,msg_text n = apns2.Notification(payload=payload, priority=apns2.PRIORITY_LOW) res = cli.push(n=n, device_token=token_val, topic=appBundleId) print(res.status_code) LOGGING.info("apns_推送状态：") LOGGING.info(res.status_code) if res.status_code == 200: return res.status_code else: print('apns push fail') print(res.reason) LOGGING.info('apns push fail') LOGGING.info(res.reason) return res.status_code except (ValueError, ArithmeticError): return 'The program has a numeric format exception, one of the arithmetic exceptions' except Exception as e: print(repr(e)) print('do_apns函数错误行号', e.__traceback__.tb_lineno) LOGGING.info('异常详情,errLine:{}, errMsg:{}'.format(e.__traceback__.tb_lineno, repr(e))) return repr(e) @staticmethod def do_xmpush(channel_id, uid, channel, appBundleId, token_val, event_type, n_time, msg_title, msg_text): """ android 国内小米APP消息提醒推送 """ 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, } headers = { 'Authorization': 'key={}'.format(app_secret) } response = requests.post(url, data=data, headers=headers) if response.status_code == 200: LOGGING.info('小米推送结果：{}'.format(response.json())) return response.json() @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消息提醒推送 """ 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: LOGGING.info("oppo推送返回值:{},uid:{},time:{},event:{}".format(response.json(), uid, now_time, event_type)) if event_type == 606 or event_type == '606': PushObject.jpush_transparent_transmission(msg_title, msg_text, appBundleId, jg_token_val) return response.json() @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): """ 异步APP图片推送 """ try: image_url = aws_s3_client.generate_presigned_url('get_object', Params={'Bucket': bucket, 'Key': key}, ExpiresIn=3600) LOGGING.info('推送图片url:{}'.format(image_url)) if push_type == 0: PushObject.ios_apns_push(uid, appBundleId, token_val, n_time, event_type, msg_title, msg_text, uid, channel, image_url) elif push_type == 1: PushObject.android_fcm_push(uid, appBundleId, token_val, n_time, event_type, msg_title, msg_text, uid, channel, image_url) elif push_type == 3: huawei_push_object = HuaweiPushObject() 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) except Exception as e: LOGGING.info('图片推送异常,errLine:{}, errMsg:{}'.format(e.__traceback__.tb_lineno, repr(e))) @staticmethod def get_push_url(**params): """ 获取推送URL，设备根本当前返回结果进行数据上传 @return: re_data """ re_data = {'code': 0, 'msg': 'success'} if params['is_st'] == 0 or params['is_st'] == 2: re_data['msg'] = 'success 0 or 2' for up in params['uid_set_push_list']: if up['push_type'] == 0: # ios apns up['do_apns_code'] = params['code_dict']['code_date']['do_apns_code'] elif up['push_type'] == 1: # android gcm up['do_fcm_code'] = params['code_dict']['code_date']['do_fcm_code'] elif up['push_type'] == 2: # android jpush up['do_jpush_code'] = params['code_dict']['code_date']['do_jpush_code'] elif up['push_type'] == 4: # android jpush up['do_xmpush_code'] = params['code_dict']['code_date']['do_xmpush_code'] elif up['push_type'] == 5: # android jpush up['do_vivopush_code'] = params['code_dict']['code_date']['do_vivopush_code'] elif up['push_type'] == 7: # android jpush up['do_meizupush_code'] = params['code_dict']['code_date']['do_meizupush_code'] del up['push_type'] del up['userID_id'] del up['userID__NickName'] del up['lang'] del up['tz'] del up['uid_set__nickname'] del up['uid_set__detect_interval'] del up['uid_set__detect_group'] re_data['re_list'] = params['uid_set_push_list'] elif params['is_st'] == 1: key_name = '{uid}/{channel}/{filename}.jpeg' \ .format(uid=params['uid'], channel=params['channel'], filename=params['n_time']) re_args = {'Key': key_name} if params['region'] == 2: # 2:国内 re_args['Bucket'] = 'push' else: # 1：国外 re_args['Bucket'] = 'foreignpush' response_url = DevicePushService.generate_s3_url(params['aws_s3_client'], re_args) re_data['img_push'] = response_url elif params['is_st'] == 3: img_url_list = [] if params['region'] == 2: # 2:国内 re_args = {'Bucket': 'push'} else: # 1：国外 re_args = {'Bucket': 'foreignpush'} for i in range(params['is_st']): key_name = '{uid}/{channel}/{filename}_{st}.jpeg'. \ format(uid=params['uid'], channel=params['channel'], filename=params['n_time'], st=i) re_args['Key'] = key_name response_url = DevicePushService.generate_s3_url(params['aws_s3_client'], re_args) img_url_list.append(response_url) re_data['img_url_list'] = img_url_list re_data['msg'] = 'success 3' return re_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 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): """ 是否进行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 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): LOGGING.info('APP推送提醒不在自定义时间内：{}'.format(push_time_config)) return False # APP接收提醒，判断识别类型是否勾选提醒 push_result = DevicePushService.is_type_push(event_type, event_tag, app_event_types) LOGGING.info('APP推送提醒是否执行：{}'.format(push_result)) return push_result except Exception as e: LOGGING.info('判断是否执行APP推送异常,errLine:{}, errMsg:{}'.format(e.__traceback__.tb_lineno, repr(e))) return True @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