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 jpush
import requests
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
from Object.RedisObject import RedisObject
from Service.CommonService import CommonService
from Service.VivoPushService.push_admin.APIMessage import PushMessage
from Service.VivoPushService.push_admin.APISender import APISender


class PushObject:
    # 推送对象
    @staticmethod
    def get_msg_title(app_bundle_id, nickname):
        """
        获取推送消息标题
        @param app_bundle_id: app包id
        @param nickname: 设备名
        @return: msg_title
        """
        if app_bundle_id in APP_BUNDLE_DICT.keys():
            msg_title = APP_BUNDLE_DICT[app_bundle_id] + '(' + nickname + ')'
        else:
            msg_title = nickname
        return msg_title

    @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: None
        """
        logger = logging.getLogger('info')
        try:
            pem_path = os.path.join(BASE_DIR, APNS_CONFIG[app_bundle_id]['pem_path'])
            logger.info('apns推送app_bundle_id:{}, pem_path:{}'.format(app_bundle_id, pem_path))
            cli = apns2.APNSClient(mode=APNS_MODE, client_cert=pem_path)
            alert = apns2.PayloadAlert(title=msg_title, body=msg_text, launch_image=launch_image)
            push_data = {'alert': 'Motion', '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
                         }
            payload = apns2.Payload(alert=alert, custom=push_data, sound='default', category='myCategory',
                                    mutable_content=True)
            n = apns2.Notification(payload=payload, priority=apns2.PRIORITY_LOW)
            res = cli.push(n=n, device_token=token_val, topic=app_bundle_id)
            assert res.status_code == 200
        except Exception as e:
            logger.info('--->IOS推送异常{}'.format(repr(e)))
            return repr(e)

    @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: None
        """
        logger = logging.getLogger('info')
        try:
            serverKey = FCM_CONFIG[app_bundle_id]
            push_service = FCMNotification(api_key=serverKey)
            push_data = {'alert': 'Motion', '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
                         }
            result = push_service.notify_single_device(registration_id=token_val, message_title=msg_title,
                                                       message_body=msg_text, data_message=push_data,
                                                       extra_kwargs={'default_sound': True,
                                                                     'default_vibrate_timings': True,
                                                                     'default_light_settings': True,
                                                                     }
                                                       )
            logger.info('fcm推送结果:{}'.format(result))
        except Exception as e:
            return repr(e)

    @staticmethod
    def android_jpush(nickname, app_bundle_id, token_val, n_time, event_type, msg_title, msg_text):
        """
        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: 推送内容
        @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_data = {'alert': 'Motion', 'msg': '', 'sound': 'sound.aif', 'zpush': '1',
                         'received_at': n_time, 'event_time': n_time, 'event_type': event_type, 'nickname': nickname
                         }
            android = jpush.android(title=msg_title, big_text=msg_text, alert=msg_text, extras=push_data,
                                    priority=1, style=1, alert_type=7
                                    )
            push.notification = jpush.notification(android=android)
            push.platform = jpush.all_
            res = push.send()
            assert res.status_code == 200
        except Exception as e:
            return repr(e)

    @staticmethod
    def android_xmpush(nickname, app_bundle_id, token_val, n_time, event_type, msg_title, msg_text,
                       uid='', channel='1', image=''):
        """
        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 uid: uid
        @param channel: 通道
        @param image: 推送图片链接
        @return: None
        """
        logger = logging.getLogger('info')
        try:
            url = 'https://api.xmpush.xiaomi.com/v3/message/regid'
            app_secret = XMPUSH_CONFIG[app_bundle_id]
            payload = {'alert': 'Motion', 'msg': '', 'sound': 'sound.aif', 'zpush': '1',
                       'received_at': n_time, 'event_time': n_time, 'event_type': event_type, 'nickname': nickname,
                       'uid': uid, 'channel': channel
                       }
            data = {
                'title': msg_title,
                'description': msg_text,
                'payload': 'payload',
                'restricted_package_name': app_bundle_id,
                'registration_id': token_val,
            }
            # 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
        except Exception as e:
            return repr(e)

    @staticmethod
    def android_vivopush(app_bundle_id, token_val, n_time, event_type, msg_title, msg_text,
                         uid='', channel='1', image='', skip_type='1', nickname=''):
        """
        vivo 推送
        @param app_bundle_id: app包id
        @param token_val: 推送token
        @param event_type: 消息类型 （0：运营类消息，1：系统类消息。默认为 0）
        @param msg_title: 推送标题
        @param msg_text: 推送内容
        @param push_mode: 推送模式 （0：正式推送；1：测试推送，默认为0）
        @param m_code: 用户请求唯一标识 最大 64 字符
        @param skip_type: 跳转类型(1：打开 APP 首页 2：打开链接 3：自定义 4:打开 app 内指定页面)
        @param n_time: 当前时间
        @param nickname: 设备昵称
        @param uid: uid
        @param image: 推送图片链接
        @param channel: 通道
        @return: None
        """
        logger = logging.getLogger('info')
        try:
            authToken = 'authToken_' + app_bundle_id
            redisObj = RedisObject()
            # 获取redis里面的authToken
            redis_authToken = redisObj.get_data(key=authToken)
            if msg_title == '':
                msg_title = '周视'
            if redis_authToken is not False:
                app_secret = VIVOPUSH_CONFIG[app_bundle_id]['Secret']
                sender_send = APISender(app_secret)
                sender_send.set_token(redis_authToken)
            else:
                skip_type = int(skip_type)
                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,有效期3个小时
                redisObj = RedisObject()
                redisObj.set_data(key=authToken, val=rec['authToken'], expire=10800)
                sender_send = APISender(app_secret)
                sender_send.set_token(rec['authToken'])
            push_data = {'alert': 'Motion', '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
                         }
            # 推送
            message = PushMessage() \
                .reg_id(token_val) \
                .title(msg_title) \
                .content(msg_text) \
                .push_mode(1) \
                .notify_type(1) \
                .skip_type(skip_type) \
                .request_id('123456') \
                .classification(1) \
                .client_custom_map(**push_data) \
                .message_dict()
            rec2 = sender_send.send(message)
            logger.info('vivo推送结果:{}'.format(rec2))
        except Exception as e:
            logger.info('vivo推送异常:{}'.format(e))

    @staticmethod
    def android_oppopush(nickname, app_bundle_id, token_val, n_time, event_type, msg_title, msg_text,
                         uid='', channel='1', image=''):
        """
        android oppo 推送
        @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: None
        """
        logger = logging.getLogger('info')
        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'
            push_data = {'alert': 'Motion', '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,
                }
            }
            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()))
            assert response.status_code == 200
        except Exception as e:
            return repr(e)