Python apple PUSH APNSWrapper支持字典以及群发

APNSWrapper是苹果推送通知服务的Python包装。使用起来也不错。不过发现使用property的时候不支持字典,查阅代码:
notifications.py APNSProperty类中处理property数据

if not isinstance(data, (int, str, list, tuple, float)):
    raise APNSValueError, "Data argument should be string, number, list of tuple"

可以看出,并不支持字典,由于业务需要,我们需要字典支持。

if not isinstance(data, (int, str, list, dict, tuple, float)):
    raise APNSValueError, "Data argument should be string, number, list of tuple"

以支持字典。
随后在def _build(self)方法中增加代码,需json支持:

if isinstance(self.data, dict):
    return "%s%s" % (name, json.write(self.data))

这样我们可以定义类似如下的数据:
{‘type’:’1′,’url’:'http://www.google.com.hk’}来支持我们的业务。

经过查看代码,发现APNSWrapper支持单条发送,而APNS支持一次连接,多次发送的模式。
修改APNSNotificationWrapper

class APNSNotificationWrapper(object):
    """
    This object wrap a list of APNS tuples. You should use
    .append method to add notifications to the list. By usint
    method .notify() all notification will send to the APNS server.
    """
    sandbox = True
    apnsHost = 'gateway.push.apple.com'
    apnsSandboxHost = 'gateway.sandbox.push.apple.com'
    apnsPort = 2195
    payloads = None
    connection = None
    debug_ssl = False
    kapnsconnection = None  #新建连接对象

    def __init__(self, certificate = None, sandbox = True, debug_ssl = False, \
                    force_ssl_command = False):
        self.debug_ssl = debug_ssl
        self.force_ssl_command = False
        self.connection = APNSConnection(certificate = certificate, \
                            force_ssl_command = self.force_ssl_command, debug = self.debug_ssl)
        self.sandbox = sandbox
        self.payloads = []


    def append(self, payload = None):
        """Append payload to wrapper"""
        if not isinstance(payload, APNSNotification):
            raise APNSTypeError, "Unexpected argument type. Argument should be an instance of APNSNotification object"
        self.payloads.append(payload)


    def count(self):
        """Get count of payloads
        """
        return len(self.payloads)


    def notifyclose(self):  #增加关闭连接
        self.kapnsconnection.close()

    def notify(self):  #保持连接

        apnsConnection = self.connection

        if self.sandbox != True:
            apnsHost = self.apnsHost
        else:
            apnsHost = self.apnsSandboxHost
        if not self.kapnsconnection:
            self.kapnsconnection = apnsConnection.connect(apnsHost, self.apnsPort)

        return True

    def send(self):  #发送
        """
        Send nofification to APNS:
            1) prepare all internal variables to APNS Payout JSON
            2) make connection to APNS server and send notification
        """
        payloads = [o.payload() for o in self.payloads]
        payloadsLen = sum([len(p) for p in payloads])
        messages = []
        offset = 0

        if len(payloads) == 0:
            return False

        for p in payloads:
            plen = len(p)
            messages.append(struct.pack('%ds' % plen, p))
            offset += plen

        message = "".join(messages)
        self.kapnsconnection.write(message)

这样,我们就支持了连续发送的功能。
使用方法:

import sys,os,binascii
import json
import APNSWrapper

dt = 'b9d98721 b5586b61 a00fbfa0 d61a0339 54e1bfb8 faaae3df c5a1382a 96e1a95a'
deviceToken = binascii.unhexlify(dt.replace(' ',''))
wrapper = APNSWrapper.APNSNotificationWrapper("/data0/pushservice/apns/certificates/wanjibaodiangaoshoupian_1122_production_push_certificates.pem", False)
wrapper.notify()  #建立连接
while True:
    message = APNSWrapper.APNSNotification()
    message.token(deviceToken)
    message.alert("抽烟去" + str(i))
    print("抽烟去" + str(i))
    message.sound()
    message.badge(1)
    cc = {'type':'1','url':'http://www.google.com.hk'}
    property = APNSWrapper.APNSProperty("fljt",cc)
    message.appendProperty(property)
    wrapper.append(message)
    wrapper.send() #发送
    i = i + 1
    if i == 10:
        break
wrapper.notifyclose() #关闭连接

deviceToken可以为不同的。
下载
安装

python PATH/setup.py install