Rivian oder status script

.env

# rivian.com credentials:
USERNAME=
PASSWORD=

# optional twilio.com credentials. If set, will send an SMS when a change in status is detected:
TWILIO_ACCOUNT_SID=
TWILIO_AUTH_TOKEN=
TWILIO_FROM_PHONE_NUMBER=
TWILIO_TO_PHONE_NUMBER=

main.py

import time
from datetime import datetime
import argparse
import requests
from dotenv import dotenv_values
from twilio.rest import Client
from dataclasses import dataclass

conf = dotenv_values()

LOGIN_URL = "https://auth.rivianservices.com/auth/api/v1/token/auth"
GQL_ORDERS_URL = "https://rivian.com/api/gql/orders/graphql/"
GQL_T2D_URL = "https://rivian.com/api/gql/t2d/graphql/"
CONTENT_TYPE = 'application/json;charset=UTF-8'
CLIENT_ID = "rivian.mobile.sc12bjxe8lmhkul"
CLIENT_SECRET = "rlL058p5kipkZr0C85KrdA4AZ0QBNVh75zXWwEWf"
DC_CID = "account--9dfce4c4-dbd1-4e70-8735-12e9c776afbf--c202d410-ed84-48b5-8c5b-4b91d1a14648"
SLEEP_TIME = 60 * 10

session = None
last_order_status = None
last_transaction_status = None

@dataclass
class TransactionStatus:
    titleAndReg: str
    tradeIn: str
    finance: str
    delivery: str
    insurance: str
    documentUpload: str
    contracts: str
    payment: str


def create_session():
    headers = {'Content-Type': CONTENT_TYPE}
    payload = {
        "username": conf['USERNAME'],
        "pwd": conf['PASSWORD'],
        "source": "mobile",
        "grant_type": "password",
        "client_id": CLIENT_ID,
        "client_secret": CLIENT_SECRET
    }
    global session
    session = requests.Session()

    response = session.post(url=LOGIN_URL, headers=headers, json=payload)
    result = response.json()
    return result['access_token'], result['refresh_token']


def create_csrf_token():
    headers = {
        'Content-Type': CONTENT_TYPE,
        'dc-cid': DC_CID
    }
    payload = '{"query":"mutation createCsrfToken { createCsrfToken { __typename csrfToken } }"}'
    response = session.post(url=GQL_ORDERS_URL, headers=headers, data=payload)
    return response.json()['data']['createCsrfToken']['csrfToken']


def create_graph_session(csrf_token, access_token, refresh_token):
    headers = {
        'Content-Type': CONTENT_TYPE,
        'csrf-token': csrf_token,
        'dc-cid': DC_CID
    }
    payload = """{
        "query":"mutation loginWithToken($tokens: CredentialTokensInput!) { loginWithToken(tokens: $tokens) { __typename userId } }",
        "variables":{
            "tokens":{
                "clientId":"{client_id}",
                "accessToken":"{access_token}",
                "refreshToken":"{refresh_token}"
            }
        }
    }""".replace("{client_id}", CLIENT_ID) \
        .replace("{access_token}", access_token) \
        .replace("{refresh_token}", refresh_token)
    response = session.post(url=GQL_ORDERS_URL, headers=headers, data=payload)
    return response.headers['a-sess'], response.headers['u-sess']


def get_order_status(csrf_token, a_sess, u_sess):
    headers = {
        'Content-Type': CONTENT_TYPE,
        'dc-cid': DC_CID,
        'csrf-token': csrf_token,
        'a-sess': a_sess,
        'u-sess': u_sess
    }
    payload = "{\"query\":\"query {\\n\\tuser {\\n  \\t\\torderSnapshots (filterTypes: [PRE_ORDER, VEHICLE, RETAIL]) {\\n\\t\\t    id\\n\\t\\t    state\\n\\t\\t    configurationStatus\\n\\t\\t    fulfillmentSummaryStatus\\n\\t\\t    eligibleForNewVehicleInventoryMarketplace\\n\\t\\t    vehicleId\\n  \\t\\t}  \\n\\t}\\n}\\n\",\"variables\":{}}"
    response = session.post(url=GQL_ORDERS_URL, headers=headers, data=payload)
    return response.json()['data']['user']['orderSnapshots'][-1]

