diff --git a/.gitignore b/.gitignore index 4ec1f2b..cad92a6 100644 --- a/.gitignore +++ b/.gitignore @@ -99,4 +99,13 @@ venv.bak/ /site # mypy -.mypy_cache/ \ No newline at end of file +.mypy_cache/ + +# intellij +.idea/ + +# vscode +.vscode/ + +#test script +test.py \ No newline at end of file diff --git a/CHANGELOG.md b/CHANGELOG.md index b6501bb..c1bc4fa 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,13 @@ All notable changes to this project will be documented in this file. The format is based on [Keep a Changelog](http://keepachangelog.com/) and this project adheres to [Semantic Versioning](http://semver.org/). +## 1.2.0 - 2019-08-08 +### Added +- Created packages and letters classes +- Isoformat parsed datetime for delivery and planned dates +- Added properties for is_delivered and delivery_today +- Renamed methods to get_X with X: deliveries, distributions and letters + ## 1.0.2 - 2018-05-28 ### Fixed - Traceback when no deliveryDate is found in shipment diff --git a/README.md b/README.md index dc8b244..efc74bf 100644 --- a/README.md +++ b/README.md @@ -19,19 +19,29 @@ from postnl_api import PostNL_API # Login using your jouw.postnl.nl credentials postnl = PostNL_API('email@domain.com', 'password') -# Get relevant shipments -shipments = postnl.get_relevant_shipments() +# Get relevant deliveries +print("Getting relevant deliveries") +rel_deliveries = postnl.get_relevant_deliveries() +for delivery in rel_deliveries: + print(delivery.debug_string) + +# Get relevant deliveries +print("Getting all deliveries") +all_deliveries = postnl.get_deliveries() +for delivery in all_deliveries: + print(delivery.debug_string) + +# Get relevant deliveries +print("Getting all distributions (sent packages)") +distributions = postnl.get_distributions() +for distribution in distributions: + print(distribution.debug_string) -for shipment in shipments: - name = shipment['settings']['title'] - status = shipment['status']['formatted']['short'] - status = postnl.parse_datetime(status, '%d-%m-%Y', '%H:%M') - - print (shipment['key'] + ' ' + name + ' ' + status) - # Get letters +print("Getting all letters, if that function is turned on") letters = postnl.get_letters() -print (letters) +for letter in letters: + print(letter.debug_string) ``` ## Miscellaneous 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..44b1c6d --- /dev/null +++ b/postnl_api/items/letter.py @@ -0,0 +1,18 @@ +from datetime import datetime + + +class Letter(object): + def __init__(self, data, documents): + self.id = data.get("barcode") + self.delivery_date = datetime.fromisoformat(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 __str__(self): + return f"{self.id} {self.status_message} {self.delivery_date.date()}" diff --git a/postnl_api/items/package.py b/postnl_api/items/package.py new file mode 100644 index 0000000..d68684e --- /dev/null +++ b/postnl_api/items/package.py @@ -0,0 +1,41 @@ +from datetime import datetime + + +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 = datetime.fromisoformat( + 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 = datetime.fromisoformat( + 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") + + @property + def is_delivered(self): + return self.status == "Delivered" + + @property + def delivery_today(self): + return self.delivery_date.date() == datetime.today().date() + + def __str__(self): + return f"{self.id} {self.name} {self.type} {self.status} {self.status_message} {self.delivery_date.date()} {self.planned_date} {self.planned_from} {self.planned_to}" diff --git a/postnl_api/postnl_api.py b/postnl_api/postnl_api.py index d264046..34aee56 100644 --- a/postnl_api/postnl_api.py +++ b/postnl_api/postnl_api.py @@ -1,265 +1,223 @@ """ Python wrapper for the PostNL API """ +import logging +import time from datetime import datetime, timedelta -import re +from urllib.parse import unquote import requests -BASE_URL = 'https://jouw.postnl.nl' +from postnl_api.items.package import Package +from postnl_api.items.letter import Letter -AUTHENTICATE_URL = BASE_URL + '/mobile/token' -SHIPMENTS_URL = BASE_URL + '/mobile/api/shipments' -PROFILE_URL = BASE_URL + '/mobile/api/profile' -LETTERS_URL = BASE_URL + '/mobile/api/letters' -VALIDATE_LETTERS_URL = BASE_URL + '/mobile/api/letters/validation' +BASE_URL = "https://jouw.postnl.nl" -DEFAULT_HEADER = { - 'api-version': '4.7', - 'user-agent': 'PostNL/1 CFNetwork/889.3 Darwin/17.2.0', -} +AUTHENTICATE_URL = BASE_URL + "/mobile/token" +SHIPMENTS_URL = BASE_URL + "/mobile/api/shipments" +PROFILE_URL = BASE_URL + "/mobile/api/profile" +LETTERS_URL = BASE_URL + "/mobile/api/letters" +VALIDATE_LETTERS_URL = BASE_URL + "/mobile/api/letters/validation" +## PROFILE_URL +## isMyMailAvailable: true / false +## hasPendingMyMailValidation: true / false -class UnauthorizedException(Exception): - pass - - -class PostNL_API(object): - """ Interface class for the PostNL API """ - - def __init__(self, user, password): - """ 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']) - - def _is_token_expired(self): - """ Check if access token is expired """ - if (datetime.now() > self._token_expires_at): - self._refresh_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() - - self._access_token = data['access_token'] - - def parse_datetime(self, text, dateFormat='%d-%m-%Y', timeFormat='%H:%M'): - - def parse_date(date): - return datetime.strptime(date.group(1) - .replace(' ', '')[:-6], '%Y-%m-%dT%H:%M:%S').strftime(dateFormat) - - def parse_time(date): - return datetime.strptime(date.group(1) - .replace(' ', '')[:-6], '%Y-%m-%dT%H:%M:%S').strftime(timeFormat) - - text = re.sub(r'{(?:Date|dateAbs):(.*?)}', parse_date, text) - text = re.sub(r'{(?:time):(.*?)}', parse_time, text) - - return text - - def get_shipments(self): - """ Retrieve shipments """ - - self._is_token_expired() - - headers = { - 'authorization': 'Bearer ' + self._access_token - } - - response = requests.request( - 'GET', SHIPMENTS_URL, headers={**headers, **DEFAULT_HEADER}) - - if response.status_code == 401: - self._refresh_access_token() - shipments = self.get_shipments() - else: - shipments = response.json() - - return shipments - - def get_shipment(self, shipment_id): - """ Retrieve single shipment by id """ - - self._is_token_expired() +## VALIDATE_LETTERS_URL +## status: Validated - headers = { - 'authorization': 'Bearer ' + self._access_token - } +## CHANGE NAME +## PATCH /mobile/api/shipments/{package-key} - response = requests.request( - 'GET', SHIPMENTS_URL + '/' + shipment_id, headers={**headers, **DEFAULT_HEADER}) +## DELETE SHIPMENT +## DELETE /mobile/api/shipments/{package-key} - if response.status_code == 401: - self._refresh_access_token() - shipments = self.get_shipment(shipment_id) - else: - shipments = response.json() +## DELETE LETTER +## DELETE /mobile/api/letters/{letter-barcode} - return shipments +DEFAULT_HEADER = {"api-version": "4.16", "X-Client-Library": "python-postnl-api"} - def get_profile(self): - """ Retrieve profile """ +REFRESH_RATE = 120 - self._is_token_expired() +_LOGGER = logging.getLogger(__name__) - headers = { - 'authorization': 'Bearer ' + self._access_token - } - response = requests.request( - 'GET', PROFILE_URL, headers={**headers, **DEFAULT_HEADER}) +class PostNL_API(object): + """ Interface class for the PostNL API """ - if response.status_code == 401: - self._refresh_access_token() - profile = self.get_profile() - else: - profile = response.json() + def __init__(self, user, password, refresh_rate=REFRESH_RATE): + """ Constructor """ + self._user = user + self._password = password + self._deliveries = {} + self._distributions = {} + self._letters = {} + self._letters_activated = False + self._last_refresh = None + self._refresh_rate = refresh_rate + self._request_login() + + @property + def _token_expired(self): + """ checks whether or not the token is expired """ + return datetime.now() > self._token_expires_at + + def _update(self): + """ Update the cache """ + current_time = int(time.time()) + last_refresh = 0 if self._last_refresh is None else self._last_refresh + + if current_time >= (last_refresh + self._refresh_rate): + self._update_packages() + self._update_letter_status() + self._update_letters() + self._last_refresh = int(time.time()) + + def _update_packages(self): + """ Retrieve packages """ + packages = self._request_update(SHIPMENTS_URL) + if packages is False: + return + + self._deliveries = {} + self._distributions = {} + + for package in packages: + if package.get("settings").get("box") == "Sender": + self._distributions[package["key"]] = Package(package) + else: + self._deliveries[package["key"]] = Package(package) + + def _update_letters(self): + """ Retrieve letters """ + if self._letters_activated is False: + return - return profile + letters = self._request_update(LETTERS_URL) + if letters is False: + return - def validate_letters(self): - """ Retrieve letter validation status """ + self._letters = {} - self._is_token_expired() + for letter in letters: + documents = self._request_update(LETTERS_URL + "/" + letter["barcode"]) + self._letters[letter["barcode"]] = Letter(letter, documents) + + def _update_letter_status(self): + """ update the state of being able to see letters """ + validate = self._request_update(VALIDATE_LETTERS_URL) + + if validate.get("status") == "Validated": + self._letters_activated = True + + def get_relevant_deliveries(self): + """ filter shipments to today's and future shipments """ + self._update() + return [ + d + for d in self._deliveries.values() + if (not d.is_delivered) or (d.is_delivered and d.delivery_today) + ] + + def get_deliveries(self): + """ Get all packages to be delivered to you """ + self._update() + return self._deliveries.values() + + def get_distributions(self): + """ Get all packages submitted by you """ + self._update() + return self._distributions.values() + def get_letters(self): + """ Get all letters to be delivered to you """ + self._update() + return self._letters.values() + + @property + def is_letters_activated(self): + """ Return if letters are activated or not """ + return self._letters_activated + + def _request_update(self, url, count=0, max=3): + """ Perform a request to update information """ + if self._token_expired: + self._request_access_token() headers = { - 'authorization': 'Bearer ' + self._access_token + "authorization": "Bearer " + self._access_token, + "Content-Type": "application/json", } - - response = requests.request( - 'GET', VALIDATE_LETTERS_URL, headers={**headers, **DEFAULT_HEADER}) + response = requests.request("GET", url, headers={**headers, **DEFAULT_HEADER}) if response.status_code == 401: - self._refresh_access_token() - validation = self.validate_letters() - else: - validation = response.json() + count += 1 + _LOGGER.debug(f"Access denied. Failed to refresh, attempt {count} of {max}.") + self._request_update(url, count, max) - return validation + if response.status_code != 200: + _LOGGER.error("Unable to perform request " + str(response.content)) + return False - def get_letters(self): - """ Retrieve letters """ + return response.json() - self._is_token_expired() + def _request_login(self): + headers = {"Content-Type": "application/x-www-form-urlencoded"} - 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() - - # TODO Add validation / exception handling - # if letters['type'] == 'ProfileValidationFeatureMissing': - # _LOGGER.error(letters['message']) - # return [] + try: + response = requests.request( + "POST", + AUTHENTICATE_URL, + data=payload, + headers={**headers, **DEFAULT_HEADER}, + ) + data = response.json() - return letters + except Exception: + raise (UnauthorizedException()) - def get_letter(self, letter_id): - """ Retrieve single letter by id """ + if "error" in data: + raise UnauthorizedException(data["error"]) - self._is_token_expired() + 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) + ) - headers = { - 'authorization': 'Bearer ' + self._access_token - } + def _request_access_token(self): + """ Refresh access_token """ + headers = {"Content-Type": "application/x-www-form-urlencoded"} + payload = {"grant_type": "refresh_token", "refresh_token": self._refresh_token} response = requests.request( - 'GET', LETTERS_URL + '/' + letter_id, headers={**headers, **DEFAULT_HEADER}) + "POST", + AUTHENTICATE_URL, + data=payload, + 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) + data = response.json() + if response.status_code != 200: + self._request_login() else: - raise Exception('Unknown Error') - - return letter + 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) + ) - 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) - - return relevant_shipments - - def get_relevant_letters(self): - """ Retrieve letters with a future delivery date """ - - letters = self.get_letters() - relevant_letters = [] - - 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..8b881dc 100644 --- a/postnl_api/test_postnl_api.py +++ b/postnl_api/test_postnl_api.py @@ -5,33 +5,32 @@ def main(): """Main function.""" parser = argparse.ArgumentParser(description="Run the test for PostNL_API") - parser.add_argument( - 'username', type=str, - help="Your username (email address)") - parser.add_argument( - 'password', type=str, - help="Your password") + parser.add_argument("username", type=str, help="Your username (email address)") + parser.add_argument("password", type=str, help="Your password") args = parser.parse_args() 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_deliveries() + print("Number of packages to be delivered: ", len(packages)) + print("Listing packages:") + [print(p) for p in packages] - # Get letters - print("Getting letters") - letters = postnl.get_letters() - print("Number of letters: ", len(letters)) - print("Listing letters:") - print (letters) + packages = api.get_distributions() + print("Number of packages to be distributed: ", len(packages)) + print("Listing packages:") + [print(p) for p in packages] + if api.is_letters_activated(): + letters = api.get_letters() + print("Number of letters: ", len(letters)) + print("Listing letters:") + [print(l) for l in letters] -if __name__ == '__main__': + +if __name__ == "__main__": main() diff --git a/setup.py b/setup.py index 24a394a..f9ce487 100644 --- a/setup.py +++ b/setup.py @@ -1,12 +1,14 @@ from setuptools import setup, find_packages -setup(name='postnl_api', - version='1.0.2', - 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', - author_email='mick@imick.nl', - license='MIT', - install_requires=['requests>=2.0'], - packages=find_packages(), - zip_safe=True) +setup( + name="postnl_api", + version="1.2.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", + author_email="mick@imick.nl", + license="MIT", + install_requires=["requests>=2.0"], + packages=find_packages(), + zip_safe=True, +)