Использование концентраторов уведомлений с Python

Статья
06/01/2023

Можно обращаться ко всем функциям Центров уведомлений из серверной части Java, PHP, Python или Ruby, используя интерфейс REST Центра уведомлений в соответствии с описанием в статье MSDN Интерфейсы API REST концентраторов уведомлений.

Примечание

Это образец эталонной реализации для отправки уведомлений в Python, который не является официально поддерживаемым пакетом SDK концентраторов уведомлений Python. Этот образец был создан с помощью Python 3.4.

В этой статье показано, как:

Построение клиента REST для функций концентраторов уведомлений на языке Python.
Отправка уведомлений с помощью интерфейса API REST концентратора уведомления Python.
Получение дампа запросов и ответов HTTP REST для отладки и образовательных целей.

Можно воспользоваться указаниями в разделе Приступая к работе с центрами уведомлений для выбранной мобильной платформы, которая реализует серверную часть на языке Python.

Примечание

Область действия образца ограничена отправкой уведомлений. Образец не выполняет функции управления регистрацией.

Интерфейс клиента

Интерфейс основного клиента предоставляет те же методы, что и пакет SDK центров уведомлений для .NET. Этот интерфейс дает возможность напрямую переводить все руководства и примеры, доступные на этом сайте в настоящий момент и пополняемые интернет-сообществом.

Полный код доступен в примере оболочки REST Python.

Например, чтобы создать клиента, необходимо выполнить следующие действия.

isDebug = True
hub = NotificationHub("myConnectionString", "myNotificationHubName", isDebug)

Чтобы отправить всплывающее уведомление Windows:

wns_payload = """<toast><visual><binding template=\"ToastText01\"><text id=\"1\">Hello world!</text></binding></visual></toast>"""
hub.send_windows_notification(wns_payload)

Реализация

Если вы еще этого не делали, выполните шаги, описанные в Учебника по началу работы, до последнего раздела, в котором вам нужно реализовать серверную часть.

Подробные сведения о реализации полноценной оболочки REST можно найти в MSDN. В этом разделе описана реализация основных действий на языке Python, необходимых для доступа к конечным точкам REST службы "Центры уведомлений" и отправки уведомлений.

Проанализируйте строку подключения
Создайте маркер проверки подлинности
Отправка уведомления с помощью HTTP REST API

Проанализируйте строку подключения

Ниже показан основной класс, реализующий клиента, конструктор которого выполняет анализ строки подключения:

class NotificationHub:
    API_VERSION = "?api-version=2013-10"
    DEBUG_SEND = "&test"

    def __init__(self, connection_string=None, hub_name=None, debug=0):
        self.HubName = hub_name
        self.Debug = debug

        # Parse connection string
        parts = connection_string.split(';')
        if len(parts) != 3:
            raise Exception("Invalid ConnectionString.")

        for part in parts:
            if part.startswith('Endpoint'):
                self.Endpoint = 'https' + part[11:].lower()
            if part.startswith('SharedAccessKeyName'):
                self.SasKeyName = part[20:]
            if part.startswith('SharedAccessKey'):
                self.SasKeyValue = part[16:]

Создание маркера безопасности

Подробные сведения о создании токенов безопасности можно найти здесь. Для создания маркера на основе универсального кода ресурса (URI) текущего запроса и учетных данных, извлеченных из строки подключения, добавьте следующие методы в класс NotificationHub.

@staticmethod
def get_expiry():
    # By default returns an expiration of 5 minutes (=300 seconds) from now
    return int(round(time.time() + 300))


@staticmethod
def encode_base64(data):
    return base64.b64encode(data)


def sign_string(self, to_sign):
    key = self.SasKeyValue.encode('utf-8')
    to_sign = to_sign.encode('utf-8')
    signed_hmac_sha256 = hmac.HMAC(key, to_sign, hashlib.sha256)
    digest = signed_hmac_sha256.digest()
    encoded_digest = self.encode_base64(digest)
    return encoded_digest


