From ff4e786bfb65d456246beca42acbd069a0284dcd Mon Sep 17 00:00:00 2001 From: Jylpah Date: Sun, 30 Dec 2018 00:12:26 +0200 Subject: [PATCH] Accidentially uploaded the old file --- blitzutils.py.bak | 794 ---------------------------------------------- 1 file changed, 794 deletions(-) delete mode 100644 blitzutils.py.bak diff --git a/blitzutils.py.bak b/blitzutils.py.bak deleted file mode 100644 index e1815e2..0000000 --- a/blitzutils.py.bak +++ /dev/null @@ -1,794 +0,0 @@ -#!/usr/bin/python3.7 - -import sys, os, json, time, base64, urllib, inspect, hashlib -import asyncio, aiofiles, aiohttp, aiosqlite, lxml -from bs4 import BeautifulSoup - -MAX_RETRIES= 3 -SLEEP = 2 -DEBUG = False -VERBOSE = False -SILENT = False -UMASK= os.umask(0) -os.umask(UMASK) - -# def chkJSONaccountList(account_list: list) -> bool: -# """"Check String for being a valid accountID list JSON file""" -# try: -# if int(account_list[0]) > 0: -# return True -# except: -# cls.debug("JSON check failed") -# return False - -def setDebug(debug: bool): - global DEBUG, VERBOSE, SILENT - if debug != None: - DEBUG = debug - if DEBUG: - VERBOSE = True - SILENT = False - -def setVerbose(verbose: bool): - global VERBOSE, SILENT - if verbose != None: - VERBOSE = verbose - if VERBOSE: - SILENT = False - -def setSilent(silent: bool): - global DEBUG, VERBOSE, SILENT - if silent != None: - SILENT = silent - if SILENT: - VERBOSE = False - DEBUG = False - -def verbose(msg = ""): - """Print a message""" - if VERBOSE: - print(msg) - return None - -def verbose_std(msg = ""): - """Print a message""" - if not SILENT: - print(msg) - return None - -def printWaiter(force = False): - if not DEBUG and (not SILENT or force): - print('.', end='', flush=True) - -def printNewline(force = False): - if not DEBUG and (not SILENT or force): - print('', flush=True) - -def debug(msg = ""): - """print a conditional debug message""" - if DEBUG: - curframe = inspect.currentframe() - calframe = inspect.getouterframes(curframe, 2) - caller = calframe[1][3] - print('DEBUG: ' + caller + '(): ' + msg) - return None - -def error(msg = ""): - """Print an error message""" - curframe = inspect.currentframe() - calframe = inspect.getouterframes(curframe, 2) - caller = calframe[1][3] - print('ERROR: ' + caller + '(): ' + msg) - return None -def NOW() -> int: - return int(time.time()) - -async def saveJSON(filename: str, json_data: dict, sort_keys = False) -> bool: - """Save JSON data into file""" - try: - dirname = os.path.dirname(filename) - if (dirname != '') and not os.path.isdir(dirname): - os.makedirs(dirname, 0o770-UMASK) - async with aiofiles.open(filename,'w', encoding="utf8") as outfile: - await outfile.write(json.dumps(json_data, ensure_ascii=False, indent=4, sort_keys=sort_keys)) - return True - except Exception as err: - error(str(err)) - return False - -async def readJSON(filename: str, chkJSONfunc = None): - try: - async with aiofiles.open(filename) as fp: - json_data = json.loads(await fp.read()) - if (chkJSONfunc == None): - debug("JSON file content not checked: " + filename) - return json_data - elif chkJSONfunc(json_data): - debug("JSON File is valid: " + filename) - return json_data - else: - debug('JSON File has invalid content: ' + filename) - except Exception as err: - error('Unexpected error when reading file: ' + filename + ' : ' + str(type(err)) + ' : '+ str(err)) - return None - -async def getUrlJSON(session: aiohttp.ClientSession, url: str, chkJSONfunc = None, max_tries = MAX_RETRIES) -> dict: - """Retrieve (GET) an URL and return JSON object""" - try: - debug(url) - ## To avoid excessive use of servers - for retry in range(1,max_tries+1): - async with session.get(url) as resp: - if resp.status == 200: - debug('HTTP request OK') - json_resp = await resp.json() - if (chkJSONfunc == None) or chkJSONfunc(json_resp): - # debug("Received valid JSON: " + str(json_resp)) - return json_resp - else: - debug('Received JSON error. URL: ' + url) - # Sometimes WG API returns JSON error even a retry gives valid JSON - # return None - if retry == max_tries: - raise aiohttp.ClientError('Request failed: ' + str(resp.status) + ' JSON Response: ' + str(json_resp) ) - verbose('Retrying URL [' + str(retry) + '/' + str(max_tries) + ']: ' + url ) - await asyncio.sleep(SLEEP) - - except aiohttp.ClientError as err: - error("Could not retrieve URL: " + url) - error(str(err)) - except asyncio.CancelledError as err: - error('Queue gets cancelled while still working.') - - except Exception as err: - error('Unexpected Exception: ' + str(err)) - return None - -def bldDictHierarcy(d : dict, key : str, value) -> dict: - """Build hierarcical dict based on multi-level key separated with """ - try: - key_hier = key.split('.') - sub_key = key_hier.pop(0) - if len(key_hier) == 0: - d[sub_key] = value - elif sub_key not in d: - d[sub_key] = bldDictHierarcy({}, key_hier, value) - else: - d[sub_key] = bldDictHierarcy(d[sub_key], key_hier, value) - - return d - except KeyError as err: - error('Key not found: ' + str(err)) - except Exception as err: - error(str(err)) - return None - - -class WG: - - URL_WG_clanInfo = 'clans/info/?application_id=' - #URL_WG_playerTankList = 'tanks/stats/?fields=tank_id%2Clast_battle_time&application_id=' - URL_WG_playerTankList = 'tanks/stats/?fields=account_id%2Ctank_id%2Clast_battle_time%2Cbattle_life_time%2Call&application_id=' - URL_WG_playerTankStats = 'tanks/stats/?application_id=' - URL_WG_accountID = 'account/list/?fields=account_id%2Cnickname&application_id=' - URL_WG_playerStats = 'account/info/?application_id=' - SQLITE_STATS_CACHE = '.stats_cache.db' - - sql_create_player_stats_tbl = """CREATE TABLE IF NOT EXISTS player_stats ( - account_id INTEGER NOT NULL, - date INTEGER NOT NULL, - stat TEXT, - value FLOAT - ); """ - - sql_create_player_tank_stats_tbl = """CREATE TABLE IF NOT EXISTS player_tank_stats ( - account_id INTEGER NOT NULL, - tank_id INTEGER DEFAULT NULL, - date INTEGER NOT NULL, - stat TEXT, - value FLOAT - ); """ - - sql_select_player_stats = """SELECT value FROM player_stats ORDERBY date ASC - WHERE account_id = {} AND stat = {} - AND date >= {} LIMIT 1;""" - - - maps = { - "Random": "Random map", - "amigosville": "Falls Creek", - "asia": "Lost Temple", - "canal": "Canal", - "canyon": "Canyon", - "desert_train": "Desert Sands", - "erlenberg": "Middleburg", - "faust": "Faust", - "fort": "Macragge", - "grossberg": "Dynasty's Pearl", - "himmelsdorf": "Himmelsdorf", - "italy": "Vineyards", - "karelia": "Rockfield", - "karieri": "Copperfield", - "lake": "Mirage", - "lumber": "Alpenstadt", - "malinovka": "Winter Malinovka", - "medvedkovo": "Dead Rail", - "milbase": "Yamato Harbor", - "mountain": "Black Goldville", - "north": "North", - "ordeal": "Trial by Fire", - "pliego": "Castilla", - "port": "Port Bay", - "rock": "Mayan Ruins", - "rudniki": "Mines", - "savanna": "Oasis Palms", - "skit": "Naval Frontier", - "test": "World of Ducks", - "tutorial": "Proving Grounds" - } - - nations = [ 'ussr', 'germany', 'usa', 'china', 'france', 'uk', 'japan', 'other'] - nation_id = { - 'ussr' : 0, - 'germany' : 1, - 'usa' : 2, - 'china' : 3, - 'france' : 4, - 'uk' : 5, - 'japan' : 6, - 'other' : 7 - } - - tank_type = [ 'lightTank', 'mediumTank', 'heavyTank', 'AT-SPG' ] - tank_type_id = { - 'lightTank' : 0, - 'mediumTank' : 1, - 'heavyTank' : 2, - 'AT-SPG' : 3 - } - - URL_WG_server = { - 'eu' : 'https://api.wotblitz.eu/wotb/', - 'ru' : 'https://api.wotblitz.ru/wotb/', - 'na' : 'https://api.wotblitz.com/wotb/', - 'asia' : 'https://api.wotblitz.asia/wotb/' - } - - accountID_server= { - 'ru' : range(0, int(5e8)), - 'eu' : range(int(5e8), int(10e8)), - 'na' : range(int(1e9),int(2e9)), - 'asia': range(int(2e9),int(4e9)) - } - - def __init__(self, WG_appID = None, tankopedia_fn = None, maps_fn = None): - - self.WG_appID = WG_appID - self.tanks = None - - if (tankopedia_fn != None): - if os.path.exists(tankopedia_fn) and os.path.isfile(tankopedia_fn): - try: - with open(tankopedia_fn, 'rt', encoding='utf8') as f: - self.tanks = json.loads(f.read()) - except Exception as err: - error('Could not read tankopedia: ' + tankopedia_fn + '\n' + str(err)) - else: - verbose('Could not find Tankopedia file: ' + tankopedia_fn) - if (maps_fn != None): - if os.path.exists(maps_fn) and os.path.isfile(maps_fn): - try: - with open(maps_fn, 'rt', encoding='utf8') as f: - self.maps = json.loads(f.read()) - except Exception as err: - error('Could not read maps file: ' + maps_fn + '\n' + str(err)) - else: - verbose('Could not find maps file: ' + maps_fn) - if self.WG_appID != None: - self.session = aiohttp.ClientSession() - debug('WG aiohttp session initiated') - else: - self.session = None - debug('WG aiohttp session NOT initiated') - - self.cache = None - self.statsQ = None - self.statCacheTask = None - #if os.path.isfile(self.SQLITE_STATS_CACHE): - if False: - try: - self.cache = aiosqlite.connect('SQLITE_STATS_CACHE') - self.cache.execute(self.sql_create_player_stats_tbl) - self.cache.execute(self.sql_create_player_tank_stats_tbl) - self.cache.commit() - self.statsQ = asyncio.Queue() - self.statCacheTask = asyncio.create_task(self.statsSaver()) - - except Exception as err: - error(str(err)) - - async def close(self): - if self.statsQ != None: - await self.statsQ.join() - self.statCacheTask.cancel() - await asyncio.gather(*self.statCacheTask) - - if self.session != None: - await self.session.close() - - - ## Class methods ------------------------------ - - - @classmethod - def getServer(cls, accountID: int) -> str: - """Get Realm/server of an account based on account ID""" - if accountID > 1e9: - if accountID > 2e9: - return 'asia' - return 'na' - else: - if accountID < 5e8: - return 'ru' - return 'eu' - return None - - @classmethod - def getMap(cls, mapStr: str) -> str: - """Return map name from short map string in replays""" - try: - return cls.maps[mapStr] - except: - error('Map ' + mapStr + ' not found') - return None - - @classmethod - def getMapUserStrs(cls) -> str: - return cls.maps.keys() - - @classmethod - def chkJSONcontent(cls, json_obj, check = None) -> bool: - try: - if (check == 'player') and (not cls.chkJSONplayerStats(json_obj)): - debug('Checking player JSON failed.') - return False - elif (check == 'tank') and (not cls.chkJSONtank(json_obj)): - debug('Checking tank JSON failed.') - return False - # elif (check == 'clan') and (not chkJSONclan(json_obj)): - # debug('Checking clan JSON failed.') - # return False - elif (check == 'tankList') and (not cls.chkJSONtankList(json_obj)): - debug('Checking tank list JSON failed.') - return False - # elif (check == 'accountlist') and (not cls.chkJSONaccountList(json_obj)): - # cls.debug('Checking account list JSON failed.') - # return False - except (TypeError, ValueError) as err: - debug(str(err)) - return False - return True - - @classmethod - def chkJSONstatus(cls, json_resp: dict) -> bool: - try: - if json_resp['status'] == 'ok': - return True - elif json_resp['status'] == 'error': - error(str(json_resp['error']['code']) + ' : ' + json_resp['error']['message'] ) - return False - else: - debug('Unknown status-code: ' + json_resp['status']) - return False - except KeyError as err: - error('No field found in JSON data: ' + str(err)) - except: - error("JSON format error") - return False - - @classmethod - def chkJSONgetAccountID(cls, json_resp: dict) -> bool: - try: - if cls.chkJSONstatus(json_resp): - if (json_resp['meta']['count'] > 0): - return True - except KeyError as err: - error('Key :' + str(err) + ' not found') - except Exception as err: - error(str(err)) - return False - - # @classmethod - # def chkJSONplayer(cls, json_resp: dict) -> bool: - # """"Check String for being a valid Player JSON file""" - # try: - # if cls.chkJSONstatus(json_resp): - # if int(json_resp[0]['account_id']) > 0: - # return True - # except KeyError as err: - # error('Key :' + str(err) + ' not found') - # except: - # debug("JSON check failed") - # return False - - @classmethod - def chkJSONtank(cls, json_resp: dict) -> bool: - """"Check String for being a valid Tank JSON file""" - try: - if cls.chkJSONstatus(json_resp): - if int(json_resp[0]['tank_id']) > 0: - return True - except KeyError as err: - error('Key :' + str(err) + ' not found') - except: - debug("JSON check failed") - return False - - - @classmethod - def chkJSONplayerStats(cls, json_resp: dict) -> bool: - """"Check String for being a valid Tank JSON file""" - try: - if cls.chkJSONstatus(json_resp): - for acc in json_resp['data']: - if json_resp['data'][acc] != None: - return True - except KeyError as err: - error('Key :' + str(err) + ' not found') - except: - debug("JSON check failed") - return False - - @classmethod - def chkJSONtankList(cls, json_resp: dict) -> bool: - """"Check String for being a valid Tank list JSON file""" - try: - if cls.chkJSONstatus(json_resp): - accountID = next(iter(json_resp['data'])) - if int(json_resp['data'][accountID][0]['tank_id']) > 0: - debug('JSON tank list check OK') - return True - except Exception as err: - debug('JSON check FAILED: ' + str(type(err)) + ' : '+ str(err) + ' : ' + str(json_resp) ) - return False - - ## Methods -------------------------------------------------- - async def loadTanks(self, tankopedia_fn: str): - """Load tanks from tankopedia JSON""" - if tankopedia_fn != None: - try: - async with aiofiles.open(tankopedia_fn, 'rt', encoding='utf8') as f: - self.tanks = json.loads(await f.read()) - return True - except Exception as err: - error('Could not read tankopedia: ' + tankopedia_fn + '\n' + str(err)) - return False - - def getUrlClanInfo(self, server: str, clanID: int) -> str: - return self.URL_WG_server[server] + self.URL_WG_clanInfo + self.WG_appID + '&clan_id=' + str(clanID) - - def getUrlPlayerTankList(self, accountID: int) -> str: - server = self.getServer(accountID) - return self.URL_WG_server[server] + self.URL_WG_playerTankList + self.WG_appID + '&account_id=' + str(accountID) - - def getUrlPlayerTankStats(self, accountID, tankID, fields) -> str: - server = self.getServer(accountID) - url = self.URL_WG_server[server] + self.URL_WG_playerTankStats + self.WG_appID + '&account_id=' + str(accountID) + '&tank_id=' + str(tankID) - if (fields != None) and (len(fields) > 0): - field_str = '%2C' + '%2C'.join(fields) - else: - field_str = '' - return url + '&fields=tank_id' + field_str - - def getUrlPlayerStats(self, accountID, fields) -> str: - server = self.getServer(accountID) - return self.URL_WG_server[server] + self.URL_WG_playerStats + self.WG_appID + '&account_id=' + str(accountID) + '&fields=' + '%2C'.join(fields) - - def getUrlAccountID(self, nickname, server) -> int: - try: - return self.URL_WG_server[server] + self.URL_WG_accountID + self.WG_appID + '&search=' + urllib.parse.quote(nickname) - except Exception as err: - print('ERROR: getUrlAccountID(): ' + str(err)) - return None - - async def getAccountID(self, nickname: str) -> int: - """Get WG account_id for a nickname""" - try: - nick, server = nickname.split('@') - debug(nick + ' @ '+ server) - server = server.lower() - if nick == None or server == None: - raise ValueError('Invalid nickname given: ' + nickname) - url = self.getUrlAccountID(nick, server) - json_data = await getUrlJSON(self.session, url) - for res in json_data['data']: - if res['nickname'].lower() == nick.lower(): - return res['account_id'] - raise ValueError('No WG account_id found: ' + nickname) - - except Exception as err: - error(str(err)) - return None - - async def getPlayerTankStats(self, accountID: int, tankID : int, fields: list) -> dict: - """Get player's stats (WR, # of battles) in a tank""" - # debug('Started') - if self.session == None: - error('Session must be initialized first') - sys.exit(1) - try: - debug('AccountID: ' + str(accountID) + ' TankID: ' + str(tankID)) - url = self.getUrlPlayerTankStats(accountID, tankID, fields) - json_data = await getUrlJSON(self.session, url, self.chkJSONtankList) - if json_data != None: - debug('JSON Response received: ' + str(json_data)) - return json_data['data'][str(accountID)][0] - except Exception as err: - error(err) - return None - - async def getPlayerStats(self, accountID: int, fields: list, battle_time = None) -> dict: - """Get player's global stats """ - # debug('Started') - fields_missing = fields - cached_stats = None - if self.cache != None: - cached_stats, fields_missing = await self.getCachedPlayerStats(accountID, fields) - - if fields_missing != None: - if self.session == None: - error('Session must be initialized first') - sys.exit(1) - try: - debug('AccountID: ' + str(accountID) + ' Fields: ' + ', '.join(fields_missing)) - url = self.getUrlPlayerStats(accountID, fields_missing) - json_data = await getUrlJSON(self.session, url, self.chkJSONplayerStats) - if json_data != None: - debug('JSON Response received: ' + str(json_data)) - stats = json_data['data'][str(accountID)] - if self.statsQ != None: - self.statsQ.put([accountID, 'player', stats]) - return self.mergePlayerStats(stats, cached_stats) - except Exception as err: - error('Unexpected Exception: ' + str(type(err)) + ' : ' + str(err)) - else: - return cached_stats - return None - - async def getCachedPlayerStats(self, accountID, fields): - if self.cache == None: - error('No stats cache available') - return None, fields - try: - res_tmp = {} - missing = [] - for field in fields: - curr = self.cache.execute(self.sql_select_player_stats.format(accountID, field, NOW() )) - res_tmp[field] = curr.fetchone() - if res_tmp[field] == None: - res_tmp.pop(field, None) - missing.append(field) - else: - res_tmp[field] = res_tmp[field][0] - - res = {} - for field in res_tmp.keys(): - res = bldDictHierarcy(res, field, res_tmp[field]) - - if len(missing) == 0: - missing = None - if len(res.keys()) == 0: - res = None - return res, missing - except KeyError as err: - error('Key not found: ' + str(err)) - except Exception as err: - error(str(err)) - return None, fields - - def mergePlayerStats(self, stats1: dict, stats2: dict) -> dict: - try: - if stats2 == None: return stats1 - for keyA in stats2: - if keyA not in stats1: - stats1[keyA] = stats2[keyA] - else: - for keyB in stats2[keyA]: - stats1[keyA][keyB] = stats2[keyA][keyB] - return stats1 - except KeyError as err: - error('Key not found: ' + str(err)) - return None - - - def getTankData(self, tank_id: int, field: str): - if self.tanks == None: - return None - try: - return self.tanks['data'][str(tank_id)][field] - except KeyError as err: - error('Key not found: ' + str(err)) - return None - - async def statsSaver(self): - while True: - stats = await self.statsQ.get() - try: - return stats ## FIX - - - - except Exception as err: - error(str(err)) - return None - -class WoTinspector: - URL_WI = 'https://wotinspector.com' - URL_REPLAYS = URL_WI + '/en/replays/sort/ut/page/' - URL_REPLAY_DL = '/en/replays/download/' - URL_REPLAY_UL = '/api/replay/upload?' - URL_REPLAY_INFO = '/api/replay/upload?details=full&key=' - - REPLAY_N = 1 - - def __init__(self): - self.session = aiohttp.ClientSession() - - async def close(self): - if self.session != None: - await self.session.close() - - async def getReplayJSON(self, replay_id: str): - json_resp = await getUrlJSON(self.session, self.URL_WI + self.URL_REPLAY_INFO + replay_id, None) - try: - if self.chkJSONreplay(json_resp): - return json_resp - else: - return None - except Exception as err: - error('Unexpected Exception: ' + str(type(err)) + ' : ' + str(err) ) - return None - - async def postReplay(self, data, filename = 'Replay', account_id = 0, title = 'Replay', priv = False, N = None): - try: - N = N if N != None else self.REPLAY_N - msg_str = 'Replay[' + str(N) + ']: ' - self.REPLAY_N += 1 - - hash = hashlib.md5() - hash.update(data) - replay_id = hash.hexdigest() - - ## Testing if the replay has already been posted - json_resp = await self.getReplayJSON(replay_id) - if json_resp != None: - debug(msg_str + 'Already uploaded: ' + title) - return json_resp - - params = { - 'title' : title, - 'private' : (1 if priv else 0), - 'uploaded_by' : account_id, - 'details' : 'full', - 'key' : replay_id - } - - - url = self.URL_WI + self.URL_REPLAY_UL + urllib.parse.urlencode(params, quote_via=urllib.parse.quote) - #debug('URL: ' + url) - headers ={'Content-type': 'application/x-www-form-urlencoded'} - payload = { 'file' : (filename, base64.b64encode(data)) } - except Exception as err: - error(msg_str + 'Unexpected Exception: ' + str(type(err)) + ' : ' + str(err) ) - return None - - for retry in range(MAX_RETRIES): - debug(msg_str + 'Posting: ' + title + ' Try #: ' + str(retry + 1) + '/' + str(MAX_RETRIES) ) - try: - async with self.session.post(url, headers=headers, data=payload) as resp: - debug(msg_str + 'HTTP response: '+ str(resp.status)) - if resp.status == 200: - debug('HTTP POST 200 = Success. Reading response data') - json_resp = await resp.json() - if json_resp.get('status', None) == None: - error(msg_str +' : ' + title + ' : Received invalid JSON') - elif (json_resp['status'] == 'error'): - error(msg_str + json_resp['error']['message'] + ' : ' + title) - else: - debug('Response data read') - verbose(msg_str + title + ' posted') - return json_resp - else: - debug(msg_str + 'Got HTTP/' + str(resp.status) + ' Retrying... ' + str(retry)) - await asyncio.sleep(SLEEP) - except Exception as err: - error(str(err)) - await asyncio.sleep(SLEEP) - return None - - @classmethod - def getUrlReplayListing(cls, page : int): - return cls.URL_REPLAYS + str(page) + '?vt=#filters' - - @classmethod - def getReplayLinks(cls, doc: str): - """Get replay download links from WoTinspector.com replay listing page""" - try: - soup = BeautifulSoup(doc, 'lxml') - links = soup.find_all('a') - replay_links = set() - for tag in links: - link = tag.get('href',None) - if (link is not None) and (link.find(cls.URL_REPLAY_DL) == 0 ): - replay_links.add(link) - debug('Adding replay link:' + link) - except Exception as err: - debug(str(err)) - return replay_links - - @classmethod - def getReplayID(cls, url): - return url.rsplit('/', 1)[-1] - - @classmethod - def chkJSONreplay(cls, json_resp): - """"Check String for being a valid JSON file""" - try: - if ('status' in json_resp) and json_resp['status'] == 'ok' and ('data' in json_resp) and json_resp['data'] != None: - debug("JSON check OK") - return True - except KeyError as err: - error('Key :' + str(err) + ' not found') - except: - debug("JSON check failed: " + str(json_resp)) - return False - -class BlitzStars: - - URL_BlitzStars = 'https://www.blitzstars.com' - URL_playerStats = URL_BlitzStars + '/api/playerstats' - URL_playerTankStats = URL_BlitzStars + '/api/tanks' - URL_tankAverages = URL_BlitzStars + '/tankaverages.json' - URL_activeplayers = URL_BlitzStars + '/api/playerstats/activeinlast30days' - URL_playersTankStats= URL_BlitzStars + '/api/top/forjylpah?' - - def __init__(self): - self.session = aiohttp.ClientSession() - - async def close(self): - if self.session != None: - await self.session.close() - - @classmethod - def getUrlTankAvgs(cls): - return cls.URL_tankAverages - - @classmethod - def getUrlPlayerStats(cls, accountID: int): - return cls.URL_playerStats + '/' + str(accountID) - - @classmethod - def getUrlPlayerTankStats(cls, accountID: int, tankID: int): - return cls.URL_playerTankStats + '/' + str(accountID) + '/' + str(tankID) - - @classmethod - def getUrlPlayersTankStats(cls, accountID: int): - return cls.URL_playersTankStats + '&accountId=' + str(accountID) - - @classmethod - def getUrlActivePlayers(cls): - return cls.URL_activeplayers - - @classmethod - def chkJSONtankStats(cls, json_resp): - """"Check String for being a valid JSON file""" - try: - # print(str(json_resp['data'])) - if ('data' in json_resp): - debug("JSON check OK") - return True - except KeyError as err: - error('Key :' + str(err) + ' not found') - except: - debug("JSON check failed") - return False