telboon/darktide-shop-checker
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Feat(store): Add marks store monitoring

Browse Source
This commit is contained in:
Samuel Pua
2023-01-08 22:13:40 +08:00
parent 5ed18cc28f
commit f8de63af7d
1 changed files with 70 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files

84
main.py
View File

@@ -53,6 +53,48 @@ def main():
# Print credit store details
for store_item in credit_store_details_list:
print_str += create_store_item_details(store_item, characters, chosen_char_index, filter_name_list, filter_score_str, "Credit Store (Ordo Dockets)")
checked_marks_store = False
if datetime.datetime.utcnow().hour == 1 or not checked_marks_store:
checked_marks_store = True
# Get marks store details
marks_store_details_list = get_marks_store_details(access_jwt, account_id, characters[chosen_char_index]["id"], characters[chosen_char_index]["archetype"])
# Print marks store details
for store_item in marks_store_details_list:
print_str += create_store_item_details(store_item, characters, chosen_char_index, filter_name_list, filter_score_str, "Marks Store (Contract)")
if webhook_url != "":
splitted = print_str.split("====================================")
split_num = 2
for i in range(0, len(splitted), split_num):
requests.post(webhook_url, data="\n".join(splitted[i:i+split_num]))
print(print_str)
last_sleep_min = 60 - datetime.datetime.now().minute % 60
print("Sleeping for %d minutes" % (last_sleep_min/3))
time.sleep(last_sleep_min*60/3)
# Refresh token
refresh_jwt, access_jwt = update_refresh_jwt(refresh_jwt)
print("Sleeping for %d minutes" % (last_sleep_min/3))
time.sleep(last_sleep_min*60/3)
# Refresh token
refresh_jwt, access_jwt = update_refresh_jwt(refresh_jwt)
print("Sleeping for %d minutes" % (last_sleep_min/3))
time.sleep(last_sleep_min*60/3)
# Extra sleep to prevent clock desync
print("Sleeping for 1 minute")
time.sleep(60)
def create_store_item_details(store_item: dict, characters: list, chosen_char_index: int, filter_name_list: list, filter_score_str: str, store_name: str) -> str:
print_str = ""
item_name_display = STORE_ITEMS_DETAILS[store_item["description"]["id"]]["display_name"]
item_base_score = store_item["description"]["overrides"]["baseItemLevel"]
@@ -63,13 +105,15 @@ def main():
matched = True
break
if not matched:
continue
return ""
if filter_score_str != "" and item_base_score < int(filter_score_str):
continue
return ""
item_name = STORE_ITEMS_DETAILS[store_item["description"]["id"]]
print_str += "====================================\n"
print_str += "User: %s - %s (%s)\n" % (characters[chosen_char_index]["name"], characters[chosen_char_index]["archetype"], store_name)
print_str += "Item: %s (Base Stats Score: %d)\n" % (item_name["display_name"], item_base_score)
# try:
# print_str += "Description: %s\n" % item_name["description"]
@@ -97,32 +141,7 @@ def main():
except:
pass
if webhook_url != "":
splitted = print_str.split("====================================")
split_num = 2
for i in range(0, len(splitted), split_num):
requests.post(webhook_url, data="\n".join(splitted[i:i+split_num]))
print(print_str)
last_sleep_min = 60 - datetime.datetime.now().minute % 60
print("Sleeping for %d minutes" % (last_sleep_min/3))
time.sleep(last_sleep_min*60/3)
# Refresh token
refresh_jwt, access_jwt = update_refresh_jwt(refresh_jwt)
print("Sleeping for %d minutes" % (last_sleep_min/3))
time.sleep(last_sleep_min*60/3)
# Refresh token
refresh_jwt, access_jwt = update_refresh_jwt(refresh_jwt)
print("Sleeping for %d minutes" % (last_sleep_min/3))
time.sleep(last_sleep_min*60/3)
# Extra sleep to prevent clock desync
print("Sleeping for 1 minute")
time.sleep(60)
pass
return print_str
def update_refresh_jwt(jwt: str) -> (str, str):
url = "https://bsp-auth-prod.atoma.cloud/queue/refresh"
@@ -158,6 +177,15 @@ def get_credit_store_details(access_jwt: str, account_id: str, character_id: str
if res.status_code == 200:
return res.json()["personal"]
def get_marks_store_details(access_jwt: str, account_id: str, character_id: str, archetype: str) -> dict:
url = "https://bsp-td-prod.atoma.cloud/store/storefront/marks_store_%s?accountId=%s&personal=true&characterId=%s" % (archetype, account_id, character_id)
headers = {
"Authorization": "Bearer " + access_jwt
}
res = requests.get(url, headers=headers)
if res.status_code == 200:
return res.json()["personal"]