1
0

Ruff
Some checks failed
Verify Python project can be installed, loaded and have version checked / Test (push) Failing after 22s
Run Python tests (through Pytest) / Test (push) Failing after 24s

This commit is contained in:
Jon Michael Aanes 2025-01-22 17:31:00 +01:00
parent 809e75ad27
commit 13658958e4
4 changed files with 43 additions and 27 deletions

View File

@ -1,16 +1,17 @@
import logging
from . import parcelsapp
from . import secrets
from . import http
from . import http, parcelsapp, secrets
logger = logging.getLogger(__name__)
def main():
logging.basicConfig()
logger.setLevel('INFO')
parcelsapp_client: parcelsapp.ParcelsAppClient = parcelsapp.ParcelsAppClient(secrets.PARCELS_API_KEY)
parcelsapp_client: parcelsapp.ParcelsAppClient = parcelsapp.ParcelsAppClient(
secrets.PARCELS_API_KEY,
)
http.initialize_server(parcelsapp_client)

View File

@ -1,5 +1,6 @@
import dataclasses
@dataclasses.dataclass
class TrackingNumberEntry:
number: str

View File

@ -1,10 +1,10 @@
from bottle import route, run, template
from . import parcelsapp
from . import database
from . import database, parcelsapp
PARCELSAPP_CLIENT: parcelsapp.ParcelsAppClient | None = None
TEMPLATE = '''
TEMPLATE = """
<doctype HTML>
<html>
<head>
@ -96,7 +96,8 @@ a {
</body>
</html>
'''
"""
@route('/')
def index():
@ -105,13 +106,16 @@ def index():
tracking_results = PARCELSAPP_CLIENT.get_tracking_status(tracking_numbers)
tracking_results_by_id = {result.tracking_number: result for result in tracking_results}
tracking_results_by_id = {
result.tracking_number: result for result in tracking_results
}
derps = [(e, tracking_results_by_id.get(e.number)) for e in tracking_entries]
derps.sort(key=lambda x: x[1].latest_state().date,reverse=True)
derps.sort(key=lambda x: x[1].latest_state().date, reverse=True)
return template(TEMPLATE, tracking_results=derps)
def initialize_server(parcelsapp_client: parcelsapp.ParcelsAppClient):
global PARCELSAPP_CLIENT
PARCELSAPP_CLIENT = parcelsapp_client

View File

@ -1,11 +1,11 @@
import requests
import time
import datetime
import yaml
import dataclasses
import datetime
import logging
import time
from collections.abc import Iterator
import requests
logger = logging.getLogger(__name__)
URL_TRACKING = 'https://parcelsapp.com/api/v3/shipments/tracking'
@ -14,6 +14,7 @@ target_country = 'Denmark'
TRACKING_STATUS_CHECKING_INTERVAL = 1
@dataclasses.dataclass(frozen=True)
class ParcelState:
date: datetime.datetime
@ -34,14 +35,16 @@ class ParcelInfo:
def latest_state(self) -> ParcelState:
return max(self.states, key=lambda state: state.date)
class ParcelsAppClient:
class ParcelsAppClient:
def __init__(self, api_key: str):
self.api_key = api_key
def _request_json(self, method: str, url: str, **kwargs) -> dict:
request_json_data = {'apiKey': self.api_key, **kwargs}
response = requests.request(method=method,url=URL_TRACKING, json=request_json_data)
response = requests.request(
method=method, url=URL_TRACKING, json=request_json_data,
)
response.raise_for_status()
json_data = response.json()
if 'error' in json_data:
@ -50,7 +53,7 @@ class ParcelsAppClient:
return json_data
def check_tracking_status(self, uuid: str) -> dict:
""" Function to check tracking status with UUID"""
"""Function to check tracking status with UUID"""
json_data = self._request_json('GET', URL_TRACKING, uuid=uuid)
if json_data['done']:
logger.info('Tracking complete')
@ -61,7 +64,10 @@ class ParcelsAppClient:
return self.check_tracking_status(uuid)
def _get_tracking_status_to_json(self, tracking_ids: list[str]) -> dict:
shipments = [{'trackingId': id, 'language': 'en', 'country': target_country} for id in tracking_ids]
shipments = [
{'trackingId': id, 'language': 'en', 'country': target_country}
for id in tracking_ids
]
# Initiate tracking request
@ -71,19 +77,23 @@ class ParcelsAppClient:
return self.check_tracking_status(json_data['uuid'])
def get_tracking_status(self, tracking_ids: list[str]) -> Iterator[ParcelInfo]:
if len(tracking_ids) == 0:
return
for parcel_json in self._get_tracking_status_to_json(tracking_ids)['shipments']:
yield ParcelInfo(
tracking_number = parcel_json['trackingId'],
tracking_url = parcel_json['externalTracking'][0]['url'],
status = parcel_json['status'],
destination = parcel_json.get('destination'),
origin = parcel_json.get('origin'),
states = [ParcelState(status=s['status'],
tracking_number=parcel_json['trackingId'],
tracking_url=parcel_json['externalTracking'][0]['url'],
status=parcel_json['status'],
destination=parcel_json.get('destination'),
origin=parcel_json.get('origin'),
states=[
ParcelState(
status=s['status'],
date=datetime.datetime.fromisoformat(s['date']),
carrier=s['carrier']) for s in parcel_json['states']],
carrier=s['carrier'],
)
for s in parcel_json['states']
],
)