Kumi
0e0b24daba
The commit updates the icon path used in the notify-send command inside the Tracker class. The path to the parcel delivery icon is modified to correctly locate the image file.
118 lines
No EOL
4.3 KiB
Python
118 lines
No EOL
4.3 KiB
Python
import logging
|
|
import subprocess
|
|
import time
|
|
import importlib
|
|
from pathlib import Path
|
|
from typing import Optional, Tuple, Never
|
|
|
|
from .database import Database
|
|
from trackers.base import BaseTracker
|
|
|
|
from pykeydelivery import KeyDelivery
|
|
|
|
class Tracker:
|
|
def __init__(self):
|
|
logging.basicConfig(
|
|
format="%(asctime)s %(levelname)s: %(message)s",
|
|
level=logging.DEBUG,
|
|
datefmt="%Y-%m-%d %H:%M:%S",
|
|
)
|
|
|
|
self.find_apis()
|
|
|
|
def find_apis(self):
|
|
logging.debug("Finding APIs")
|
|
|
|
self.apis = []
|
|
|
|
for api in Path(__file__).parent.parent.glob("trackers/*.py"):
|
|
if api.name in ("__init__.py", "base.py"):
|
|
continue
|
|
|
|
logging.debug(f"Found API {api.stem}")
|
|
|
|
module = importlib.import_module(f"trackers.{api.stem}")
|
|
|
|
if "tracker" in module.__dict__:
|
|
tracker = module.tracker
|
|
logging.debug(f"Found tracker {api.stem}")
|
|
try:
|
|
carriers = tracker.supported_carriers()
|
|
api = tracker()
|
|
|
|
for carrier, priority in carriers:
|
|
self.apis.append((carrier, priority, api))
|
|
except:
|
|
logging.exception(f"Error loading tracker {name}")
|
|
|
|
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):
|
|
if api_carrier == "*" or api_carrier == carrier:
|
|
logging.debug(f"Using API {api.__class__.__name__} for {tracking_number} with carrier {carrier}")
|
|
return list(api.get_status(tracking_number, carrier))
|
|
|
|
def notify(self, title: str, message: str, urgency: str = "normal", timeout: Optional[int] = 5000) -> None:
|
|
logging.debug(f"Sending notification: {title} - {message}")
|
|
|
|
command = [
|
|
"notify-send",
|
|
"-a", "trackbert",
|
|
"-u", urgency,
|
|
"-i", str(Path(__file__).parent.parent / "assets" / "parcel-delivery-icon.webp"),
|
|
]
|
|
|
|
if timeout:
|
|
command += ["-t", str(timeout)]
|
|
|
|
command = command + [title, message]
|
|
|
|
try:
|
|
subprocess.run(command)
|
|
|
|
except FileNotFoundError:
|
|
logging.warning("notify-send not found, not sending notification")
|
|
|
|
def start_loop(self) -> Never:
|
|
logging.debug("Starting loop")
|
|
|
|
while True:
|
|
for shipment in self.db.get_shipments():
|
|
shipment_id = shipment.id
|
|
tracking_number = shipment.tracking_number
|
|
carrier = shipment.carrier
|
|
description = shipment.description
|
|
|
|
logging.debug(f"Checking shipment {tracking_number} with carrier {carrier}")
|
|
|
|
latest_known_event = self.db.get_latest_event(shipment_id)
|
|
|
|
events = self.query_api(tracking_number, carrier)
|
|
events = sorted(events, key=lambda x: x.event_time, reverse=True)
|
|
|
|
if latest_known_event:
|
|
logging.debug(f"Latest known event for {tracking_number}: {latest_known_event.event_description} - {latest_known_event.event_time}")
|
|
else:
|
|
logging.debug(f"No known events for {tracking_number}")
|
|
|
|
logging.debug(f"Latest upstream event for {tracking_number}: {events[0].event_description} - {events[0].event_time}")
|
|
|
|
latest = True
|
|
|
|
for event in events:
|
|
if latest_known_event is None or event.event_time > latest_known_event.event_time:
|
|
event.shipment_id = shipment_id
|
|
self.db.write_event(event)
|
|
|
|
logging.info(f"New event for {tracking_number}: {event.event_description} - {event.event_time}")
|
|
self.notify(f"New event for {description or tracking_number}", event.event_description + " - " + event.event_time, urgency="critical" if latest else "normal")
|
|
|
|
latest = False
|
|
|
|
time.sleep(300)
|
|
|
|
def start(self):
|
|
self.db = Database('sqlite:///trackbert.db')
|
|
self.notify("Trackbert", "Starting up")
|
|
self.start_loop() |