add pyutils

.gitmodules

@@ -0,0 +1,3 @@
+[submodule "pyutils"]
+	path = pyutils
+	url = https://github.com/Jylpah/pyutils.git

blitzutils.py

@@ -36,163 +36,132 @@ _progress_obj = None
 UMASK = os.umask(0)
 os.umask(UMASK)
 
-## -----------------------------------------------------------
-#### Class ThrottledClientSession(aiohttp.ClientSession)
-#
-#  Rate-limited async http client session
-#
-#  Inherits aiohttp.ClientSession 
-## -----------------------------------------------------------
+# -----------------------------------------------------------
+# Class ThrottledClientSession(aiohttp.ClientSession)
+# -----------------------------------------------------------
 
-from typing import Optional, Union
-import aiohttp
-import asyncio
-import time
-import logging
-import re
 
 class ThrottledClientSession(aiohttp.ClientSession):
     """Rate-throttled client session class inherited from aiohttp.ClientSession)"""
+    # MIN_SLEEP = 0.1
 
-    def __init__(self, rate_limit: float = 0, filters: list[str] = list() , 
-                limit_filtered: bool = False, re_filter: bool = False, *args,**kwargs) -> None: 
-        assert isinstance(rate_limit, (int, float)),   "rate_limit has to be float"
-        assert isinstance(filters, list),       "filters has to be list"
-        assert isinstance(limit_filtered, bool),"limit_filtered has to be bool"
-        assert isinstance(re_filter, bool),     "re_filter has to be bool"
+    def __init__(self, rate_limit: float = None, *args, **kwargs) -> None:
+        super().__init__(*args, **kwargs)
+        self.rate_limit = None
+        self._fillerTask = None
+        self._queue = None
+        self._start_time = None
+        self._count = 0
+        self._start_time = time.time()
 
-        super().__init__(*args,**kwargs)
+        if rate_limit != None:
+            if rate_limit <= 0:
+                raise ValueError('rate_limit must be positive')
+            self.rate_limit = rate_limit
+            self._queue = asyncio.Queue(min(2, int(rate_limit)+1))
+            self._fillerTask = asyncio.create_task(self._filler())
 
-        self.rate_limit     : float
-        self._fillerTask    : Optional[asyncio.Task]    = None
-        self._queue         : Optional[asyncio.Queue]   = None
-        self._start_time    : float = time.time()
-        self._count         : int = 0
-        self._limit_filtered: bool = limit_filtered
-        self._re_filter     : bool = re_filter
-        self._filters       : list[Union[str, re.Pattern]] = list()
-
-        if re_filter:
-            for filter in filters:
-                self._filters.append(re.compile(filter))
+    def _get_sleep(self) -> list:
+        if self.rate_limit == None:
+            return None
         else:
-            for filter in filters:
-                self._filters.append(filter)
-        self.set_rate_limit(rate_limit)
-
-
-    def _get_sleep(self) -> float:        
-        if self.rate_limit > 0:
             return 1/self.rate_limit
-        return 0
-
 
     def get_rate(self) -> float:
         """Return rate of requests"""
         return self._count / (time.time() - self._start_time)
 
-
-    def get_stats(self) -> dict[str, float]:
+    def get_stats(self):
         """Get session statistics"""
-        res = {'rate' : self.get_rate(), 'rate_limit': self.rate_limit, 'count' : self._count }
+        res = {'rate': self.get_rate(), 'rate_limit': self.rate_limit,
+                                     'count': self._count}
         return res
 
-
-    def get_stats_str(self) -> str:
+    def get_stats_str(self):
         """Print session statistics"""
-        return f"rate limit: {str(self.rate_limit if self.rate_limit != None else '-')} \
-                rate:   {0:.1f}.format(self.get_rate()) requests: {str(self._count)}"
+        return 'rate limit: ' + str(self.rate_limit if self.rate_limit != None else '-') + \
+                ' rate: ' + \
+                    "{0:.1f}".format(self.get_rate()) + \
+                                     ' requests: ' + str(self._count)
 
-
-    def reset_counters(self) -> dict[str, float]:
+    def reset_counters(self):
         """Reset rate counters and return current results"""
         res = self.get_stats()
         self._start_time = time.time()
         self._count = 0
         return res
 
