ASJPush/Service/PushService.py

# -*- 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 jpush
import requests
import jwt
import httpx

from firebase_admin import messaging
from pyfcm import FCMNotification

from AnsjerPush.config import APP_BUNDLE_DICT, 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 Model.models import UidPushModel
from Object.enums.ConstantEnum import ConstantEnum
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))