toyota.py

#!/usr/bin/env python3

import os
import sys
import json
import logging
import pathlib
import argparse

import requests

from datetime import datetime

from typing import Optional, Callable


logging.basicConfig(
    stream=sys.stderr,
    level=logging.WARN
)


DATE_STRUCT: dict[str, dict] = {
    "steps": {},
    "deliveries": {}
}


class TME:
    HEADERS_BASE = {
        "x-tme-brand": "toyota",
        "x-tme-lc": "en-gb"
    }

    TOKEN_HEADER = "x-tme-token"

    def __init__(self, username: str, password: str):
        self._username = username
        self._password = password
        self._token = ""
        self._uuid = ""

    def _login(self):
        logging.info("Login in...")
        req = requests.post(
            "https://ssoms.toyota-europe.com/authenticate",
            headers=self.HEADERS_BASE,
            json={
                "username": self._username,
                "password": self._password
            },
            timeout=10
        )

        if not req.ok:
            raise ValueError(f"Invalid auth ({req.status_code}): {req.text}")

        response = req.json()

        self._token = response["token"]
        self._uuid = response["customerProfile"]["uuid"]

    def orders(self) -> list[str]:
        if not self._token:
            self._login()

        logging.info("Getting orders...")
        req = requests.get(
            "https://weblos.toyota-europe.com/leads/ordered",
            params={
                "displayPreApprovedCars": "true",
                "displayVOTCars": "true"
            },
            headers={self.TOKEN_HEADER: self._token, **self.HEADERS_BASE},
            timeout=10
        )

        if not req.ok:
            raise ValueError(f"Invalid response ({req.status_code}): {req.text}")

        response = req.json()

        return [x['id'] for x in response]

    def order_details(self, order_id: str) -> dict:
        if not self._token:
            self._login()

        logging.info(f"Getting order {order_id} details...")
        req = requests.get(
            f"https://cpb2cs.toyota-europe.com/api/orderTracker/user/{self._uuid}/orderStatus/{order_id}",
            headers={self.TOKEN_HEADER: self._token, **self.HEADERS_BASE},
            timeout=10
        )

        if not req.ok:
            raise ValueError(f"Invalid response ({req.status_code}): {req.text}")

        return req.json()