-
-    def set_rate_limit(self, rate_limit: float = 0) -> float:
-        assert rate_limit is not None, "rate_limit must not be None" 
-        assert isinstance(rate_limit, (int,float)) and rate_limit >= 0, "rate_limit has to be type of 'float' >= 0"
-        
-        self.rate_limit = rate_limit
-        if rate_limit > 0:
-            self._queue     = asyncio.Queue(int(rate_limit)+1) 
-            if self._fillerTask is not None: 
-                self._fillerTask.cancel()  
-            self._fillerTask = asyncio.create_task(self._filler())
-        return self.rate_limit
-        
+    def set_rate_limit(self, rate_limit: float = 1):
+        if rate_limit >= 0:
+            self.rate_limit = rate_limit
+            return self.rate_limit
+        return None
 
     async def close(self) -> None:
         """Close rate-limiter's "bucket filler" task"""
         # DEBUG
-        logging.debug(self.get_stats_str())
+        debug(self.get_stats_str())
         try:
-            if self._fillerTask is not None:
+            if self._fillerTask != None:
                 self._fillerTask.cancel()
-                await asyncio.wait_for(self._fillerTask, timeout=0.5)
-        except asyncio.TimeoutError as err:
-            logging.error(str(err))
+            await asyncio.wait_for(self._fillerTask, timeout=0.5)
+        except (asyncio.CancelledError, asyncio.TimeoutError):
+            pass
         await super().close()
 
-    
-    async def _filler(self) -> None:
+    async def _filler_simple(self):
         """Filler task to fill the leaky bucket algo"""
         try:
-            if self._queue is None:
-                return None            
-            logging.debug('SLEEP: ' + str(self._get_sleep()))
+            if self._queue == None:
+                return
+            sleep = self._get_sleep()
+            while True:
+                self._queue.put(None)
+                await asyncio.sleep(sleep)
+        except asyncio.CancelledError:
+            debug('Cancelled')
+        except Exception as err:
+            error(exception=err)
+
+    async def _filler(self):
+        """Filler task to fill the leaky bucket algo"""
+        try:
+            if self._queue == None:
+                return
+            sleep = self._get_sleep()
+            # debug('SLEEP: ' + str(sleep))
             updated_at = time.monotonic()
-            extra_increment : float = 0
+            fraction = 0
+            extra_increment = 0
             for i in range(0, self._queue.maxsize):
-                await self._queue.put(i)
+                self._queue.put_nowait(i)
             while True:
                 if not self._queue.full():
                     now = time.monotonic()
                     increment = self.rate_limit * (now - updated_at)
-                    items_2_add = int(min(self._queue.maxsize - self._queue.qsize(), int(increment + extra_increment)))
-                    extra_increment = (increment + extra_increment) % 1
-                    for i in range(0,items_2_add):
+                    fraction += increment % 1
+                    extra_increment = fraction // 1
+                    items_2_add = int(
+                        min(self._queue.maxsize - self._queue.qsize(), int(increment) + extra_increment))
+                    fraction = fraction % 1
+                    for i in range(0, items_2_add):
                         self._queue.put_nowait(i)
                     updated_at = now
-                await asyncio.sleep(self._get_sleep())
+                await asyncio.sleep(sleep)
         except asyncio.CancelledError:
-            logging.debug('Cancelled')
+            debug('Cancelled')
         except Exception as err:
-            logging.error(str(err))
-        return None
+            error(exception=err)
 
-
-    async def _request(self, *args,**kwargs) -> aiohttp.ClientResponse:
+    async def _request(self, *args, **kwargs):
         """Throttled _request()"""
-        if self._queue is not None and self.is_limited(*args):  
+        if self._queue != None:
             await self._queue.get()
             self._queue.task_done()
         self._count += 1
-        return await super()._request(*args,**kwargs)
+        return await super()._request(*args, **kwargs)
 
 
-    def is_limited(self, *args: str) -> bool:
-        """Check wether the rate limit should be applied"""
-        try:
-            url: str = args[1]
-            for filter in self._filters:
-                if isinstance(filter, re.Pattern) and filter.match(url) is not None:
-                    return self._limit_filtered
-                elif isinstance(filter, str) and url.startswith(filter):
-                    return self._limit_filtered
-                    
-            return not self._limit_filtered
-        except Exception as err:
-            logging.error(str(err))
-        return True    
-
-# -----------------------------------------------------------
-# Class AsyncLogger()
-# -----------------------------------------------------------
-
 class AsyncLogger():
     """Async file logger"""
 
@@ -527,17 +496,18 @@ def get_date_str(timestamp:int = NOW(), date_format = '%Y%m%d_%H%M%S'):
     return None
 
 
