ShipmentNotifier/ShipmentNotifierAmazonAPI.py

120 lines
5.9 KiB
Python

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