def generate_sas_token(self):
    target_uri = self.Endpoint + self.HubName
    my_uri = urllib.parse.quote(target_uri, '').lower()
    expiry = str(self.get_expiry())
    to_sign = my_uri + '\n' + expiry
    signature = urllib.parse.quote(self.sign_string(to_sign))
    auth_format = 'SharedAccessSignature sig={0}&se={1}&skn={2}&sr={3}'
    sas_token = auth_format.format(signature, expiry, self.SasKeyName, my_uri)
    return sas_token

Отправка уведомления с помощью HTTP REST API

Примечание

Служба push-уведомлений (Майкрософт) (MPNS) устарела и больше не поддерживается.

Сначала следует определить класс, представляющий уведомление.

class Notification:
    def __init__(self, notification_format=None, payload=None, debug=0):
        valid_formats = ['template', 'apple', 'gcm',
                         'windows', 'windowsphone', "adm", "baidu"]
        if not any(x in notification_format for x in valid_formats):
            raise Exception(
                "Invalid Notification format. " +
                "Must be one of the following - 'template', 'apple', 'gcm', 'windows', 'windowsphone', 'adm', 'baidu'")

        self.format = notification_format
        self.payload = payload

        # array with keynames for headers
        # Note: Some headers are mandatory: Windows: X-WNS-Type, WindowsPhone: X-NotificationType
        # Note: For Apple you can set Expiry with header: ServiceBusNotification-ApnsExpiry
        # in W3C DTF, YYYY-MM-DDThh:mmTZD (for example, 1997-07-16T19:20+01:00).
        self.headers = None

Этот класс представляет собой контейнер для собственного текста уведомления либо набор свойств шаблонного уведомления, а также набор заголовков, содержащих свойства формата (собственная платформа или шаблон) и специальные свойства платформы (например, свойство срока действия Apple и заголовки WNS).

Обратитесь к документации по REST API для службы "Центры уведомлений" и изучите форматы специализированных платформ уведомлений, чтобы узнать обо всех доступных параметрах.

Имея в распоряжении этот класс, запишите методы отправки уведомлений внутри класса NotificationHub.

def make_http_request(self, url, payload, headers):
    parsed_url = urllib.parse.urlparse(url)
    connection = http.client.HTTPSConnection(
        parsed_url.hostname, parsed_url.port)

    if self.Debug > 0:
        connection.set_debuglevel(self.Debug)
        # adding this querystring parameter gets detailed information about the PNS send notification outcome
        url += self.DEBUG_SEND
        print("--- REQUEST ---")
        print("URI: " + url)
        print("Headers: " + json.dumps(headers, sort_keys=True,
                                       indent=4, separators=(' ', ': ')))
        print("--- END REQUEST ---\n")

    connection.request('POST', url, payload, headers)
    response = connection.getresponse()

    if self.Debug > 0:
        # print out detailed response information for debugging purpose
        print("\n\n--- RESPONSE ---")
        print(str(response.status) + " " + response.reason)
        print(response.msg)
        print(response.read())
        print("--- END RESPONSE ---")

    elif response.status != 201:
        # Successful outcome of send message is HTTP 201 - Created
        raise Exception(
            "Error sending notification. Received HTTP code " + str(response.status) + " " + response.reason)

    connection.close()


def send_notification(self, notification, tag_or_tag_expression=None):
    url = self.Endpoint + self.HubName + '/messages' + self.API_VERSION

    json_platforms = ['template', 'apple', 'gcm', 'adm', 'baidu']

    if any(x in notification.format for x in json_platforms):
        content_type = "application/json"
        payload_to_send = json.dumps(notification.payload)
    else:
        content_type = "application/xml"
        payload_to_send = notification.payload

    headers = {
        'Content-type': content_type,
        'Authorization': self.generate_sas_token(),
        'ServiceBusNotification-Format': notification.format
    }

    if isinstance(tag_or_tag_expression, set):
        tag_list = ' || '.join(tag_or_tag_expression)
    else:
        tag_list = tag_or_tag_expression

    # add the tags/tag expressions to the headers collection
    if tag_list != "":
        headers.update({'ServiceBusNotification-Tags': tag_list})

    # add any custom headers to the headers collection that the user may have added
    if notification.headers is not None:
        headers.update(notification.headers)

    self.make_http_request(url, payload_to_send, headers)