-def rebase_file_args(current_dir : str, files: list[str]) -> list[str]:
-    """Rebase file command line params after moving working dir to the script's dir""" 
-    assert isinstance(files, list), "files has to be a list"
-
-    if (files[0] == '-') or (files[0] == 'db:'):
-        return [ files[0] ] 
-    else:
-        return [ os.path.join(current_dir, fn) for fn in files ]
+def rebase_file_args(current_dir, files):
+    """REbase file command line params after moving working dir to the script's dir""" 
+    if isinstance(files, list):    
+        if (files[0] == '-') or (files[0] == 'db:'):
+            return files
+        else:
+            return [ os.path.join(current_dir, fn) for fn in files ]
+    elif isinstance(files, str):
+        return os.path.join(current_dir, files)
 
 
-async def read_int_list(filename: str) -> list[int]:
+async def read_int_list(filename: str) -> list():
     """Read file to a list and return list of integers in the input file"""
     
     input_list = []
@@ -1626,8 +1596,6 @@ class WG:
 
     async def store_tank_stats(self, key: list , stats_data: list, update_time: int):
         """Save tank stats into cache"""
-        assert self.cache is not None, "cache must be initiated"
-        
         try:
             account_id  = key[0]
             tank_ids    = set(key[1])
@@ -1649,8 +1617,6 @@ class WG:
 
     async def store_player_stats(self, key: list , stats_data: list, update_time: int):
         """Save player stats into cache"""
-        assert self.cache is not None, "cache must be initiated"
-        
         try:
             account_id  = key[0]
             if stats_data != None:
@@ -1667,8 +1633,6 @@ class WG:
 
     async def store_player_achievements(self, key: list , stats_data: list, update_time: int):
         """Save player stats into cache"""
-        assert self.cache is not None, "cache must be initiated"
-        
         try:
             account_id  = key[0]
             if stats_data != None:
@@ -1684,8 +1648,6 @@ class WG:
 
 
     async def get_cached_tank_stats(self, account_id: int, tank_ids: list, fields: list ):
-        assert self.cache is not None, "cache must be initiated"
-
         try:
             # test for cacheDB existence
             debug('Trying cached stats first')
@@ -1799,27 +1761,17 @@ class WoTinspector:
     URL_REPLAY_VIEW = URL_WI +'/en/view/'
     URL_REPLAY_UL   = 'https://api.wotinspector.com/replay/upload?'
     URL_REPLAY_INFO = 'https://api.wotinspector.com/replay/upload?details=full&key='
+    
     URL_TANK_DB     = "https://wotinspector.com/static/armorinspector/tank_db_blitz.js"
 
     REPLAY_N = 1
-    DEFAULT_RATE_LIMIT : float = 10/60  # 10 requests / min
-    DEFAULT_TOKEN : str = 'a5428af49bf7495986577d59b0e5bcff'
 
-    def __init__(self, rate_limit: float = DEFAULT_RATE_LIMIT, auth_token: Optional[str] = None):
-
-        headers : Optional[dict[str, str]] = None
-
-        if auth_token is None:
-            auth_token = self.DEFAULT_TOKEN        
-        headers = dict()
-        headers['Authorization'] = 'Token ' + auth_token
-        self.session = ThrottledClientSession(rate_limit=rate_limit, 
-                                                filters=[self.URL_REPLAY_LIST, self.URL_REPLAY_INFO], 
-                                                re_filter=False, limit_filtered=True, headers = headers)
+    def __init__(self, rate_limit: int = 30):
+        self.session = ThrottledClientSession(rate_limit=rate_limit)
         
 
-    async def close(self) -> None:
-        if self.session is not None:
+    async def close(self):
+        if self.session != None:
             debug('Closing aiohttp session')
             await self.session.close()
         
@@ -1883,7 +1835,7 @@ class WoTinspector:
             hash.update(data)
             replay_id = hash.hexdigest()
 
-            ##  Testing if the replay has already been posted
+            # Testing if the replay has already been posted
             json_resp = await self.get_replay_JSON(replay_id)
             if json_resp != None:
                 debug('Already uploaded: ' + title, id=N)
@@ -1898,7 +1850,7 @@ class WoTinspector:
             } 
 
             url = self.URL_REPLAY_UL + urllib.parse.urlencode(params, quote_via=urllib.parse.quote)
-            #debug('URL: ' + url)
+            # debug('URL: ' + url)
             headers ={'Content-type':  'application/x-www-form-urlencoded'}
             payload = { 'file' : (filename, base64.b64encode(data)) }
         except Exception as err:
@@ -1923,6 +1875,8 @@ class WoTinspector:
             except Exception as err:
                 debug(exception=err, id=N)
             await asyncio.sleep(SLEEP)
+            _id  = self.read_replay_id(json_resp)
+            json_resp['_id']  =  _id
             
         debug(' Could not post replay: ' + title, id=N)
         return json_resp
