# -*- coding: utf-8 -*- """ @Time ： 2022/5/19 11:43 @Auth ： Locky @File ：PushService.py @IDE ：PyCharm """ import hashlib import json import logging import os import time import apns2 import firebase_admin import jpush import requests from firebase_admin import messaging from pyfcm import FCMNotification from AnsjerPush.config import APP_BUNDLE_DICT, APNS_MODE, BASE_DIR, APNS_CONFIG, FCM_CONFIG, JPUSH_CONFIG, XMPUSH_CONFIG \ , VIVOPUSH_CONFIG, OPPOPUSH_CONFIG, MEIZUPUSH_CONFIG, CONFIG_INFO, HONORPUSH_CONFIG, DATA_PUSH_EVENT_TYPE_LIST, \ CONFIG_TEST from AnsjerPush.packages import apns2 import jwt import httpx from Model.models import UidPushModel from Object.RedisObject import RedisObject from Object.S3Email import S3Email from Object.enums.ConstantEnum import ConstantEnum from Object.enums.EventTypeEnum import EventTypeEnumObj from Service.CommonService import CommonService from Service.VivoPushService.push_admin.APIMessage import PushMessage from Service.VivoPushService.push_admin.APISender import APISender from AnsjerPush.config import LOGGER TIME_LOGGER = logging.getLogger('time') class PushObject: # 推送对象 @staticmethod def get_msg_title(nickname): """ 获取推送消息标题 @param nickname: 设备名 @return: msg_title """ return nickname @staticmethod def get_gateway_msg_text(n_time, tz, lang, alarm): """ 获取网关推送消息内容 @param n_time: 当前时间 @param tz: 时区 @param lang: 语言 @param alarm: 警报 @return: msg_text """ n_date = CommonService.get_now_time_str(n_time=n_time, tz=tz, lang=lang) if lang == 'cn': msg_text = '{} 日期:{}'.format(alarm, n_date) else: msg_text = '{} date:{}'.format(alarm, n_date) return msg_text @staticmethod def get_ai_msg_text(channel, n_time, lang, tz, label): """ 获取AI推送内容 @param channel: 通道 @param n_time: 当前时间 @param lang: 语言 @param tz: 时区 @param label: 识别到的标签 @return: ai_msg_text """ n_date = CommonService.get_now_time_str(n_time=n_time, tz=tz, lang=lang) if lang == 'cn': msg = '摄像头AI识别到了{}'.format(label) ai_msg_text = '{msg} 通道:{channel} 日期:{date}'.format(msg=msg, channel=channel, date=n_date) else: msg = 'Camera AI recognizes {}'.format(label) ai_msg_text = '{msg} channel:{channel} date:{date}'.format(msg=msg, channel=channel, date=n_date) return ai_msg_text @staticmethod def get_low_power_msg_text(channel, n_time, lang, tz, electricity, is_sys=0): """ 获取低电量推送内容 @param channel: 通道 @param n_time: 当前时间 @param lang: 语言 @param tz: 时区 @param electricity: 电量 @param is_sys: 是否为系统消息 @return: low_power_msg_text """ n_date = CommonService.get_now_time_str(n_time=n_time, tz=tz, lang=lang) if lang == 'cn': alarm = '剩余电量 ' + electricity if is_sys: low_power_msg_text = '{} 通道:{}'.format(alarm, channel) else: low_power_msg_text = '{} 通道:{} 日期:{}'.format(alarm, channel, n_date) else: alarm = 'Battery remaining ' + electricity if is_sys: low_power_msg_text = '{} channel:{}'.format(alarm, channel) else: low_power_msg_text = '{} channel:{} date:{}'.format(alarm, channel, n_date) return low_power_msg_text @staticmethod def ios_apns_push(nickname, app_bundle_id, token_val, n_time, event_type, msg_title, msg_text, uid='', channel='1', launch_image=None): """ ios apns 推送 @param nickname: 设备昵称 @param app_bundle_id: app包id @param token_val: 推送token @param n_time: 当前时间 @param event_type: 事件类型 @param msg_title: 推送标题 @param msg_text: 推送内容 @param uid: uid @param channel: 通道 @param launch_image: 推送图片链接 @return: bool """ pem_path = os.path.join(BASE_DIR, APNS_CONFIG[app_bundle_id]['pem_path']) LOGGER.info('IOS推送: uid:{}, app_bundle_id:{}, pem_path:{}, msg_text:'.format( uid, app_bundle_id, pem_path, msg_text)) try: apns_url = "https://api.sandbox.push.apple.com" if CONFIG_INFO == CONFIG_TEST else "https://api.push.apple.com" url = f"{apns_url}/3/device/{token_val}" jump_type = CommonService.get_jump_type(event_type) push_data = { 'alert': msg_text, 'msg': '', 'sound': '', 'zpush': '1', 'uid': uid, 'channel': channel, 'received_at': n_time, 'event_time': n_time, 'event_type': event_type, 'nickname': nickname, 'image_url': launch_image, 'jump_type': jump_type } sound = 'call_phone.mp3' if event_type in DATA_PUSH_EVENT_TYPE_LIST else 'default' body = { "aps": { "alert": { "title": msg_title, "body": msg_text, "launch-image": launch_image }, "sound": sound, "category": "myCategory", "mutable-content": 1 }, "custom": push_data } headers = { "apns-topic": app_bundle_id, "apns-push-type": "alert" } with httpx.Client(http2=True,timeout=10,cert=pem_path,) as client: res = client.post(url, headers=headers, json=body) if res.status_code == 200: LOGGER.info(f"{uid} iOS 推送成功, token:{token_val}") return True else: LOGGER.error(f"{uid} iOS 推送失败，状态码: {res.status_code}, 原因: {res.text}") return False except Exception as e: LOGGER.error(f"{uid} iOS 推送异常: {repr(e)}, 行数: {e.__traceback__.tb_lineno}") return False @staticmethod def ios_p8_push(nickname, app_bundle_id, token_val, n_time, event_type, msg_title, msg_text, uid='', channel='1', launch_image=None): """ iOS推送 (P8证书模式) @param nickname: 设备昵称 @param app_bundle_id: app包id @param token_val: 推送 token @param n_time: 当前时间 @param event_type: 事件类型 @param msg_title: 推送标题 @param msg_text: 推送内容 @param uid:uid @param channel: 通道 @param launch_image: 推送图片链接 @return: bool """ if app_bundle_id not in ConstantEnum.IOS_P8_CONFIG.value: return PushObject.ios_apns_push( nickname=nickname, app_bundle_id=app_bundle_id, token_val=token_val, n_time=n_time, event_type=event_type, msg_title=msg_title, msg_text=msg_text, uid=uid, channel=channel, launch_image=launch_image ) else: LOGGER.info("进入 ios_p8_push 方法，准备开始推送") try: team_id = ConstantEnum.IOS_P8_CONFIG.value[app_bundle_id]['team_id'] key_id = ConstantEnum.IOS_P8_CONFIG.value[app_bundle_id]['key_id'] p8_path = os.path.join(BASE_DIR, ConstantEnum.IOS_P8_CONFIG.value[app_bundle_id]['pem_path']) with open(p8_path, "r") as f: private_key = f.read() now = int(time.time()) token = jwt.encode( {"iss": team_id, "iat": now}, private_key, algorithm="ES256", headers={"kid": key_id} ) if isinstance(token, bytes): token = token.decode("utf-8") apns_url = "https://api.sandbox.push.apple.com" if CONFIG_INFO == CONFIG_TEST else "https://api.push.apple.com" url = f"{apns_url}/3/device/{token_val}" jump_type = CommonService.get_jump_type(event_type) push_data = { 'alert': msg_text, 'msg': '', 'sound': '', 'zpush': '1', 'uid': uid, 'channel': channel, 'received_at': n_time, 'event_time': n_time, 'event_type': event_type, 'nickname': nickname, 'image_url': launch_image, 'jump_type': jump_type } sound = 'call_phone.mp3' if event_type in DATA_PUSH_EVENT_TYPE_LIST else 'default' body = { "aps": { "alert": { "title": msg_title, "body": msg_text, "launch-image": launch_image }, "sound": sound, "category": "myCategory", "mutable-content": 1 }, "custom": push_data } headers = { "authorization": f"bearer {token}", "apns-topic": app_bundle_id, "apns-push-type": "alert" } with httpx.Client(http2=True, timeout=10,) as client: res = client.post(url, headers=headers, json=body) if res.status_code == 200: LOGGER.info(f"{uid} iOS p8 推送成功, token:{token_val}") return True else: LOGGER.error(f"{uid} iOS p8 推送失败，状态码: {res.status_code}, 原因: {res.text}") return False except Exception as e: LOGGER.error(f"{uid} iOS p8 推送异常: {repr(e)}, 行数: {e.__traceback__.tb_lineno}") return False @staticmethod def android_fcm_push(nickname, app_bundle_id, token_val, n_time, event_type, msg_title, msg_text, uid='', channel='1', image=''): """ android fcm 推送 @param nickname: 设备昵称 @param app_bundle_id: app包id @param token_val: 推送token @param n_time: 当前时间 @param event_type: 事件类型 @param msg_title: 推送标题 @param msg_text: 推送内容 @param uid: uid @param channel: 通道 @param image: 推送图片链接 @return: bool """ try: serverKey = FCM_CONFIG[app_bundle_id] push_service = FCMNotification(api_key=serverKey) push_data = {'alert': msg_text, 'msg': '', 'sound': 'sound.aif', 'zpush': '1', 'image': image, 'received_at': n_time, 'event_time': n_time, 'event_type': event_type, 'nickname': nickname, 'uid': uid, 'channel': channel } if event_type in DATA_PUSH_EVENT_TYPE_LIST: push_data['priority'] = 'high' push_data['content_available'] = True push_data['direct_boot_ok'] = True result = push_service.notify_single_device(registration_id=token_val, message_title=msg_title, message_body=msg_text, data_message=push_data, click_action='android.intent.action.VIEW', extra_kwargs={'default_sound': True, 'default_vibrate_timings': True, 'default_light_settings': True, }, ) TIME_LOGGER.info('uid:{}fcm推送结果:{}'.format(uid, result)) return True except Exception as e: TIME_LOGGER.error('uid:{}fcm推送异常:{}'.format(uid, repr(e))) return False @staticmethod def android_fcm_push_v1(nickname, app_bundle_id, token_val, n_time, event_type, msg_title, msg_text, uid='', channel='1', image=''): """ android fcm 推送 @param nickname: 设备昵称 @param app_bundle_id: app包id @param token_val: 推送token @param n_time: 当前时间 @param event_type: 事件类型 @param msg_title: 推送标题 @param msg_text: 推送内容 @param uid: uid @param channel: 通道 @param image: 推送图片链接 @return: bool """ try: event_type = str(event_type) n_time = str(n_time) # 跳转类型 jump_type = str(CommonService.get_jump_type(event_type)) # 推送数据类型必须为字符串,否则抛ValueError('Message.data must not contain non-string values.')异常 push_data = {'alert': msg_text, 'msg': '', 'sound': 'sound.aif', 'zpush': '1', 'received_at': n_time, 'event_time': n_time, 'event_type': event_type, 'nickname': nickname, 'uid': uid, 'channel': channel, 'jump_type': jump_type } if event_type in DATA_PUSH_EVENT_TYPE_LIST: push_data['priority'] = 'high' push_data['content_available'] = True push_data['direct_boot_ok'] = True message = messaging.Message( notification=messaging.Notification( title=msg_title, body=msg_text, image=image ), data=push_data, token=token_val, ) # Send a message to the device corresponding to the provided # registration token. result = messaging.send(message) LOGGER.info('uid:{}fcm推送结果:{}'.format(uid, result)) return True except Exception as e: LOGGER.info('uid:{}fcm推送异常:{}'.format(uid, repr(e))) return False @staticmethod def android_jpush(nickname, app_bundle_id, token_val, n_time, event_type, msg_title, msg_text, channel=1): """ android 极光 推送 @param nickname: 设备昵称 @param app_bundle_id: app包id @param token_val: 推送token @param n_time: 当前时间 @param event_type: 事件类型 @param msg_title: 推送标题 @param msg_text: 推送内容 @param channel: 设备通道 @return: bool """ try: # app_key = JPUSH_CONFIG[app_bundle_id]['Key'] # master_secret = JPUSH_CONFIG[app_bundle_id]['Secret'] # # 换成各自的app_key和master_secret # _jpush = jpush.JPush(app_key, master_secret) # push = _jpush.create_push() # push.audience = jpush.registration_id(token_val) # if event_type in DATA_PUSH_EVENT_TYPE_LIST: # channel_id = '111934' # else: # channel_id = '1' # push_data = {'alert': msg_text, 'msg': '', 'sound': 'sound.aif', 'zpush': '1', 'uid': nickname, # 'received_at': n_time, 'event_time': n_time, 'event_type': event_type, 'nickname': nickname, # 'channel': channel # } # android = jpush.android(title=msg_title, big_text=msg_text, alert=msg_text, extras=push_data, # priority=1, style=1, alert_type=7, channel_id=channel_id # ) # push.notification = jpush.notification(android=android) # push.platform = jpush.all_ # res = push.send() # assert res.status_code == 200 return True except Exception as e: LOGGER.info('极光推送异常:{}'.format(repr(e))) return False @staticmethod def jpush(nickname, app_bundle_id, token_val, n_time, event_type, msg_title, msg_text, channel=1): """ android 极光 推送 @param nickname: 设备昵称 @param app_bundle_id: app包id @param token_val: 推送token @param n_time: 当前时间 @param event_type: 事件类型 @param msg_title: 推送标题 @param msg_text: 推送内容 @param channel: 设备通道 @return: bool """ try: app_key = JPUSH_CONFIG[app_bundle_id]['Key'] master_secret = JPUSH_CONFIG[app_bundle_id]['Secret'] # 换成各自的app_key和master_secret _jpush = jpush.JPush(app_key, master_secret) push = _jpush.create_push() push.audience = jpush.registration_id(token_val) if event_type in DATA_PUSH_EVENT_TYPE_LIST: channel_id = '111934' else: channel_id = '1' # 跳转类型 jump_type = CommonService.get_jump_type(event_type) push_data = {'alert': msg_text, 'msg': '', 'sound': 'sound.aif', 'zpush': '1', 'uid': nickname, 'received_at': n_time, 'event_time': n_time, 'event_type': event_type, 'nickname': nickname, 'channel': channel, 'jump_type': jump_type } android = jpush.android(title=msg_title, big_text=msg_text, alert=msg_text, extras=push_data, priority=1, style=1, alert_type=7, channel_id=channel_id ) push.notification = jpush.notification(android=android) push.platform = jpush.all_ res = push.send() assert res.status_code == 200 LOGGER.info('极光推送响应:{}, 参数:{}, 令牌:{}'.format(res, push_data, token_val)) return True except Exception as e: LOGGER.info('极光推送异常:{}'.format(repr(e))) return False @staticmethod def android_xmpush(channel_id, nickname, app_bundle_id, token_val, n_time, event_type, msg_title, msg_text, uid='', channel='1', image=''): """ android 小米 推送 @param channel_id: 通知通道 @param nickname: 设备昵称 @param app_bundle_id: app包id @param token_val: 推送token @param n_time: 当前时间 @param event_type: 事件类型 @param msg_title: 推送标题 @param msg_text: 推送内容 @param uid: uid @param channel: 通道 @param image: 推送图片链接 @return: bool """ try: url = 'https://api.xmpush.xiaomi.com/v3/message/regid' app_secret = XMPUSH_CONFIG[app_bundle_id] # 跳转类型 jump_type = CommonService.get_jump_type(event_type) data = { 'title': msg_title, 'description': msg_text, 'payload': 'payload', 'restricted_package_name': app_bundle_id, 'registration_id': token_val, 'extra.channel_id': channel_id, 'extra.alert': msg_text, 'extra.msg': '', 'extra.sound': 'sound.aif', 'extra.zpush': '1', 'extra.received_at': n_time, 'extra.event_time': n_time, 'extra.event_type': event_type, 'extra.nickname': nickname, 'extra.uid': uid, 'extra.channel': channel, 'extra.jump_type': jump_type } # if image: # data['extra.notification_style_type'] = 2 # data['extra.notification_bigPic_uri'] = image headers = { 'Authorization': 'key={}'.format(app_secret) } response = requests.post(url, data=data, headers=headers) LOGGER.info("小米推送返回值:{}".format(response.json())) assert response.status_code == 200 return True except Exception as e: LOGGER.info("小米推送异常:{}".format(repr(e))) return False @staticmethod def android_vivopush(token_val, n_time, event_type, msg_title, msg_text, app_bundle_id='', uid='', channel='1', image='', nickname='', appBundleId='', jg_token_val=''): """ vivo 推送(不支持图片) @param app_bundle_id: app包名 @param appBundleId： app包名 @param token_val: 推送token @param jg_token_val: 极光推送token @param event_type: 事件类型 @param msg_title: 推送标题 @param msg_text: 推送内容 @param n_time: 当前时间 @param nickname: 设备昵称 @param uid: uid @param image: 推送图片链接 @param channel: 通道 @return: bool """ try: app_bundle_id = app_bundle_id if app_bundle_id != '' else appBundleId # 获取redis里面的authToken if msg_title == '': msg_title = APP_BUNDLE_DICT[app_bundle_id] app_id = VIVOPUSH_CONFIG[app_bundle_id]['ID'] app_key = VIVOPUSH_CONFIG[app_bundle_id]['Key'] app_secret = VIVOPUSH_CONFIG[app_bundle_id]['Secret'] sender = APISender(app_secret) rec = sender.get_token(app_id, app_key) # 鉴权接口调用获得authToken sender_send = APISender(app_secret) sender_send.set_token(rec['authToken']) # 跳转类型 jump_type = CommonService.get_jump_type(event_type) push_data = {'alert': msg_text, 'msg': '', 'sound': 'sound.aif', 'zpush': '1', 'image': image, 'received_at': n_time, 'event_time': n_time, 'event_type': event_type, 'nickname': nickname, 'uid': uid, 'channel': channel, 'jump_type': jump_type } # 获取唯一标识符 uid_push_qs = UidPushModel.objects.filter(token_val=token_val).values('m_code') m_code = uid_push_qs[0]['m_code'] if uid_push_qs[0]['m_code'] else '' # 推送 push_mode: 推送模式 （0：正式推送；1：测试推送，默认为0） # 推送 event_type: 消息类型 （0：运营类消息，1：系统类消息。默认为 0） # 推送 skip_type: 跳转类型(1：打开 APP 首页 2：打开链接 3：自定义 4:打开 app 内指定页面) activity = 'vpushscheme://com.vivo.pushvideo/detail' message = PushMessage() \ .reg_id(token_val) \ .title(msg_title) \ .content(msg_text) \ .push_mode(0) \ .notify_type(3) \ .skip_type(4) \ .skip_content(activity) \ .request_id(m_code) \ .classification(1) \ .client_custom_map(**push_data) \ .message_dict() rec = sender_send.send(message) LOGGER.info('vivo推送结果:{}, 设备uid:{}'.format(rec, uid)) if rec['result'] == 0 and event_type in DATA_PUSH_EVENT_TYPE_LIST: PushObject.jpush_transparent_transmission(msg_title, msg_text, app_bundle_id, jg_token_val, push_data) return True except Exception as e: LOGGER.error('vivo推送异常,uid:{},error_line:{},error_msg:{}'. format(uid, e.__traceback__.tb_lineno, repr(e))) return False @staticmethod def android_oppopush(channel_id, nickname, app_bundle_id, token_val, n_time, event_type, msg_title, msg_text, uid='', channel='1', image='', jg_token_val=''): """ android oppo 推送 @param channel_id: 通知通道id @param nickname: 设备昵称 @param app_bundle_id: app包id @param token_val: 推送token @param jg_token_val: 推送token @param n_time: 当前时间 @param event_type: 事件类型 @param msg_title: 推送标题 @param msg_text: 推送内容 @param uid: uid @param channel: 通道 @param image: 推送图片链接 @return: bool """ try: """ android 国内oppo APP消息提醒推送 """ app_key = OPPOPUSH_CONFIG[app_bundle_id]['Key'] master_secret = OPPOPUSH_CONFIG[app_bundle_id]['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': msg_text, 'msg': '', 'sound': 'sound.aif', 'zpush': '1', 'received_at': n_time, 'event_time': n_time, 'event_type': event_type, 'nickname': nickname, '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': 1, 'click_action_activity': OPPOPUSH_CONFIG[app_bundle_id]['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) LOGGER.info("oppo推送返回值:{}".format(response.json())) if response.status_code == 200 and event_type in DATA_PUSH_EVENT_TYPE_LIST: PushObject.jpush_transparent_transmission(msg_title, msg_text, app_bundle_id, jg_token_val, extra_data) return True except Exception as e: LOGGER.info("oppo推送异常:{}".format(repr(e))) return False @staticmethod def android_meizupush(token_val, n_time, event_type, msg_title, msg_text, uid='', channel='1', app_bundle_id='', appBundleId='', nickname='', image=''): """ android 魅族推送(不支持图片) @param app_bundle_id: app包名 @param appBundleId: app包名 @param token_val: 推送token @param event_type: 消息类型 （0：运营类消息，1：系统类消息。默认为 0） @param msg_title: 推送标题 @param msg_text: 推送内容 @param n_time: 当前时间 @param nickname: 设备昵称 @param uid: uid @param image: 推送图片链接 @param channel: 通道 @return: bool """ try: # 获取包和AppSecret app_bundle_id = app_bundle_id if app_bundle_id != '' else appBundleId appId = MEIZUPUSH_CONFIG[app_bundle_id]['ID'] appSecret = MEIZUPUSH_CONFIG[app_bundle_id]['AppSecret'] url = 'https://server-api-push.meizu.com/garcia/api/server/push/varnished/pushByPushId' # 跳转类型 jump_type = CommonService.get_jump_type(event_type) extra_data = {'alert': msg_text, 'msg': '', 'sound': 'sound.aif', 'zpush': '1', 'received_at': n_time, 'event_time': n_time, 'event_type': event_type, 'nickname': nickname, 'uid': uid, 'channel': channel, 'jump_type': jump_type } # 转换为json格式 extra_data = json.dumps(extra_data) if msg_title == '': msg_title = APP_BUNDLE_DICT[app_bundle_id] # 拼接发送内容 activity = MEIZUPUSH_CONFIG[app_bundle_id]['click_action_activity'] # clickType点击动作, 0打开应用, 1打开应用页面, 2打开url页面, 3应用客户端自定义 messageJson = '{"clickTypeInfo": {"activity": "%s",' \ '"clickType": 1, "parameters": %s },"extra": {},' % (activity, extra_data) noticeBarInfo = ('"noticeBarInfo": {"title": "%s", "content": "%s"},' % (msg_title, msg_text)) noticeExpandInfo = '"noticeExpandInfo": {"noticeExpandType": 0}, "pushTimeInfo": {"validTime": 24}}' messageJson += noticeBarInfo messageJson += noticeExpandInfo data_meizu = { 'appId': appId, 'pushIds': token_val, 'messageJson': messageJson } # 魅族MD5加密,生成密钥 sign = CommonService.getMD5Sign(data=data_meizu, key=appSecret) data = { 'appId': appId, 'messageJson': messageJson, 'sign': sign, 'pushIds': token_val, } # 进行推送 response = requests.post(url, data=data) LOGGER.info("魅族推送结果:{}".format(response.json())) return True except Exception as e: LOGGER.info("魅族推送异常:{}".format(repr(e))) return False @staticmethod def android_honorpush(token_val, n_time, event_type, msg_title, msg_text, uid='', channel='1', image='', app_bundle_id='', appBundleId='', channel_id='', nickname=''): """ android honor 推送 @param channel_id: 通知通道id @param nickname: 设备昵称 @param app_bundle_id: app包id @param appBundleId: app包id @param token_val: 推送token @param n_time: 当前时间 @param event_type: 事件类型 @param msg_title: 推送标题 @param msg_text: 推送内容 @param uid: uid @param channel: 通道 @param image: 推送图片链接 @return: bool """ app_bundle_id = appBundleId if appBundleId else app_bundle_id try: client_id = HONORPUSH_CONFIG[app_bundle_id]['client_id'] client_secret = HONORPUSH_CONFIG[app_bundle_id]['client_secret'] app_id = HONORPUSH_CONFIG[app_bundle_id]['app_id'] get_access_token_url = 'https://iam.developer.hihonor.com/auth/token' post_data = { 'grant_type': 'client_credentials', 'client_id': client_id, 'client_secret': client_secret } headers = {'Content-Type': 'application/x-www-form-urlencoded'} access_token_response = requests.post(get_access_token_url, data=post_data, headers=headers) access_result = access_token_response.json() authorization_token = 'Bearer ' + access_result['access_token'] # 发送推送 push_url = 'https://push-api.cloud.hihonor.com/api/v1/{}/sendMessage'.format(app_id) headers = {'Content-Type': 'application/json', 'Authorization': authorization_token, 'timestamp': str(int(time.time()) * 1000)} # 跳转类型 jump_type = CommonService.get_jump_type(event_type) extra_data = {'alert': msg_text, 'msg': '', 'sound': 'sound.aif', 'zpush': '1', 'received_at': n_time, 'event_time': n_time, 'event_type': str(event_type), 'nickname': nickname, 'uid': uid, 'channel': channel, 'title': msg_title, 'body': msg_text, 'jump_type': jump_type } # 通知推送 push_data = { "android": { "notification": { "body": msg_text, "title": msg_title, "importance": "NORMAL", "clickAction": { "type": 3, } }, "targetUserType": 0, "data": json.dumps(extra_data) }, "token": [token_val] } LOGGER.info("uid:{},时间:{},荣耀推送发送内容:{}".format(uid, n_time, push_data)) response = requests.post(push_url, json=push_data, headers=headers) LOGGER.info("uid:{},时间:{},荣耀推送通知返回值:{}".format(uid, n_time, response.json())) # 一键通话透传推送 if int(event_type) in DATA_PUSH_EVENT_TYPE_LIST: push_data = { "data": json.dumps(extra_data), "token": [token_val] } response = requests.post(push_url, json=push_data, headers=headers) LOGGER.info("uid:{},时间:{},荣耀透传推送返回值:{}".format(uid, n_time, response.json())) return True except Exception as e: LOGGER.info("荣耀推送异常:error_line:{},error_msg:{}".format(e.__traceback__.tb_lineno, repr(e))) return False @staticmethod def jpush_transparent_transmission(msg_title, msg_text, app_bundle_id, token_val, extra_data): """ android 极光透传 @param msg_title: 推送标题 @param msg_text: 推送内容 @param token_val: 推送token @param app_bundle_id: app包id @param extra_data: 额外数据 @return: None """ try: app_key = JPUSH_CONFIG[app_bundle_id]['Key'] master_secret = JPUSH_CONFIG[app_bundle_id]['Secret'] # 换成各自的app_key和master_secret _jpush = jpush.JPush(app_key, master_secret) push = _jpush.create_push() push.audience = jpush.registration_id(token_val) push.message = jpush.message(msg_content=msg_text, title=msg_title, extras=extra_data) push.platform = jpush.all_ res = push.send() LOGGER.info('极光透传，结果：{}，参数：{}'.format(res, extra_data)) except Exception as e: LOGGER.info('jpush_transparent_transmission极光透传异常：errLine:{}, errMsg:{}, 参数:{}'.format( e.__traceback__.tb_lineno, repr(e), extra_data))