From 6adcc49a6d05d56891fb37424a86b29c631eaad8 Mon Sep 17 00:00:00 2001 From: Jylpah Date: Wed, 3 Oct 2018 23:13:33 +0800 Subject: [PATCH] Update --- WG_utils.py | 294 +++++++++++++++++++++++++++++++--------------------- 1 file changed, 173 insertions(+), 121 deletions(-) diff --git a/WG_utils.py b/WG_utils.py index b4ccc61..4484a45 100644 --- a/WG_utils.py +++ b/WG_utils.py @@ -1,17 +1,104 @@ #!/usr/bin/python3.7 -import sys, json, asyncio, aiofiles, aiohttp, base64, urllib, inspect +import sys, os, json, asyncio, aiofiles, aiohttp, base64, urllib, inspect MAX_RETRIES= 3 SLEEP = 2 +DEBUG = False +VERBOSE = 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 + if debug != None: + DEBUG = debug + if DEBUG: VERBOSE = True + +def setVerbose(verbose: bool): + global VERBOSE + if verbose != None: + VERBOSE = verbose + +def verbose(msg = ""): + """Print a message""" + if VERBOSE: + curframe = inspect.currentframe() + calframe = inspect.getouterframes(curframe, 2) + caller = calframe[1][3] + print(caller + '(): ' + msg) + return None + +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 + +async def saveJSON(filename: str, json_data: dict, sort_keys = False) -> bool: + """Save JSON data into file""" + try: + dirname = os.path.dirname(filename) + if 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 getUrlJSON(session: aiohttp.ClientSession, url: str, chkJSONfunc = None) -> dict: + """Retrieve (GET) an URL and return JSON object""" + try: + debug(url) + ## To avoid excessive use of servers + for retry in range(1,MAX_RETRIES+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") + return json_resp + else: + debug('Received JSON error') + if retry == MAX_RETRIES: + raise aiohttp.ClientError('Request failed: ' + str(resp.status) ) + verbose('Retrying URL [' + str(retry) + ']: ' + url ) + await asyncio.sleep(SLEEP) + + except Exception as err: + error("Could not retrieve URL: " + url) + 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_accountID = 'account/list/?fields=account_id%2Cnickname&application_id=' - - DEBUG = False - + maps = { "Random": "Random map", "amigosville": "Falls Creek", @@ -44,7 +131,11 @@ class WG: "test": "World of Ducks", "tutorial": "Proving Grounds" } - + + nations = [ 'ussr', 'germany', 'usa', 'china', 'france', 'uk', 'japan', 'other'] + + tank_type = [ 'lightTank', 'mediumTank', 'heavyTank', 'AT-SPG' ] + URL_WG_server = { 'eu' : 'https://api.wotblitz.eu/wotb/', 'ru' : 'https://api.wotblitz.ru/wotb/', @@ -59,7 +150,7 @@ class WG: 'asia': range(int(2e9),int(4e9)) } - def __init__(self, WG_appID: str, tankopedia_fn: str, maps_fn = None): + def __init__(self, WG_appID = None, tankopedia_fn = None, maps_fn = None): self.WG_appID = WG_appID self.tanks = None @@ -68,14 +159,14 @@ class WG: with open(tankopedia_fn, 'rt', encoding='utf8') as f: self.tanks = json.loads(f.read()) except Exception as err: - self.error('Could not read tankopedia: ' + tankopedia_fn + '\n' + str(err)) + error('Could not read tankopedia: ' + tankopedia_fn + '\n' + str(err)) if maps_fn != None: try: with open(maps_fn, 'rt', encoding='utf8') as f: self.maps = json.loads(f.read()) except Exception as err: - self.error('Could not read maps file: ' + maps_fn + '\n' + str(err)) + error('Could not read maps file: ' + maps_fn + '\n' + str(err)) self.session = aiohttp.ClientSession() # self.session = None @@ -85,10 +176,20 @@ class WG: @classmethod def getServer(cls, accountID: int) -> str: """Get Realm/server of an account based on account ID""" - for server in cls.accountID_server.keys(): - if accountID in cls.accountID_server[server]: - return server - print('ERROR: AccountID not in range: ' + str(accountID)) + # for server in cls.accountID_server.keys(): + # if accountID in cls.accountID_server[server]: + # return server + # print('ERROR: AccountID not in range: ' + str(accountID)) + + # faster, but can fail for negatives + if accountID > 1e9: + if accountID > 2e9: + return 'asia' + return 'na' + else: + if accountID < 5e8: + return 'ru' + return 'eu' return None @classmethod @@ -97,97 +198,59 @@ class WG: try: return cls.maps[mapStr] except: - cls.error('Map ' + mapStr + ' not found') + error('Map ' + mapStr + ' not found') return None @classmethod def getMapUserStrs(cls) -> str: return cls.maps.keys() - # @classmethod - # def getUrlTankAvgs(cls): - # return cls.URL_tankAverages - - @classmethod - def debug(cls, msg = ""): - """print a conditional debug message""" - if cls.DEBUG: - curframe = inspect.currentframe() - calframe = inspect.getouterframes(curframe, 2) - caller = calframe[1][3] - print('DEBUG: ' + caller + '(): ' + msg) - return None - - @classmethod - def error(cls, msg = ""): - """Print an error message""" - curframe = inspect.currentframe() - calframe = inspect.getouterframes(curframe, 2) - caller = calframe[1][3] - print('ERROR: ' + caller + '(): ' + msg) - return None - - # @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, tankID: int, account_list: list): - # return cls.URL_playersTankStats + 'tankId=' + str(tankID) + '&accountIds=' + ','.join(str(x) for x in account_list) - - # @classmethod - # def getUrlActivePlayers(cls): - # return cls.URL_activeplayers - @classmethod def chkJSONresponse(cls, json_resp: dict) -> bool: try: if ('status' in json_resp) and (json_resp['status'] == 'error'): - cls.debug(str(json_resp['error']['code']) + ' : ' + json_resp['error']['message'] ) + debug(str(json_resp['error']['code']) + ' : ' + json_resp['error']['message'] ) return False return True except: - cls.error("JSON format error") + error("JSON format error") + return False + + @classmethod + def chkJSONgetAccountID(cls, json_resp: dict) -> bool: + try: + if (json_resp['status'] == 'ok'): + return True + else: + error_msg = str(json_resp['error']['code']) + ' : ' + json_resp['error']['message'] + raise ValueError('Received JSON error: ' + error_msg) + except Exception as err: + error(str(err)) return False @classmethod def chkJSONcontent(cls, json_obj, check = None) -> bool: try: if (check == 'player') and (not cls.chkJSONplayer(json_obj)): - cls.debug('Checking player JSON failed.') + debug('Checking player JSON failed.') return False elif (check == 'tank') and (not cls.chkJSONtank(json_obj)): - cls.debug('Checking tank JSON failed.') + 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)): - cls.debug('Checking tank list JSON failed.') + 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: - cls.debug(str(err)) + debug(str(err)) return False return True -# def chkJSONclan(clanInfo: dict) -> bool: -# """"Check String for being a valid Clan JSON file""" -# try: -# clanID = clanInfo['data'].keys()[0] -# member_count = clanInfo['data'][clanID]['members_count'] -# if int(member_count) > 0: -# return True -# except: -# print('ERROR: chkJSONclan:') -# return False - @classmethod def chkJSONplayer(cls, playerInfo: dict) -> bool: """"Check String for being a valid Player JSON file""" @@ -196,7 +259,7 @@ class WG: if int(accountID) > 0: return True except: - cls.debug("JSON check failed") + debug("JSON check failed") return False @classmethod @@ -206,7 +269,7 @@ class WG: if int(tankInfo[0]['tank_id']) > 0: return True except: - cls.debug("JSON check failed") + debug("JSON check failed") return False @classmethod @@ -216,18 +279,9 @@ class WG: if int(tankInfo['account_ids'][0]) > 0: return True except: - cls.debug("JSON check failed") + debug("JSON check failed") return False - # def chkJSONaccountList(cls, account_list: list) -> bool: - # """"Check String for being a valid accountID list JSON file""" - # try: - # if int(account_list[0]) > 0: - # return True - # except: - # debug("JSON check failed") - # return False - @classmethod def chkJSONtankList(cls, tankList: dict) -> bool: """"Check String for being a valid Tank list JSON file""" @@ -235,10 +289,10 @@ class WG: accountID = next(iter(tankList['data'])) tankID = tankList['data'][accountID][0]['tank_id'] if int(tankID) > 0: - cls.debug('JSON tank list check OK') + debug('JSON tank list check OK') return True except Exception as err: - cls.error('JSON check FAILED: ' + str(err)) + error('JSON check FAILED: ' + str(err)) return False ## Methods -------------------------------------------------- @@ -250,36 +304,9 @@ class WG: self.tanks = json.loads(await f.read()) return True except Exception as err: - self.error('Could not read tankopedia: ' + tankopedia_fn + '\n' + str(err)) + error('Could not read tankopedia: ' + tankopedia_fn + '\n' + str(err)) return False - - async def getUrlJSON(self, url: str) -> dict: - """Retrieve (GET) an URL and return JSON object""" - try: - self.debug(url) - ## To avoid excessive use of servers - for retry in range(1,MAX_RETRIES+1): - async with self.session.get(url) as resp: - if resp.status == 200: - self.debug('HTTP request OK') - json_resp = await resp.json() - if self.chkJSONresponse(json_resp): - self.debug("Received valid JSON") - return json_resp - else: - self.debug('Received JSON error') - if retry == MAX_RETRIES: - self.error("Could not retrieve URL: " + url) - raise aiohttp.ClientError('Request failed: ' + str(resp.status_code) ) - print('Retrying URL [' + str(retry) + ']: ' + url ) - await asyncio.sleep(SLEEP) - - except (TypeError, ValueError, aiohttp.ClientError) as err: - self.debug(url) - self.debug(str(err)) - #return None - return None - + def getUrlClanInfo(self, server: str, clanID: int): return self.URL_WG_server[server] + self.URL_WG_clanInfo + self.WG_appID + '&clan_id=' + str(clanID) @@ -293,26 +320,51 @@ class WG: 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('@') - self.debug(nick + ' @ '+ server) + debug(nick + ' @ '+ server) server = server.lower() if nick == None or server == None: - self.error('Invalid nickname given: ' + nickname) - return None + raise ValueError('Invalid nickname given: ' + nickname) url = self.getUrlAccountID(nick, server) - json_data = await self.getUrlJSON(url) - # self.debug(str(json_data)) + json_data = await getUrlJSON(self.session, url) for res in json_data['data']: if res['nickname'].lower() == nick.lower(): return res['account_id'] - self.error('No account_id found: ' + nickname) - return None + raise ValueError('No WG account_id found: ' + nickname) except Exception as err: - print('ERROR: Account ID not valid or server not given: ' + nick + ' ' + str(err)) + error(str(err)) return None + +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?' + + @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, tankID: int, account_list: list): + return cls.URL_playersTankStats + 'tankId=' + str(tankID) + '&accountIds=' + ','.join(str(x) for x in account_list) + + @classmethod + def getUrlActivePlayers(cls): + return cls.URL_activeplayers