@@ -1960,6 +1914,11 @@ class WoTinspector:
         return replay_links
     
 
+    @classmethod
+    def get_replay_id(cls, url):
+        return url.rsplit('/', 1)[-1]
+
+
     @classmethod
     def read_replay_id(cls, json_replay):
         """Read replay_id from replay JSON file""" 
@@ -1970,18 +1929,14 @@ class WoTinspector:
             error(exception=err)
         return None
 
-
     @classmethod
-    def get_replay_id(cls, url):
-        return url.rsplit('/', 1)[-1]
-
-
-    @classmethod
-    def chk_JSON_replay(cls, json_resp) -> bool:
+    def chk_JSON_replay(cls, json_resp):
         """"Check String for being a valid JSON file"""
         try:
+            # FIX The replay can have allies/enemies i.e. be useful for 
+            # team composition analysis, but does not have player contribution
             if ('status' in json_resp) and json_resp['status'] == 'ok' and \
-                (get_JSON_value(json_resp, key='data.summary.exp_base') != None) :
+                (get_JSON_value(json_resp, key='data.summary.winner_team') != None) :
                 debug("JSON check OK")
                 return True 
         except KeyError as err:
@@ -1990,9 +1945,6 @@ class WoTinspector:
             debug("Replay JSON check failed: " + str(json_resp))
         return False      
 
-# -----------------------------------------------------------
-# Class BlitzStars 
-# -----------------------------------------------------------
 
 class BlitzStars:
 

upload_wotb_replays.py

@@ -1,8 +1,8 @@
 #!/usr/bin/env python3
 ## PYTHON VERSION MUST BE 3.7 OR HIGHER
 
-import json, argparse, sys, os, io, asyncio, aiofiles, aioconsole
-import logging, re, configparser, zipfile
+import json, argparse, inspect, sys, os, io, base64, aiohttp, urllib, asyncio, aiofiles, aioconsole
+import logging, re, concurrent.futures, configparser, motor.motor_asyncio, ssl, zipfile
 import blitzutils as bu
 from blitzutils import WG
 from blitzutils import WoTinspector
@@ -21,17 +21,17 @@ SKIPPED_N 	= 0
 ERROR_N 	= 0
 WIurl='https://wotinspector.com/api/replay/upload?'
 WG_appID  = '81381d3f45fa4aa75b78a7198eb216ad'
-wg : WG | None = None
-wi : WoTinspector | None = None
+wg = None
+wi = None
 
-async def main(argv) -> int:
+async def main(argv):
 	global wg, wi
 	current_dir = os.getcwd()
 	# set the directory for the script
 	os.chdir(os.path.dirname(sys.argv[0]))
 	
 	# options defaults
-	OPT_WORKERS_N  	= 2
+	OPT_WORKERS_N  	= 5
 	WG_ID 			= None
 	WG_ACCOUNT 		= None 	# format: nick@server, where server is either 'eu', 'ru', 'na', 'asia' or 'china'. 
 	  					# China is not supported since WG API stats are not available there
@@ -43,6 +43,12 @@ async def main(argv) -> int:
 			bu.debug('Reading config file: ' + FILE_CONFIG)
 			config = configparser.ConfigParser()
 			config.read(FILE_CONFIG)
+			try:
+				if 'UPLOADER' in config.sections():
+					configUploader 	= config['UPLOADER']
+					OPT_WORKERS_N = configUploader.getint('opt_workers', OPT_WORKERS_N)
+			except (KeyError, configparser.NoSectionError) as err:
+				bu.error(exception=err)
 				
 			try:
 				if 'WG' in config.sections():
@@ -76,19 +82,18 @@ async def main(argv) -> int:
 	bu.set_log_level(args.silent, args.verbose, args.debug)
 
 	wg = WG(WG_appID, args.tankopedia, args.mapfile)
-	wi = WoTinspector()
+	wi = WoTinspector(rate_limit=WG_RATE_LIMIT)
 
-	if args.account is not None:
+	if args.account != None:
 		args.accountID = await wg.get_account_id(args.account)
 		bu.debug('WG  account_id: ' + str(args.accountID))
 
-	if args.accountID is None: 
+	if args.accountID == None: 
 		args.accountID = 0
 
 	try:
-		queue : asyncio.Queue = asyncio.Queue()	
+		queue  = asyncio.Queue()	
 		# rebase file arguments due to moving the working directory to the script location
-		bu.verbose_std("replays.wotinspector.com rate limits replays to 6 / minute")
 		args.files = bu.rebase_file_args(current_dir, args.files)
 
 		tasks = []
@@ -118,7 +123,7 @@ async def main(argv) -> int:
 		await wg.session.close()
 		await wi.close()
 
