120 lines
5.9 KiB
Python
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
|