package-tracking/package_tracking/parcelsapp.py

import dataclasses
import datetime
import logging
import time
from collections.abc import Iterator

import requests

logger = logging.getLogger(__name__)

URL_TRACKING = 'https://parcelsapp.com/api/v3/shipments/tracking'

target_country = 'Denmark'

TRACKING_STATUS_CHECKING_INTERVAL = 1


@dataclasses.dataclass(frozen=True)
class ParcelState:
    date: datetime.datetime
    status: str
    carrier: str | None


@dataclasses.dataclass(frozen=True)
class ParcelInfo:
    tracking_number: str
    tracking_url: str
    status: str
    destination: str | None
    origin: str | None
    states: list[ParcelState]
    # TODO: More fields

    def latest_state(self) -> ParcelState:
        return max(self.states, key=lambda state: state.date)

class ParcelsApiError(RuntimeError):
    pass

class ParcelsAppClient:
    def __init__(self, api_key: str):
        assert api_key is not None, 'Missing API Key'
        self.api_key = api_key

    def _request_json(self, method: str, url: str, **kwargs) -> dict:
        request_json_data = {'apiKey': self.api_key, 'language': 'en', **kwargs}
        response = requests.request(
            method=method, url=url, json=request_json_data,
        )
        response.raise_for_status()
        json_data = response.json()
        if 'error' in json_data:
            msg = 'Error from endpoint: {}'.format(json_data['error'])
            raise ParcelsApiError(msg)
        return json_data

    def check_tracking_status(self, uuid: str) -> dict:
        """Function to check tracking status with UUID."""
        json_data = self._request_json('GET', URL_TRACKING, uuid=uuid)
        if json_data['done']:
            logger.info('Tracking complete')
            return json_data
        else:
            logger.info('Tracking in progress...')
            time.sleep(TRACKING_STATUS_CHECKING_INTERVAL)
            return self.check_tracking_status(uuid)

    def _get_tracking_status_to_json(self, tracking_ids: list[str]) -> dict:
        shipments = [
            {'trackingId': tracking_id, 'country': target_country}
            for tracking_id in tracking_ids
        ]

        # Initiate tracking request
        json_data = self._request_json('POST', URL_TRACKING, shipments=shipments)
        if json_data.get('done'):
            return json_data

        return self.check_tracking_status(json_data['uuid'])

    def get_tracking_status(self, tracking_ids: list[str]) -> Iterator[ParcelInfo]:
        if len(tracking_ids) == 0:
            return

        for parcel_json in self._get_tracking_status_to_json(tracking_ids)['shipments']:
            yield ParcelInfo(
                tracking_number=parcel_json['trackingId'],
                tracking_url=parcel_json['externalTracking'][0]['url'],
                status=parcel_json['status'],
                destination=parcel_json.get('destination'),
                origin=parcel_json.get('origin'),
                states=[
                    ParcelState(
                        status=s['status'],
                        date=datetime.datetime.fromisoformat(s['date']),
                        carrier=s['carrier'],
                    )
                    for s in parcel_json['states']
                ],
            )