<s> """<STR_LIT>""" <EOL> import codecs <EOL> import json <EOL> import ssl <EOL> import struct <EOL> import socket <EOL> import time <EOL> from contextlib import closing <EOL> from binascii import unhexlify <EOL> from django . core . exceptions import ImproperlyConfigured <EOL> from . import NotificationError <EOL> from . settings import PUSH_NOTIFICATIONS_SETTINGS as SETTINGS <EOL> class APNSError ( NotificationError ) : <EOL> pass <EOL> class APNSServerError ( APNSError ) : <EOL> def __init__ ( self , status , identifier ) : <EOL> super ( APNSServerError , self ) . __init__ ( status , identifier ) <EOL> self . status = status <EOL> self . identifier = identifier <EOL> class APNSDataOverflow ( APNSError ) : <EOL> pass <EOL> def _apns_create_socket ( address_tuple ) : <EOL> certfile = SETTINGS . get ( "<STR_LIT>" ) <EOL> if not certfile : <EOL> raise ImproperlyConfigured ( <EOL> '<STR_LIT>' <EOL> ) <EOL> try : <EOL> with open ( certfile , "<STR_LIT:r>" ) as f : <EOL> f . read ( ) <EOL> except Exception as e : <EOL> raise ImproperlyConfigured ( "<STR_LIT>" % ( certfile , e ) ) <EOL> ca_certs = SETTINGS . get ( "<STR_LIT>" ) <EOL> sock = socket . socket ( ) <EOL> sock = ssl . wrap_socket ( sock , ssl_version = ssl . PROTOCOL_TLSv1 , certfile = certfile , ca_certs = ca_certs ) <EOL> sock . connect ( address_tuple ) <EOL> return sock <EOL> def _apns_create_socket_to_push ( ) : <EOL> return _apns_create_socket ( ( SETTINGS [ "<STR_LIT>" ] , SETTINGS [ "<STR_LIT>" ] ) ) <EOL> def _apns_create_socket_to_feedback ( ) : <EOL> return _apns_create_socket ( ( SETTINGS [ "<STR_LIT>" ] , SETTINGS [ "<STR_LIT>" ] ) ) <EOL> def _apns_pack_frame ( token_hex , payload , identifier , expiration , priority ) : <EOL> token = unhexlify ( token_hex ) <EOL> frame_len = <NUM_LIT:3> * <NUM_LIT:5> + len ( token ) + len ( payload ) + <NUM_LIT:4> + <NUM_LIT:4> + <NUM_LIT:1> <EOL> frame_fmt = "<STR_LIT>" % ( len ( token ) , len ( payload ) ) <EOL> frame = struct . pack ( <EOL> frame_fmt , <EOL> <NUM_LIT:2> , frame_len , <EOL> <NUM_LIT:1> , len ( token ) , token , <EOL> <NUM_LIT:2> , len ( payload ) , payload , <EOL> <NUM_LIT:3> , <NUM_LIT:4> , identifier , <EOL> <NUM_LIT:4> , <NUM_LIT:4> , expiration , <EOL> <NUM_LIT:5> , <NUM_LIT:1> , priority ) <EOL> return frame <EOL> def _apns_check_errors ( sock ) : <EOL> timeout = SETTINGS [ "<STR_LIT>" ] <EOL> if timeout is None : <EOL> return <EOL> saved_timeout = sock . gettimeout ( ) <EOL> try : <EOL> sock . settimeout ( timeout ) <EOL> data = sock . recv ( <NUM_LIT:6> ) <EOL> if data : <EOL> command , status , identifier = struct . unpack ( "<STR_LIT>" , data ) <EOL> assert command == <NUM_LIT:8> , "<STR_LIT>" <EOL> if status != <NUM_LIT:0> : <EOL> raise APNSServerError ( status , identifier ) <EOL> except socket . timeout : <EOL> pass <EOL> except ssl . SSLError as e : <EOL> if "<STR_LIT>" not in e . message : <EOL> raise <EOL> finally : <EOL> sock . settimeout ( saved_timeout ) <EOL> def _apns_send ( token , alert , badge = None , sound = None , category = None , content_available = False , <EOL> action_loc_key = None , loc_key = None , loc_args = [ ] , extra = { } , identifier = <NUM_LIT:0> , <EOL> expiration = None , priority = <NUM_LIT:10> , socket = None ) : <EOL> data = { } <EOL> aps_data = { } <EOL> if action_loc_key or loc_key or loc_args : <EOL> alert = { "<STR_LIT:body>" : alert } if alert else { } <EOL> if action_loc_key : <EOL> alert [ "<STR_LIT>" ] = action_loc_key <EOL> if loc_key : <EOL> alert [ "<STR_LIT>" ] = loc_key <EOL> if loc_args : <EOL> alert [ "<STR_LIT>" ] = loc_args <EOL> if alert is not None : <EOL> aps_data [ "<STR_LIT>" ] = alert <EOL> if badge is not None : <EOL> aps_data [ "<STR_LIT>" ] = badge <EOL> if sound is not None : <EOL> aps_data [ "<STR_LIT>" ] = sound <EOL> if category is not None : <EOL> aps_data [ "<STR_LIT>" ] = category <EOL> if content_available : <EOL> aps_data [ "<STR_LIT>" ] = <NUM_LIT:1> <EOL> data [ "<STR_LIT>" ] = aps_data <EOL> data . update ( extra ) <EOL> json_data = json . dumps ( data , separators = ( "<STR_LIT:U+002C>" , "<STR_LIT::>" ) , sort_keys = True ) . encode ( "<STR_LIT:utf-8>" ) <EOL> max_size = SETTINGS [ "<STR_LIT>" ] <EOL> if len ( json_data ) > max_size : <EOL> raise APNSDataOverflow ( "<STR_LIT>" % ( max_size ) ) <EOL> expiration_time = expiration if expiration is not None else int ( time . time ( ) ) + <NUM_LIT> <EOL> frame = _apns_pack_frame ( token , json_data , identifier , expiration_time , priority ) <EOL> if socket : <EOL> socket . write ( frame ) <EOL> else : <EOL> with closing ( _apns_create_socket_to_push ( ) ) as socket : <EOL> socket . write ( frame ) <EOL> _apns_check_errors ( socket ) <EOL> def _apns_read_and_unpack ( socket , data_format ) : <EOL> length = struct . calcsize ( data_format ) <EOL> data = socket . recv ( length ) <EOL> if data : <EOL> return struct . unpack_from ( data_format , data , <NUM_LIT:0> ) <EOL> else : <EOL> return None <EOL> def _apns_receive_feedback ( socket ) : <EOL> expired_token_list = [ ] <EOL> header_format = '<STR_LIT>' <EOL> has_data = True <EOL> while has_data : <EOL> try : <EOL> header_data = _apns_read_and_unpack ( socket , header_format ) <EOL> if header_data is not None : <EOL> timestamp , token_length = header_data <EOL> token_format = '<STR_LIT>' % token_length <EOL> device_token = _apns_read_and_unpack ( socket , token_format ) <EOL> if device_token is not None : <EOL> expired_token_list . append ( ( timestamp , device_token [ <NUM_LIT:0> ] ) ) <EOL> else : <EOL> has_data = False <EOL> except socket . timeout : <EOL> pass <EOL> except ssl . SSLError as e : <EOL> if "<STR_LIT>" not in e . message : <EOL> raise <EOL> return expired_token_list <EOL> def apns_send_message ( registration_id , alert , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> _apns_send ( registration_id , alert , ** kwargs ) <EOL> def apns_send_bulk_message ( registration_ids , alert , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> with closing ( _apns_create_socket_to_push ( ) ) as socket : <EOL> for identifier , registration_id in enumerate ( registration_ids ) : <EOL> _apns_send ( registration_id , alert , identifier = identifier , socket = socket , ** kwargs ) <EOL> _apns_check_errors ( socket ) <EOL> def apns_fetch_inactive_ids ( ) : <EOL> """<STR_LIT>""" <EOL> with closing ( _apns_create_socket_to_feedback ( ) ) as socket : <EOL> inactive_ids = [ ] <EOL> for tStamp , registration_id in _apns_receive_feedback ( socket ) : <EOL> inactive_ids . append ( codecs . encode ( registration_id , '<STR_LIT>' ) ) <EOL> return inactive_ids </s>