-	return 0
+	return None
 
 
 async def mkReplayQ(queue : asyncio.Queue, files : list, title : str):
@@ -132,13 +137,13 @@ async def mkReplayQ(queue : asyncio.Queue, files : list, title : str):
 			if not line: 
 				break
 			else:
-				if (p_replayfile.match(line) is not None):
+				if (p_replayfile.match(line) != None):
 					await queue.put(await mkQueueItem(line, title))
 	else:
 		for fn in files:
 			if fn.endswith('"'):
 				fn = fn[:-1]  
-			if os.path.isfile(fn) and (p_replayfile.match(fn) is not None):
+			if os.path.isfile(fn) and (p_replayfile.match(fn) != None):
 				await queue.put(await mkQueueItem(fn, title))
 				bu.debug('File added to queue: ' + fn)
 			elif os.path.isdir(fn):
@@ -146,7 +151,7 @@ async def mkReplayQ(queue : asyncio.Queue, files : list, title : str):
 					for entry in dirEntry:
 						try:
 							bu.debug('Found: ' + entry.name)
-							if entry.is_file() and (p_replayfile.match(entry.name) is not None): 
+							if entry.is_file() and (p_replayfile.match(entry.name) != None): 
 								bu.debug(entry.name)
 								await queue.put(await mkQueueItem(entry.path, title))
 								bu.debug('File added to queue: ' + entry.path)
@@ -170,9 +175,6 @@ async def replayWorker(queue: asyncio.Queue, workerID: int, account_id: int, pri
 	"""Async Worker to process the replay queue"""
 	global SKIPPED_N
 	global ERROR_N
-	global wi
-
-	assert wi is not None, "wi cannot be None"
 
 	while True:
 		item = await queue.get()
@@ -208,7 +210,7 @@ async def replayWorker(queue: asyncio.Queue, workerID: int, account_id: int, pri
 				filename = os.path.basename(filename)
 				bu.debug(msg_str + 'File:  ' + filename)
 				json_resp = await wi.post_replay(await fp.read(), filename, account_id, title, priv, N)
-				if json_resp is not None:
+				if json_resp != None:
 					if (await bu.save_JSON(replay_json_fn,json_resp)):
 						if wi.chk_JSON_replay(json_resp):
 							if not bu.debug(msg_str + 'Replay saved OK: ' + filename):
@@ -233,7 +235,7 @@ async def replayWorker(queue: asyncio.Queue, workerID: int, account_id: int, pri
 def getTitle_old(replayfile: str, title: str, i : int) -> str:
 	global wg
 
-	if title is None:
+	if title == None:
 		try:
 			filename = os.path.basename(replayfile)	
 			bu.debug(filename)
@@ -245,8 +247,8 @@ def getTitle_old(replayfile: str, title: str, i : int) -> str:
 			p = re.compile('\\d{8}_\\d{4}_.*?(' + '|'.join(tank_userStrs) + ')_(' + '|'.join(map_usrStrs) + ')(?:-\\d)?\\.wotbreplay$')
 			
 			m = p.match(filename)
-			if (m is not None):
-				if wg.tanks is not None:
+			if (m != None):
+				if wg.tanks != None:
 					tank = m.group(1)
 					if tank in wg.tanks['userStr']:
 						tank = wg.tanks['userStr'][tank]
@@ -271,9 +273,7 @@ def getTitle_old(replayfile: str, title: str, i : int) -> str:
 
 def getTitle(replayfile: str, title: str = None, i : int = 0) -> str:
 	global wg
-	assert wg is not None, "wg must not be None"
-
-	if title is None:
+	if title == None:
 		try:
 			filename = os.path.basename(replayfile)	
 			bu.debug(filename)
@@ -289,10 +289,10 @@ def getTitle(replayfile: str, title: str = None, i : int = 0) -> str:
 						#player = metadata_json['playerName']						
 						tank_name = wg.get_tank_name(metadata_json['vehicleCompDescriptor'])
 						map_name = wg.get_map(metadata_json['mapName'])
-						bu.debug('Tank: ' + tank_name if tank_name is not None else "NOT FOUND IN tanks.json" +  ' Map: ' + map_name)
+						bu.debug('Tank: ' + tank_name if tank_name != None else "NOT FOUND IN tanks.json" +  ' Map: ' + map_name)
 					except Exception as err:
 						bu.error(exception = err)
-			if (tank_name is not None) and  (map_name is not None):
+			if (tank_name != None) and  (map_name != None):
 				title = tank_name + ' @ ' + map_name
 			else:
 				title = re.sub('\\.wotbreplay$', '', filename)