def get_8steps_status(csrf_token, a_sess, u_sess, order_id):
    headers = {
        'Content-Type': CONTENT_TYPE,
        'dc-cid': DC_CID,
        'csrf-token': csrf_token,
        'a-sess': a_sess,
        'u-sess': u_sess
    }
    payload = "{\"query\":\"query transactionStatus($orderId: ID!) {\\n  transactionStatus(orderId: $orderId) {\\n    titleAndReg {\\n      sourceStatus {\\n        status\\n      }\\n    }\\n    tradeIn {\\n      sourceStatus {\\n        status\\n      }\\n    }\\n    finance {\\n      sourceStatus {\\n        status\\n      }\\n    }\\n    delivery {\\n      sourceStatus {\\n        status\\n      }\\n    }\\n    insurance {\\n      sourceStatus {\\n        status\\n      }\\n    }\\n    documentUpload {\\n      sourceStatus {\\n        status\\n      }\\n    }\\n    contracts {\\n      sourceStatus {\\n        status\\n      }\\n    }\\n    payment {\\n      sourceStatus {\\n        status\\n      }\\n    }\\n  }\\n}\\n\",\"variables\":{\"orderId\":\"{order_id}\"}}".replace("{order_id}",order_id)
    response = session.post(url=GQL_T2D_URL, headers=headers, data=payload)
    json = response.json()['data']['transactionStatus']
    return TransactionStatus(
        json['titleAndReg']['sourceStatus']['status'],
        json['tradeIn']['sourceStatus']['status'],
        json['finance']['sourceStatus']['status'],
        json['delivery']['sourceStatus']['status'],
        json['insurance']['sourceStatus']['status'],
        json['documentUpload']['sourceStatus']['status'],
        json['contracts']['sourceStatus']['status'],
        json['payment']['sourceStatus']['status']
    )


def send_text(status):
    if conf['TWILIO_ACCOUNT_SID'] == '':
        return

    client = Client(conf['TWILIO_ACCOUNT_SID'], conf['TWILIO_AUTH_TOKEN'])
    client.messages .create(
        body='Change in status!\n{}'.format(status),
        from_=conf['TWILIO_FROM_PHONE_NUMBER'],
        to=conf['TWILIO_TO_PHONE_NUMBER']
    )


def multiple_status_checks():
    while True:
        single_status_check();
        time.sleep(SLEEP_TIME)


def single_status_check():
    access_token, refresh_token = create_session()
    csrf_token = create_csrf_token()
    a_sess, u_sess = create_graph_session(csrf_token, access_token, refresh_token)
    order_status = get_order_status(csrf_token, a_sess, u_sess)
    dt_string = datetime.now().strftime("%d/%m/%Y %H:%M:%S")
    order_status_message = "{}:\n\tstate: {}\n\tconfigurationStatus: {}\n\tfulfillmentSummaryStatus: {}\n\teligibleForNewVehicleInventoryMarketplace: {}\n\tvehicleId: {}".format(
        dt_string,
        order_status['state'],
        order_status['configurationStatus'],
        order_status['fulfillmentSummaryStatus'],
        order_status['eligibleForNewVehicleInventoryMarketplace'],
        order_status['vehicleId'])
    print(order_status_message)

    global last_order_status
    if last_order_status is not None and last_order_status != order_status:
        send_text(order_status_message)
    last_order_status = order_status

    try:
        transaction_status = get_8steps_status(csrf_token, a_sess, u_sess, order_status['id'])
        transaction_status_message = "{}:\n\ttitleAndReg: {}\n\ttradeIn: {}\n\tfinance: {}\n\tinsurance: {}\n\tdocumentUpload: {}\n\tcontracts: {}\n\tpayment: {}\n\tdelivery: {}".format(
            dt_string,
            transaction_status.titleAndReg,
            transaction_status.tradeIn,
            transaction_status.finance,
            transaction_status.insurance,
            transaction_status.documentUpload,
            transaction_status.contracts,
            transaction_status.payment,
            transaction_status.delivery)
        print(transaction_status_message)

        global last_transaction_status
        if last_transaction_status is not None and last_transaction_status != transaction_status:
            send_text(transaction_status_message)
        last_transaction_status = transaction_status
    except Exception as e:
        pass


if __name__ == '__main__':
    parser = argparse.ArgumentParser()
    parser.add_argument('-l', '--loop', action='store_true', help="Continually check on a loop")
    args = parser.parse_args()

    if args.loop:
        multiple_status_checks()
    else:
        single_status_check()