diff --git a/pyproject.toml b/pyproject.toml index 5c6d3a7..04738ad 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "trackbert" -version = "0.2.3" +version = "0.2.4" authors = [ { name="Kumi Mitterer", email="trackbert@kumi.email" }, ] @@ -23,6 +23,7 @@ dependencies = [ "glsapi", "fedextrack", "dhltrack", + "dpdtrack", "sqlalchemy", "alembic", "python-dateutil", diff --git a/src/trackbert/classes/tracker.py b/src/trackbert/classes/tracker.py index 0d33085..c430bd4 100644 --- a/src/trackbert/classes/tracker.py +++ b/src/trackbert/classes/tracker.py @@ -130,7 +130,7 @@ class Tracker: logging.exception(f"Error querying API for {shipment.tracking_number}") return - events = sorted(events, key=lambda x: x.event_time, reverse=True) + events = sorted(events, key=lambda x: x.event_time) if not events: logging.debug(f"No events found for {shipment.tracking_number}") @@ -154,7 +154,7 @@ class Tracker: ): event.shipment_id = shipment.id self.db.write_event(event) - self.notify_event(shipment, event, event == events[0]) + self.notify_event(shipment, event, event == events[-1]) def start_loop(self) -> Never: logging.debug("Starting loop") diff --git a/src/trackbert/trackers/dpd.py b/src/trackbert/trackers/dpd.py new file mode 100644 index 0000000..6f3bffa --- /dev/null +++ b/src/trackbert/trackers/dpd.py @@ -0,0 +1,43 @@ +from .base import BaseTracker +from ..classes.database import Event + +import json + +from datetime import datetime + +from dpdtrack.classes.api import DPD as DPDAPI + + +class DPD(BaseTracker): + def __init__(self, *args, **kwargs): + pass + + def get_status(self, tracking_number, carrier): + api = DPDAPI() + status = api.tracking(tracking_number) + + events = status["data"][0]["lifecycle"]["entries"] + + for event in events: + if "depotData" in event and event["depotData"] is not None: + event_location = f"[{', '.join(event['depotData'])}] " + else: + event_location = "" + + event_timestamp = datetime.strptime(event["datetime"], "%Y%m%d%H%M%S") + event_time = event_timestamp.strftime("%Y-%m-%d %H:%M:%S") + + yield Event( + shipment_id=0, + event_time=event_time, + event_description=f"{event_location}{event['state']['text']}", + raw_event=json.dumps(event), + ) + + def supported_carriers(self): + return [ + ("dpd", 100, "DPD (Austria)"), + ] + + +tracker = DPD