# -*- 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\ , VIVOPUSH_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))