From ad44b1e1ccc14e2e9ed06b8801c4b134f897015e Mon Sep 17 00:00:00 2001 From: Kumi Date: Wed, 30 Aug 2023 10:59:13 +0200 Subject: [PATCH] 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. --- README.md | 10 ++- pyproject.toml | 3 +- src/trackbert/__main__.py | 93 ++++++++++++++++++++++----- src/trackbert/classes/http.py | 21 ++++++ src/trackbert/classes/tracker.py | 47 ++++++++++---- src/trackbert/config.dist.ini | 4 ++ src/trackbert/trackers/base.py | 16 +++-- src/trackbert/trackers/dhl.py | 29 +++++++-- src/trackbert/trackers/fedex.py | 4 +- src/trackbert/trackers/gls.py | 2 +- src/trackbert/trackers/keydelivery.py | 18 ++++-- src/trackbert/trackers/postat.py | 2 +- 12 files changed, 197 insertions(+), 52 deletions(-) create mode 100644 src/trackbert/classes/http.py diff --git a/README.md b/README.md index f1d1847..6361dc6 100644 --- a/README.md +++ b/README.md @@ -50,7 +50,15 @@ tracking providers. To add a new shipment, run `trackbert --tracking-number --carrier `. 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 diff --git a/pyproject.toml b/pyproject.toml index 826be10..5c6d3a7 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -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] diff --git a/src/trackbert/__main__.py b/src/trackbert/__main__.py index 4745271..887ccd6 100644 --- a/src/trackbert/__main__.py +++ b/src/trackbert/__main__.py @@ -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()) diff --git a/src/trackbert/classes/http.py b/src/trackbert/classes/http.py new file mode 100644 index 0000000..48a91e8 --- /dev/null +++ b/src/trackbert/classes/http.py @@ -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") diff --git a/src/trackbert/classes/tracker.py b/src/trackbert/classes/tracker.py index c00637b..0d33085 100644 --- a/src/trackbert/classes/tracker.py +++ b/src/trackbert/classes/tracker.py @@ -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() \ No newline at end of file + self.notify("Trackbert", "Starting up") + await self.start_loop_async() diff --git a/src/trackbert/config.dist.ini b/src/trackbert/config.dist.ini index 8d78ae8..b791b17 100644 --- a/src/trackbert/config.dist.ini +++ b/src/trackbert/config.dist.ini @@ -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 \ No newline at end of file diff --git a/src/trackbert/trackers/base.py b/src/trackbert/trackers/base.py index e48d1bb..33ee8ec 100644 --- a/src/trackbert/trackers/base.py +++ b/src/trackbert/trackers/base.py @@ -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. diff --git a/src/trackbert/trackers/dhl.py b/src/trackbert/trackers/dhl.py index 9a225d1..4df651a 100644 --- a/src/trackbert/trackers/dhl.py +++ b/src/trackbert/trackers/dhl.py @@ -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"), ] diff --git a/src/trackbert/trackers/fedex.py b/src/trackbert/trackers/fedex.py index 89e38c4..ef33eaf 100644 --- a/src/trackbert/trackers/fedex.py +++ b/src/trackbert/trackers/fedex.py @@ -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"), ] diff --git a/src/trackbert/trackers/gls.py b/src/trackbert/trackers/gls.py index 24817d5..ce63bb3 100644 --- a/src/trackbert/trackers/gls.py +++ b/src/trackbert/trackers/gls.py @@ -27,7 +27,7 @@ class GLS(BaseTracker): def supported_carriers(self): return [ - ("gls", 100), + ("gls", 100, "GLS"), ] diff --git a/src/trackbert/trackers/keydelivery.py b/src/trackbert/trackers/keydelivery.py index 8fb468f..46feabb 100644 --- a/src/trackbert/trackers/keydelivery.py +++ b/src/trackbert/trackers/keydelivery.py @@ -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 diff --git a/src/trackbert/trackers/postat.py b/src/trackbert/trackers/postat.py index 8fe104f..b74b9e9 100644 --- a/src/trackbert/trackers/postat.py +++ b/src/trackbert/trackers/postat.py @@ -30,7 +30,7 @@ class PostAT(BaseTracker): def supported_carriers(self): return [ - ("austrian_post", 100), + ("austrian_post", 100, "Austrian Post"), ]