Add support for DPD Romania

Allow wrapping response in Shipment/Event objects
This commit is contained in:
Kumi 2023-09-21 14:15:26 +02:00
parent 0856d4cc0f
commit e0c980b0c3
Signed by: kumi
GPG key ID: ECBCC9082395383F
5 changed files with 138 additions and 12 deletions

View file

@ -17,6 +17,9 @@ classifiers = [
"License :: OSI Approved :: MIT License", "License :: OSI Approved :: MIT License",
"Operating System :: OS Independent", "Operating System :: OS Independent",
] ]
dependencies = [
"beautifulsoup4"
]
[project.urls] [project.urls]
"Homepage" = "https://kumig.it/kumitterer/dpdtrack" "Homepage" = "https://kumig.it/kumitterer/dpdtrack"

View file

@ -1,2 +1,4 @@
from .http import HTTPRequest from .http import HTTPRequest
from .api import DPD from .api import DPDAT as DPD
from .api import DPDAT, DPDRO
from .shipment import Shipment, Event

View file

@ -1,16 +1,31 @@
from hashlib import md5 from hashlib import md5
from configparser import ConfigParser from configparser import ConfigParser
from urllib.parse import urlencode from urllib.parse import urlencode
from datetime import datetime
import json 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" SEARCH = "https://www.mydpd.at/jws.php/parcel/search"
VERIFY = "https://www.mydpd.at/jws.php/parcel/verify" 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: if postal_code is None:
endpoint = self.SEARCH endpoint = self.SEARCH
payload = tracking_number payload = tracking_number
@ -22,6 +37,82 @@ class DPD:
request.add_json_payload(payload) request.add_json_payload(payload)
response = request.execute() 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

View file

@ -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

25
test.py
View file

@ -7,7 +7,7 @@ from dpdtrack import *
class TestHTTPRequest(TestCase): class TestHTTPRequest(TestCase):
def test_http_request(self): def test_http_request(self):
http = HTTPRequest("https://httpbin.org/get") http = HTTPRequest("https://httpbin.kumi.systems/get")
response = http.execute() response = http.execute()
self.assertEqual(response["headers"]["User-Agent"], http.USER_AGENT) self.assertEqual(response["headers"]["User-Agent"], http.USER_AGENT)
@ -17,13 +17,24 @@ class TestDPD(TestCase):
def test_tracking(self): def test_tracking(self):
tracking_number = "01155036780055" tracking_number = "01155036780055"
response = self.api.tracking(tracking_number) response = self.api.tracking(tracking_number, wrap=True)
self.assertEqual(response["state"], "success") self.assertTrue(response.events)
self.assertEqual(response["data"][0]["pno"], tracking_number) self.assertEqual(response.tracking_number, tracking_number)
def test_tracking_with_postal_code(self): def test_tracking_with_postal_code(self):
tracking_number = "01155036780055" tracking_number = "01155036780055"
postal_code = "8010" postal_code = "8010"
response = self.api.tracking(tracking_number, postal_code) response = self.api.tracking(tracking_number, postal_code=postal_code, wrap=True)
self.assertEqual(response["state"], "success") self.assertTrue(response.events)
self.assertEqual(response["data"][0]["pno"], tracking_number) 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")