From 6bf26e7ec9b0ce6d49a45e4c63e11c45e86f551a Mon Sep 17 00:00:00 2001 From: Peter Nijssen Date: Sun, 14 Jul 2019 22:18:44 +0200 Subject: [PATCH] Added objects. Validate letters. Add distributions --- .gitignore | 5 +- postnl_api/__init__.py | 2 +- postnl_api/items/__init__.py | 0 postnl_api/items/letter.py | 25 +++ postnl_api/items/package.py | 48 +++++ postnl_api/postnl_api.py | 321 +++++++++++++++------------------- postnl_api/test_postnl_api.py | 58 ++++-- setup.py | 2 +- 8 files changed, 262 insertions(+), 199 deletions(-) create mode 100644 postnl_api/items/__init__.py create mode 100644 postnl_api/items/letter.py create mode 100644 postnl_api/items/package.py diff --git a/.gitignore b/.gitignore index 4ec1f2b..3f1913d 100644 --- a/.gitignore +++ b/.gitignore @@ -99,4 +99,7 @@ venv.bak/ /site # mypy -.mypy_cache/ \ No newline at end of file +.mypy_cache/ + +# intellij +.idea/ \ No newline at end of file diff --git a/postnl_api/__init__.py b/postnl_api/__init__.py index c487ece..92b2667 100644 --- a/postnl_api/__init__.py +++ b/postnl_api/__init__.py @@ -1 +1 @@ -from postnl_api.postnl_api import PostNL_API, UnauthorizedException \ No newline at end of file +from .postnl_api import PostNL_API, UnauthorizedException diff --git a/postnl_api/items/__init__.py b/postnl_api/items/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/postnl_api/items/letter.py b/postnl_api/items/letter.py new file mode 100644 index 0000000..2954f1c --- /dev/null +++ b/postnl_api/items/letter.py @@ -0,0 +1,25 @@ +class Letter(object): + + def __init__(self, data, documents): + self.id = data.get('barcode') + self.delivery_date = data.get('expectedDeliveryDate') + self.status_message = None + self.image = None + + if data.get('phase') is not None: + self.status_message = data.get('phase').get('message') + + if len(documents.get('documents')) > 0: + self.image = documents.get('documents')[0].get('link') + "?type=png" + + def id(self): + return self.id + + def delivery_date(self): + return self.delivery_date + + def status_message(self): + return self.status_message + + def image(self): + return self.image diff --git a/postnl_api/items/package.py b/postnl_api/items/package.py new file mode 100644 index 0000000..6cd0c31 --- /dev/null +++ b/postnl_api/items/package.py @@ -0,0 +1,48 @@ +class Package(object): + + def __init__(self, data): + self.id = data.get('key') + self.name = data.get('title') + self.type = data.get('settings').get('box') + self.status = data.get('status').get('deliveryStatus') + self.status_message = data.get('status').get('phase').get('message') + self.delivery_date = data.get('status').get('delivery').get('deliveryDate') + self.planned_date = None + self.planned_from = None + self.planned_to = None + if data.get('status').get('enroute') is not None and data.get('status').get('enroute').get('timeframe') is not None: + self.planned_date = data.get('status').get('enroute').get('timeframe').get('date') + self.planned_from = data.get('status').get('enroute').get('timeframe').get('from') + self.planned_to = data.get('status').get('enroute').get('timeframe').get('to') + + self.url = data.get('status').get('webUrl') + + def id(self): + return self.id + + def name(self): + return self.name + + def type(self): + return self.type + + def status(self): + return self.status + + def status_message(self): + return self.status_message + + def delivery_date(self): + return self.delivery_date + + def planned_date(self): + return self.planned_date + + def planned_from(self): + return self.planned_from + + def planned_to(self): + return self.planned_to + + def url(self): + return self.url diff --git a/postnl_api/postnl_api.py b/postnl_api/postnl_api.py index d264046..09721a4 100644 --- a/postnl_api/postnl_api.py +++ b/postnl_api/postnl_api.py @@ -1,10 +1,15 @@ """ Python wrapper for the PostNL API """ +import logging +import time from datetime import datetime, timedelta -import re +from urllib.parse import unquote import requests +from postnl_api.items.package import Package +from postnl_api.items.letter import Letter + BASE_URL = 'https://jouw.postnl.nl' AUTHENTICATE_URL = BASE_URL + '/mobile/token' @@ -13,253 +18,205 @@ LETTERS_URL = BASE_URL + '/mobile/api/letters' VALIDATE_LETTERS_URL = BASE_URL + '/mobile/api/letters/validation' +## PROFILE_URL +## isMyMailAvailable: true / false +## hasPendingMyMailValidation: true / false + +## VALIDATE_LETTERS_URL +## status: Validated + +## CHANGE NAME +## PATCH /mobile/api/shipments/{package-key} + +## DELETE SHIPMENT +## DELETE /mobile/api/shipments/{package-key} + +## DELETE LETTER +## DELETE /mobile/api/letters/{letter-barcode} + DEFAULT_HEADER = { - 'api-version': '4.7', - 'user-agent': 'PostNL/1 CFNetwork/889.3 Darwin/17.2.0', + 'api-version': '4.16', + 'X-Client-Library': 'python-postnl-api', } +REFRESH_RATE = 120 -class UnauthorizedException(Exception): - pass +_LOGGER = logging.getLogger(__name__) class PostNL_API(object): """ Interface class for the PostNL API """ - def __init__(self, user, password): + def __init__(self, user, password, refresh_rate=REFRESH_RATE): """ Constructor """ self._user = user self._password = password - - payload = { - 'grant_type': 'password', - 'client_id': 'pwIOSApp', - 'username': self._user, - 'password': self._password - } - - try: - response = requests.request( - 'POST', AUTHENTICATE_URL, data=payload, headers=DEFAULT_HEADER) - data = response.json() - - except Exception: - raise(UnauthorizedException()) - - if 'error' in data: - raise UnauthorizedException(data['error']) - - self._access_token = data['access_token'] - self._refresh_token = data['refresh_token'] - self._token_expires_in = data['expires_in'] - self._token_expires_at = datetime.now( - ) + timedelta(0, data['expires_in']) + self._delivery = {} + self._distribution = {} + self._letters = {} + self._letters_activated = False + self._last_refresh = None + self._refresh_rate = refresh_rate + self._request_login() def _is_token_expired(self): """ Check if access token is expired """ - if (datetime.now() > self._token_expires_at): - self._refresh_access_token() + if datetime.now() > self._token_expires_at: + self._request_access_token() return True return False - def _refresh_access_token(self): - """ Refresh access_token """ - - payload = { - 'grant_type': 'refresh_token', - 'client_id': 'pwIOSApp', - 'refresh_token': self._refresh_token - } - - response = requests.request( - 'POST', AUTHENTICATE_URL, data=payload, headers=DEFAULT_HEADER) - - data = response.json() + def update(self): + """ Update the cache """ + current_time = int(time.time()) + last_refresh = 0 if self._last_refresh is None else self._last_refresh - self._access_token = data['access_token'] + if current_time >= (last_refresh + self._refresh_rate): + self.update_packages() + self.update_letter_status() + self.update_letters() - def parse_datetime(self, text, dateFormat='%d-%m-%Y', timeFormat='%H:%M'): + self._last_refresh = int(time.time()) - def parse_date(date): - return datetime.strptime(date.group(1) - .replace(' ', '')[:-6], '%Y-%m-%dT%H:%M:%S').strftime(dateFormat) + def update_packages(self): + """ Retrieve packages """ + packages = self._request_update(SHIPMENTS_URL) + if packages is False: + return - def parse_time(date): - return datetime.strptime(date.group(1) - .replace(' ', '')[:-6], '%Y-%m-%dT%H:%M:%S').strftime(timeFormat) + self._delivery = {} + self._distribution = {} - text = re.sub(r'{(?:Date|dateAbs):(.*?)}', parse_date, text) - text = re.sub(r'{(?:time):(.*?)}', parse_time, text) + for package in packages: + if package.get("settings").get("box") == "Sender": + self._distribution[package['key']] = Package(package) + else: + self._delivery[package['key']] = Package(package) - return text + def update_letters(self): + """ Retrieve letters """ + if self._letters_activated is False: + return - def get_shipments(self): - """ Retrieve shipments """ + letters = self._request_update(LETTERS_URL) + if letters is False: + return - self._is_token_expired() + self._letters = {} - headers = { - 'authorization': 'Bearer ' + self._access_token - } + for letter in letters: + documents = self._request_update(LETTERS_URL + "/" + letter['barcode']) + self._letters[letter['barcode']] = Letter(letter, documents) - response = requests.request( - 'GET', SHIPMENTS_URL, headers={**headers, **DEFAULT_HEADER}) + def update_letter_status(self): + validate = self._request_update(VALIDATE_LETTERS_URL) - if response.status_code == 401: - self._refresh_access_token() - shipments = self.get_shipments() - else: - shipments = response.json() + if validate.get('status') == 'Validated': + self._letters_activated = True - return shipments + def get_delivery(self): + """ Get all packages to be delivered to you """ + self.update() - def get_shipment(self, shipment_id): - """ Retrieve single shipment by id """ + return self._delivery.values() - self._is_token_expired() + def get_distribution(self): + """ Get all packages submitted by you """ + self.update() - headers = { - 'authorization': 'Bearer ' + self._access_token - } + return self._distribution.values() - response = requests.request( - 'GET', SHIPMENTS_URL + '/' + shipment_id, headers={**headers, **DEFAULT_HEADER}) - - if response.status_code == 401: - self._refresh_access_token() - shipments = self.get_shipment(shipment_id) - else: - shipments = response.json() + def get_letters(self): + """ Get all letters to be delivered to you """ + self.update() - return shipments + return self._letters.values() - def get_profile(self): - """ Retrieve profile """ + def is_letters_activated(self): + """ Return if letters are activated or not """ + return self._letters_activated + def _request_update(self, url): + """ Perform a request to update information """ self._is_token_expired() headers = { - 'authorization': 'Bearer ' + self._access_token + 'authorization': 'Bearer ' + self._access_token, + 'Content-Type': 'application/json', } - response = requests.request( - 'GET', PROFILE_URL, headers={**headers, **DEFAULT_HEADER}) + response = requests.request('GET', url, headers={**headers, **DEFAULT_HEADER}) if response.status_code == 401: - self._refresh_access_token() - profile = self.get_profile() - else: - profile = response.json() - - return profile + _LOGGER.debug("Access denied. Failed to refresh?") + self._request_access_token() + self._request_update(url) - def validate_letters(self): - """ Retrieve letter validation status """ + if response.status_code != 200: + _LOGGER.error("Unable to perform request " + str(response.content)) + return False - self._is_token_expired() + return response.json() + def _request_login(self): headers = { - 'authorization': 'Bearer ' + self._access_token + 'Content-Type': 'application/x-www-form-urlencoded', } - response = requests.request( - 'GET', VALIDATE_LETTERS_URL, headers={**headers, **DEFAULT_HEADER}) - - if response.status_code == 401: - self._refresh_access_token() - validation = self.validate_letters() - else: - validation = response.json() - - return validation - - def get_letters(self): - """ Retrieve letters """ - - self._is_token_expired() - - headers = { - 'authorization': 'Bearer ' + self._access_token + payload = { + 'grant_type': 'password', + 'client_id': 'pwAndroidApp', + 'username': self._user, + 'password': self._password } - response = requests.request( - 'GET', LETTERS_URL, headers={**headers, **DEFAULT_HEADER}) - - if response.status_code == 401: - self._refresh_access_token() - letters = self.get_letters() - else: - letters = response.json() + try: + response = requests.request( + 'POST', AUTHENTICATE_URL, data=payload, headers={**headers, **DEFAULT_HEADER}) + data = response.json() - # TODO Add validation / exception handling - # if letters['type'] == 'ProfileValidationFeatureMissing': - # _LOGGER.error(letters['message']) - # return [] + except Exception: + raise (UnauthorizedException()) - return letters + if 'error' in data: + raise UnauthorizedException(data['error']) - def get_letter(self, letter_id): - """ Retrieve single letter by id """ + self._access_token = data['access_token'] + self._refresh_token = unquote(data['refresh_token']) + self._token_expires_in = data['expires_in'] + self._token_expires_at = datetime.now() + timedelta(0, (int(data['expires_in']) - 20)) - self._is_token_expired() + def _request_access_token(self): + """ Refresh access_token """ headers = { - 'authorization': 'Bearer ' + self._access_token + 'Content-Type': 'application/x-www-form-urlencoded', } - response = requests.request( - 'GET', LETTERS_URL + '/' + letter_id, headers={**headers, **DEFAULT_HEADER}) - - if response.status_code == 200: - letter = response.json() - elif response.status_code == 401: - self._refresh_access_token() - letter = self.get_letter(letter_id) - else: - raise Exception('Unknown Error') - - return letter - - def get_relevant_shipments(self): - """ Retrieve not delivered shipments and shipments delivered today """ - - shipments = self.get_shipments() - relevant_shipments = [] - - for shipment in shipments: - - # Check if package is not delivered yet - if not shipment['status']['isDelivered']: - relevant_shipments.append(shipment) - continue - - # Check if package has been delivered today - if shipment['status']['delivery'] and \ - shipment['status']['delivery']['deliveryDate']: - delivery_date = datetime.strptime( - shipment['status']['delivery']['deliveryDate'][:19], "%Y-%m-%dT%H:%M:%S") - - if delivery_date.date() == datetime.today().date(): - relevant_shipments.append(shipment) + payload = { + 'grant_type': 'refresh_token', + 'refresh_token': self._refresh_token + } - return relevant_shipments + response = requests.request( + 'POST', AUTHENTICATE_URL, data=payload, headers={**headers, **DEFAULT_HEADER}) - def get_relevant_letters(self): - """ Retrieve letters with a future delivery date """ + data = response.json() - letters = self.get_letters() - relevant_letters = [] + if response.status_code != 200: + self._request_login() + else: + self._access_token = data['access_token'] + self._refresh_token = unquote(data['refresh_token']) + self._token_expires_in = data['expires_in'] + self._token_expires_at = datetime.now() + timedelta(0, (int(data['expires_in']) - 20)) - for letter in letters: - # Check if letter is scheduled for delivery in the future - if letter['expectedDeliveryDate']: - expected_delivery_date = datetime.strptime( - letter['expectedDeliveryDate'][:19], "%Y-%m-%dT%H:%M:%S") +class UnauthorizedException(Exception): + pass - if expected_delivery_date.date() >= datetime.today().date(): - relevant_letters.append(letter) - return relevant_letters +class PostnlApiException(Exception): + pass diff --git a/postnl_api/test_postnl_api.py b/postnl_api/test_postnl_api.py index 82be8cf..abc7341 100644 --- a/postnl_api/test_postnl_api.py +++ b/postnl_api/test_postnl_api.py @@ -15,22 +15,52 @@ def main(): username = args.username password = args.password # Login using your jouw.postnl.nl credentials - postnl = PostNL_API(username, password) + api = PostNL_API(username, password, 5) - # Get relevant shipments - print("Getting shipments") - shipments = postnl.get_relevant_shipments() - print("Number of shipments: ", len(shipments)) - print("Listing shipments:") - for shipment in shipments: - print (shipment['key']) + # Get packages + print("Get packages") + packages = api.get_delivery() + print("Number of packages to be delivered: ", len(packages)) + print("Listing packages:") + for package in packages: + print("ID: " + package.id) + print("Name: " + package.name) + print("Type: " + package.type) + print("Status: " + package.status) + print("Status Message: " + package.status_message) + print("URL: " + package.url) + if package.planned_date is not None: + print("Planned date: " + package.planned_date) + print("Planned from: " + package.planned_from) + print("Planned to: " + package.planned_to) + if package.delivery_date is not None: + print("Delivery date: " + package.delivery_date) - # Get letters - print("Getting letters") - letters = postnl.get_letters() - print("Number of letters: ", len(letters)) - print("Listing letters:") - print (letters) + packages = api.get_distribution() + print("Number of packages submitted: ", len(packages)) + print("Listing packages:") + for package in packages: + print("ID: " + package.id) + print("Name: " + package.name) + print("Type: " + package.type) + print("Status: " + package.status) + print("Status Message: " + package.status_message) + print("URL: " + package.url) + if package.planned_date is not None: + print("Planned date: " + package.planned_date) + print("Planned from: " + package.planned_from) + print("Planned to: " + package.planned_to) + if package.delivery_date is not None: + print("Delivery date: " + package.delivery_date) + + if api.is_letters_activated() is True: + letters = api.get_letters() + print("Number of letters: ", len(letters)) + print("Listing letters:") + for letter in letters: + print("ID: " + letter.id) + print("Image URL: " + letter.image) + print("Status Message: " + letter.status_message) if __name__ == '__main__': diff --git a/setup.py b/setup.py index 24a394a..9a20c17 100644 --- a/setup.py +++ b/setup.py @@ -1,7 +1,7 @@ from setuptools import setup, find_packages setup(name='postnl_api', - version='1.0.2', + version='2.0.0', description='Python wrapper for the PostNL API, a way to track packages using their online portal', url='https://github.com/imicknl/python-postnl-api', author='Mick Vleeshouwer',