diff --git a/iksm.py b/iksm.py index 6371c21..077c5db 100644 --- a/iksm.py +++ b/iksm.py @@ -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[0-9a-f]{40})\b.*revision_info_not_set\"\),.*?=\"(?P\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) diff --git a/s3s.py b/s3s.py old mode 100755 new mode 100644 index 4500734..6232ff2 --- a/s3s.py +++ b/s3s.py @@ -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"): diff --git a/utils.py b/utils.py index d133eae..7f223c7 100644 --- a/utils.py +++ b/utils.py @@ -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[0-9a-f]{40})\b.*revision_info_not_set\"\),.*?=\"(?P\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)