ShipmentNotifier/ShipmentNotifierAmazonAPI.py

import requests
import yaml

from ShipmentNotifierLogger import log

SETTINGS = yaml.safe_load(open('../ShipmentNotifierSettings.yaml'))

def getAccessToken(settings=SETTINGS):
    with requests.post(
        'https://api.amazon.com/auth/o2/token', 
        {
            'grant_type': 'refresh_token', 
            'refresh_token': settings['REFRESH_TOKEN'],
            'client_id': settings['CLIENT_ID'],
            'client_secret': settings['CLIENT_SECRET'],
        }
    ) as accessToken:
        if accessToken.status_code == 200:
            accessTokenJSON = accessToken.json()

    return accessTokenJSON['access_token']

def getProductName(MSKU, maxRetries=50, settings=SETTINGS):
    for attempt in range(maxRetries):
        with requests.get(
            settings['SPAPI_ENDPOINT'] + '/listings/2021-08-01/items/' + settings['SELLER_ID'] + f'/{MSKU}',
            headers = {
                'x-amz-access-token': getAccessToken(),
            },
            params = {
                'marketplaceIds': settings['MARKETPLACE_ID']
            }, 
        ) as productNameResponse:
            if productNameResponse.status_code == 200:
                productName = productNameResponse.json()['summaries'][0]['itemName']
                if attempt + 1 > 1:
                    log(f"SP-API call for product name completed on attempt {attempt + 1}", 'info')
                break
            else:
                log(f"SP-API call for product name failed, status code: {productNameResponse.status_code} JSON response: {productNameResponse.json()}", 'error')
                log(f'Headers: {productNameResponse.headers}', 'debug')

    return productName

def getInboundShipmentPlans(maxRetries=50, settings=SETTINGS):
    for attempt in range(maxRetries):
        with requests.get(
            settings['SPAPI_ENDPOINT'] + '/inbound/fba/2024-03-20/inboundPlans?pageSize=10&sortBy=CREATION_TIME&sortOrder=DESC&status=SHIPPED',
            headers = {
                'x-amz-access-token': getAccessToken(),
            }
        ) as inboundShipmentPlansResponse:
            if inboundShipmentPlansResponse.status_code == 200:
                inboundShipmentPlans = inboundShipmentPlansResponse.json()['inboundPlans']
                if attempt + 1 > 1:
                    log(f"SP-API call for inbound shipment plans completed on attempt {attempt + 1}", 'info')
                break
            else:
                log(f"SP-API call for inbound shipment plans failed, status code: {inboundShipmentPlansResponse.status_code} JSON response: {inboundShipmentPlansResponse.json()}", 'error')
                log(f'Headers: {inboundShipmentPlansResponse.headers}', 'debug')

    return inboundShipmentPlans

def getInboundShipmentPlan(inboundShipmentPlanID, maxRetries=50, settings=SETTINGS):
    for attempt in range(maxRetries):
        with requests.get(
            settings['SPAPI_ENDPOINT'] + f'/inbound/fba/2024-03-20/inboundPlans/{inboundShipmentPlanID}/items',
            headers = {
                'x-amz-access-token': getAccessToken(),
            }
        ) as inboundShipmentPlanResponse:
            if inboundShipmentPlanResponse.status_code == 200:
                inboundShipmentPlan = inboundShipmentPlanResponse.json()
                if attempt + 1 > 1:
                    log(f"SP-API call for inbound shipment plan completed on attempt {attempt + 1}", 'info')
                break
            else:
                log(f"SP-API call for inbound shipment plan failed, status code: {inboundShipmentPlanResponse.status_code} JSON response: {inboundShipmentPlanResponse.json()}", 'error')
                log(f'Headers: {inboundShipmentPlanResponse.headers}', 'debug')

    return inboundShipmentPlan

def getInboundShipmentData(inboundPlanId, maxRetries=50, settings=SETTINGS):
    inboundShipmentData = {inboundPlanId: {'destinations': [], 'shipmentIDs': []}}

    for attempt in range(maxRetries):
        with requests.get(
            settings['SPAPI_ENDPOINT'] + f'/inbound/fba/2024-03-20/inboundPlans/{inboundPlanId}/placementOptions',
            headers = {
                'x-amz-access-token': getAccessToken(),
            }
        ) as inboundPlanShipmentDataResponse:
            if inboundPlanShipmentDataResponse.status_code == 200:
                if attempt + 1 > 1:
                    log(f"SP-API call for inbound shipment data completed on attempt {attempt + 1}", 'info')
                break
            else:
                log(f"SP-API call for inbound shipment data failed, status code: {inboundPlanShipmentDataResponse.status_code} JSON response: {inboundPlanShipmentDataResponse.json()}", 'error')
                log(f'Headers: {inboundPlanShipmentDataResponse.headers}', 'debug')

    for shipmentID in inboundPlanShipmentDataResponse.json()['placementOptions'][0]['shipmentIds']:
        for attempt in range(maxRetries):
            with requests.get(
                settings['SPAPI_ENDPOINT'] + f'/inbound/fba/2024-03-20/inboundPlans/{inboundPlanId}/shipments/{shipmentID}',
                headers = {
                    'x-amz-access-token': getAccessToken(),
                }
            ) as individualShipmentDataResponse:
                if inboundPlanShipmentDataResponse.status_code == 200:
                    inboundShipmentData[inboundPlanId]['shipmentIDs'].append(individualShipmentDataResponse.json()['shipmentConfirmationId'])
                    inboundShipmentData[inboundPlanId]['destinations'].append(individualShipmentDataResponse.json()['destination']['warehouseId'])
                    if attempt + 1 > 1:
                        log(f"SP-API call for individual shipment data completed on attempt {attempt + 1}", 'info')
                    break
                else:
                    log(f"SP-API call for individual shipment data failed, status code: {individualShipmentDataResponse.status_code} JSON response: {individualShipmentDataResponse.json()}", 'error')
                    log(f'Headers: {individualShipmentDataResponse.headers}', 'debug')

    return inboundShipmentData