Update

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",
@@ -45,6 +132,10 @@ class WG:
         "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():
+        # for server in cls.accountID_server.keys():
-            if accountID in cls.accountID_server[server]:
+        #     if accountID in cls.accountID_server[server]:
-                return server
+        #         return server
-        print('ERROR: AccountID not in range: ' + str(accountID))
+        # 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)
 
@@ -294,25 +321,50 @@ class WG:
             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)
+                raise ValueError('Invalid nickname given: ' + nickname)
-                return None
             url = self.getUrlAccountID(nick, server)
-            json_data = await self.getUrlJSON(url)
+            json_data = await getUrlJSON(self.session, url)
-            # self.debug(str(json_data))
             for res in json_data['data']:
                 if res['nickname'].lower() == nick.lower(): 
                     return res['account_id']
-            self.error('No account_id found: ' + nickname)
+            raise ValueError('No WG account_id found: ' + nickname)
-            return None
             
         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