mirror of
https://github.com/frozenpandaman/s3s.git
synced 2026-09-28 12:16:26 -05:00
release v0.0.1 of s3s!!! ᔦꙬᔨ
This commit is contained in:
eli fessler
363
iksm.py
Normal file
363
iksm.py
Normal file
@@ -0,0 +1,363 @@
|
||||
# (ↄ) 2017-2022 eli fessler (frozenpandaman), clovervidia
|
||||
# https://github.com/frozenpandaman/s3s
|
||||
# License: GPLv3
|
||||
|
||||
import requests, json, re, sys
|
||||
import os, base64, hashlib
|
||||
import uuid, time, random, string
|
||||
from bs4 import BeautifulSoup
|
||||
|
||||
session = requests.Session()
|
||||
s3s_version = "unknown"
|
||||
nsoapp_version = "2.2.0"
|
||||
|
||||
# functions in this file & call stack:
|
||||
# get_nsoapp_version()
|
||||
# get_web_view_ver()
|
||||
# log_in() -> get_session_token()
|
||||
# get_gtoken() -> call_imink_api() -> f
|
||||
# get_bullet()
|
||||
# enter_tokens()
|
||||
|
||||
# place config.txt in same directory as script (bundled or not)
|
||||
if getattr(sys, 'frozen', False):
|
||||
app_path = os.path.dirname(sys.executable)
|
||||
elif __file__:
|
||||
app_path = os.path.dirname(__file__)
|
||||
config_path = os.path.join(app_path, "config.txt")
|
||||
|
||||
|
||||
def get_nsoapp_version():
|
||||
'''Fetches the current Nintendo Switch Online app version from the Apple App Store.'''
|
||||
|
||||
try:
|
||||
page = requests.get("https://apps.apple.com/us/app/nintendo-switch-online/id1234806557")
|
||||
soup = BeautifulSoup(page.text, 'html.parser')
|
||||
elt = soup.find("p", {"class": "whats-new__latest__version"})
|
||||
ver = elt.get_text().replace("Version ","").strip()
|
||||
return ver
|
||||
except:
|
||||
global nsoapp_version
|
||||
return nsoapp_version
|
||||
|
||||
|
||||
def get_web_view_ver(): # TODO
|
||||
pass
|
||||
|
||||
|
||||
def log_in(ver):
|
||||
'''Logs in to a Nintendo Account and returns a session_token.'''
|
||||
|
||||
global s3s_version
|
||||
s3s_version = ver
|
||||
|
||||
auth_state = base64.urlsafe_b64encode(os.urandom(36))
|
||||
|
||||
auth_code_verifier = base64.urlsafe_b64encode(os.urandom(32))
|
||||
auth_cv_hash = hashlib.sha256()
|
||||
auth_cv_hash.update(auth_code_verifier.replace(b"=", b""))
|
||||
auth_code_challenge = base64.urlsafe_b64encode(auth_cv_hash.digest())
|
||||
|
||||
app_head = {
|
||||
'Host': 'accounts.nintendo.com',
|
||||
'Connection': 'keep-alive',
|
||||
'Cache-Control': 'max-age=0',
|
||||
'Upgrade-Insecure-Requests': '1',
|
||||
'User-Agent': 'Mozilla/5.0 (Linux; Android 11; Pixel 5) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/94.0.4606.61 Mobile Safari/537.36',
|
||||
'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,image/apng,*/*;q=0.8n',
|
||||
'DNT': '1',
|
||||
'Accept-Encoding': 'gzip,deflate,br',
|
||||
}
|
||||
|
||||
body = {
|
||||
'state': auth_state,
|
||||
'redirect_uri': 'npf71b963c1b7b6d119://auth',
|
||||
'client_id': '71b963c1b7b6d119',
|
||||
'scope': 'openid user user.birthday user.mii user.screenName',
|
||||
'response_type': 'session_token_code',
|
||||
'session_token_code_challenge': auth_code_challenge.replace(b"=", b""),
|
||||
'session_token_code_challenge_method': 'S256',
|
||||
'theme': 'login_form'
|
||||
}
|
||||
|
||||
url = 'https://accounts.nintendo.com/connect/1.0.0/authorize'
|
||||
r = session.get(url, headers=app_head, params=body)
|
||||
|
||||
post_login = r.history[0].url
|
||||
|
||||
print("\nMake sure you have fully read the \"Token generation\" section of the readme before proceeding. To manually input a token instead, enter \"skip\" at the prompt below.")
|
||||
print("\nNavigate to this URL in your browser:")
|
||||
print(post_login)
|
||||
print("Log in, right click the \"Select this account\" button, copy the link address, and paste it below:")
|
||||
while True:
|
||||
try:
|
||||
use_account_url = input("")
|
||||
if use_account_url == "skip":
|
||||
return "skip"
|
||||
session_token_code = re.search('de=(.*)&', use_account_url)
|
||||
return get_session_token(session_token_code.group(1), auth_code_verifier)
|
||||
except KeyboardInterrupt:
|
||||
print("\nBye!")
|
||||
sys.exit(1)
|
||||
except AttributeError:
|
||||
print("Malformed URL. Please try again, or press Ctrl+C to exit.")
|
||||
print("URL:", end=' ')
|
||||
except KeyError: # session_token not found
|
||||
print("\nThe URL has expired. Please log out and back into your Nintendo Account and try again.")
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
def get_session_token(session_token_code, auth_code_verifier):
|
||||
'''Helper function for log_in().'''
|
||||
|
||||
nsoapp_version = get_nsoapp_version()
|
||||
|
||||
app_head = {
|
||||
'User-Agent': 'OnlineLounge/' + nsoapp_version + ' NASDKAPI Android',
|
||||
'Accept-Language': 'en-US',
|
||||
'Accept': 'application/json',
|
||||
'Content-Type': 'application/x-www-form-urlencoded',
|
||||
'Content-Length': '540',
|
||||
'Host': 'accounts.nintendo.com',
|
||||
'Connection': 'Keep-Alive',
|
||||
'Accept-Encoding': 'gzip'
|
||||
}
|
||||
|
||||
body = {
|
||||
'client_id': '71b963c1b7b6d119',
|
||||
'session_token_code': session_token_code,
|
||||
'session_token_code_verifier': auth_code_verifier.replace(b"=", b"")
|
||||
}
|
||||
|
||||
url = 'https://accounts.nintendo.com/connect/1.0.0/api/session_token'
|
||||
|
||||
r = session.post(url, headers=app_head, data=body)
|
||||
return json.loads(r.text)["session_token"]
|
||||
|
||||
|
||||
def get_gtoken(f_gen_url, session_token, ver):
|
||||
'''Provided the session_token, returns a GameWebToken and account info.'''
|
||||
|
||||
nsoapp_version = get_nsoapp_version()
|
||||
|
||||
global s3s_version
|
||||
s3s_version = ver
|
||||
|
||||
app_head = {
|
||||
'Host': 'accounts.nintendo.com',
|
||||
'Accept-Encoding': 'gzip',
|
||||
'Content-Type': 'application/json',
|
||||
'Content-Length': '436',
|
||||
'Accept': 'application/json',
|
||||
'Connection': 'Keep-Alive',
|
||||
'User-Agent': 'OnlineLounge/' + nsoapp_version + ' NASDKAPI Android' # CHANGE
|
||||
}
|
||||
|
||||
body = {
|
||||
'client_id': '71b963c1b7b6d119',
|
||||
'session_token': session_token,
|
||||
'grant_type': 'urn:ietf:params:oauth:grant-type:jwt-bearer-session-token'
|
||||
}
|
||||
|
||||
url = "https://accounts.nintendo.com/connect/1.0.0/api/token"
|
||||
r = requests.post(url, headers=app_head, json=body)
|
||||
id_response = json.loads(r.text)
|
||||
|
||||
# get user info
|
||||
try:
|
||||
app_head = {
|
||||
'User-Agent': 'NASDKAPI; Android',
|
||||
'Content-Type': 'application/json',
|
||||
'Accept': 'application/json',
|
||||
'Authorization': f'Bearer {id_response["access_token"]}',
|
||||
'Host': 'api.accounts.nintendo.com',
|
||||
'Connection': 'Keep-Alive',
|
||||
'Accept-Encoding': 'gzip'
|
||||
}
|
||||
except:
|
||||
print("Not a valid authorization request. Please delete config.txt and try again.")
|
||||
print("Error from Nintendo (in api/token step):")
|
||||
print(json.dumps(id_response, indent=2))
|
||||
sys.exit(1)
|
||||
|
||||
url = "https://api.accounts.nintendo.com/2.0.0/users/me"
|
||||
r = requests.get(url, headers=app_head)
|
||||
user_info = json.loads(r.text)
|
||||
|
||||
user_nickname = user_info["nickname"]
|
||||
user_lang = user_info["language"]
|
||||
user_country = user_info["country"]
|
||||
|
||||
# get access token
|
||||
app_head = {
|
||||
'X-Platform': 'Android',
|
||||
'X-ProductVersion': nsoapp_version,
|
||||
'Content-Type': 'application/json; charset=utf-8',
|
||||
'Content-Length': '1074',
|
||||
'Connection': 'Keep-Alive',
|
||||
'Accept-Encoding': 'gzip',
|
||||
'User-Agent': 'com.nintendo.znca/' + nsoapp_version + '(Android/7.1.2)',
|
||||
}
|
||||
|
||||
body = {}
|
||||
try:
|
||||
idToken = id_response["id_token"]
|
||||
f, uuid, timestamp = call_imink_api(idToken, 1, f_gen_url)
|
||||
|
||||
parameter = {
|
||||
'f': f,
|
||||
'language': user_lang,
|
||||
'naBirthday': user_info["birthday"],
|
||||
'naCountry': user_country,
|
||||
'naIdToken': idToken,
|
||||
'requestId': uuid,
|
||||
'timestamp': timestamp
|
||||
}
|
||||
except SystemExit:
|
||||
sys.exit(1)
|
||||
except:
|
||||
print("Error(s) from Nintendo:")
|
||||
print(json.dumps(id_response, indent=2))
|
||||
print(json.dumps(user_info, indent=2))
|
||||
sys.exit(1)
|
||||
body["parameter"] = parameter
|
||||
|
||||
url = "https://api-lp1.znc.srv.nintendo.net/v3/Account/Login"
|
||||
r = requests.post(url, headers=app_head, json=body)
|
||||
splatoon_token = json.loads(r.text)
|
||||
|
||||
try:
|
||||
idToken = splatoon_token["result"]["webApiServerCredential"]["accessToken"]
|
||||
f, uuid, timestamp = call_imink_api(idToken, 2, f_gen_url)
|
||||
except:
|
||||
print("Error from Nintendo (in Account/Login step):")
|
||||
print(json.dumps(splatoon_token, indent=2))
|
||||
sys.exit(1)
|
||||
|
||||
# get web service token
|
||||
try:
|
||||
app_head = {
|
||||
'X-Platform': 'Android',
|
||||
'X-ProductVersion': nsoapp_version,
|
||||
'Authorization': f'Bearer {splatoon_token["result"]["webApiServerCredential"]["accessToken"]}',
|
||||
'Content-Type': 'application/json; charset=utf-8',
|
||||
'Content-Length': '391',
|
||||
'Accept-Encoding': 'gzip',
|
||||
'User-Agent': 'com.nintendo.znca/' + nsoapp_version + '(Android/7.1.2)'
|
||||
}
|
||||
except:
|
||||
print("Error from Nintendo (in Account/Login step):")
|
||||
print(json.dumps(splatoon_token, indent=2))
|
||||
sys.exit(1)
|
||||
|
||||
body = {}
|
||||
parameter = {
|
||||
'f': f,
|
||||
'id': 4834290508791808,
|
||||
'registrationToken': idToken,
|
||||
'requestId': uuid,
|
||||
'timestamp': timestamp
|
||||
}
|
||||
body["parameter"] = parameter
|
||||
|
||||
url = "https://api-lp1.znc.srv.nintendo.net/v2/Game/GetWebServiceToken"
|
||||
r = requests.post(url, headers=app_head, json=body)
|
||||
web_service_resp = json.loads(r.text)
|
||||
|
||||
try:
|
||||
web_service_token = web_service_resp["result"]["accessToken"]
|
||||
except:
|
||||
print("Error from Nintendo (in Game/GetWebServiceToken step):")
|
||||
print(json.dumps(web_service_resp, indent=2))
|
||||
sys.exit(1)
|
||||
|
||||
return web_service_token, user_nickname, user_lang, user_country
|
||||
|
||||
|
||||
def get_bullet(web_service_token, web_view_ver, app_user_agent, user_lang, user_country):
|
||||
'''Returns a bulletToken.'''
|
||||
|
||||
app_head = {
|
||||
'Content-Length': '0',
|
||||
'Content-Type': 'application/json',
|
||||
'Accept-Language': user_lang,
|
||||
'User-Agent': app_user_agent,
|
||||
'X-Web-View-Ver': web_view_ver,
|
||||
'X-NACOUNTRY': user_country,
|
||||
'Accept': '*/*',
|
||||
'Origin': 'https://api.lp1.av5ja.srv.nintendo.net',
|
||||
'X-Requested-With': 'com.nintendo.znca'
|
||||
}
|
||||
app_cookies = {
|
||||
'_gtoken': web_service_token # X-GameWebToken
|
||||
}
|
||||
url = "https://api.lp1.av5ja.srv.nintendo.net/api/bullet_tokens"
|
||||
r = requests.post(url, headers=app_head, cookies=app_cookies)
|
||||
bullet_resp = json.loads(r.text)
|
||||
|
||||
print("***")
|
||||
print(r.text)
|
||||
|
||||
if r.status_code == 401:
|
||||
print("Unauthorized error (ERROR_INVALID_GAME_WEB_TOKEN). Cannot fetch tokens at this time.")
|
||||
sys.exit(1)
|
||||
elif r.status_code == 403:
|
||||
print("Forbidden error (ERROR_OBSOLETE_VERSION). Cannot fetch tokens at this time.")
|
||||
sys.exit(1)
|
||||
elif r.status_code == 204: # No Content, USER_NOT_REGISTERED
|
||||
print("Cannot access SplatNet 3 without having played online.")
|
||||
sys.exit(1)
|
||||
else:
|
||||
try:
|
||||
bullet_token = bullet_resp["bulletToken"]
|
||||
except:
|
||||
print("Error from Nintendo (in api/bullet_tokens step):")
|
||||
print(json.dumps(bullet_resp, indent=2))
|
||||
sys.exit(1)
|
||||
|
||||
return bullet_token
|
||||
|
||||
|
||||
def call_imink_api(id_token, step, f_gen_url):
|
||||
'''Passes in an naIdToken to the imink API and fetches the response (comprised of an f token, UUID, and timestamp).'''
|
||||
|
||||
try:
|
||||
api_head = {
|
||||
'User-Agent': f's3s/{s3s_version}',
|
||||
'Content-Type': 'application/json; charset=utf-8'
|
||||
}
|
||||
api_body = {
|
||||
'token': id_token,
|
||||
'hashMethod': step
|
||||
}
|
||||
api_response = requests.post(f_gen_url, data=json.dumps(api_body), headers=api_head)
|
||||
resp = json.loads(api_response.text)
|
||||
|
||||
f = resp["f"]
|
||||
uuid = resp["request_id"]
|
||||
timestamp = resp["timestamp"]
|
||||
return f, uuid, timestamp
|
||||
except:
|
||||
try: # if api_response never gets set
|
||||
if api_response.text:
|
||||
print(u"Error from the imink API:\n{}".format(json.dumps(json.loads(api_response.text), indent=2, ensure_ascii=False)))
|
||||
else:
|
||||
print(f"Error from the imink API: Error {api_response.status_code}.")
|
||||
except:
|
||||
pass
|
||||
sys.exit(1)
|
||||
|
||||
def enter_tokens():
|
||||
'''Prompts the user to enter a gtoken and bulletToken.'''
|
||||
|
||||
print("Go to the page below to find instructions to obtain your gtoken and bulletToken:\nhttps://github.com/frozenpandaman/s3s/wiki/mitmproxy-instructions\n")
|
||||
|
||||
new_gtoken = input("Enter your gtoken: ")
|
||||
while len(new_gtoken) != 124:
|
||||
new_gtoken = input("Invalid token - length should be 124 characters. Try again.\nEnter your gtoken: ")
|
||||
|
||||
new_bullettoken = input("Enter your bulletToken: ")
|
||||
while len(new_bullettoken) != 926:
|
||||
new_bullettoken = input("Invalid token - length should be 926 characters. Try again.\nEnter your bulletToken: ")
|
||||
|
||||
return new_gtoken, new_bullettoken
|
||||
852
s3s.py
Executable file
852
s3s.py
Executable file
@@ -0,0 +1,852 @@
|
||||
#!/usr/bin/env python
|
||||
# s3s (ↄ) 2022 eli fessler (frozenpandaman)
|
||||
# Based on splatnet2statink (ↄ) 2017-2022 eli fessler (frozenpandaman), clovervidia
|
||||
# https://github.com/frozenpandaman/s3s
|
||||
# License: GPLv3
|
||||
|
||||
import sys, os, requests, json, time, datetime, argparse, msgpack
|
||||
from PIL import Image, ImageDraw
|
||||
from packaging import version
|
||||
import iksm
|
||||
|
||||
A_VERSION = "0.0.1"
|
||||
|
||||
print(f"s3s v{A_VERSION}")
|
||||
|
||||
# CONFIG.TXT CREATION
|
||||
if getattr(sys, 'frozen', False): # place config.txt in same directory as script (bundled or not)
|
||||
app_path = os.path.dirname(sys.executable)
|
||||
elif __file__:
|
||||
app_path = os.path.dirname(__file__)
|
||||
config_path = os.path.join(app_path, "config.txt")
|
||||
|
||||
try:
|
||||
config_file = open(config_path, "r")
|
||||
CONFIG_DATA = json.load(config_file)
|
||||
config_file.close()
|
||||
except (IOError, ValueError):
|
||||
print("Generating new config file.")
|
||||
CONFIG_DATA = {"api_key": "", "acc_loc": "", "gtoken": "", "bullettoken": "", "session_token": "", "f_gen": "https://api.imink.app/f"}
|
||||
config_file = open(config_path, "w")
|
||||
config_file.seek(0)
|
||||
config_file.write(json.dumps(CONFIG_DATA, indent=4, sort_keys=True, separators=(',', ': ')))
|
||||
config_file.close()
|
||||
config_file = open(config_path, "r")
|
||||
CONFIG_DATA = json.load(config_file)
|
||||
config_file.close()
|
||||
|
||||
# SET GLOBALS
|
||||
API_KEY = CONFIG_DATA["api_key"] # for stat.ink
|
||||
USER_LANG = CONFIG_DATA["acc_loc"][:5] # nintendo account info
|
||||
USER_COUNTRY = CONFIG_DATA["acc_loc"][-2:] # nintendo account info
|
||||
GTOKEN = CONFIG_DATA["gtoken"] # for accessing splatnet
|
||||
BULLETTOKEN = CONFIG_DATA["bullettoken"] # for accessing splatnet
|
||||
SESSION_TOKEN = CONFIG_DATA["session_token"] # for nintendo login
|
||||
F_GEN_URL = CONFIG_DATA["f_gen"] # endpoint for generating f (imink API by default)
|
||||
# UNIQUE_ID = CONFIG_DATA["app_unique_id"] # NPLN player ID
|
||||
|
||||
GRAPHQL_URL = "https://api.lp1.av5ja.srv.nintendo.net/api/graphql"
|
||||
WEB_VIEW_VER = '1.0.0-5e2bcdfb' # TODO - parse from js, don't hardcode
|
||||
|
||||
# SET HTTP HEADERS
|
||||
if "app_user_agent" in CONFIG_DATA:
|
||||
APP_USER_AGENT = str(CONFIG_DATA["app_user_agent"])
|
||||
else:
|
||||
APP_USER_AGENT = 'Mozilla/5.0 (Linux; Android 11; Pixel 5) ' \
|
||||
'AppleWebKit/537.36 (KHTML, like Gecko) ' \
|
||||
'Chrome/94.0.4606.61 Mobile Safari/537.36'
|
||||
|
||||
graphql_head = {
|
||||
'Authorization': f'Bearer {BULLETTOKEN}',
|
||||
'Accept-Language': USER_LANG,
|
||||
'User-Agent': APP_USER_AGENT,
|
||||
'X-Web-View-Ver': WEB_VIEW_VER,
|
||||
'Content-Type': 'application/json',
|
||||
'Accept': '*/*',
|
||||
'Origin': 'https://api.lp1.av5ja.srv.nintendo.net',
|
||||
'X-Requested-With': 'com.nintendo.znca',
|
||||
'Referer': f'https://api.lp1.av5ja.srv.nintendo.net/?lang={USER_LANG}&na_country={USER_COUNTRY}&na_lang={USER_LANG}',
|
||||
'Accept-Encoding': 'gzip, deflate'
|
||||
}
|
||||
|
||||
# SHA256 hash database for SplatNet 3 GraphQL queries
|
||||
# full list: https://github.com/samuelthomas2774/nxapi/blob/main/src/api/splatnet3-types.ts#L28-L98
|
||||
translate_rid = {
|
||||
'LatestBattleHistoriesQuery': '7d8b560e31617e981cf7c8aa1ca13a00', # blank vars - query1
|
||||
'VsHistoryDetailQuery': 'cd82f2ade8aca7687947c5f3210805a6', # req "vsResultId" - query2
|
||||
'CoopHistoryQuery': '817618ce39bcf5570f52a97d73301b30', # blank vars - query1
|
||||
'CoopHistoryDetailQuery': 'f3799a033f0a7ad4b1b396f9a3bafb1e', # req "coopHistoryDetailId" - query2
|
||||
'HomeQuery': 'dba47124d5ec3090c97ba17db5d2f4b3', # blank vars
|
||||
}
|
||||
|
||||
|
||||
def set_noun(which):
|
||||
'''Returns the term to be used when referring to the type of results in question.'''
|
||||
|
||||
if which == "both":
|
||||
return "battles/jobs"
|
||||
elif which == "salmon":
|
||||
return "jobs"
|
||||
else: # "ink"
|
||||
return "battles"
|
||||
|
||||
|
||||
def b64d(string):
|
||||
'''Base64 decode a string and cut off the SplatNet prefix.'''
|
||||
|
||||
thing_id = base64.b64decode(string).decode('utf-8')
|
||||
thing_id = thing_id.replace("VsStage-", "")
|
||||
thing_id = thing_id.replace("Weapon-", "")
|
||||
thing_id = thing_id.replace("CoopStage-", "")
|
||||
thing_id = thing_id.replace("CoopGrade-", "")
|
||||
return int(thing_id)
|
||||
|
||||
|
||||
def gen_graphql_body(sha256hash, varname=None, varvalue=None):
|
||||
'''Generates a JSON dictionary, specifying information to retrieve, to send with GraphQL requests.'''
|
||||
great_passage = {
|
||||
"extensions": {
|
||||
"persistedQuery": {
|
||||
"sha256Hash": sha256hash,
|
||||
"version": 1
|
||||
}
|
||||
},
|
||||
"variables": {}
|
||||
}
|
||||
|
||||
if varname != None and varvalue != None:
|
||||
great_passage["variables"][varname] = varvalue
|
||||
|
||||
return json.dumps(great_passage)
|
||||
|
||||
|
||||
def write_config(tokens):
|
||||
'''Writes config file and updates the global variables.'''
|
||||
|
||||
config_file = open(config_path, "w")
|
||||
config_file.seek(0)
|
||||
config_file.write(json.dumps(tokens, indent=4, sort_keys=True, separators=(',', ': ')))
|
||||
config_file.close()
|
||||
|
||||
config_file = open(config_path, "r")
|
||||
CONFIG_DATA = json.load(config_file)
|
||||
|
||||
global API_KEY
|
||||
API_KEY = CONFIG_DATA["api_key"]
|
||||
global USER_LANG
|
||||
USER_LANG = CONFIG_DATA["acc_loc"][:5]
|
||||
global USER_COUNTRY
|
||||
USER_COUNTRY = CONFIG_DATA["acc_loc"][-2:]
|
||||
global GTOKEN
|
||||
GTOKEN = CONFIG_DATA["gtoken"]
|
||||
global BULLETTOKEN
|
||||
BULLETTOKEN = CONFIG_DATA["bullettoken"]
|
||||
global SESSION_TOKEN
|
||||
SESSION_TOKEN = CONFIG_DATA["session_token"]
|
||||
|
||||
config_file.close()
|
||||
|
||||
|
||||
def custom_key_exists(key, value=True):
|
||||
'''Checks if a given custom key exists in config.txt and is set to the specified value (true by default).'''
|
||||
|
||||
# https://github.com/frozenpandaman/s3s/wiki/config-keys
|
||||
if key not in ["ignore_private", "app_user_agent"]:
|
||||
print("(!) checking unexpected custom key")
|
||||
return True if key in config_data and config_data[key].lower() == str(value).lower() else False
|
||||
|
||||
|
||||
def prefetch_checks():
|
||||
'''Queries the SplatNet 3 homepage to check if our gtoken cookie and bulletToken are still valid, otherwise regenerate.'''
|
||||
|
||||
if SESSION_TOKEN == "" or GTOKEN == "" or BULLETTOKEN == "":
|
||||
gen_new_tokens("blank")
|
||||
|
||||
sha = translate_rid["HomeQuery"]
|
||||
test = requests.post(GRAPHQL_URL, data=gen_graphql_body(sha), headers=graphql_head, cookies=dict(_gtoken=GTOKEN))
|
||||
if test.status_code != 200:
|
||||
gen_new_tokens("expiry")
|
||||
|
||||
|
||||
def gen_new_tokens(reason, force=False):
|
||||
'''Attempts to generate new tokens when the saved ones have expired.'''
|
||||
|
||||
manual_entry = False
|
||||
if force != True: # unless we force our way through
|
||||
if reason == "blank":
|
||||
print("Blank token(s).")
|
||||
elif reason == "expiry":
|
||||
print("The stored tokens have expired.")
|
||||
else:
|
||||
print("Cannot access SplatNet 3 without having played online.")
|
||||
sys.exit(1)
|
||||
|
||||
if SESSION_TOKEN == "":
|
||||
print("Please log in to your Nintendo Account to obtain your session_token.")
|
||||
new_token = iksm.log_in(A_VERSION)
|
||||
if new_token == None:
|
||||
print("There was a problem logging you in. Please try again later.")
|
||||
elif new_token == "skip":
|
||||
manual_entry = True
|
||||
else:
|
||||
print("\nWrote session_token to config.txt.")
|
||||
CONFIG_DATA["session_token"] = new_token
|
||||
write_config(CONFIG_DATA)
|
||||
elif SESSION_TOKEN == "skip":
|
||||
manual_entry = True
|
||||
|
||||
if manual_entry: # no session_token ever gets stored
|
||||
print("\nYou have opted against automatic token generation and must manually input your tokens.\n")
|
||||
new_gtoken, new_bullettoken = iksm.enter_tokens()
|
||||
else:
|
||||
print("Attempting to generate new gtoken and bulletToken...")
|
||||
new_gtoken, acc_name, acc_lang, acc_country = iksm.get_gtoken(F_GEN_URL, SESSION_TOKEN, A_VERSION)
|
||||
new_bullettoken = iksm.get_bullet(new_gtoken, WEB_VIEW_VER, APP_USER_AGENT, acc_lang, acc_country)
|
||||
CONFIG_DATA["gtoken"] = new_gtoken # valid for 2 hours
|
||||
CONFIG_DATA["bullettoken"] = new_bullettoken # valid for 2 hours
|
||||
CONFIG_DATA["acc_loc"] = acc_lang + "|" + acc_country
|
||||
write_config(CONFIG_DATA)
|
||||
|
||||
if manual_entry:
|
||||
print("Wrote tokens to config.txt.")
|
||||
else:
|
||||
print(f"Wrote tokens for {acc_name} to config.txt.")
|
||||
|
||||
|
||||
def fetch_json(which, separate=False, exportall=False):
|
||||
'''Returns results JSON from SplatNet 3, including a combined dict for ink battles + SR jobs if requested.'''
|
||||
|
||||
if exportall and not separate:
|
||||
print("fetch_json() must be called with separate=True if using exportall.")
|
||||
sys.exit(1)
|
||||
|
||||
prefetch_checks()
|
||||
|
||||
ink_list, salmon_list = [], []
|
||||
parent_files = []
|
||||
|
||||
sha_list = []
|
||||
if which == "both" or which == "ink":
|
||||
sha_list.append(translate_rid["LatestBattleHistoriesQuery"])
|
||||
else:
|
||||
sha_list.append(None)
|
||||
if which == "both" or which == "salmon":
|
||||
sha_list.append(translate_rid["CoopHistoryQuery"])
|
||||
else:
|
||||
sha_list.append(None)
|
||||
|
||||
for sha in sha_list:
|
||||
if sha != None:
|
||||
battle_ids, job_ids = [], []
|
||||
|
||||
query1 = requests.post(GRAPHQL_URL, data=gen_graphql_body(sha), headers=graphql_head, cookies=dict(_gtoken=GTOKEN))
|
||||
query1_resp = json.loads(query1.text)
|
||||
|
||||
# prefetch_checks() ensures this exists
|
||||
try: # ink battle
|
||||
for b in query1_resp["data"]["latestBattleHistories"]["historyGroups"]["nodes"][0]["historyDetails"]["nodes"]:
|
||||
battle_ids.append(b["id"])
|
||||
except: # salmon run job
|
||||
for j in query1_resp["data"]["coopResult"]["historyGroups"]["nodes"][0]["historyDetails"]["nodes"]:
|
||||
job_ids.append(j["id"])
|
||||
|
||||
for bid in battle_ids:
|
||||
query2_b = requests.post(GRAPHQL_URL,
|
||||
data=gen_graphql_body(translate_rid["VsHistoryDetailQuery"], "vsResultId", bid),
|
||||
headers=graphql_head,
|
||||
cookies=dict(_gtoken=GTOKEN))
|
||||
query2_resp_b = json.loads(query2_b.text)
|
||||
ink_list.append(query2_resp_b)
|
||||
|
||||
for jid in job_ids:
|
||||
query2_j = requests.post(GRAPHQL_URL,
|
||||
data=gen_graphql_body(translate_rid["CoopHistoryDetailQuery"], "coopHistoryDetailId", jid),
|
||||
headers=graphql_head,
|
||||
cookies=dict(_gtoken=GTOKEN))
|
||||
query2_resp_j = json.loads(query2_j.text)
|
||||
salmon_list.append(query2_resp_j)
|
||||
|
||||
parent_files.append(query1_resp)
|
||||
else: # sha = None (we don't want to get the specified result type)
|
||||
pass
|
||||
|
||||
if exportall:
|
||||
return parent_files, ink_list, salmon_list
|
||||
elif separate:
|
||||
return ink_list, salmon_list
|
||||
else:
|
||||
return ink_list + salmon_list
|
||||
|
||||
|
||||
def update_salmon_profile():
|
||||
''' Updates stat.ink Salmon Run stats/profile.'''
|
||||
|
||||
pass # TODO
|
||||
|
||||
# prefetch_checks()
|
||||
|
||||
# old code
|
||||
# .../api/coop_results - need stat.ink s3 support
|
||||
# results_list = requests.get(url, headers=graphql_head, cookies=dict(_gtoken=GTOKEN))
|
||||
# data = json.loads(results_list.text)
|
||||
# profile = data["summary"]
|
||||
|
||||
# payload = {
|
||||
# "work_count": profile["card"]["job_num"],
|
||||
# "total_golden_eggs": profile["card"]["golden_ikura_total"],
|
||||
# "total_eggs": profile["card"]["ikura_total"],
|
||||
# "total_rescued": profile["card"]["help_total"],
|
||||
# "total_point": profile["card"]["kuma_point_total"]
|
||||
# # TODO - other new things - fish scales, xtrawave stuff?, etc.
|
||||
# }
|
||||
|
||||
# url = "https://stat.ink/api/v2/salmon-stats" # TODO - need stat.ink s3 support
|
||||
# auth = {'Authorization': f'Bearer {API_KEY}'}
|
||||
# updateprofile = requests.post(url, headers=auth, data=payload)
|
||||
|
||||
# if updateprofile.ok:
|
||||
# print("Successfully updated your Salmon Run profile.")
|
||||
# else:
|
||||
# print("Could not update your Salmon Run profile. Error from stat.ink:")
|
||||
# print(updateprofile.text)
|
||||
|
||||
|
||||
def prepare_battle_result(battle):
|
||||
'''Converts the Nintendo JSON format for a Turf War/Ranked battle to the stat.ink one.'''
|
||||
|
||||
payload = {}
|
||||
battle = battle["data"]["vsHistoryDetail"]
|
||||
|
||||
## MODE ##
|
||||
##########
|
||||
mode = battle["vsMode"]["mode"]
|
||||
if mode == "REGULAR":
|
||||
payload["lobby"] = "regular"
|
||||
elif mode == "BANKARA":
|
||||
if battle["bankaraMatch"]["mode"] == "OPEN":
|
||||
payload["lobby"] = "bankara_open"
|
||||
elif battle["bankaraMatch"]["mode"] == "CHALLENGE":
|
||||
payload["lobby"] = "bankara_challenge"
|
||||
elif mode == "PRIVATE":
|
||||
payload["lobby"] = "private"
|
||||
# TODO - splatfest stuff
|
||||
|
||||
## RULE ##
|
||||
##########
|
||||
rule = battle["vsRule"]["rule"]
|
||||
if rule == "TURF_WAR":
|
||||
payload["rule"] = "nawabari"
|
||||
elif rule == "AREA":
|
||||
payload["rule"] = "area"
|
||||
elif rule == "LOFT":
|
||||
payload["rule"] = "yagura"
|
||||
elif rule == "GOAL":
|
||||
payload["rule"] = "hoko"
|
||||
elif rule == "CLAM":
|
||||
payload["rule"] = "asari"
|
||||
|
||||
## STAGE ##
|
||||
###########
|
||||
# hardcoded in alpha
|
||||
stage_id = b64d(battle["vsStage"]["id"])
|
||||
if stage_id == 1:
|
||||
payload["stage"] = "gorge" # yunohana
|
||||
elif stage_id == 2:
|
||||
payload["stage"] = "alley" # gonzui
|
||||
elif stage_id == 3:
|
||||
payload["stage"] = "market" # yagara
|
||||
elif stage_id == 4:
|
||||
payload["stage"] = "spillway" # mategai
|
||||
elif stage_id == 6:
|
||||
payload["stage"] = "metalworks" # namero
|
||||
elif stage_id == 10:
|
||||
payload["stage"] = "bridge" # masaba
|
||||
elif stage_id == 11:
|
||||
payload["stage"] = "museum" # kinmedai
|
||||
elif stage_id == 12:
|
||||
payload["stage"] = "resort" # mahimahi
|
||||
elif stage_id == 13:
|
||||
payload["stage"] = "academy" # amabi
|
||||
elif stage_id == 14:
|
||||
payload["stage"] = "shipyard" # chozame
|
||||
elif stage_id == 15:
|
||||
payload["stage"] = "mart" # zatou
|
||||
elif stage_id == 16:
|
||||
payload["stage"] = "world" # sumeshi
|
||||
|
||||
## WEAPON, K/D, TURF INKED ##
|
||||
############
|
||||
for teammate in battle["myTeam"]["players"]:
|
||||
if player["isMyself"] == True:
|
||||
payload["weapon"] = b64d(player["weapon"]["id"])
|
||||
payload["kill"] = player["result"]["kill"]
|
||||
payload["assist"] = player["result"]["assist"]
|
||||
payload["kill_or_assist"] = payload["kill"] + payload["assist"]
|
||||
payload["death"] = player["result"]["death"]
|
||||
payload["special"] = player["result"]["special"]
|
||||
# payload["inked"] = player["paint"] # TODO - check how bonus works?? two diff values?
|
||||
# player["result"]["noroshiTry"] = ultra signal attempts
|
||||
payload["species"] = player["species"].lower() # not supported for now
|
||||
|
||||
|
||||
## RESULT ##
|
||||
############
|
||||
result = battle["myTeam"]["judgement"]
|
||||
if result == "WIN":
|
||||
payload["result"] = "win"
|
||||
elif result == "LOSE":
|
||||
payload["result"] = "lose"
|
||||
elif result == "DRAW":
|
||||
payload["result"] = "draw"
|
||||
|
||||
## LEVEL ## TODO
|
||||
###########
|
||||
# payload["level_before"] = ...
|
||||
# payload["level_after"] = ...
|
||||
|
||||
## CASH ## TODO
|
||||
##########
|
||||
# payload["cash_before"] = ...
|
||||
# payload["cash_after"] = ...
|
||||
|
||||
## START/END TIMES ##
|
||||
#####################
|
||||
time_string = battle["playedTime"] # UTC
|
||||
utc_time = datetime.datetime.strptime(time_string, "%Y-%m-%dT%H:%M:%SZ")
|
||||
epoch_time = int((utc_time - datetime.datetime(1970, 1, 1)).total_seconds())
|
||||
payload["start_at"] = epoch_time
|
||||
payload["end_at"] = epoch_time + battle["duration"]
|
||||
|
||||
## RANK IN TEAM ## TODO
|
||||
##################
|
||||
# payload["rank_in_team"] = 1, 2, 3, 4
|
||||
|
||||
# Turf War only (NOT TRICOLOR)
|
||||
if mode == "REGULAR":
|
||||
payload["our_team_percent"] = float(battle["myTeam"]["result"]["paintRatio"]) * 100
|
||||
payload["their_team_percent"] = float(battle["otherTeams"][0]["result"]["paintRatio"]) * 100
|
||||
|
||||
our_team_inked, their_team_inked = 0, 0
|
||||
for player in battle["myTeam"]["players"]:
|
||||
our_team_inked += player["paint"]
|
||||
for player in battle["otherTeams"][0]["players"]:
|
||||
their_team_inked += player["paint"]
|
||||
payload["our_team_inked"] = our_team_inked
|
||||
payload["their_team_inked"] = their_team_inked
|
||||
|
||||
# Anarchy Battles only
|
||||
if mode == "BANKARA":
|
||||
payload["our_team_count"] = battle["myTeam"]["result"]["score"]
|
||||
payload["their_team_count"] = battle["otherTeams"][0]["result"]["score"]
|
||||
payload["knockout"] = "yes" if battle["knockout"] != "NEITHER" else "no" # or check if either == 100
|
||||
|
||||
# TODO old code - have to get from parent request
|
||||
# try: # not present in all modes
|
||||
# rank_after = battle["udemae"]["name"].lower()
|
||||
# rank_before = battle["player_result"]["player"]["udemae"]["name"].lower()
|
||||
# rank_s_plus_num_after = battle["udemae"]["s_plus_number"]
|
||||
# rank_s_plus_num_before = battle["player_result"]["player"]["udemae"]["s_plus_number"]
|
||||
# except:
|
||||
# rank_after, rank_before, rank_s_plus_num_after, rank_s_plus_num_before = None, None, None, None
|
||||
|
||||
# payload["rank_after"] = rank_after
|
||||
# payload["rank_before"] = rank_before
|
||||
# payload["rank_after_s_plus"] = rank_s_plus_num_after
|
||||
# payload["rank_before_s_plus"] = rank_s_plus_num_before
|
||||
|
||||
# payload["rank_before_exp"] = ...
|
||||
# payload["rank_after_exp"] = ...
|
||||
|
||||
payload["automated"] = "yes" # data was not manually entered!
|
||||
payload["splatnet_json"] = json.dumps(battle)
|
||||
|
||||
return payload
|
||||
|
||||
|
||||
def prepare_job_result(battle):
|
||||
'''Converts the Nintendo JSON format for a Salmon Run job to the stat.ink one.'''
|
||||
|
||||
pass # TODO
|
||||
# combo of set_teammates() + salmon_post_shift()
|
||||
# set payload["splatnet_json"]
|
||||
|
||||
|
||||
def post_result(data, isblackout, istestrun):
|
||||
'''Uploads battle/job JSON to stat.ink, and prints the returned URL or error message..'''
|
||||
|
||||
try:
|
||||
results = data["results"] # list of dictionaries
|
||||
except KeyError:
|
||||
results = [data] # single battle/job
|
||||
|
||||
# filter down to one battle at a time
|
||||
for i in range(len(results)):
|
||||
if results[i].has_key("battle_number"): # ink battle
|
||||
payload = prepare_battle_result(results[i])
|
||||
elif results[i].has_key("job_id"): # salmon run job
|
||||
payload = prepare_job_result(results[i])
|
||||
else: # shouldn't happen
|
||||
print("Ill-formatted JSON while uploading. Exiting.")
|
||||
sys.exit(1)
|
||||
|
||||
if payload["lobby"] == "private" and custom_key_exists("ignore_private"):
|
||||
continue
|
||||
|
||||
# TODO - do stuff with isblackout - check for salmon run too??
|
||||
|
||||
s3s_values = {'agent': '\u0073\u0033\u0073', 'agent_version': A_VERSION} # lol
|
||||
s3s_values["agent_variables"] = {} # TODO - monitoring mode or not
|
||||
payload.update(s3s_values)
|
||||
|
||||
if payload["agent"][0:3] != os.path.basename(__file__)[:-3]:
|
||||
print("Could not upload. Please contact @frozenpandaman on GitHub for assistance.")
|
||||
sys.exit(1)
|
||||
|
||||
if istestrun:
|
||||
payload["test"] = "yes"
|
||||
|
||||
# post
|
||||
url = "https://stat.ink/api/v3/battle"
|
||||
auth = {'Authorization': f'Bearer {API_KEY}', 'Content-Type': 'application/x-msgpack'}
|
||||
postbattle = requests.post(url, headers=auth, data=msgpack.packb(payload), allow_redirects=False)
|
||||
|
||||
# response
|
||||
headerloc = postbattle.headers.get('location')
|
||||
if headerloc != None:
|
||||
if postbattle.status_code == 302: # redirect
|
||||
print(f"Battle already uploaded - {headerloc}")
|
||||
else: # 200 OK
|
||||
print(f"Battle uploaded to {headerloc}")
|
||||
else: # error
|
||||
print("Error uploading battle. Message from server:")
|
||||
print(postbattle.content.decode('utf-8'))
|
||||
|
||||
|
||||
def check_for_updates():
|
||||
'''Checks the script version against the repo, reminding users to update if available.'''
|
||||
|
||||
print("Auto-update is not yet implemented. Please update the script regularly with `git pull`.")
|
||||
# TODO
|
||||
|
||||
|
||||
def check_statink_key():
|
||||
'''Checks if a valid length API key has been provided and, if not, prompts the user to enter one.'''
|
||||
|
||||
if API_KEY == "skip":
|
||||
return
|
||||
elif len(API_KEY) != 43:
|
||||
new_api_key = ""
|
||||
while len(new_api_key.strip()) != 43 and new_api_key.strip() != "skip":
|
||||
if new_api_key.strip() == "" and API_KEY.strip() == "":
|
||||
new_api_key = input("stat.ink API key: ")
|
||||
else:
|
||||
print("Invalid stat.ink API key. Please re-enter it below.")
|
||||
new_api_key = input("stat.ink API key: ")
|
||||
CONFIG_DATA["api_key"] = new_api_key
|
||||
write_config(CONFIG_DATA)
|
||||
return
|
||||
|
||||
|
||||
def get_num_results(which):
|
||||
'''I/O for getting number of battles/jobs to upload.'''
|
||||
|
||||
noun = set_noun(which)
|
||||
try:
|
||||
n = int(input(f"Number of recent {noun} to upload (0-50)? "))
|
||||
except ValueError:
|
||||
print("Please enter an integer between 0 and 50. Exiting.")
|
||||
sys.exit(1)
|
||||
if n < 1:
|
||||
print("Exiting without uploading anything.")
|
||||
sys.exit(0)
|
||||
elif n > 50:
|
||||
print(f"SplatNet 3 only stores the 50 most recent {noun}. Exiting.")
|
||||
sys.exit(1)
|
||||
else:
|
||||
return n
|
||||
|
||||
|
||||
def check_if_missing(which, isblackout, istestrun):
|
||||
'''Checks for unuploaded battles, and uploads any that are found.'''
|
||||
|
||||
noun = set_noun(which)
|
||||
print(f"Checking if there are previously-unuploaded {noun}...")
|
||||
|
||||
urls = []
|
||||
if which == "both" or which == "ink":
|
||||
urls.append("https://stat.ink/api/v2/user-battle?only=splatnet_number&count=100") # TODO - update for s3
|
||||
else:
|
||||
urls.append(None)
|
||||
if which == "both" or which == "salmon":
|
||||
urls.append("https://stat.ink/api/v2/user-salmon?only=splatnet_number&count=100") # TODO - update for s3
|
||||
else:
|
||||
urls.append(None)
|
||||
|
||||
noun = "battles" # first (and maybe only)
|
||||
for url in urls:
|
||||
if url != None:
|
||||
printed = False
|
||||
auth = {'Authorization': f'Bearer {API_KEY}'}
|
||||
resp = requests.get(url, headers=auth)
|
||||
try:
|
||||
statink_uploads = json.loads(resp.text)
|
||||
except:
|
||||
print(f"Encountered an error while checking recently-uploaded {noun}. Is stat.ink down?")
|
||||
sys.exit(1)
|
||||
|
||||
# ! fetch from online
|
||||
results = fetch_json(which)
|
||||
|
||||
splatnet_results = [] # 50 recent battles/jobs on splatnet
|
||||
for i, result in reversed(list(enumerate(results))):
|
||||
try:
|
||||
num = int(result["battle_number"])
|
||||
except KeyError:
|
||||
try:
|
||||
num = int(result["job_id"])
|
||||
except:
|
||||
print(f"Ill-formatted JSON while checking missing {noun}. Exiting.")
|
||||
sys.exit(1)
|
||||
splatnet_results.append(num)
|
||||
if num not in statink_uploads: # one of the splatnet entries isn't on stat.ink (unuploaded)
|
||||
if not printed:
|
||||
printed = True
|
||||
print(f"Previously-unuploaded {noun} detected. Uploading now...")
|
||||
post_result(results[i], isblackout, istestrun)
|
||||
|
||||
if not printed:
|
||||
print(f"No previously-unuploaded {noun} found.")
|
||||
|
||||
noun = "jobs" # for second run through the loop
|
||||
|
||||
|
||||
def monitor_battles(which, secs, isblackout, istestrun):
|
||||
'''Monitors JSON for changes/new battles and uploads them.'''
|
||||
|
||||
# ! fetch from online
|
||||
battles_results, jobs_results = fetch_json(which, True) # separate the jsons
|
||||
|
||||
cached_battles = []
|
||||
cached_jobs = []
|
||||
for result in reversed(battles_results):
|
||||
cached_battles.append(int(result["battle_number"]))
|
||||
for result in reversed(jobs_results):
|
||||
cached_jobs.append(int(result["job_id"]))
|
||||
|
||||
# counters
|
||||
battle_wins, battle_losses, splatfest_wins, splatfest_losses, mirror_matches = [0]*5 # init all to 0
|
||||
job_successes, job_failures = [0]*2
|
||||
|
||||
mins = str(round(float(secs)/60.0, 2))
|
||||
mins = mins[:-2] if mins[-2:] == ".0" else mins
|
||||
print(f"Waiting for new {set_noun(which)}... (checking every {mins} minutes)")
|
||||
|
||||
try:
|
||||
while True:
|
||||
for i in range(secs, -1, -1):
|
||||
sys.stdout.write(f"Press Ctrl+C to exit. {i} ")
|
||||
sys.stdout.flush()
|
||||
time.sleep(1)
|
||||
sys.stdout.write("\r")
|
||||
|
||||
# ! fetch from online
|
||||
ink_results, salmon_results = fetch_json(which, True)
|
||||
|
||||
if which == "both" or which == "ink":
|
||||
for i, result in reversed(list(enumerate(battles_results))):
|
||||
if int(result["battle_number"]) not in cached_battles:
|
||||
if result["game_mode"]["key"] == "private" and custom_key_exists("ignore_private"):
|
||||
pass
|
||||
else:
|
||||
outcome = "Won" if result["my_team_result"]["key"] == "victory" else "Lost"
|
||||
splatfest_match = True if result["game_mode"]["key"] in ["fes_solo", "fes_team"] else False
|
||||
if splatfest_match: # keys will exist
|
||||
our_key = result["my_team_fes_theme"]["key"]
|
||||
their_key = result["other_team_fes_theme"]["key"]
|
||||
mirror_match = True if our_key == their_key else False
|
||||
if outcome == "Won":
|
||||
battle_wins += 1
|
||||
if splatfest_match and not mirror_match:
|
||||
splatfest_wins += 1
|
||||
else: # Lost
|
||||
battle_losses += 1
|
||||
if splatfest_match and not mirror_match:
|
||||
splatfest_losses += 1
|
||||
if splatfest_match and mirror_match:
|
||||
mirror_matches += 1
|
||||
# stagename = "..." # TODO - get short name??
|
||||
dt = datetime.datetime.fromtimestamp(int(result["start_time"])).strftime('%I:%M:%S %p').lstrip("0")
|
||||
# print(f"New battle result detected at {dt}! ({stagename}, {outcome})")
|
||||
print(f"New battle result detected at {dt}! ({outcome})") # TODO - temp workaround
|
||||
cached_battles.append(int(result["battle_number"]))
|
||||
post_result(result, isblackout, istestrun)
|
||||
|
||||
if which == "both" or which == "salmon":
|
||||
for i, result in reversed(list(enumerate(jobs_results))):
|
||||
if int(result["job_id"]) not in cached_jobs:
|
||||
if result["game_mode"]["key"] == "private" and custom_key_exists("ignore_private"):
|
||||
pass
|
||||
else:
|
||||
outcome = "Success" if result["job_result"]["is_clear"] == True else "Failure"
|
||||
if outcome == "Success":
|
||||
job_successes += 1
|
||||
else: # Failure
|
||||
job_failures += 1
|
||||
# stagename = "..." # TODO - get short name??
|
||||
dt = datetime.datetime.fromtimestamp(int(result["start_time"])).strftime('%I:%M:%S %p').lstrip("0")
|
||||
# print(f"New job result detected at {dt}! ({stagename}, {outcome})")
|
||||
print(f"New job result detected at {dt}! ({outcome})") # TODO - temp workaround
|
||||
cached_jobs.append(int(result["job_id"]))
|
||||
post_result(result, isblackout, istestrun)
|
||||
|
||||
except KeyboardInterrupt:
|
||||
# print(f"\nChecking to see if there are unuploaded {set_noun(which)} before exiting...")
|
||||
# TODO
|
||||
# do update_salmon_profile() at end if salmon run
|
||||
print("Checking for unuploaded results before exiting is not yet implemented.")
|
||||
print("Bye!")
|
||||
|
||||
|
||||
def main():
|
||||
'''Main process, including I/O and setup.'''
|
||||
|
||||
# setup
|
||||
#######
|
||||
check_for_updates()
|
||||
# check_statink_key() # TODO - temp. disabled
|
||||
|
||||
# argparse stuff
|
||||
################
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument("-M", dest="N", required=False, nargs="?", action="store",
|
||||
help="monitoring mode; pull data every N secs (default: 300)", const=300)
|
||||
parser.add_argument("-r", required=False, action="store_true",
|
||||
help="retroactively post unuploaded battles/jobs")
|
||||
parser.add_argument("-nsr", required=False, action="store_true",
|
||||
help="do not check for Salmon Run jobs")
|
||||
parser.add_argument("-osr", required=False, action="store_true",
|
||||
help="only check for Salmon Run jobs")
|
||||
parser.add_argument("--blackout", required=False, action="store_true",
|
||||
help="black out names on scoreboard result images")
|
||||
|
||||
parser.add_argument("-t", required=False, action="store_true",
|
||||
help="dry run for testing (won't post to stat.ink)")
|
||||
parser.add_argument("-i", dest="filename", required=False, help=argparse.SUPPRESS)
|
||||
parser.add_argument("-o", required=False, action="store_true", help=argparse.SUPPRESS)
|
||||
|
||||
parser_result = parser.parse_args()
|
||||
|
||||
# regular args
|
||||
n_value = parser_result.N
|
||||
check_old = parser_result.r
|
||||
only_ink = parser_result.nsr # ink battles ONLY
|
||||
only_salmon = parser_result.osr # salmon run ONLY
|
||||
blackout = parser_result.blackout
|
||||
|
||||
# testing/dev stuff
|
||||
test_run = parser_result.t
|
||||
filename = parser_result.filename # intended for results json or battle files
|
||||
outfile = parser_result.o # output to local files
|
||||
|
||||
# i/o checks
|
||||
############
|
||||
if only_ink and only_salmon:
|
||||
print("That doesn't make any sense! :) Exiting.")
|
||||
sys.exit(1)
|
||||
|
||||
if filename and len(sys.argv) > 2:
|
||||
print("Cannot use -i with other arguments. Exiting.")
|
||||
sys.exit(1)
|
||||
elif outfile and len(sys.argv) > 2:
|
||||
print("Cannot use -o with other arguments. Exiting.")
|
||||
sys.exit(1)
|
||||
|
||||
secs = -1
|
||||
if n_value != None:
|
||||
try:
|
||||
secs = int(parser_result.N)
|
||||
except ValueError:
|
||||
print("Number provided must be an integer. Exiting.")
|
||||
sys.exit(1)
|
||||
if secs < 0:
|
||||
print("No.")
|
||||
sys.exit(1)
|
||||
elif secs < 60:
|
||||
print("Minimum number of seconds in monitoring mode is 60. Exiting.")
|
||||
sys.exit(1)
|
||||
|
||||
# exporting results to file
|
||||
###########################
|
||||
if outfile:
|
||||
print("\nFetching your JSON files to export locally...")
|
||||
parents, results, coop_results = fetch_json("both", True, True) # will do prefetch_checks()
|
||||
|
||||
cwd = os.getcwd()
|
||||
export_dir = os.path.join(cwd, f'export-{int(time.time())}')
|
||||
if not os.path.exists(export_dir):
|
||||
os.makedirs(export_dir)
|
||||
|
||||
if parents != None:
|
||||
with open(os.path.join(cwd, export_dir, "overview.json"), "x") as fout:
|
||||
json.dump(parents, fout)
|
||||
print("Created overview.json with general info about your battle and job stats.")
|
||||
|
||||
if results != None:
|
||||
with open(os.path.join(cwd, export_dir, "results.json"), "x") as fout:
|
||||
json.dump(results, fout)
|
||||
print("Created results.json with detailed recent battles stats (up to 50).")
|
||||
|
||||
if coop_results != None:
|
||||
with open(os.path.join(cwd, export_dir, "coop_results.json"), "x") as fout:
|
||||
json.dump(coop_results, fout)
|
||||
print("Created coop_results.json with detailed recent Salmon Run job stats (up to 50).")
|
||||
|
||||
print("\nHave fun playing Splatoon 3! :) Bye!")
|
||||
sys.exit(0)
|
||||
|
||||
# manual json upload
|
||||
####################
|
||||
if filename:
|
||||
if not os.path.exists(filename):
|
||||
argparse.ArgumentParser().error(f"File {filename} does not exist!") # exit
|
||||
with open(filename) as data_file:
|
||||
try:
|
||||
data = json.load(data_file)
|
||||
except ValueError:
|
||||
print("Could not decode JSON object in this file.")
|
||||
sys.exit(1)
|
||||
post_result(data, blackout, test_run) # one or multiple
|
||||
sys.exit(0)
|
||||
|
||||
# regular run
|
||||
#############
|
||||
which = "ink" if only_ink else "salmon" if only_salmon else "both"
|
||||
|
||||
# ---
|
||||
# TEMP. UNTIL STAT.INK SUPPORTS S3
|
||||
print("\nNOTE: stat.ink does not yet support Splatoon 3. Re-run the script with the -o flag " \
|
||||
"to save your current SplatNet 3 results (up to most recent 50 battles & jobs) to JSON files, " \
|
||||
"which can later be manually uploaded with -i file.json. This message will be removed when support is added.")
|
||||
sys.exit(0)
|
||||
# ---
|
||||
|
||||
if which != "ink":
|
||||
update_salmon_profile()
|
||||
|
||||
if check_old:
|
||||
check_if_missing(which, blackout, test_run)
|
||||
if secs != -1: # monitoring mode
|
||||
monitor_battles(which, secs, blackout, test_run)
|
||||
else: # regular mode (no -M)
|
||||
n = get_num_results(which)
|
||||
print("Pulling data from online...")
|
||||
|
||||
# ! fetch from online
|
||||
results = fetch_json(which)
|
||||
|
||||
for i in reversed(range(n)):
|
||||
post_result(results[i], blackout, test_run)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user