# -*-coding:utf-8-*- # # Copyright 2020. Huawei Technologies Co., Ltd. All rights reserved. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. import json from Service.HuaweiPushService.push_admin import _messages import six class MessageSerializer(json.JSONEncoder): """ Use https://docs.python.org/3/library/json.html to do serialization The serializer should serialize the following messages: _messages.Message _messages.Notification _messages.ApnsConfig _messages.WebPushConfig _messages.WebPushNotification _messages.WebPushNotificationAction _messages.WebPushHMSOptions _messages.AndroidConfig _messages.AndroidNotification _messages.AndroidClickAction _messages.BadgeNotification """ def default(self, message): """ :param message: The push message :return: formatted push messages """ result = { 'data': message.data, 'notification': MessageSerializer.encode_notification(message.notification), 'android': MessageSerializer.encode_android_config(message.android), 'apns': MessageSerializer.encode_apns_config(message.apns), 'webpush': MessageSerializer.encode_webpush_config(message.web_push), 'token': message.token, 'topic': message.topic, 'condition': message.condition } result = MessageSerializer.remove_null_values(result) return result @classmethod def remove_null_values(cls, dict_value): return {k: v for k, v in dict_value.items() if v not in [None, [], {}]} @classmethod def encode_notification(cls, notification): """ An example: { "title":"Big News", "body":"This is a Big News!", "image":"https://res.vmallres.com/pimages//common/config/logo/SXppnESYv4K11DBxDFc2_0.png" } :param notification: :return: """ if notification is None: return None if not isinstance(notification, _messages.Notification): raise ValueError('Message.notification must be an instance of Notification class.') result = { 'title': notification.title, 'body': notification.body, 'image': notification.image } return cls.remove_null_values(result) @classmethod def encode_android_config(cls, android_config): """ An example: { "android":{ "collapse_key":-1, "urgency":"HIGH", "ttl":"1448s", "bi_tag":"Trump", "fast_app_target":1, "notification": {} } :param android_config: :return: """ if android_config is None: return None if not isinstance(android_config, _messages.AndroidConfig): raise ValueError('Message.android must be an instance of AndroidConfig class.') result = { 'collapse_key': android_config.collapse_key, 'urgency': android_config.urgency, 'ttl': android_config.ttl, 'bi_tag': android_config.bi_tag, 'fast_app_target': android_config.fast_app_target, 'data': android_config.data, 'notification': MessageSerializer.encode_android_notification(android_config.notification), 'category': android_config.category } return cls.remove_null_values(result) @classmethod def encode_android_notification(cls, notification): """ "notification":{ "title":"Noti in Noti title", "body":"Noti in Noti body", "icon":"https://res.vmallres.com/pimages//common/config/logo/SXppnESYv4K11DBxDFc2.png", "color":"#AACCDD", "default_sound":true, "tag":"tagBoom", "importance":"PRIORITY_HIGH", "click_action":{ "type":2, "url":"https://www.huawei.com" }, "body_loc_key":"M.String.body", "body_loc_args":[ "Boy", "Dog" ], "title_loc_key":"M.String.title", "title_loc_args":[ "Girl", "Cat" ], "channel_id":"RingRing", "notify_summary":"Some Summary", "style":2, "big_title":"Big Boom Title", "big_body":"Big Boom Body", "notify_id":486, "group":"Espace", "badge":{ "add_num":99, "set_num":99, "class":"Classic" }, "ticker":"i am a ticker", "auto_cancel":false, "when":"2019-11-05", "use_default_vibrate":true, "use_default_light":false, "visibility":"PUBLIC", "vibrate_config":["1.5","2","3"], "light_settings":{ "color":{ "alpha":0, "red":0, "green":1, "blue":1 }, "light_on_duration":"3.5", "light_off_duration":"5S" }, "foreground_show":true } :param notification: :return: """ if notification is None: return None if not isinstance(notification, _messages.AndroidNotification): raise ValueError('Message.AndroidConfig.notification must be an instance of AndroidNotification class.') result = { "title": notification.title, "body": notification.body, "icon": notification.icon, "color": notification.color, "sound": notification.sound, "default_sound": notification.default_sound, "tag": notification.tag, "importance": notification.importance, "multi_lang_key": notification.multi_lang_key, "click_action": MessageSerializer.encode_android_click_action(notification.click_action), "body_loc_key": notification.body_loc_key, "body_loc_args": notification.body_loc_args, "title_loc_key": notification.title_loc_key, "title_loc_args": notification.title_loc_args, "channel_id": notification.channel_id, "notify_summary": notification.notify_summary, "image": notification.image, "style": notification.style, "big_title": notification.big_title, "big_body": notification.big_body, "notify_id": notification.notify_id, "group": notification.group, "badge": MessageSerializer.encode_android_badge(notification.badge), "ticker": notification.ticker, "auto_cancel": notification.auto_cancel, "when": notification.when, "use_default_vibrate": notification.use_default_vibrate, "use_default_light": notification.use_default_light, "visibility": notification.visibility, "vibrate_config": notification.vibrate_config, "light_settings": MessageSerializer.encode_android_light_settings(notification.light_settings), "foreground_show": notification.foreground_show } result = cls.remove_null_values(result) return result @classmethod def encode_android_click_action(cls, click_action): """ "click_action":{ "type":2, "url":"https://www.huawei.com" } "click_action":{ "type":1, "intent":"https://www.huawei.com", "action":"" } :param click_action: _messages.AndroidClickAction :return: """ if click_action is None: return None if not isinstance(click_action, _messages.AndroidClickAction): raise ValueError('Message.AndroidConfig.AndroidNotification.click_action must be an instance\ of AndroidClickAction class.') result = { "type": click_action.action_type, "intent": click_action.intent, "url": click_action.url, "action": click_action.action } result = cls.remove_null_values(result) return result @classmethod def encode_android_badge(cls, badge): """ refer to: _messages.AndroidBadgeNotification "badge":{ "add_num":99, "set_num":99, "class":"Classic" } :param badge: :return: """ if badge is None: return None if not isinstance(badge, _messages.AndroidBadgeNotification): raise ValueError('Message.AndroidConfig.AndroidNotification.badge must be an instance\ of AndroidBadgeNotification class.') result = { "add_num": badge.add_num, "set_num": badge.set_num, "class": badge.clazz } result = cls.remove_null_values(result) return result @classmethod def encode_android_light_settings(cls, android_light_settings): """ refer to: _messages.AndroidLightSettings "light_settings":{ "color":{ "alpha":0, "red":0, "green":1, "blue":1 }, "light_on_duration":"3.5", "light_off_duration":"5S" } :param android_light_settings: _messages.AndroidLightSettings :return: """ if android_light_settings is None: return None if not isinstance(android_light_settings, _messages.AndroidLightSettings): raise ValueError('Message.AndroidConfig.AndroidNotification.android_light_settings must be an instance\ of AndroidLightSettings class.') result = { "color": MessageSerializer.encode_android_light_settings_color(android_light_settings.color), "light_on_duration": android_light_settings.light_on_duration, "light_off_duration": android_light_settings.light_off_duration } result = cls.remove_null_values(result) return result @classmethod def encode_android_light_settings_color(cls, color): """ "color":{ "alpha":0, "red":0, "green":1, "blue":1 } :param color: _messages.AndroidLightSettingsColor :return: """ if color is None: return None if not isinstance(color, _messages.AndroidLightSettingsColor): raise ValueError('Message.AndroidConfig.AndroidNotification.android_light_settings.color must be an instance\ of AndroidLightSettingsColor class.') result = { "alpha": color.alpha, "red": color.red, "green": color.green, "blue": color.blue } result = cls.remove_null_values(result) return result @classmethod def encode_webpush_config(cls, webpush_config): """ "webpush":{ "headers":{ ... }, "notification":{ ... }, "hms_options":{ ... } } :param webpush_config: refer to _messages.WebPushConfig :return: """ if webpush_config is None: return None if not isinstance(webpush_config, _messages.WebPushConfig): raise ValueError('Message.webpush must be an instance of WebPushConfig class.') result = { "headers": MessageSerializer.encode_webpush_config_headers(webpush_config.headers), "notification": MessageSerializer.encode_webpush_config_notification(webpush_config.notification), "hms_options": MessageSerializer.encode_webpush_config_hms_options(webpush_config.hms_options), } result = cls.remove_null_values(result) return result @classmethod def encode_webpush_config_headers(cls, webpush_headers): """ "headers":{ "ttl":"990", "urgency":"very-low", "topic":"12313ceshi" } :param webpush_headers: _messages.WebPushHeader :return: """ if webpush_headers is None: return None if not isinstance(webpush_headers, _messages.WebPushHeader): raise ValueError('Message.webpush.headers must be an instance of WebPushHeader class.') result = { "ttl": webpush_headers.ttl, "urgency": webpush_headers.urgency, "topic": webpush_headers.topic, } result = cls.remove_null_values(result) return result @classmethod def encode_webpush_config_notification(cls, webpush_notification): """ "notification":{ "title":"notication string", "body":"web push body", "actions":[ { "action":"", "icon":"https://res.vmallres.com/pimages//common/config/logo/SXppnESYv4K11DBxDFc2.png", "title":"string" } ], "badge":"string", "dir":"auto", "icon":"https://res.vmallres.com/pimages//common/config/logo/SXppnESYv4K11DBxDFc2.png", "image":"string", "lang":"string", "renotify":true, "requireInteraction":true, "silent":true, "tag":"string", "timestamp":1545201266, "vibrate":[1,2,3] } :param webpush_notification: refer to _messages.WebPushNotification :return: """ if webpush_notification is None: return None if not isinstance(webpush_notification, _messages.WebPushNotification): raise ValueError('Message.webpush.notification must be an instance of WebPushNotification class.') result = { "title": webpush_notification.title, "body": webpush_notification.body, "actions": [MessageSerializer.encode_webpush_notification_action(_) for _ in webpush_notification.actions], "badge": webpush_notification.badge, "dir": webpush_notification.dir, "icon": webpush_notification.icon, "image": webpush_notification.image, "lang": webpush_notification.lang, "renotify": webpush_notification.renotify, "require_interaction": webpush_notification.require_interaction, "silent": webpush_notification.silent, "tag": webpush_notification.tag, "timestamp": webpush_notification.timestamp, "vibrate": webpush_notification.vibrate } result = cls.remove_null_values(result) return result @classmethod def encode_webpush_notification_action(cls, webpush_notification_action): """ "actions":[ { "action":"", "icon":"https://res.vmallres.com/pimages//common/config/logo/SXppnESYv4K11DBxDFc2.png", "title":"string" } ], :param webpush_notification_action: refer to _messages.WebPushNotificationAction :return: """ if webpush_notification_action is None: return None if not isinstance(webpush_notification_action, _messages.WebPushNotificationAction): raise ValueError('Message.webpush.notification.action must be an instance of \ WebPushNotificationAction class.') result = { "action": webpush_notification_action.action, "icon": webpush_notification_action.icon, "title": webpush_notification_action.title } result = cls.remove_null_values(result) return result @classmethod def encode_webpush_config_hms_options(cls, webpush_hms_options): """ "hms_options":{ "link":"https://www.huawei.com/" } :param webpush_hms_options: refer to _messages.WebPushHMSOptions :return: """ if webpush_hms_options is None: return None if not isinstance(webpush_hms_options, _messages.WebPushHMSOptions): raise ValueError('Message.webpush.hmsoptions must be an instance of \ WebPushHMSOptions class.') result = { "link": webpush_hms_options.link } result = cls.remove_null_values(result) return result @classmethod def encode_apns_config(cls, apns_config): """ Encode APNs config into JSON :param apns_config: :return: """ if apns_config is None: return None if not isinstance(apns_config, _messages.APNsConfig): raise ValueError('Message.apns_config must be an instance of _messages.APNsConfig class.') result = { 'headers': apns_config.headers, 'payload': cls.encode_apns_payload(apns_config.payload), 'hms_options': cls.encode_apns_hms_options(apns_config.apns_hms_options) } return cls.remove_null_values(result) @classmethod def encode_apns_payload(cls, apns_payload): """Encodes an ``APNSPayload`` instance into JSON.""" if apns_payload is None: return None if not isinstance(apns_payload, _messages.APNsPayload): raise ValueError('APNSConfig.payload must be an instance of _messages.APNsPayload class.') result = { 'aps': cls.encode_apns_payload_aps(apns_payload.aps) } for key, value in apns_payload.custom_data.items(): result[key] = value return cls.remove_null_values(result) @classmethod def encode_apns_payload_aps(cls, apns_payload_aps): """Encodes an ``Aps`` instance into JSON.""" if not isinstance(apns_payload_aps, _messages.APNsAps): raise ValueError('APNSPayload.aps must be an instance of _messages.APNsAps class.') result = { 'alert': cls.encode_apns_payload_alert(apns_payload_aps.alert), 'badge': apns_payload_aps.badge, 'sound': apns_payload_aps.sound, 'category': apns_payload_aps.category, 'thread-id': apns_payload_aps.thread_id } if apns_payload_aps.content_available is True: result['content-available'] = 1 if apns_payload_aps.mutable_content is True: result['mutable-content'] = 1 if apns_payload_aps.custom_data is not None: if not isinstance(apns_payload_aps.custom_data, dict): raise ValueError('Aps.custom_data must be a dict.') for key, val in apns_payload_aps.custom_data.items(): if key in result: raise ValueError('Multiple specifications for {0} in Aps.'.format(key)) result[key] = val return cls.remove_null_values(result) @classmethod def encode_apns_payload_alert(cls, apns_payload_alert): """Encodes an ``ApsAlert`` instance into JSON.""" if apns_payload_alert is None: return None if isinstance(apns_payload_alert, six.string_types): return apns_payload_alert if not isinstance(apns_payload_alert, _messages.APNsAlert): raise ValueError('Aps.alert must be a string or an instance of _messages.APNsAlert class.') result = { 'title': apns_payload_alert.title, 'body': apns_payload_alert.body, 'title-loc-key': apns_payload_alert.title_loc_key, 'title-loc-args': apns_payload_alert.title_loc_args, 'loc-key': apns_payload_alert.loc_key, 'loc-args': apns_payload_alert.loc_args, 'action-loc-key': apns_payload_alert.action_loc_key, 'launch-image': apns_payload_alert.launch_image } if result.get('loc-args') and not result.get('loc-key'): raise ValueError( 'ApsAlert.loc_key is required when specifying loc_args.') if result.get('title-loc-args') and not result.get('title-loc-key'): raise ValueError( 'ApsAlert.title_loc_key is required when specifying title_loc_args.') if apns_payload_alert.custom_data is not None: if not isinstance(apns_payload_alert.custom_data, dict): raise ValueError('ApsAlert.custom_data must be a dict.') for key, val in apns_payload_alert.custom_data.items(): result[key] = val return cls.remove_null_values(result) @classmethod def encode_apns_hms_options(cls, apns_hms_options): """ :param apns_hms_options: """ if apns_hms_options is None: return None if not isinstance(apns_hms_options, _messages.APNsHMSOptions): raise ValueError('Aps.alert must be a string or an instance of _messages.APNsHMSOptions class.') result = { 'target_user_type': apns_hms_options.target_user_type, } return cls.remove_null_values(result)