michi.vonah/themepark-alerts
1
0
Fork 0
mirror of https://github.com/michivonah/themepark-alerts.git synced 2025-12-22 20:36:27 +01:00
Code Issues Projects Releases Packages Wiki Activity Actions

improved exception & error handling #1

Browse source
This commit is contained in:
Michi 2024-10-12 17:30:47 +02:00
parent 5e07460615
commit 0988256239
1 changed files with 50 additions and 25 deletions
Download patch file Download diff file Expand all files Collapse all files

75
main.py
View file

@ -26,41 +26,63 @@ def sendMessage(message, notificationType):
header = {
"Authorization":f"Bearer {accessToken}"
}
response = requests.post(url=endpoint, headers=header, data=message)
else: response = requests.post(url=endpoint, data=message)
try:
response = requests.post(url=endpoint, headers=header, data=message)
except Exception as error:
raise Exception(f"Got an error while sending the notification: {error}")
else:
try:
response = requests.post(url=endpoint, data=message)
except Exception as error:
raise Exception(f"Got an error while sending the notification: {error}")
return response
case "discord" | _:
endpoint = os.getenv('DISCORD_WEBHOOK')
data = {
"content": message,
}
response = requests.post(url=endpoint, json=data)
return response
try:
response = requests.post(url=endpoint, json=data)
return response
except Exception as error:
raise Exception(f"Got an error while sending the notification: {error}")
# Check for the current waiting times
def checkTimes(subscribedAttractions, themepark):
endpoint = "https://api.wartezeiten.app/v1/waitingtimes"
try:
endpoint = "https://api.wartezeiten.app/v1/waitingtimes"
header = {
"language":"de",
"park":themepark
}
header = {
"language":"de",
"park":themepark
}
req = requests.get(url=endpoint, headers=header)
result = req.json()
attractions = result
for attraction in attractions:
if attraction["code"] in subscribedAttractions:
if attraction["status"] == "opened":
refreshTime = 30
if not attraction["code"] in currentTimes: currentTimes[attraction["code"]] = attraction["waitingtime"];
if currentTimes[attraction["code"]] > attraction["waitingtime"]:
sendMessage(f"Waiting time of {attraction['name']} sank to {attraction['waitingtime']} Minutes!", notificationType)
elif currentTimes[attraction["code"]] < attraction["waitingtime"]:
sendMessage(f"Waiting time for {attraction['name']} increased to {attraction['waitingtime']} Minutes!", notificationType)
currentTimes[attraction["code"]] = attraction["waitingtime"]
req = requests.get(url=endpoint, headers=header)
except:
raise Exception(f"API Request to endpoint {endpoint} failed.")
try:
result = req.json()
except:
raise Exception("Format of API response is invalid. (JSON expected)")
try:
attractions = result
for attraction in attractions:
if isinstance(attraction, dict) and "code" in attraction:
if attraction["code"] in subscribedAttractions:
if attraction["status"] == "opened":
refreshTime = 30
if not attraction["code"] in currentTimes: currentTimes[attraction["code"]] = attraction["waitingtime"];
if currentTimes[attraction["code"]] > attraction["waitingtime"]:
sendMessage(f"Waiting time of {attraction['name']} sank to {attraction['waitingtime']} Minutes!", notificationType)
elif currentTimes[attraction["code"]] < attraction["waitingtime"]:
sendMessage(f"Waiting time for {attraction['name']} increased to {attraction['waitingtime']} Minutes!", notificationType)
currentTimes[attraction["code"]] = attraction["waitingtime"]
else:
refreshTime = 180
else:
refreshTime = 180
print(f"Info: Attraction was skipt because it has an invalid data structure. Affacted attraction: {attraction}")
except Exception as error:
raise Exception(f"Got an error while checking for differences since the last API call. Error: {error}")
# Main Loop
# Checks every 30 seconds for changes in the waiting times of the subscribed attractions
@ -70,7 +92,10 @@ if __name__ == '__main__':
print("By Michi von Ah")
print("Big thanks to the wartezeiten.app API!")
while True:
checkTimes(subscribedAttractions, themepark)
print(f"Checked for updates at {time.strftime('%H:%M:%S', time.localtime())}")
try:
checkTimes(subscribedAttractions, themepark)
print(f"Checked for updates at {time.strftime('%H:%M:%S', time.localtime())}")
except Exception as error:
raise Exception(error)
time.sleep(refreshTime)