ASJPush/Service/PushService.py

# -*- coding: utf-8 -*-
"""
@Time ： 2022/5/19 11:43
@Auth ： Locky
@File ：PushService.py
@IDE ：PyCharm
"""
import logging
import os

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
from Service.CommonService import CommonService


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: 推送内容
        @return: None
        """
        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)
            assert response.status_code == 200
        except Exception as e:
            return repr(e)