From e0c980b0c3be1ee25523e167a3277153d142034b Mon Sep 17 00:00:00 2001 From: Kumi Date: Thu, 21 Sep 2023 14:15:26 +0200 Subject: [PATCH] Add support for DPD Romania Allow wrapping response in Shipment/Event objects --- pyproject.toml | 3 + src/dpdtrack/classes/__init__.py | 4 +- src/dpdtrack/classes/api.py | 99 ++++++++++++++++++++++++++++++-- src/dpdtrack/classes/shipment.py | 19 ++++++ test.py | 25 +++++--- 5 files changed, 138 insertions(+), 12 deletions(-) create mode 100644 src/dpdtrack/classes/shipment.py diff --git a/pyproject.toml b/pyproject.toml index 4f4580a..7392c91 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -17,6 +17,9 @@ classifiers = [ "License :: OSI Approved :: MIT License", "Operating System :: OS Independent", ] +dependencies = [ + "beautifulsoup4" +] [project.urls] "Homepage" = "https://kumig.it/kumitterer/dpdtrack" diff --git a/src/dpdtrack/classes/__init__.py b/src/dpdtrack/classes/__init__.py index d5681a0..f7ff27f 100644 --- a/src/dpdtrack/classes/__init__.py +++ b/src/dpdtrack/classes/__init__.py @@ -1,2 +1,4 @@ from .http import HTTPRequest -from .api import DPD \ No newline at end of file +from .api import DPDAT as DPD +from .api import DPDAT, DPDRO +from .shipment import Shipment, Event \ No newline at end of file diff --git a/src/dpdtrack/classes/api.py b/src/dpdtrack/classes/api.py index 1c0ba9b..1727d11 100644 --- a/src/dpdtrack/classes/api.py +++ b/src/dpdtrack/classes/api.py @@ -1,16 +1,31 @@ from hashlib import md5 from configparser import ConfigParser from urllib.parse import urlencode +from datetime import datetime import json -from .http import HTTPRequest +import bs4 + +from .http import HTTPRequest +from .shipment import Shipment, Event + +class DPDAT: + """ DPD Austria API + + This API is used to track packages in Austria. It also seems to work for + packages in Germany, but this is not extensively tested. + """ -class DPD: SEARCH = "https://www.mydpd.at/jws.php/parcel/search" VERIFY = "https://www.mydpd.at/jws.php/parcel/verify" - def tracking(self, tracking_number: str, postal_code: str = None): + def tracking(self, tracking_number: str, **kwargs): + """ Search for a tracking number """ + + postal_code = kwargs.get("postal_code", None) + wrap = kwargs.get("wrap", False) + if postal_code is None: endpoint = self.SEARCH payload = tracking_number @@ -22,6 +37,82 @@ class DPD: request.add_json_payload(payload) response = request.execute() - return response + if not wrap: + return response + shipment = Shipment() + shipment.tracking_number = response["data"][0]["pno"] + shipment.courier = self.__class__.__name__ + + shipment.events = [] + + for event in response["data"][0]["lifecycle"]["entries"]: + event_obj = Event() + + if "depotData" in event and event["depotData"] is not None: + event_obj.location = ", ".join(event['depotData']) + else: + event_obj.location = None + + event_obj.timestamp = datetime.strptime(event["datetime"], "%Y%m%d%H%M%S") + event_obj.description = event['state']['text'] + event_obj.raw = json.dumps(event) + + shipment.events.append(event_obj) + + shipment.raw = json.dumps(response) + + return shipment + +class DPDRO: + """ DPD Romania API """ + + URL = "https://tracking.dpd.ro/?shipmentNumber=%s&language=%s" + + def tracking(self, tracking_number: str, **kwargs): + """ Search for a tracking number """ + + language = kwargs.get("language", "en") + wrap = kwargs.get("wrap", False) + + request = HTTPRequest(self.URL % (tracking_number, language)) + response = request.execute(False).decode() + + if not wrap: + return response + + response = bs4.BeautifulSoup(response, features="html.parser") + + shipment = Shipment() + + header_table = response.find("span", {"class", "spanTableHeader"}) + shipment.tracking_number = header_table.text.split()[0] + shipment.courier = self.__class__.__name__ + + if remote_data := header_table.find("a"): + remote_courier = remote_data.get("href") + if "dpd.de" in remote_courier: + remote_courier = "DPDDE" + elif "mydpd.at" in remote_courier: + remote_courier = "DPDAT" + + shipment.remote = [(remote_data.text, remote_courier)] + + shipment.events = [] + + data_table = response.find("table", {"class": "standard-table"}) + + for row in data_table.find_all("tr")[1:]: + event_obj = Event() + + date, time, event_obj.description, event_obj.location = row.find_all("td") + + event_obj.timestamp = datetime.strptime(f"{date.text} {time.text}", "%d.%m.%Y %H:%M:%S") + event_obj.raw = row.prettify() + + shipment.events.append(event_obj) + + shipment.raw = response.prettify() + + return shipment \ No newline at end of file diff --git a/src/dpdtrack/classes/shipment.py b/src/dpdtrack/classes/shipment.py new file mode 100644 index 0000000..7c7bc1c --- /dev/null +++ b/src/dpdtrack/classes/shipment.py @@ -0,0 +1,19 @@ +from typing import List, Optional, Tuple +from datetime import datetime + +class Event: + """ A class representing an individual event. """ + + timestamp: datetime + description: str + location: str + raw: str + +class Shipment: + """ A class representing a shipment. """ + + tracking_number: str + courier: str + events: Optional[List[Event]] = None + foreign: Optional[Tuple[str, str]] = None + raw: str \ No newline at end of file diff --git a/test.py b/test.py index 1b8ce1a..f1d28fc 100644 --- a/test.py +++ b/test.py @@ -7,7 +7,7 @@ from dpdtrack import * class TestHTTPRequest(TestCase): def test_http_request(self): - http = HTTPRequest("https://httpbin.org/get") + http = HTTPRequest("https://httpbin.kumi.systems/get") response = http.execute() self.assertEqual(response["headers"]["User-Agent"], http.USER_AGENT) @@ -17,13 +17,24 @@ class TestDPD(TestCase): def test_tracking(self): tracking_number = "01155036780055" - response = self.api.tracking(tracking_number) - self.assertEqual(response["state"], "success") - self.assertEqual(response["data"][0]["pno"], tracking_number) + response = self.api.tracking(tracking_number, wrap=True) + self.assertTrue(response.events) + self.assertEqual(response.tracking_number, tracking_number) def test_tracking_with_postal_code(self): tracking_number = "01155036780055" postal_code = "8010" - response = self.api.tracking(tracking_number, postal_code) - self.assertEqual(response["state"], "success") - self.assertEqual(response["data"][0]["pno"], tracking_number) \ No newline at end of file + response = self.api.tracking(tracking_number, postal_code=postal_code, wrap=True) + self.assertTrue(response.events) + self.assertEqual(response.tracking_number, tracking_number) + +class DPDROTest(TestCase): + def setUp(self): + self.api = DPDRO() + + def test_tracking(self): + tracking_number = "80720052822" + response = self.api.tracking(tracking_number, wrap=True) + self.assertTrue(response.events) + self.assertEqual(response.tracking_number, tracking_number) + self.assertEqual(response.remote[0][1], "DPDDE") \ No newline at end of file