HTML server
This commit is contained in:
parent
ddd9d6e21f
commit
9c26fe7f9d
|
@ -1,5 +1,4 @@
|
||||||
import logging
|
import logging
|
||||||
import yaml
|
|
||||||
|
|
||||||
from . import parcelsapp
|
from . import parcelsapp
|
||||||
from . import secrets
|
from . import secrets
|
||||||
|
@ -7,25 +6,13 @@ from . import http
|
||||||
|
|
||||||
logger = logging.getLogger(__name__)
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
def main_cli(parcelsapp_client: parcelsapp.ParcelsAppClient):
|
|
||||||
tracking_ids = [
|
|
||||||
'00157128965207138207',
|
|
||||||
'00057151273127784840',
|
|
||||||
]
|
|
||||||
shipment_statuses = parcelsapp_client.get_tracking_status(tracking_ids)
|
|
||||||
print(yaml.dump(shipment_statuses['shipments']))
|
|
||||||
|
|
||||||
|
|
||||||
def main():
|
def main():
|
||||||
logging.basicConfig()
|
logging.basicConfig()
|
||||||
logger.setLevel('INFO')
|
logger.setLevel('INFO')
|
||||||
|
|
||||||
parcelsapp_client: parcelsapp.ParcelsAppClient = parcelsapp.ParcelsAppClient(secrets.PARCELS_API_KEY)
|
parcelsapp_client: parcelsapp.ParcelsAppClient = parcelsapp.ParcelsAppClient(secrets.PARCELS_API_KEY)
|
||||||
|
|
||||||
if True:
|
http.initialize_server(parcelsapp_client)
|
||||||
main_cli(parcelsapp_client)
|
|
||||||
else:
|
|
||||||
http.initialize_server()
|
|
||||||
|
|
||||||
|
|
||||||
if __name__ == '__main__':
|
if __name__ == '__main__':
|
||||||
|
|
33
package_tracking/database.py
Normal file
33
package_tracking/database.py
Normal file
|
@ -0,0 +1,33 @@
|
||||||
|
import dataclasses
|
||||||
|
|
||||||
|
@dataclasses.dataclass
|
||||||
|
class TrackingNumberEntry:
|
||||||
|
number: str
|
||||||
|
name: str
|
||||||
|
|
||||||
|
def __post_init__(self):
|
||||||
|
assert ' ' not in self.number
|
||||||
|
assert '\t' not in self.number
|
||||||
|
assert '\n' not in self.number
|
||||||
|
assert ' ' not in self.name
|
||||||
|
assert '\t' not in self.name
|
||||||
|
assert '\n' not in self.name
|
||||||
|
|
||||||
|
|
||||||
|
FILEPATH = 'output/entries_db.txt'
|
||||||
|
|
||||||
|
|
||||||
|
def get_tracking_numbers() -> list[TrackingNumberEntry]:
|
||||||
|
with open(FILEPATH) as f:
|
||||||
|
lines = f.read().split('\n')
|
||||||
|
|
||||||
|
lines = [line.split(' ') for line in lines if len(line) > 0]
|
||||||
|
return [TrackingNumberEntry(line[0], line[1]) for line in lines]
|
||||||
|
|
||||||
|
|
||||||
|
def add_tracking_number(tracking_number: TrackingNumberEntry) -> None:
|
||||||
|
with open(FILEPATH, 'a') as f:
|
||||||
|
f.write(tracking_number.number)
|
||||||
|
f.write(' ')
|
||||||
|
f.write(tracking_number.name)
|
||||||
|
f.write('\n')
|
|
@ -1,15 +1,52 @@
|
||||||
from bottle import route, run, template
|
from bottle import route, run, template
|
||||||
from . import parcelsapp
|
from . import parcelsapp
|
||||||
|
from . import database
|
||||||
|
|
||||||
PARCELSAPP_CLIENT: parcelsapp.ParcelsAppClient | None = None
|
PARCELSAPP_CLIENT: parcelsapp.ParcelsAppClient | None = None
|
||||||
|
|
||||||
TEMPLATE = '''
|
TEMPLATE = '''
|
||||||
|
<doctype HTML>
|
||||||
|
<html>
|
||||||
|
<head>
|
||||||
|
<style type="text/css">
|
||||||
|
|
||||||
|
|
||||||
|
</style>
|
||||||
|
</head>
|
||||||
|
<body>
|
||||||
|
|
||||||
|
<h1>My very own thingy</h1>
|
||||||
|
|
||||||
|
<main>
|
||||||
|
|
||||||
|
<div class="tracking-grid">
|
||||||
|
% for entry, tracking_data in tracking_results:
|
||||||
|
<div class="tracking-name"><a href="{{tracking_data.tracking_url}}">{{ entry.name }}/{{entry.number}}</a></div>
|
||||||
|
<div class="tracking-status">{{ tracking_data.status }}</div>
|
||||||
|
<div class="tracking-status">{{ tracking_data.latest_state().status }}</div>
|
||||||
|
<div class="tracking-status">{{ tracking_data.latest_state().date }}</div>
|
||||||
|
% end
|
||||||
|
</div>
|
||||||
|
|
||||||
|
</main>
|
||||||
|
|
||||||
|
</body>
|
||||||
|
</html>
|
||||||
'''
|
'''
|
||||||
|
|
||||||
@route('/')
|
@route('/')
|
||||||
def index():
|
def index():
|
||||||
|
tracking_entries = database.get_tracking_numbers()
|
||||||
|
tracking_numbers = [e.number for e in tracking_entries]
|
||||||
|
|
||||||
return template(TEMPLATE, name=name)
|
tracking_results = PARCELSAPP_CLIENT.get_tracking_status(tracking_numbers)
|
||||||
|
|
||||||
|
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)
|
||||||
|
|
||||||
|
return template(TEMPLATE, tracking_results=derps)
|
||||||
|
|
||||||
def initialize_server(parcelsapp_client: parcelsapp.ParcelsAppClient):
|
def initialize_server(parcelsapp_client: parcelsapp.ParcelsAppClient):
|
||||||
global PARCELSAPP_CLIENT
|
global PARCELSAPP_CLIENT
|
||||||
|
|
|
@ -1,6 +1,10 @@
|
||||||
import requests
|
import requests
|
||||||
import time
|
import time
|
||||||
|
import datetime
|
||||||
|
import yaml
|
||||||
|
import dataclasses
|
||||||
import logging
|
import logging
|
||||||
|
from collections.abc import Iterator
|
||||||
|
|
||||||
logger = logging.getLogger(__name__)
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
@ -10,6 +14,26 @@ target_country = 'Denmark'
|
||||||
|
|
||||||
TRACKING_STATUS_CHECKING_INTERVAL = 1
|
TRACKING_STATUS_CHECKING_INTERVAL = 1
|
||||||
|
|
||||||
|
@dataclasses.dataclass(frozen=True)
|
||||||
|
class ParcelState:
|
||||||
|
date: datetime.datetime
|
||||||
|
status: str
|
||||||
|
carrier: str | None
|
||||||
|
|
||||||
|
|
||||||
|
@dataclasses.dataclass(frozen=True)
|
||||||
|
class ParcelInfo:
|
||||||
|
tracking_number: str
|
||||||
|
tracking_url: str
|
||||||
|
status: str
|
||||||
|
destination: str | None
|
||||||
|
origin: str | None
|
||||||
|
states: list[ParcelState]
|
||||||
|
# TODO: More fields
|
||||||
|
|
||||||
|
def latest_state(self) -> ParcelState:
|
||||||
|
return max(self.states, key=lambda state: state.date)
|
||||||
|
|
||||||
class ParcelsAppClient:
|
class ParcelsAppClient:
|
||||||
|
|
||||||
def __init__(self, api_key: str):
|
def __init__(self, api_key: str):
|
||||||
|
@ -18,7 +42,6 @@ class ParcelsAppClient:
|
||||||
def _request_json(self, method: str, url: str, **kwargs) -> dict:
|
def _request_json(self, method: str, url: str, **kwargs) -> dict:
|
||||||
request_json_data = {'apiKey': self.api_key, **kwargs}
|
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)
|
||||||
print(response)
|
|
||||||
response.raise_for_status()
|
response.raise_for_status()
|
||||||
json_data = response.json()
|
json_data = response.json()
|
||||||
if 'error' in json_data:
|
if 'error' in json_data:
|
||||||
|
@ -37,7 +60,7 @@ class ParcelsAppClient:
|
||||||
time.sleep(TRACKING_STATUS_CHECKING_INTERVAL)
|
time.sleep(TRACKING_STATUS_CHECKING_INTERVAL)
|
||||||
return self.check_tracking_status(uuid)
|
return self.check_tracking_status(uuid)
|
||||||
|
|
||||||
def get_tracking_status(self, tracking_ids: list[str]):
|
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
|
# Initiate tracking request
|
||||||
|
@ -45,5 +68,22 @@ class ParcelsAppClient:
|
||||||
json_data = self._request_json('POST', URL_TRACKING, shipments=shipments)
|
json_data = self._request_json('POST', URL_TRACKING, shipments=shipments)
|
||||||
if json_data.get('done'):
|
if json_data.get('done'):
|
||||||
return json_data
|
return json_data
|
||||||
|
|
||||||
return self.check_tracking_status(json_data['uuid'])
|
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'],
|
||||||
|
date=datetime.datetime.fromisoformat(s['date']),
|
||||||
|
carrier=s['carrier']) for s in parcel_json['states']],
|
||||||
|
)
|
||||||
|
|
Loading…
Reference in New Issue
Block a user