def send_apple_notification(self, payload, tags=""):
    nh = Notification("apple", payload)
    self.send_notification(nh, tags)


def send_google_notification(self, payload, tags=""):
    nh = Notification("gcm", payload)
    self.send_notification(nh, tags)


def send_adm_notification(self, payload, tags=""):
    nh = Notification("adm", payload)
    self.send_notification(nh, tags)


def send_baidu_notification(self, payload, tags=""):
    nh = Notification("baidu", payload)
    self.send_notification(nh, tags)


def send_mpns_notification(self, payload, tags=""):
    nh = Notification("windowsphone", payload)

    if "<wp:Toast>" in payload:
        nh.headers = {'X-WindowsPhone-Target': 'toast',
                      'X-NotificationClass': '2'}
    elif "<wp:Tile>" in payload:
        nh.headers = {'X-WindowsPhone-Target': 'tile',
                      'X-NotificationClass': '1'}

    self.send_notification(nh, tags)


def send_windows_notification(self, payload, tags=""):
    nh = Notification("windows", payload)

    if "<toast>" in payload:
        nh.headers = {'X-WNS-Type': 'wns/toast'}
    elif "<tile>" in payload:
        nh.headers = {'X-WNS-Type': 'wns/tile'}
    elif "<badge>" in payload:
        nh.headers = {'X-WNS-Type': 'wns/badge'}

    self.send_notification(nh, tags)


def send_template_notification(self, properties, tags=""):
    nh = Notification("template", properties)
    self.send_notification(nh, tags)

Указанные выше методы отправляют запрос HTTP POST к конечной точке "/messages" центра уведомлений с соответствующим текстом и заголовками для отправки уведомления.

Включение подробного ведения журнала с помощью свойства отладки

Если включить свойство отладки при инициализации центра уведомлений, в журнал будут записаны подробные сведения о дампе запроса и ответа HTTP, а также детальные выходные данные об отправке уведомлений. Свойство TestSend службы "Центры уведомлений" возвращает подробные сведения о результатах отправки уведомлений. Чтобы его использовать, выполните инициализацию с помощью приведенного кода.

hub = NotificationHub("myConnectionString", "myNotificationHubName", isDebug)

В результате к URL-адресу типа HTTP запроса на отправку для концентратора уведомлений прикрепляется строка запроса "test".

Завершение работы с учебником

Теперь вы можете завершить работу с учебником по началу работы, отправив уведомление из серверной части Python.

Инициализируйте клиент концентратора уведомлений (замените строку подключения и имя концентратора в соответствии с инструкциями в Учебника по началу работы):

hub = NotificationHub("myConnectionString", "myNotificationHubName")

Затем добавьте код отправки, определяемый целевой мобильной платформой. В этом примере добавлены методы более высокого уровня, чтобы включить отправку уведомлений для определенной платформы, например send_windows_notification (для Windows), send_apple_notification (для Apple) и т. д.

Магазин Windows и Windows Phone 8.1 (без Silverlight)

wns_payload = """<toast><visual><binding template=\"ToastText01\"><text id=\"1\">Test</text></binding></visual></toast>"""
hub.send_windows_notification(wns_payload)

Windows Phone 8.0 и 8.1 Silverlight

hub.send_mpns_notification(toast)

iOS

alert_payload = {
    'data':
        {
            'msg': 'Hello!'
        }
}
hub.send_apple_notification(alert_payload)

Android

gcm_payload = {
    'data':
        {
            'msg': 'Hello!'
        }
}
hub.send_google_notification(gcm_payload)

Kindle Fire

adm_payload = {
    'data':
        {
            'msg': 'Hello!'
        }
}
hub.send_adm_notification(adm_payload)

Baidu

baidu_payload = {
    'data':
        {
            'msg': 'Hello!'
        }
}
hub.send_baidu_notification(baidu_payload)

После выполнения кода Python на целевом устройстве должно отобразиться уведомление.

Примеры

Включение свойства `debug`

Если включить флаг отладки во время инициализации центра уведомлений (NotificationHub), вы увидите подробный дамп HTTP-запроса и ответа HTTP наряду с данными NotificationOutcome, которые указывают, какие именно заголовки HTTP переданы в запросе и какой ответ HTTP был получен от центра уведомлений.

