59 lines
1.7 KiB
Python
59 lines
1.7 KiB
Python
|
import requests
|
||
|
import time
|
||
|
import logging
|
||
|
import yaml
|
||
|
from . import secrets
|
||
|
|
||
|
logger = logging.getLogger(__name__)
|
||
|
|
||
|
URL_TRACKING = 'https://parcelsapp.com/api/v3/shipments/tracking'
|
||
|
|
||
|
target_country = 'Denmark'
|
||
|
|
||
|
TRACKING_STATUS_CHECKING_INTERVAL = 1
|
||
|
|
||
|
def request_json(method: str, url: str, **kwargs) -> dict:
|
||
|
response = requests.request(method=method,url=URL_TRACKING, json={'apiKey': secrets.PARCELS_API_KEY, **kwargs})
|
||
|
print(response)
|
||
|
response.raise_for_status()
|
||
|
json_data = response.json()
|
||
|
if 'error' in json_data:
|
||
|
msg = 'Error from endpoint: {}'.format(json_data['error'])
|
||
|
raise RuntimeError(msg)
|
||
|
return json_data
|
||
|
|
||
|
def check_tracking_status(uuid: str) -> dict:
|
||
|
""" Function to check tracking status with UUID"""
|
||
|
json_data = request_json('GET', URL_TRACKING, uuid=uuid)
|
||
|
if json_data['done']:
|
||
|
logger.info('Tracking complete')
|
||
|
return json_data
|
||
|
else:
|
||
|
logger.info('Tracking in progress...')
|
||
|
time.sleep(TRACKING_STATUS_CHECKING_INTERVAL)
|
||
|
return check_tracking_status(uuid)
|
||
|
|
||
|
def get_tracking_status(tracking_ids: list[str]):
|
||
|
shipments = [{'trackingId': id, 'language': 'en', 'country': target_country} for id in tracking_ids]
|
||
|
|
||
|
# Initiate tracking request
|
||
|
|
||
|
json_data = request_json('POST', URL_TRACKING, shipments=shipments)
|
||
|
if json_data.get('done'):
|
||
|
return json_data
|
||
|
return check_tracking_status(json_data['uuid'])
|
||
|
|
||
|
def main():
|
||
|
logging.basicConfig()
|
||
|
logger.setLevel('INFO')
|
||
|
|
||
|
tracking_ids = [
|
||
|
'00157128965207138207',
|
||
|
'00057151273127784840',
|
||
|
]
|
||
|
shipment_statuses = get_tracking_status(tracking_ids)
|
||
|
print(yaml.dump(shipment_statuses['shipments']))
|
||
|
|
||
|
if __name__ == '__main__':
|
||
|
main()
|