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 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))