Снимок экрана: окно консоли, где красным цветом выделены подробные сведения о дампе запросов и ответов HTTP, а также сообщения с результатами уведомлений.

Вы увидите подробный результат центра уведомлений, например сведения о том,

когда сообщение успешно отправлено в службу push-уведомлений.
```
<Outcome>The Notification was successfully sent to the Push Notification System</Outcome>
```
Если какое-либо push-уведомление не достигло адресата, отобразятся следующие выходные данные (указывающие, что, вероятнее всего, регистрации для доставки уведомлений не были найдены из-за несоответствующих тегов).
```
'<NotificationOutcome xmlns="http://schemas.microsoft.com/netservices/2010/10/servicebus/connect" xmlns:i="https://www.w3.org/2001/XMLSchema-instance"><Success>0</Success><Failure>0</Failure><Results i:nil="true"/></NotificationOutcome>'
```

Рассылка всплывающих уведомлений для Windows

Обратите внимание на заголовки, которые отправляются при рассылке всплывающих уведомлений клиенту Windows.

hub.send_windows_notification(wns_payload)

Снимок экрана: окно консоли, где красным цветом выделены подробные сведения о запросе HTTP, формате оповещений Служебной шины и значения типов XWNS.

Отправка уведомления с указанием тега (или регулярного выражения)

Обратите внимание на заголовок HTTP Tags, который будет добавлен к HTTP-запросу (в приведенном ниже примере уведомление отправляется только для регистраций с полезными данными "sports").

hub.send_windows_notification(wns_payload, "sports")

Снимок экрана: окно консоли, где красным цветом выделены подробные сведения о запросе HTTP, формате оповещений Служебной шины, тег оповещения Служебной шины и значения типа XWNS.

Отправка уведомления с указанием нескольких тегов

Обратите внимание на изменения заголовка Tags HTTP при отправке нескольких тегов.

tags = {'sports', 'politics'}
hub.send_windows_notification(wns_payload, tags)

Снимок экрана: окно консоли, где красным цветом выделены подробные сведения о запросе HTTP, формате оповещений Служебной шины, несколько тегов оповещения Служебной шины и значения типа XWNS.

Шаблон уведомления

Обратите внимание, что происходит изменение заголовка Format HTTP и текст полезных данных отправляется в составе запроса HTTP.

Сторона клиента — зарегистрированный шаблон:

var template = @"<toast><visual><binding template=""ToastText01""><text id=""1"">$(greeting_en)</text></binding></visual></toast>";

Серверная сторона — отправка полезных данных:

template_payload = {'greeting_en': 'Hello', 'greeting_fr': 'Salut'}
hub.send_template_notification(template_payload)

Снимок экрана: окно консоли, где красным цветом выделены подробные сведения о запросе HTTP, значениях типа содержимого и формата оповещений Служебной шины.

Next Steps

В этой статье рассмотрено создание простого клиента REST Python для службы "Центры уведомлений". Здесь можно выполнять следующие действия:

Скачать полный пример программы-оболочки REST Python, содержащий весь код из этой статьи.
Продолжить изучение функции тегов в центрах уведомлений с помощью учебника по передаче экстренных новостей
Продолжить изучение функции шаблонов центров уведомлений в учебнике по передаче локализованных экстренных новостей

Использование концентраторов уведомлений с Python

Интерфейс клиента

Реализация

Проанализируйте строку подключения

Создание маркера безопасности

Отправка уведомления с помощью HTTP REST API

Включение подробного ведения журнала с помощью свойства отладки

Завершение работы с учебником

Магазин Windows и Windows Phone 8.1 (без Silverlight)

Windows Phone 8.0 и 8.1 Silverlight

iOS

Android

Kindle Fire

Baidu

Примеры

Включение свойства debug

Рассылка всплывающих уведомлений для Windows

Отправка уведомления с указанием тега (или регулярного выражения)

Отправка уведомления с указанием нескольких тегов

Шаблон уведомления

Next Steps

Дополнительные ресурсы

Включение свойства `debug`