Merge branch 'master' into master

This commit is contained in:
eli
2022-10-26 16:42:55 -10:00
committed by GitHub
3 changed files with 236 additions and 119 deletions

166
iksm.py
View File

@@ -2,42 +2,119 @@
# https://github.com/frozenpandaman/s3s
# License: GPLv3
import requests, json, re, sys
import os, base64, hashlib
import base64, hashlib, json, os, re, sys
import requests
from bs4 import BeautifulSoup
session = requests.Session()
S3S_VERSION = "unknown"
NSOAPP_VERSION = "2.3.1"
S3S_VERSION = "unknown"
NSOAPP_VERSION = "unknown"
NSOAPP_VER_FALLBACK = "2.3.1"
WEB_VIEW_VERSION = "unknown"
WEB_VIEW_VER_FALLBACK = "1.0.0-5644e7a2" # fallback for current splatnet 3 ver
SPLATNET3_URL = "https://api.lp1.av5ja.srv.nintendo.net"
# functions in this file & call stack:
# get_nsoapp_version()
# log_in() -> get_session_token()
# get_gtoken() -> call_imink_api() -> f
# get_bullet()
# enter_tokens()
# - get_nsoapp_version()
# - get_web_view_ver()
# - log_in() -> get_session_token()
# - get_gtoken() -> call_f_api()
# - 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")
session = requests.Session()
def get_nsoapp_version():
'''Fetches the current Nintendo Switch Online app version from the Apple App Store.'''
'''Fetches the current Nintendo Switch Online app version from the Apple App Store and sets it globally.'''
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
if NSOAPP_VERSION != "unknown": # already set
return NSOAPP_VERSION
else:
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()
NSOAPP_VERSION = ver
return NSOAPP_VERSION
except: # error with web request
return NSOAPP_VER_FALLBACK
def log_in(ver):
def get_web_view_ver(bhead=[], gtoken=""):
'''Finds & parses the SplatNet 3 main.js file to fetch the current site version and sets it globally.'''
global WEB_VIEW_VERSION
if WEB_VIEW_VERSION != "unknown":
return WEB_VIEW_VERSION
else:
app_head = {
'Upgrade-Insecure-Requests': '1',
'Accept': '*/*',
'DNT': '1',
'X-AppColorScheme': 'DARK',
'X-Requested-With': 'com.nintendo.znca',
'Sec-Fetch-Site': 'none',
'Sec-Fetch-Mode': 'navigate',
'Sec-Fetch-User': '?1',
'Sec-Fetch-Dest': 'document'
}
app_cookies = {
'_dnt': '1' # Do Not Track
}
if bhead:
app_head["User-Agent"] = bhead.get("User-Agent")
app_head["Accept-Encoding"] = bhead.get("Accept-Encoding")
app_head["Accept-Language"] = bhead.get("Accept-Language")
if gtoken:
app_cookies["_gtoken"] = gtoken # X-GameWebToken
home = requests.get(SPLATNET3_URL, headers=app_head, cookies=app_cookies)
if home.status_code != 200:
return WEB_VIEW_VER_FALLBACK
soup = BeautifulSoup(home.text, "html.parser")
main_js = soup.select_one("script[src*='static']")
if not main_js: # failed to parse html for main.js file
return WEB_VIEW_VER_FALLBACK
main_js_url = SPLATNET3_URL + main_js.attrs["src"]
app_head = {
'Accept': '*/*',
'X-Requested-With': 'com.nintendo.znca',
'Sec-Fetch-Site': 'same-origin',
'Sec-Fetch-Mode': 'no-cors',
'Sec-Fetch-Dest': 'script'
}
if bhead:
app_head["User-Agent"] = bhead.get("User-Agent")
app_head["Referer"] = bhead.get("Referer")
app_head["Accept-Encoding"] = bhead.get("Accept-Encoding")
app_head["Accept-Language"] = bhead.get("Accept-Language")
main_js_body = requests.get(main_js_url, headers=app_head, cookies=app_cookies)
if main_js_body.status_code != 200:
return iksm.WEB_VIEW_VER_FALLBACK
pattern = r"\b(?P<revision>[0-9a-f]{40})\b.*revision_info_not_set\"\),.*?=\"(?P<version>\d+\.\d+\.\d+)"
match = re.search(pattern, main_js_body.text)
if match is None:
return iksm.WEB_VIEW_VER_FALLBACK
version, revision = match.group("version"), match.group("revision")
ver_string = f"{version}-{revision[:8]}"
WEB_VIEW_VERSION = ver_string
return WEB_VIEW_VERSION
def log_in(ver, app_user_agent):
'''Logs in to a Nintendo Account and returns a session_token.'''
global S3S_VERSION
@@ -55,7 +132,7 @@ def log_in(ver):
'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',
'User-Agent': app_user_agent,
'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',
@@ -128,7 +205,7 @@ def get_session_token(session_token_code, auth_code_verifier):
def get_gtoken(f_gen_url, session_token, ver):
'''Provided the session_token, returns a GameWebToken and account info.'''
'''Provided the session_token, returns a GameWebToken JWT and account info.'''
nsoapp_version = get_nsoapp_version()
@@ -184,7 +261,7 @@ def get_gtoken(f_gen_url, session_token, ver):
body = {}
try:
id_token = id_response["id_token"]
f, uuid, timestamp = call_imink_api(id_token, 1, f_gen_url)
f, uuid, timestamp = call_f_api(id_token, 1, f_gen_url)
parameter = {
'f': f,
@@ -223,7 +300,7 @@ def get_gtoken(f_gen_url, session_token, ver):
except:
# retry once if 9403/9599 error from nintendo
try:
f, uuid, timestamp = call_imink_api(id_token, 1, f_gen_url)
f, uuid, timestamp = call_f_api(id_token, 1, f_gen_url)
body["parameter"]["f"] = f
body["parameter"]["requestId"] = uuid
body["parameter"]["timestamp"] = timestamp
@@ -238,7 +315,7 @@ def get_gtoken(f_gen_url, session_token, ver):
print("Re-running the script usually fixes this.")
sys.exit(1)
f, uuid, timestamp = call_imink_api(id_token, 2, f_gen_url)
f, uuid, timestamp = call_f_api(id_token, 2, f_gen_url)
# get web service token
app_head = {
@@ -270,7 +347,7 @@ def get_gtoken(f_gen_url, session_token, ver):
except:
# retry once if 9403/9599 error from nintendo
try:
f, uuid, timestamp = call_imink_api(id_token, 2, f_gen_url)
f, uuid, timestamp = call_f_api(id_token, 2, f_gen_url)
body["parameter"]["f"] = f
body["parameter"]["requestId"] = uuid
body["parameter"]["timestamp"] = timestamp
@@ -286,24 +363,25 @@ def get_gtoken(f_gen_url, session_token, ver):
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.'''
def get_bullet(web_service_token, app_user_agent, user_lang, user_country):
'''Given a gtoken, 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-Web-View-Ver': get_web_view_ver(),
'X-NACOUNTRY': user_country,
'Accept': '*/*',
'Origin': 'https://api.lp1.av5ja.srv.nintendo.net',
'Origin': SPLATNET3_URL,
'X-Requested-With': 'com.nintendo.znca'
}
app_cookies = {
'_gtoken': web_service_token # X-GameWebToken
'_gtoken': web_service_token, # X-GameWebToken
'_dnt': '1' # Do Not Track
}
url = "https://api.lp1.av5ja.srv.nintendo.net/api/bullet_tokens"
url = f'{SPLATNET3_URL}/api/bullet_tokens'
r = requests.post(url, headers=app_head, cookies=app_cookies)
if r.status_code == 401:
@@ -320,7 +398,7 @@ def get_bullet(web_service_token, web_view_ver, app_user_agent, user_lang, user_
bullet_resp = json.loads(r.text)
bullet_token = bullet_resp["bulletToken"]
except (json.decoder.JSONDecodeError, TypeError):
print("Got non-JSON response from Nintendo (in api/bullet_tokens step:")
print("Got non-JSON response from Nintendo (in api/bullet_tokens step):")
print(bullet_resp)
sys.exit(1)
except:
@@ -331,8 +409,8 @@ def get_bullet(web_service_token, web_view_ver, app_user_agent, user_lang, user_
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).'''
def call_f_api(id_token, step, f_gen_url):
'''Passes an naIdToken to the f generation API (default: imink) & fetches the response (f token, UUID, and timestamp).'''
try:
api_head = {
@@ -365,7 +443,8 @@ def call_imink_api(id_token, step, f_gen_url):
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")
print("Go to the page below to find instructions to obtain your gtoken and bulletToken:")
print("https://github.com/frozenpandaman/s3s/wiki/mitmproxy-instructions\n")
new_gtoken = input("Enter your gtoken: ")
while len(new_gtoken) != 926:
@@ -379,3 +458,8 @@ def enter_tokens():
new_bullettoken = input("Invalid token - length should be 124 characters. Try again.\nEnter your bulletToken: ")
return new_gtoken, new_bullettoken
if __name__ == "__main__":
print("This program cannot be run alone. See https://github.com/frozenpandaman/s3s")
sys.exit(0)

130
s3s.py Executable file → Normal file
View File

@@ -5,12 +5,11 @@
# License: GPLv3
import argparse, datetime, json, os, shutil, re, requests, sys, time, uuid
from concurrent.futures import ThreadPoolExecutor
import msgpack
import iksm, utils
from concurrent.futures import ThreadPoolExecutor
A_VERSION = "0.1.6"
A_VERSION = "0.1.10"
DEBUG = False
@@ -52,12 +51,11 @@ F_GEN_URL = CONFIG_DATA["f_gen"] # endpoint for generating f (imink
thread_pool = ThreadPoolExecutor(max_workers=10)
# 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'
DEFAULT_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'
APP_USER_AGENT = str(CONFIG_DATA.get("app_user_agent", DEFAULT_USER_AGENT))
def write_config(tokens):
'''Writes config file and updates the global variables.'''
@@ -87,28 +85,32 @@ def write_config(tokens):
def headbutt():
'''Return a (dynamic!) header used for GraphQL requests.'''
'''Returns a (dynamic!) header used for GraphQL requests.'''
graphql_head = {
'Authorization': f'Bearer {BULLETTOKEN}', # update every time it's called with current global var
'Accept-Language': USER_LANG,
'User-Agent': APP_USER_AGENT,
'X-Web-View-Ver': utils.get_web_view_ver(),
'X-Web-View-Ver': iksm.WEB_VIEW_VERSION,
'Content-Type': 'application/json',
'Accept': '*/*',
'Origin': 'https://api.lp1.av5ja.srv.nintendo.net',
'Origin': iksm.SPLATNET3_URL,
'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}',
'Referer': f'{iksm.SPLATNET3_URL}?lang={USER_LANG}&na_country={USER_COUNTRY}&na_lang={USER_LANG}',
'Accept-Encoding': 'gzip, deflate'
}
return graphql_head
def prefetch_checks(printout=False):
'''Queries the SplatNet 3 homepage to check if our gtoken cookie and bulletToken are still valid, otherwise regenerate.'''
'''Queries the SplatNet 3 homepage to check if our gtoken & bulletToken are still valid and regenerates them if not.'''
if printout:
print("Validating your tokens...", end='\r')
global WEB_VIEW_VERSION
WEB_VIEW_VERSION = iksm.get_web_view_ver()
if SESSION_TOKEN == "" or GTOKEN == "" or BULLETTOKEN == "":
gen_new_tokens("blank")
@@ -129,7 +131,7 @@ def gen_new_tokens(reason, force=False):
manual_entry = False
if force != True: # unless we force our way through
if reason == "blank":
print("Blank token(s).")
print("Blank token(s). ")
elif reason == "expiry":
print("The stored tokens have expired.")
else:
@@ -138,7 +140,7 @@ def gen_new_tokens(reason, force=False):
if SESSION_TOKEN == "":
print("Please log in to your Nintendo Account to obtain your session_token.")
new_token = iksm.log_in(A_VERSION)
new_token = iksm.log_in(A_VERSION, APP_USER_AGENT)
if new_token is None:
print("There was a problem logging you in. Please try again later.")
elif new_token == "skip":
@@ -159,7 +161,7 @@ def gen_new_tokens(reason, force=False):
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, utils.get_web_view_ver(), APP_USER_AGENT, acc_lang, acc_country)
new_bullettoken = iksm.get_bullet(new_gtoken, 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
@@ -172,7 +174,7 @@ def gen_new_tokens(reason, force=False):
def fetch_json(which, separate=False, exportall=False, specific=False, numbers_only=False, printout=False, skipprefetch=False):
'''Returns results JSON from SplatNet 3, including a combined dict for ink battles + SR jobs if requested.'''
'''Returns results JSON from SplatNet 3, including a combined dictionary for battles + SR jobs if requested.'''
swim = SquidProgress()
@@ -535,13 +537,28 @@ def prepare_battle_result(battle, ismonitoring, overview_data=None):
payload["our_team_inked"] = our_team_inked
payload["their_team_inked"] = their_team_inked
# Anarchy Battles only
if mode == "PRIVATE": # these don't get sent otherwise
try: # could be anarchy, maybe not
payload["knockout"] = "no" if battle["knockout"] is None or battle["knockout"] == "NEITHER" else "yes"
except:
pass
try:
payload["our_team_count"] = battle["myTeam"]["result"]["score"]
payload["their_team_count"] = battle["otherTeams"][0]["result"]["score"]
except:
pass
try:
payload["our_team_percent"] = float(battle["myTeam"]["result"]["paintRatio"]) * 100
payload["their_team_percent"] = float(battle["otherTeams"][0]["result"]["paintRatio"]) * 100
except:
pass
if mode == "BANKARA":
try:
payload["our_team_count"] = battle["myTeam"]["result"]["score"]
payload["their_team_count"] = battle["otherTeams"][0]["result"]["score"]
except TypeError: # draw - 'result' is null
except: # draw - 'result' is null
pass
payload["knockout"] = "no" if battle["knockout"] is None or battle["knockout"] == "NEITHER" else "yes"
@@ -571,8 +588,10 @@ def prepare_battle_result(battle, ismonitoring, overview_data=None):
if child["id"] == battle["id"]: # found the battle ID in the other file
full_rank = re.split('([0-9]+)', child["udemae"].lower())
was_s_plus_before = len(full_rank) > 1 # true if "before" rank is s+
payload["rank_before"] = full_rank[0]
if len(full_rank) > 1:
if was_s_plus_before:
payload["rank_before_s_plus"] = int(full_rank[1])
# anarchy battle (series) - not open
@@ -589,6 +608,8 @@ def prepare_battle_result(battle, ismonitoring, overview_data=None):
if parent["bankaraMatchChallenge"]["udemaeAfter"] is None:
payload["rank_after"] = payload["rank_before"]
if was_s_plus_before:
payload["rank_after_s_plus"] = payload["rank_before_s_plus"] # also s+ after
else:
if idx != 0:
payload["rank_after"] = payload["rank_before"]
@@ -650,12 +671,13 @@ def prepare_job_result(battle, ismonitoring, overview_data=None):
'''Converts the Nintendo JSON format for a Salmon Run job to the stat.ink one.'''
pass # stat.ink doesn't support SR yet
# combo of set_teammates() + salmon_post_shift()
# combo of set_teammates() + salmon_post_shift() - mirrors prepare_battle_result()
# specify if private or not; also ensure SR results are formatted the same in post_result()!
# set payload["splatnet_json"]
def post_result(data, ismonitoring, isblackout, istestrun, overview_data=None):
'''Uploads battle/job JSON to stat.ink, and prints the returned URL or error message..'''
'''Uploads battle/job JSON to stat.ink, and prints the returned URL or error message.'''
if isinstance(data, list): # -o export format
try:
@@ -690,7 +712,7 @@ def post_result(data, ismonitoring, isblackout, istestrun, overview_data=None):
continue
# should have been taken care of in fetch_json() but just in case...
if payload["lobby"] == "private" and utils.custom_key_exists("ignore_private", CONFIG_DATA): # TODO - also check SR?
if payload["lobby"] == "private" and utils.custom_key_exists("ignore_private", CONFIG_DATA):
continue
# TODO - isblackout stuff... for SR too
@@ -742,6 +764,7 @@ def post_result(data, ismonitoring, isblackout, istestrun, overview_data=None):
def check_for_updates():
'''Checks the script version against the repo, reminding users to update if available.'''
# TODO
print('\033[3m' + "» While s3s is in beta, please update the script regularly via " \
'`\033[91m' + "git pull" + '\033[0m' + "`." + '\033[0m' + "\n")
# try:
@@ -833,7 +856,7 @@ def fetch_and_upload_single_result(hash, noun, ismonitoring, isblackout, istestr
def check_if_missing(which, ismonitoring, isblackout, istestrun):
'''Checks for unuploaded battles, and uploads any that are found.'''
'''Checks for unuploaded battles and uploads any that are found.'''
noun = utils.set_noun(which)
print(f"Checking if there are previously-unuploaded {noun}...")
@@ -890,7 +913,7 @@ def check_if_missing(which, ismonitoring, isblackout, istestrun):
def monitor_battles(which, secs, isblackout, istestrun):
'''Monitors JSON for changes/new battles and uploads them.'''
'''Monitors SplatNet endpoint(s) for changes (new results) and uploads them.'''
if DEBUG:
print(f"* monitoring mode start - calling fetch_json() w/ which={which}")
@@ -973,7 +996,7 @@ def monitor_battles(which, secs, isblackout, istestrun):
cached_battles.append(num)
post_result(result, True, isblackout, istestrun) # True = is monitoring mode
if which in ("both", "salmon"):
if which in ("salmon"): # TODO - add in "both" when we have SR support
for num in reversed(salmon_results):
if num not in cached_jobs:
# get the full job data
@@ -983,13 +1006,14 @@ def monitor_battles(which, secs, isblackout, istestrun):
cookies=dict(_gtoken=GTOKEN))
result = json.loads(result_post.text)
if False and utils.custom_key_exists("ignore_private"): # TODO - how to check for SR private battles?
if result["data"]["coopHistoryDetail"]["jobScore"] == None \
and utils.custom_key_exists("ignore_private", CONFIG_DATA):
pass
else:
outcome = "Success" if result["job_result"]["is_clear"] == True else "Failure"
if outcome == "Success":
outcome = "Clear" if result["data"]["coopHistoryDetail"]["resultWave"] == 0 else "Defeat"
if outcome == "Clear":
job_successes += 1
else: # Failure
else:
job_failures += 1
stagename = result["data"]["coopHistoryDetail"]["coopStage"]["name"]
@@ -1005,14 +1029,20 @@ def monitor_battles(which, secs, isblackout, istestrun):
except KeyboardInterrupt:
# print(f"\nChecking to see if there are unuploaded {utils.set_noun(which)} before exiting...") # TODO
# TODO - do update_salmon_profile() at end if salmon run
# check LatestBattleHistoriesQuery against https://stat.ink/api/v3/s3s/uuid-list
# if SR:
# check CoopHistoryQuery
# update_salmon_profile()
# show job/battle/splatfest_wins, etc.
print("\n\nChecking for unuploaded results before exiting is not yet implemented.")
print("Please run s3s again with " + '\033[91m' + "-r" + '\033[0m' + " to get these battles.")
print("Bye!")
class SquidProgress:
'''Display animation while waiting.'''
'''Displays an animation of a squid swimming while waiting. :)'''
def __init__(self):
self.count = 0
@@ -1031,18 +1061,9 @@ class SquidProgress:
sys.stdout.flush()
def main():
'''Main process, including I/O and setup.'''
def parse_arguments():
'''Setup for command-line options.'''
print('\033[93m\033[1m' + "s3s" + '\033[0m\033[93m' + f" v{A_VERSION}" + '\033[0m')
# setup
#######
check_for_updates()
check_statink_key()
# argparse stuff
################
parser = argparse.ArgumentParser()
srgroup = parser.add_mutually_exclusive_group()
parser.add_argument("-M", dest="N", required=False, nargs="?", action="store",
@@ -1061,7 +1082,22 @@ def main():
help="upload local results. use `-i results.json overview.json`")
parser.add_argument("-t", required=False, action="store_true",
help="dry run for testing (won't post to stat.ink)")
parser_result = parser.parse_args()
return parser.parse_args()
def main():
'''Main process, including I/O and setup.'''
print('\033[93m\033[1m' + "s3s" + '\033[0m\033[93m' + f" v{A_VERSION}" + '\033[0m')
# setup
#######
check_for_updates()
check_statink_key()
# argparse setup
################
parser_result = parse_arguments()
# regular args
n_value = parser_result.N
@@ -1163,7 +1199,7 @@ def main():
try:
statink_uploads = json.loads(resp.text)
except:
print(f"Encountered an error while checking recently-uploaded {noun}. Is stat.ink down?")
print(f"Encountered an error while checking recently-uploaded data. Is stat.ink down?")
sys.exit(1)
to_upload = []
@@ -1201,7 +1237,7 @@ def main():
'\033[91m' + "-i results.json overview.json" + '\033[0m' + ". " \
"Or, run the script in monitoring mode (with " + '\033[91m' + "-M" + '\033[0m' \
") to capture & upload new results as you play. " \
"stat.ink does not support Salmon Run data (coop_results.json) or Splatfest battles at this time.\n")
"stat.ink does not support Salmon Run data (coop_results.json) at this time.\n")
# ---
if which in ("salmon", "both"):

View File

@@ -2,16 +2,23 @@
# https://github.com/frozenpandaman/s3s
# License: GPLv3
import base64, datetime, json, re, requests, uuid
import base64, datetime, json, re, sys, uuid
import requests
from bs4 import BeautifulSoup
import iksm
SPLATNET3_URL = "https://api.lp1.av5ja.srv.nintendo.net"
GRAPHQL_URL = "https://api.lp1.av5ja.srv.nintendo.net/api/graphql"
WEB_VIEW_VERSION = "1.0.0-216d0219" # fallback
SPLATNET3_URL = iksm.SPLATNET3_URL
GRAPHQL_URL = f'{SPLATNET3_URL}/api/graphql'
S3S_NAMESPACE = uuid.UUID('b3a2dbf5-2c09-4792-b78c-00b548b70aeb')
SUPPORTED_KEYS = [
"ignore_private",
"app_user_agent",
"force_uploads"
]
# SHA256 hash database for SplatNet 3 GraphQL queries
# full list: https://github.com/samuelthomas2774/nxapi/discussions/11#discussioncomment-3737698
# full list: https://github.com/samuelthomas2774/nxapi/discussions/11#discussioncomment-3614603
translate_rid = {
'HomeQuery': 'dba47124d5ec3090c97ba17db5d2f4b3', # blank vars
'LatestBattleHistoriesQuery': '7d8b560e31617e981cf7c8aa1ca13a00', # INK / blank vars - query1
@@ -20,29 +27,9 @@ translate_rid = {
'PrivateBattleHistoriesQuery': '38e0529de8bc77189504d26c7a14e0b8', # INK / blank vars - query1
'VsHistoryDetailQuery': '2b085984f729cd51938fc069ceef784a', # INK / req "vsResultId" - query2
'CoopHistoryQuery': '817618ce39bcf5570f52a97d73301b30', # SR / blank vars - query1
'CoopHistoryDetailQuery': 'f3799a033f0a7ad4b1b396f9a3bafb1e', # SR / req "coopHistoryDetailId" - query2
'CoopHistoryDetailQuery': 'f3799a033f0a7ad4b1b396f9a3bafb1e' # SR / req "coopHistoryDetailId" - query2
}
def get_web_view_ver():
'''Find & parse the SplatNet 3 main.js file for the current site version.'''
splatnet3_home = requests.get(SPLATNET3_URL)
soup = BeautifulSoup(splatnet3_home.text, "html.parser")
main_js = soup.select_one("script[src*='static']")
if not main_js:
return WEB_VIEW_VERSION
main_js_url = SPLATNET3_URL + main_js.attrs["src"]
main_js_body = requests.get(main_js_url)
match = re.search(r"\b(?P<revision>[0-9a-f]{40})\b.*revision_info_not_set\"\),.*?=\"(?P<version>\d+\.\d+\.\d+)", main_js_body.text)
if not match:
return WEB_VIEW_VERSION
version, revision = match.group("version"), match.group("revision")
return f"{version}-{revision[:8]}"
def set_noun(which):
'''Returns the term to be used when referring to the type of results in question.'''
@@ -56,14 +43,19 @@ def set_noun(which):
def b64d(string):
'''Base64 decode a string and cut off the SplatNet prefix.'''
'''Base64-decodes a string and cuts off the SplatNet prefix.'''
thing_id = base64.b64decode(string).decode('utf-8')
thing_id = thing_id.replace("VsStage-", "")
thing_id = thing_id.replace("VsMode-", "")
thing_id = thing_id.replace("Weapon-", "")
thing_id = thing_id.replace("CoopStage-", "")
thing_id = thing_id.replace("CoopGrade-", "")
if "Weapon-" in thing_id:
thing_id = thing_id.replace("Weapon-", "")
if len(thing_id) == 5 and thing_id[:1] == "2" and thing_id[-3:] == "900": # grizzco weapon ID from a hacker
return ""
if thing_id[:15] == "VsHistoryDetail" or thing_id[:17] == "CoopHistoryDetail":
return thing_id # string
else:
@@ -71,7 +63,7 @@ def b64d(string):
def epoch_time(time_string):
'''Converts a playedTime string into an int representing the epoch time.'''
'''Converts a playedTime string into an integer representing the epoch time.'''
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())
@@ -100,6 +92,11 @@ def custom_key_exists(key, config_data, 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", "force_uploads"]:
if key not in SUPPORTED_KEYS:
print("(!) Checking unexpected custom key")
return True if key in config_data and config_data[key].lower() == str(value).lower() else False
return str(config_data.get(key, None)).lower() == str(value).lower()
if __name__ == "__main__":
print("This program cannot be run alone. See https://github.com/frozenpandaman/s3s")
sys.exit(0)