macf/ShipmentNotifier
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Refactoring file names and structure, adding retries and proper closure to API calls, updating README, adding footer to Discord notifications

Browse Source
This commit is contained in:
Dom 2024-09-01 18:16:01 +01:00
parent 9b86936a49
commit 5dd3a06864
13 changed files with 211 additions and 187 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.gitignore vendored Normal file
View File

@ -0,0 +1 @@
logs/*

6
README.md
View File

@ -1,8 +1,8 @@
# ShipmentNotifier
Script that polls the Amazon SP-API for most recent shipments, if a shipment is found that has been created in the last 6 hours, sends a Discord notification to a provided webhook.
Script that polls the Amazon SP-API for most recent shipments, if a shipment is found that has been created in the last 14 days, sends a Discord notification to a provided webhook.
- Acceptable time delta: < 6 hours - accounts for SP-API delays and time for the shipment to transition to 'Shipped'
- Acceptable time delta: < 14 days - accounts for SP-API delays and time for the shipment to transition to 'Shipped'
- Number of shipments to look for: 10
- Stores UID of shipments that a notification has previously been sent for to avoid sending duplicate notifications
- Recommended to run every 10 minutes via cron
- Runs every 10 minutes via cron

0
SentNotifications.json → SentNotifications.json.example
View File

18
ShipmentNotifier.py
View File

@ -1,8 +1,16 @@
from log import log, cleanLogs
from timeOperations import isInboundShipmentPlanWithinSpecifiedDelta
from amazonAPI import getAccessToken, getInboundShipmentData, getInboundShipmentPlans, getInboundShipmentPlan, getProductName
from sentNotifications import isInboundShipmentPlanIDInSentNotifications, updateSentNotifications
from discordNotifications import sendDiscordNotification
from ShipmentNotifierLogger import log, cleanLogs
from ShipmentNotifierAmazonAPI import (
getInboundShipmentData,
getInboundShipmentPlans,
getInboundShipmentPlan,
getProductName
)
from ShipmentNotifierHelpers import (
isInboundShipmentPlanIDInSentNotifications,
updateSentNotifications,
sendDiscordNotification,
isInboundShipmentPlanWithinSpecifiedDelta
)
def parseInboundShipmentPlans():
log('\U0001F504 Getting inbound shipment plans...', 'info')

119
ShipmentNotifierAmazonAPI.py Normal file
View File

@ -0,0 +1,119 @@
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

75
ShipmentNotifierHelpers.py Normal file
View File

@ -0,0 +1,75 @@
import json
import yaml
import requests
from datetime import (
datetime,
timedelta
)
from ShipmentNotifierLogger import log
SETTINGS = yaml.safe_load(open('../ShipmentNotifierSettings.yaml'))
def sendDiscordNotification(*args, settings=SETTINGS):
newLine = '\n'
shipmentNotification = requests.post(
SETTINGS['DISCORD_WEBHOOK'],
json = {
"embeds": [
{
"title": ":package: New Inbound Shipment Detected :package:",
"url": f"https://sellercentral.amazon.co.uk/fba/sendtoamazon?wf={args[0]}",
"color": 4886754,
"fields": [
{
"name": "Shipment Information",
"value": f'''
> Inbound Shipment Plan ID: {args[0]}
> Creation Date: {args[1]['creationDate']}
> Destination Fulfilment Centres: {', '.join(destination for destination in args[1]['destinations'])}
> Shipments: {', '.join(f'[{shipmentID}](https://sellercentral.amazon.co.uk/fba/inbound-shipment/summary/{shipmentID}/contents)' for shipmentID in args[1]['shipmentIDs'])}
'''
},
{
"name": "Shipment Contents",
"value": f'''
{newLine.join(f"> {MSKU}: {Count}" for MSKU, Count in args[1]['contents'].items())}
'''
}
]
}
]
},
)
def updateSentNotifications(inboundPlanId):
with open('../ShipmentNotifierSentNotifications.json') as sentNotificationsJSON:
sentNotifications = json.load(sentNotificationsJSON)
sentNotifications['sentNotifications'].append(inboundPlanId)
with open('../ShipmentNotifierSentNotifications.json', mode='w') as outputSentNotifications:
outputSentNotifications.write(json.dumps(sentNotifications, indent=4))
def isInboundShipmentPlanIDInSentNotifications(inboundShipmentPlanId):
with open('../ShipmentNotifierSentNotifications.json') as sentNotificationsJSON:
sentNotifications = json.load(sentNotificationsJSON)
if inboundShipmentPlanId in sentNotifications['sentNotifications']:
return True
else:
return False
def isInboundShipmentPlanWithinSpecifiedDelta(inboundShipmentPlanCreationTime, delta=20160):
currentTime = datetime.now()
inboundShipmentPlanTime = datetime.strptime(inboundShipmentPlanCreationTime, '%Y-%m-%dT%H:%M:%SZ')
timeDelta = currentTime - inboundShipmentPlanTime
log(f'Current time: {currentTime}', 'info')
log(f'Inbound shipment plan creation time: {inboundShipmentPlanTime}', 'info')
log(f'Time delta: {timeDelta}', 'info')
if timeDelta < timedelta(minutes=delta):
return True
else:
return False

0
log.py → ShipmentNotifierLogger.py
View File

0
settings.yaml → ShipmentNotifierSettings.yaml.example
View File

77
amazonAPI.py
View File

@ -1,77 +0,0 @@
import requests
import yaml
from log import log
from discordNotifications import sendDiscordNotification
from sentNotifications import isInboundShipmentPlanIDInSentNotifications, updateSentNotifications
from timeOperations import isInboundShipmentPlanWithinSpecifiedDelta
SETTINGS = yaml.safe_load(open('settings.yaml'))
def getAccessToken(settings=SETTINGS):
accessToken = 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'],
}
)
return accessToken.json()['access_token']
def getProductName(MSKU, settings=SETTINGS):
product = 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']
},
)
return product.json()['summaries'][0]['itemName']
def getInboundShipmentPlans(settings=SETTINGS):
inboundShipmentPlans = 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(),
}
)
return inboundShipmentPlans.json()['inboundPlans']
def getInboundShipmentPlan(inboundShipmentPlanID, settings=SETTINGS):
inboundShipmentPlan = requests.get(
settings['SPAPI_ENDPOINT'] + f'/inbound/fba/2024-03-20/inboundPlans/{inboundShipmentPlanID}/items',
headers = {
'x-amz-access-token': getAccessToken(),
}
)
return inboundShipmentPlan.json()
def getInboundShipmentData(inboundPlanId, settings=SETTINGS):
inboundShipmentData = {inboundPlanId: {'destinations': [], 'shipmentIDs': []}}
inboundPlanShipmentData = requests.get(
settings['SPAPI_ENDPOINT'] + f'/inbound/fba/2024-03-20/inboundPlans/{inboundPlanId}/placementOptions',
headers = {
'x-amz-access-token': getAccessToken(),
}
)
for shipmentID in inboundPlanShipmentData.json()['placementOptions'][0]['shipmentIds']:
individualShipmentData = requests.get(
settings['SPAPI_ENDPOINT'] + f'/inbound/fba/2024-03-20/inboundPlans/{inboundPlanId}/shipments/{shipmentID}',
headers = {
'x-amz-access-token': getAccessToken(),
}
)
inboundShipmentData[inboundPlanId]['shipmentIDs'].append(individualShipmentData.json()['shipmentConfirmationId'])
inboundShipmentData[inboundPlanId]['destinations'].append(individualShipmentData.json()['destination']['warehouseId'])
return inboundShipmentData

36
discordNotifications.py
View File

@ -1,36 +0,0 @@
import yaml
import requests
SETTINGS = yaml.safe_load(open('settings.yaml'))
def sendDiscordNotification(*args, settings=SETTINGS):
newLine = '\n'
shipmentNotification = requests.post(
SETTINGS['DISCORD_WEBHOOK'],
json = {
"embeds": [
{
"title": ":package: New Inbound Shipment Detected :package:",
"url": f"https://sellercentral.amazon.co.uk/fba/sendtoamazon?wf={args[0]}",
"color": 4886754,
"fields": [
{
"name": "Shipment Information",
"value": f'''
> Inbound Shipment Plan ID: {args[0]}
> Creation Date: {args[1]['creationDate']}
> Destination Fulfilment Centres: {', '.join(destination for destination in args[1]['destinations'])}
> Shipments: {', '.join(f'[{shipmentID}](https://sellercentral.amazon.co.uk/fba/inbound-shipment/summary/{shipmentID}/contents)' for shipmentID in args[1]['shipmentIDs'])}
'''
},
{
"name": "Shipment Contents",
"value": f'''
{newLine.join(f"> {MSKU}: {Count}" for MSKU, Count in args[1]['contents'].items())}
'''
}
]
}
]
},
)

31
logs/log_2024-07-12_21:37:04.log
View File

@ -1,31 +0,0 @@
[INFO] 🔄 Getting shipments...
[INFO] Current time: 2024-07-12 21:37:04.979052
[INFO] Shipment creation time: 2024-07-12 13:07:50
[INFO] Time delta: 8:29:14.979052
[INFO] Current time: 2024-07-12 21:37:04.981162
[INFO] Shipment creation time: 2024-07-10 09:34:31
[INFO] Time delta: 2 days, 12:02:33.981162
[INFO] Current time: 2024-07-12 21:37:04.981349
[INFO] Shipment creation time: 2024-07-09 15:47:43
[INFO] Time delta: 3 days, 5:49:21.981349
[INFO] Current time: 2024-07-12 21:37:04.981507
[INFO] Shipment creation time: 2024-07-08 12:23:20
[INFO] Time delta: 4 days, 9:13:44.981507
[INFO] Current time: 2024-07-12 21:37:04.981663
[INFO] Shipment creation time: 2024-07-05 12:57:09
[INFO] Time delta: 7 days, 8:39:55.981663
[INFO] Current time: 2024-07-12 21:37:04.981816
[INFO] Shipment creation time: 2024-07-05 11:39:06
[INFO] Time delta: 7 days, 9:57:58.981816
[INFO] Current time: 2024-07-12 21:37:04.981968
[INFO] Shipment creation time: 2024-07-04 13:50:17
[INFO] Time delta: 8 days, 7:46:47.981968
[INFO] Current time: 2024-07-12 21:37:04.982168
[INFO] Shipment creation time: 2024-07-03 13:29:03
[INFO] Time delta: 9 days, 8:08:01.982168
[INFO] Current time: 2024-07-12 21:37:04.982320
[INFO] Shipment creation time: 2024-07-02 14:06:09
[INFO] Time delta: 10 days, 7:30:55.982320
[INFO] Current time: 2024-07-12 21:37:04.982470
[INFO] Shipment creation time: 2024-07-01 14:54:48
[INFO] Time delta: 11 days, 6:42:16.982470

19
sentNotifications.py
View File

@ -1,19 +0,0 @@
import json
def updateSentNotifications(inboundPlanId):
with open('SentNotifications.json') as sentNotificationsJSON:
sentNotifications = json.load(sentNotificationsJSON)
sentNotifications['sentNotifications'].append(inboundPlanId)
with open('SentNotifications.json', mode='w') as outputSentNotifications:
outputSentNotifications.write(json.dumps(sentNotifications, indent=4))
def isInboundShipmentPlanIDInSentNotifications(inboundShipmentPlanId):
with open('SentNotifications.json') as sentNotificationsJSON:
sentNotifications = json.load(sentNotificationsJSON)
if inboundShipmentPlanId in sentNotifications['sentNotifications']:
return True
else:
return False

16
timeOperations.py
View File

@ -1,16 +0,0 @@
from datetime import datetime, timedelta
from log import log
def isInboundShipmentPlanWithinSpecifiedDelta(inboundShipmentPlanCreationTime, delta=20160):
currentTime = datetime.now()
inboundShipmentPlanTime = datetime.strptime(inboundShipmentPlanCreationTime, '%Y-%m-%dT%H:%M:%SZ')
timeDelta = currentTime - inboundShipmentPlanTime
log(f'Current time: {currentTime}', 'info')
log(f'Inbound shipment plan creation time: {inboundShipmentPlanTime}', 'info')
log(f'Time delta: {timeDelta}', 'info')
if timeDelta < timedelta(minutes=delta):
return True
else:
return False