class Reporter:
    RESET = "\033[0m"
    BOLD = "\033[1m"
    INVERT = "\033[7m"
    RED = "\033[31m"
    GREEN = "\033[32m"
    YELLOW = "\033[33m"
    BLUE = "\033[34m"
    SP = "  "

    @classmethod
    def _fmt_status(cls, status: str, length: int) -> str:
        color = ""

        if status in ["notVisited", "pending"]:
            color = cls.RED
        elif status == "visited":
            color = cls.BLUE
        elif status == "inTransit":
            color = cls.YELLOW
        elif status == "current":
            color = cls.GREEN

        return f"{cls.BOLD}{color}{status}{cls.RESET}{' '*(length - len(status))}"

    @classmethod
    def _print_table(cls, table: list, fmt_data: Callable[[list, list[int]], list]):
        # Get max column length
        lengths = [0] * len(table[0])
        for data in table:
            for idx, val in enumerate(data):
                lengths[idx] = max(lengths[idx], len(val))

        # Generate format string
        fmt = f"{cls.SP}│"
        for length in lengths:
            fmt += f" {{:<{length}}} │"

        # Print headers and separator
        print(fmt.format(*table[0]))
        print(f"{cls.SP}├" + "┼".join("─"*(ln+2) for ln in lengths) + "┤")

        # Print steps
        for data in table[1:]:
            print(fmt.format(*fmt_data(data, lengths)))

    @classmethod
    def _load_dates(cls, filename: str) -> tuple[pathlib.Path, dict]:
        dates_file = pathlib.Path(filename)

        dates = DATE_STRUCT

        if dates_file.exists():
            with dates_file.open("r") as fp:
                dates = json.load(fp)

        return dates_file, dates

    @classmethod
    def _save_dates(cls, dates_file: pathlib.Path, dates: dict):
        with dates_file.open("w") as fp:
            json.dump(dates, fp)

    @classmethod
    def print_order(cls, order: dict, store_dates: Optional[bool] = False):
        if store_dates:
            dates_file, dates = cls._load_dates(f"{order['orderDetails']['orderId']}.json")

        details = order["orderDetails"]
        status = order["currentStatus"]

        print()
        print(f"{cls.SP}{cls.INVERT}{cls.BOLD} Order {details['orderId']} {cls.RESET}")
        print()
        print(f"{cls.SP}{cls.BOLD}Status{cls.RESET}: {status['currentStatus']}")
        print(f"{cls.SP}{cls.BOLD}Estimated Delivery?{cls.RESET}: {order.get('etaToFinalDestination', 'N/A')} / {status.get('estimatedDeliveryToFinalDestination', 'N/A')}")
        print()
        print(f"{cls.SP}{cls.BOLD}Call Off?{cls.RESET}: {status['callOffStatus']}")
        print(f"{cls.SP}{cls.BOLD}Delayed?{cls.RESET}: {status['isDelayed'] if status.get('isDelayed') else False}")
        print(f"{cls.SP}{cls.BOLD}Damage?{cls.RESET}: {status['damageCode'] if status.get('damageCode') else None}")
        print()
        print(f"{cls.SP}{cls.BOLD}Vehicle{cls.RESET}: {details.get('vehicleModel')}")
        print(f"{cls.SP}{cls.BOLD}Engine{cls.RESET}: {details.get('engine')}")
        print(f"{cls.SP}{cls.BOLD}Transmission{cls.RESET}: {details.get('transmission')}")
        print(f"{cls.SP}{cls.BOLD}Color Code{cls.RESET}: {details.get('vehicleExternalColor')}")
        print()
        print(f"{cls.SP}{cls.BOLD}VIN{cls.RESET}: {details.get('vin')}")

        steps = order.get("preprocessed", {}).get("steps")

        if steps:
            print()

            table = [
                [
                    "Step",
                    "Location",
                    "Status"
                ]
            ]

            if store_dates:
                table[0].append("Dates")

            for k, v in steps.items():
                table.append([
                    k,
                    v.get("location", ""),
                    v["status"]
                ])

                if store_dates:
                    if k not in dates["steps"]:
                        dates["steps"][k] = {}

                    if v["status"] not in dates["steps"][k] and v["status"] != "pending":
                        dates["steps"][k][v["status"]] = datetime.today().strftime('%Y-%m-%d')

                    table[-1].append(" | ".join(f"{kd}: {vd}" for kd, vd in dates["steps"][k].items()))

            if store_dates:
                fmt_fn = lambda data, lengths: [*data[:-2], cls._fmt_status(data[-2], lengths[-2]), data[-1]]
            else:
                fmt_fn = lambda data, lengths: [*data[:-1], cls._fmt_status(data[-1], lengths[-1])]

            cls._print_table(
                table,
                fmt_fn
            )

        else:
            print(f"\n{cls.SP}Order has no steps.")

        deliveries = order.get("intermediateDeliveries")

        if deliveries:
            print()
            table = [
                [
                    "Loc. Code",
                    "Location",
                    "Loc. Type",
                    "Transport",
                    "Visited"
                ]
            ]

            if store_dates:
                table[0].append("Dates")

            for d in deliveries:
                table.append([
                    f"{d['locationCode']}, {d['countryCode']}",
                    f"{d['locationName']}, {d['countryName']}",
                    d["destinationType"],
                    d["transportMethod"],
                    d["isVisited"]
                ])

                if store_dates:
                    if d["locationCode"] not in dates["deliveries"]:
                        dates["deliveries"][d["locationCode"]] = {}

                    if d["isVisited"] not in dates["deliveries"][d["locationCode"]] and d["isVisited"] != "notVisited":
                        dates["deliveries"][d["locationCode"]][d["isVisited"]] = datetime.today().strftime('%Y-%m-%d')

                    table[-1].append(" | ".join(f"{kd}: {vd}" for kd, vd in dates["deliveries"][d["locationCode"]].items()))

            if store_dates:
                fmt_fn = lambda data, lengths: [*data[:-2], cls._fmt_status(data[-2], lengths[-2]), data[-1]]
            else:
                fmt_fn = lambda data, lengths: [*data[:-1], cls._fmt_status(data[-1], lengths[-1])]

            cls._print_table(
                table,
                fmt_fn
            )

        else:
            print(f"\n{cls.SP}Order has no deliveries.")

        print()

        if store_dates:
            cls._save_dates(dates_file, dates)


def main(username: str, password: str, store_dates: Optional[bool] = False):
    tme = TME(username, password)

    for order_id in tme.orders():
        Reporter.print_order(tme.order_details(order_id), store_dates)


if __name__ == "__main__":
    parser = argparse.ArgumentParser()

    parser.add_argument(
        "--username", type=str,
        default=os.getenv("TOYOTA_USER"),
        help="Toyota account username"
    )
    parser.add_argument(
        "--password", type=str,
        default=os.getenv("TOYOTA_PASS"),
        help="Toyota account password"
    )
    parser.add_argument(
        "--store-dates", action="store_true",
        help=(
            "store state changes dates "
            "(keep in mind they are not reported by Toyota, "
            "they will be what is observed by the execution of this script)"
        )
    )

    args = parser.parse_args()

    if not args.username or not args.password:
        print(
            "Username and password required, use --username and --password flags"
            " or TOYOTA_USER and TOYOTA_PASS environment variables."
        )

        sys.exit(1)

    main(args.username, args.password, args.store_dates)