如何透過 Python 使用通知中樞

發行項
06/01/2023

您可以使用通知中樞 REST 介面，透過 Java/PHP/Python/Ruby 後端來存取所有通知中樞功能，如 MSDN 文章通知中樞 REST API所述。

注意

這是在 Python 實作通知傳送的範例參考實作，並非正式支援的通知中樞 Python SDK。此範例是使用 Python 3.4 建立的。

本文示範如何：

在 Python 中建置通知中樞功能的 REST 用戶端。
使用 Python 介面傳送通知到通知中樞 REST API。
取得 HTTP REST 要求/回應的傾印以用於偵錯/教學用途。

您可以遵循選定行動平台的開始使用教學課程來實作 Python 的後端部分。

注意

此範例的範圍僅限於傳送通知，不會執行任何註冊管理。

用戶端介面

主要用戶端介面提供的方法與 .NET 通知中樞 SDK中的方法相同。這個介面可讓您直接轉換此網站目前所提供，以及由網際網路社群所貢獻的所有教學課程和範例。

您可在 Python REST 包裝函式範例中找到所有可用的程式碼。

例如，若要建立用戶端：

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)

實作

如果尚未這麼做，請遵循開始使用教學課程中的指示，一路進行到您必須實作後端的最後一節。

您可以在 MSDN上找到所有實作完整 REST 包裝函式的詳細資料。本節將針對存取通知中樞 REST 端點和傳送通知所需主要步驟的 Python 實作進行說明

解析連接字串
產生授權權杖
使用 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:]

建立安全性權杖

您可以在此處找到建立安全性權杖的詳細資料。將下列方法新增至 NotificationHub 類別，以依據目前要求的 URI，以及從連接字串擷取的認證來建立權杖。

@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 傳送通知

注意

Microsoft 推播通知服務 (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)

通知中樞傳送要求 HTTP URL 會附加 "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 要求和回應傾印，還有類似以下的 NotificationOutcome，您可從中了解在要求中傳送的 HTTP 標頭，以及從通知中樞接收到的 HTTP 回應：

主控台的螢幕擷取畫面，其中具有 HTTP 要求的詳細資料，以及以紅色勾出的回應傾印和通知結果訊息。

例如，您會看到詳細的通知中樞結果。

訊息成功傳送至推播通知服務的時間。

<Outcome>The Notification was successfully sent to the Push Notification System</Outcome>

如果找不到任何推播通知的目標，您可能會看到下列輸出作為回應 (這表示找不到可傳遞通知的註冊，可能是因為註冊有一些不相符的標記)

'<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 要求的詳細資料，以及以紅色勾出的內容類型和服務匯流排通知格式。

後續步驟

本主題會說明如何針對通知中樞建立 Python REST 用戶端。從這裡您可以：

下載完整的 Python REST 包裝函式範例，其中包含本文章的所有程式碼。
繼續了解即時新聞教學課程
繼續了解當地語系化新聞教學課程