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    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'][0]

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\tvehicleId: {}".format(
        dt_string,
        order_status['state'],
        order_status['configurationStatus'],
        order_status['fulfillmentSummaryStatus'],
        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()

Instructions:

To install dependencies:

pip3 install requests python-dotenv twilio

To run:

• Set USERNAME and PASSWORD in .env
• python3 main.py

To run in a loop checking every 10 minutes:

• python3 main.py --loop

To send an SMS when running in a loop and status change is detected, set the TWILIO_* properties in .env.

Works great, thanks!

Note: on macOS Monterey (after Python2 was removed), apparently you need to run pip3 install requests python-dotenv, instead of just pip ….

Thanks @OverZealous I've updated the instructions. Also updated the code to optionally run in a loop, and optionally send an SMS when a change in status is detected.

Added support to track the 8-step status (once available).

Thanks for doing this, I'll definitely be using it. For customers with multiple order numbers (both active and cancelled), it seems to pull the first order number in the array. Unfortunately for me, I have 3 cancelled orders and 1 active order that get returned, so only the first (cancelled) order's data is returned:

04/06/2022 09:55:56: state: CANCELLED configurationStatus: CONFIGURED fulfillmentSummaryStatus: None vehicleId: None

I changed the following on line 96 in main.py to grab the last item of the array instead of the first, and it seems to be pulling my correct info.
return response.json()['data']['user']['orderSnapshots'][-1]

04/06/2022 10:24:17: state: CONFIGURED configurationStatus: CONFIGURED fulfillmentSummaryStatus: None vehicleId: None

Would it be possible to allow for a specific order number as an input, or to loop through all returned order numbers? Thanks!

@boardthatpowder Hi, nice work! Do you know if the URL's have changed? Getting a generic 404 Not Found when hitting that first auth endpoint.