apn.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-

__author__ = 'johnx'
__date__ = '2/27/13 3:36 PM'

import socket
import ssl
import struct
import json
import binascii

servers = {
    'debug_post': ('gateway.sandbox.push.apple.com', 2195),
    'release_post': ('gateway.push.apple.com', 2195),
    'debug_feedback': ('feedback.sandbox.push.apple.com', 2196),
    'release_feedback': ('feedback.push.apple.com', 2196),
}

PAYLOAD_LIMIT = 256 #256 bytes
#@see: http://developer.apple.com/library/mac/#documentation/NetworkingInternet/Conceptual/RemoteNotificationsPG/ApplePushService/ApplePushService.html#//apple_ref/doc/uid/TP40008194-CH100-SW1
def post_apn(cert_file, message, mode='release', token_data=None, device_token=None, **kwargs):
    if not isinstance(message, (str, unicode)):
        message = str(message)
    if isinstance(message, unicode):
        message = message.encode('utf-8')

    if not kwargs:
        aps_data = dict(alert=message)
    else:
#        {"aps":{"sound":"default","alert":{"body":u'message',"action-loc-key":"Click me"}}}
        aps_data = dict(alert={'body': message})

        sound = kwargs.pop('sound')
        if sound is not None:
            aps_data['sound'] = sound

        badge = kwargs.pop('badge')
        if badge is not None:
            aps_data['badge'] = badge

        aps_data['alert'].update(kwargs)


    payload = json.dumps(dict(aps=aps_data))
    assert len(payload) < PAYLOAD_LIMIT, 'payload limit exceed(256 bytes).'

    if token_data is None:
        assert device_token is not None
        token_data = binascii.a2b_hex(device_token.replace(' ', ''))
    apn_data = '\x00\x00\x20' + token_data + struct.pack('>H', len(payload)) + payload

    server = servers['%s_post' % mode]
    return _send_data(server, cert_file, apn_data)

def _send_data(server, cert_file, payload):
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    conn = ssl.wrap_socket(sock, certfile=cert_file)
    conn.connect(server)
    result = conn.send(payload)
    conn.close()
    sock.close()

    return result == len(payload)