kumi/trackbert
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

Update trackbert version to 0.2.3, add tabulate to dependencies, and refactor main function arguments in __main__.py. Also, introduce HTTPRequest class for handling HTTP requests.

Browse source
This commit is contained in:
Kumi 2023-08-30 10:59:13 +02:00
parent 751a7ad9e5
commit ad44b1e1cc
Signed by: kumi
GPG key ID: ECBCC9082395383F
12 changed files with 197 additions and 52 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

10
README.md
View file

@ -50,7 +50,15 @@ tracking providers.
To add a new shipment, run `trackbert --tracking-number <tracking-number> --carrier <carrier-id>`. Find the required carrier ID in the [KeyDelivery API management](https://app.kd100.com/api-management).
To run the main loop, run `trackbert`. This will check the status of all shipments every 5 minutes, and print the status to the console. If the status of a shipment changes, you will get a desktop notification.
To run the main loop, run `trackbert`. This will check the status of all shipments every minute, and print the status to the console. If the status of a shipment changes, you will get a desktop notification.
## Caveats
### DHL
By default, the script queries for updates for each active shipment once per minute. However, if you have the DHL API enabled, you will quickly run into the rate limit. Therefore, the script will only query for updates once per hour (`if minute == 0`). For the default 250 requests per day limit, this means that you can only track up to 10 shipments simultaneously.
You may request a higher rate limit from DHL. See the [DHL Developer Portal](https://developer.dhl.com/) for details. If you do this, you can set `ratelimited = 0` (note that 0/1 is a boolean value in the config file) in your `config.ini` to disable the rate limit.
## License

3
pyproject.toml
View file

@ -4,7 +4,7 @@ build-backend = "hatchling.build"
[project]
name = "trackbert"
version = "0.2.2"
version = "0.2.3"
authors = [
{ name="Kumi Mitterer", email="trackbert@kumi.email" },
]
@ -26,6 +26,7 @@ dependencies = [
"sqlalchemy",
"alembic",
"python-dateutil",
"tabulate",
]
[project.urls]

93
src/trackbert/__main__.py
View file

@ -1,5 +1,6 @@
from pykeydelivery import KeyDelivery
from pathlib import Path
from tabulate import tabulate
import json
import time
import subprocess
@ -18,9 +19,27 @@ def main():
# Arguments related to the tracker
parser.add_argument("--tracking-number", "-n", type=str, required=False)
parser.add_argument("--carrier", "-c", type=str, required=False)
parser.add_argument("--description", "-d", type=str, required=False)
parser.add_argument(
"--tracking-number",
"-n",
type=str,
required=False,
help="Tracking number of the shipment",
)
parser.add_argument(
"--carrier",
"-c",
type=str,
required=False,
help="Carrier code of the shipment – use --list-carriers to list all supported carriers",
)
parser.add_argument(
"--description",
"-d",
type=str,
required=False,
help="Optional description for the shipment",
)
parser.add_argument(
"--update",
"-u",
@ -36,10 +55,29 @@ def main():
help="Disable existing shipment",
)
parser.add_argument(
"--list-carriers",
"-l",
action="store_true",
required=False,
help="List supported carriers",
)
# Arguments related to the config file
parser.add_argument("--generate-config", action="store_true", required=False)
parser.add_argument("--config-file", "-C", type=str, required=False)
parser.add_argument(
"--generate-config",
action="store_true",
required=False,
help="Generate new config file",
)
parser.add_argument(
"--config-file",
"-C",
type=str,
required=False,
help="Path to the config file to use or generate (default: config.ini)",
)
args = parser.parse_args()
@ -60,26 +98,52 @@ def main():
# Load config file
if not config_file.exists():
if args.config_file and not config_file.exists():
print(f"Config file {config_file} does not exist. Use -g to generate it.")
exit(1)
tracker = Tracker(config_file)
db = Database("sqlite:///trackbert.db")
# List carriers if requested
if args.list_carriers:
print("Supported carriers:\n")
carriers = set(
[
(api[0], (api[3] if len(api) > 3 else None))
for api in tracker.apis
if not any(
[
others[1] > api[1]
for others in filter(lambda x: x[0] == api[0], tracker.apis)
]
)
]
)
print(tabulate(sorted(carriers, key=lambda x: x[0]), headers=["Code", "Name"]))
exit(0)
if args.tracking_number is not None and args.carrier is not None:
if shipment := db.get_shipment(args.tracking_number) and not args.update:
if (
shipment := tracker.db.get_shipment(args.tracking_number)
and not args.update
):
print(f"Shipment {args.tracking_number} already exists. Use -u to update.")
exit(1)
if shipment:
db.update_shipment(args.tracking_number, args.carrier, args.description)
tracker.db.update_shipment(
args.tracking_number, args.carrier, args.description
)
print(
f"Updated shipment for {args.tracking_number} with carrier {args.carrier}"
)
else:
db.create_shipment(args.tracking_number, args.carrier, args.description)
tracker.db.create_shipment(
args.tracking_number, args.carrier, args.description
)
print(
f"Created shipment for {args.tracking_number} with carrier {args.carrier}"
)
@ -88,17 +152,16 @@ def main():
if args.tracking_number is not None:
if args.disable:
if not db.get_shipment(args.tracking_number):
if not tracker.db.get_shipment(args.tracking_number):
print(f"Shipment {args.tracking_number} does not exist.")
exit(1)
db.disable_shipment(args.tracking_number)
tracker.db.disable_shipment(args.tracking_number)
print(f"Disabled shipment for {args.tracking_number}")
exit(0)
print("You must specify a carrier with -c")
exit(1)
tracker = Tracker()
asyncio.run(tracker.start_async())

21
src/trackbert/classes/http.py Normal file
View file

@ -0,0 +1,21 @@
from urllib.request import Request, urlopen
import json
class HTTPRequest(Request):
USER_AGENT = "Mozilla/5.0 (compatible; Trackbert/dev; +https://kumig.it/kumitterer/trackbert)"
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.add_header("User-Agent", self.USER_AGENT)
def execute(self, load_json: bool = True, *args, **kwargs):
response = urlopen(self, *args, **kwargs).read()
if load_json:
response = json.loads(response)
return response
def add_json_payload(self, payload: dict):
self.add_header("Content-Type", "application/json")
self.data = json.dumps(payload).encode("utf-8")

47
src/trackbert/classes/tracker.py
View file

@ -17,13 +17,15 @@ class Tracker:
loop_interval = 60
loop_timeout = 30
def __init__(self):
def __init__(self, config: Optional[PathLike] = None):
logging.basicConfig(
format="%(asctime)s %(levelname)s: %(message)s",
level=logging.DEBUG,
level=logging.WARN,
datefmt="%Y-%m-%d %H:%M:%S",
)
self._pre_start(config)
self.find_apis()
def find_apis(self):
@ -37,24 +39,32 @@ class Tracker:
logging.debug(f"Found API {api.stem}")
module = importlib.import_module(f"trackbert.trackers.{api.stem}")
try:
module = importlib.import_module(f"trackbert.trackers.{api.stem}")
except:
logging.error(f"Error loading class {api.stem}")
if "tracker" in module.__dict__:
tracker = module.tracker
logging.debug(f"Found tracker {api.stem}")
try:
api = tracker()
api = tracker(config=self.config_path)
carriers = api.supported_carriers()
for carrier, priority in carriers:
self.apis.append((carrier, priority, api))
except:
logging.exception(f"Error loading tracker {api.stem}")
for carrier in carriers:
self.apis.append((carrier[0], carrier[1], api, (carrier[2] if len(carrier) > 2 else None)))
except Exception as e:
logging.error(f"Error loading tracker {api.__class__.__name__}: {e}")
def query_api(self, tracking_number: str, carrier: str) -> list:
logging.debug(f"Querying API for {tracking_number} with carrier {carrier}")
for api_carrier, _, api in sorted(self.apis, key=lambda x: x[1], reverse=True):
for api_entry in sorted(self.apis, key=lambda x: x[1], reverse=True):
api_carrier = api_entry[0]
priority = api_entry[1]
api = api_entry[2]
name = api_entry[3] if len(api_entry) > 3 else None
if api_carrier == "*" or api_carrier == carrier:
logging.debug(
f"Using API {api.__class__.__name__} for {tracking_number} with carrier {carrier}"
@ -177,19 +187,28 @@ class Tracker:
await asyncio.sleep(self.loop_interval)
def _pre_start(self, config: Optional[PathLike] = None):
self.config_path = config
parser = ConfigParser()
parser.read(config or [])
self.database_uri = parser.get("Trackbert", "database", fallback="sqlite:///trackbert.db")
self.debug = parser.getboolean("Trackbert", "debug", fallback=False)
if self.debug:
logger = logging.getLogger()
logger.setLevel(logging.DEBUG)
self.database_uri = parser.get(
"Trackbert", "database", fallback="sqlite:///trackbert.db"
)
self.db = Database(self.database_uri)
self.loop_interval = parser.getint("Trackbert", "interval", fallback=60)
self.notify("Trackbert", "Starting up")
def start(self, config: Optional[PathLike] = None):
self._pre_start(config)
self.notify("Trackbert", "Starting up")
self.start_loop()
async def start_async(self, config: Optional[PathLike] = None):
self._pre_start(config)
await self.start_loop_async()
self.notify("Trackbert", "Starting up")
await self.start_loop_async()

4
src/trackbert/config.dist.ini
View file

@ -1,3 +1,6 @@
[Trackbert]
debug = 0
[KeyDelivery]
key = api_key
secret = api_secret
@ -9,3 +12,4 @@ secret = api_secret
[DHL]
key = api_key
secret = api_secret
ratelimited = 1

16
src/trackbert/trackers/base.py
View file

@ -1,18 +1,22 @@
from typing import Optional, Tuple, List, Generator
from ..classes.database import Event
class BaseTracker:
def __init__(self, *args, **kwargs):
pass
def get_status(self, tracking_number, carrier):
def get_status(self, tracking_number: str, carrier: str) -> Generator[Event, None, None]:
raise NotImplementedError()
def supported_carriers(self):
def supported_carriers(self) -> List[Tuple[str, int, Optional[str]]]:
"""Defines the carriers supported by this tracker.
Returns:
list: List of supported carriers as tuples of (carrier_code, priority),
where priority is an integer. The carrier with the highest priority
will be used when tracking a shipment. "*" can be used as a wildcard
to match all carriers.
list: List of supported carriers as tuples of (carrier_code, priority,
carrier_name (optional)), where priority is an integer. The carrier
with the highest priority will be used when tracking a shipment.
"*" can be used as a wildcard to match all carriers.
Raises:
NotImplementedError: When this method is not implemented by the subclass.

29
src/trackbert/trackers/dhl.py
View file

@ -7,13 +7,30 @@ from dateutil.parser import parse
import json
import logging
from datetime import datetime
from configparser import ConfigParser
class DHL(BaseTracker):
def __init__(self, *args, **kwargs):
self.api = DHLAPI.from_config("config.ini")
self.api = DHLAPI.from_config(str(kwargs.get("config")))
config = ConfigParser()
config.read(kwargs.get("config"))
self.ratelimited = config.getboolean("dhl", "ratelimited", fallback=True)
def get_status(self, tracking_number, carrier):
response = self.api.track(tracking_number)
if self.ratelimited:
if datetime.now().minute != 0:
logging.warn("Skipping DHL API call due to ratelimiting")
return
try:
response = self.api.track(tracking_number)
except Exception as e:
logging.error(f"Error getting events for {tracking_number}: {e}")
return
try:
all_events = response["shipments"][0]["events"]
@ -23,13 +40,11 @@ class DHL(BaseTracker):
logging.error(f"Error getting events for {tracking_number}: {all_events}")
return
events = sorted(
all_events, key=lambda x: x["timestamp"], reverse=True
)
events = sorted(all_events, key=lambda x: x["timestamp"], reverse=True)
for event in events:
event_time = parse(event["timestamp"]).strftime("%Y-%m-%d %H:%M:%S")
try:
event_locality = f"[{event['location']['address']['addressLocality']}] "
except KeyError:
@ -46,7 +61,7 @@ class DHL(BaseTracker):
def supported_carriers(self):
return [
("dhl", 100),
("dhl", 100, "DHL"),
]

4
src/trackbert/trackers/fedex.py
View file

@ -10,7 +10,7 @@ import logging
class FedEx(BaseTracker):
def __init__(self, *args, **kwargs):
self.api = FedExAPI.from_config("config.ini")
self.api = FedExAPI.from_config(str(kwargs.get("config")))
def get_status(self, tracking_number, carrier):
response = self.api.track_by_tracking_number(tracking_number)
@ -47,7 +47,7 @@ class FedEx(BaseTracker):
def supported_carriers(self):
return [
("fedex", 100),
("fedex", 100, "FedEx"),
]

2
src/trackbert/trackers/gls.py
View file

@ -27,7 +27,7 @@ class GLS(BaseTracker):
def supported_carriers(self):
return [
("gls", 100),
("gls", 100, "GLS"),
]

18
src/trackbert/trackers/keydelivery.py
View file

@ -1,5 +1,6 @@
from .base import BaseTracker
from ..classes.database import Event
from ..classes.http import HTTPRequest
from pykeydelivery import KeyDelivery as KeyDeliveryAPI
@ -9,7 +10,7 @@ import logging
class KeyDelivery(BaseTracker):
def __init__(self, *args, **kwargs):
self.api = KeyDeliveryAPI.from_config("config.ini")
self.api = KeyDeliveryAPI.from_config(str(kwargs.get("config")))
def get_status(self, tracking_number, carrier):
all_events = self.api.realtime(carrier, tracking_number)
@ -36,9 +37,18 @@ class KeyDelivery(BaseTracker):
)
def supported_carriers(self):
return [
("*", 1),
]
try:
request = HTTPRequest("https://app.kd100.com/console/utils/kdbm")
response = request.execute()
carriers = [
(carrier["code"], 1, carrier["name"])
for carrier in response["data"]
]
return carriers
except:
return [
("*", 1),
]
tracker = KeyDelivery

2
src/trackbert/trackers/postat.py
View file

@ -30,7 +30,7 @@ class PostAT(BaseTracker):
def supported_carriers(self):
return [
("austrian_post", 100),
("austrian_post", 100, "Austrian Post"),
]