diff --git a/analyze_wotb_replays.py b/analyze_wotb_replays.py index d931a0c..1b0d722 100644 --- a/analyze_wotb_replays.py +++ b/analyze_wotb_replays.py @@ -478,12 +478,19 @@ async def main(argv): ## Read config config = configparser.ConfigParser() config.read(FILE_CONFIG) - configGeneral = config['GENERAL'] + + configOptions = config['OPTIONS'] # WG account id of the uploader: # # Find it here: https://developers.wargaming.net/reference/all/wotb/account/list/ - WG_ID = configGeneral.getint('wg_id', None) - USE_DB = configGeneral.getboolean('use_DB', False) - + OPT_DB = configOptions.getboolean('opt_DB', False) + OPT_EXTENDED = configOptions.getboolean('opt_extended', False) + OPT_HIST = configOptions.getboolean('opt_hist', False) + OPT_STAT_FUNC = configOptions.get('opt_stat_func', fallback='tank_tier') + + configWG = config['WG'] + WG_ID = configWG.getint('wg_id', None) + WG_RATE_LIMIT = configWG.getint('wg_rate_limit', 10) + configDB = config['DATABASE'] DB_SERVER = configDB.get('db_server', 'localhost') DB_PORT = configDB.getint('db_port', 27017) @@ -502,15 +509,15 @@ async def main(argv): parser.add_argument('--output', default='plain', choices=['json', 'plain', 'db'], help='Select output mode: JSON, plain text or database') parser.add_argument('-id', dest='account_id', type=int, default=WG_ID, help='WG account_id to analyze') parser.add_argument('-a', '--account', type=str, default=None, help='WG account nameto analyze. Format: ACCOUNT_NAME@SERVER') - parser.add_argument('-x', '--extended', action='store_true', default=False, help='Print Extended stats') + parser.add_argument('-x', '--extended', action='store_true', default=OPT_EXTENDED, help='Print Extended stats') parser.add_argument('-X', '--extra_categories', choices=BattleRecordCategory.get_extra_categories(), default=None, nargs='*', help='Print Extended categories') - parser.add_argument('--hist', action='store_true', default=False, help='Print player histograms (WR/battles)') - parser.add_argument('--stat_func', default='tank_tier', choices=STAT_FUNC.keys(), help='Select how to calculate for ally/enemy performance: tank-tier stats, global player stats') + parser.add_argument('--hist', action='store_true', default=OPT_HIST, help='Print player histograms (WR/battles)') + parser.add_argument('--stat_func', default=OPT_STAT_FUNC, choices=STAT_FUNC.keys(), help='Select how to calculate for ally/enemy performance: tank-tier stats, global player stats') parser.add_argument('-u', '--url', action='store_true', default=False, help='Print replay URLs') parser.add_argument('--tankfile', type=str, default='tanks.json', help='JSON file to read Tankopedia from. Default is "tanks.json"') parser.add_argument('--mapfile', type=str, default='maps.json', help='JSON file to read Blitz map names from. Default is "maps.json"') parser.add_argument('-o','--outfile', type=str, default='-', metavar="OUTPUT", help='File to write results. Default STDOUT') - parser.add_argument('--db', action='store_true', default=USE_DB, help='Use DB - You are unlikely to have it') + parser.add_argument('--db', action='store_true', default=OPT_DB, help='Use DB - You are unlikely to have it') parser.add_argument('--filters', type=str, default=None, help='Filters for DB based analyses. MongoDB find() filter JSON format.') parser.add_argument('-d', '--debug', action='store_true', default=False, help='Debug mode') parser.add_argument('-v', '--verbose', action='store_true', default=False, help='Verbose mode') @@ -521,8 +528,8 @@ async def main(argv): bu.set_log_level(args.silent, args.verbose, args.debug) bu.set_progress_step(250) # Set the frequency of the progress dots. - wg = await WG.create(WG_APP_ID, args.tankfile, args.mapfile, True) - wi = WoTinspector() + wg = WG(WG_APP_ID, args.tankfile, args.mapfile, stats_cache=True, rate_limit=WG_RATE_LIMIT) + wi = WoTinspector(rate_limit=10) if args.account != None: args.account_id = await wg.get_account_id(args.account) @@ -936,8 +943,8 @@ async def get_wg_player_stats(stat_id_str: str) -> dict: account_id = int(stat_id_str) # 'battles' must always be there - hist_stats = [ 'all.' + x for x in histogram_fields.keys() ] - hist_stats.append('tank_id') + hist_stats = [ 'statistics.all.' + x for x in histogram_fields.keys() ] + hist_stats.append('account_id') player_stats = await wg.get_player_stats(account_id, hist_stats) #bu.debug('account_id: ' + str(account_id) + ' ' + str(player_stats)) diff --git a/blitzstats.ini.default b/blitzstats.ini.default index e73bdcc..0aaf3dd 100644 --- a/blitzstats.ini.default +++ b/blitzstats.ini.default @@ -1,16 +1,14 @@ -[GENERAL] +[OPTIONS] +## You may set your preferred commandline param defaults here +#opt_extended = False +#opt_hist = False +#opt_stat_func = tank_tier + +[WG] ## Find your WG account_id : https://developers.wargaming.net/reference/all/wotb/account/list/ -#wg_id = 0 +#wg_id = 0 +## WG API limits requests / sec to 10 for client apps +## Check https://developers.wargaming.net/documentation/guide/getting-started/ +#wg_rate_limit = 10 + -# you are unlikely have MongoDB set up so just ignore these -#[DATABASE] -#db_server = localhost -#db_port = 27017 -#db_auth = admin -#db_name = BlitzStats -#db_user = USERNAME -#db_password = PASSWORD -#db_ssl = True -#db_cert_req = 2 -#db_ssl_cert_file= cert.pem -#db_ssl_ca_file = ca.pem diff --git a/blitzutils.py b/blitzutils.py index 90d7ebb..c8c086d 100644 --- a/blitzutils.py +++ b/blitzutils.py @@ -27,6 +27,94 @@ _progress_obj = None UMASK= os.umask(0) os.umask(UMASK) +## ----------------------------------------------------------- +#### Class ThrottledClientSession(aiohttp.ClientSession) +## ----------------------------------------------------------- + +class ThrottledClientSession(aiohttp.ClientSession): + """Rate-throttled client session class inherited from aiohttp.ClientSession)""" + MIN_SLEEP = 0.1 + + def __init__(self, rate_limit: float =None, *args,**kwargs) -> None: + super().__init__(*args,**kwargs) + self.rate_limit = rate_limit + self._fillerTask = None + self._queue = None + + if rate_limit != None: + if rate_limit <= 0: + raise ValueError('rate_limit must be positive') + #(increment, sleep) = self._get_rate_increment() + self._queue = asyncio.Queue(min(2, int(rate_limit)+1)) + self._fillerTask = asyncio.create_task(self._filler(rate_limit)) + self._start_time = None + self._count = 0 + + def _get_sleep(self) -> list: + return max(1/self.rate_limit, self.MIN_SLEEP) + + async def close(self) -> None: + """Close rate-limiter's "bucket filler" task""" + # DEBUG + if self._start_time != None: + duration = time.time() - self._start_time + debug('Average WG API request rate: ' + '{:.1f}'.format(self._count / duration) + ' / sec') + if self._fillerTask != None: + self._fillerTask.cancel() + try: + await asyncio.wait_for(self._fillerTask, timeout= 3) + except asyncio.TimeoutError as err: + error(exception=err) + await super().close() + + + async def _filler(self, rate_limit: float = 1): + """Filler task to fill the leaky bucket algo""" + try: + if self._queue == None: + return + self.rate_limit = rate_limit + sleep = self._get_sleep() + debug('SLEEP: ' + str(sleep)) + updated_at = time.monotonic() + fraction = 0 + extra_increment = 0 + for i in range(0,self._queue.maxsize): + self._queue.put_nowait(i) + while True: + if not self._queue.full(): + now = time.monotonic() + increment = rate_limit * (now - updated_at) + 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(sleep) + except asyncio.CancelledError: + debug('Cancelled') + except Exception as err: + error(exception=err) + + + async def _allow(self) -> None: + if self._queue != None: + # debug + if self._start_time == None: + self._start_time = time.time() + await self._queue.get() + self._queue.task_done() + # DEBUG + self._count += 1 + return None + + + async def _request(self, *args,**kwargs): + """Throttled _request()""" + await self._allow() + return await super()._request(*args,**kwargs) def set_debug(debug: bool): global _log_level @@ -67,37 +155,39 @@ def get_log_level_str() -> str: error('Unknown log level: ' + str(_log_level)) -def verbose(msg = "", id = None): +def verbose(msg = "", id = None) -> bool: """Print a message""" if _log_level >= VERBOSE: if id == None: print(msg) else: print('[' + str(id) + ']: ' + msg) - return None + return True + return False -def verbose_std(msg = "", id = None): +def verbose_std(msg = "", id = None) -> bool: """Print a message""" if _log_level >= NORMAL: if id == None: print(msg) else: print('[' + str(id) + ']: ' + msg) - return None + return True + return False - -def debug(msg = "", id = None, exception = None): +def debug(msg = "", id = None, exception = None, force: bool = False) -> bool: """print a conditional debug message""" - if _log_level >= DEBUG: + if (_log_level >= DEBUG) or force: _print_log_msg('DEBUG', msg, exception, id) - return None + return True + return False -def error(msg = "", exception = None, id = None): +def error(msg = "", exception = None, id = None) -> bool: """Print an error message""" _print_log_msg('ERROR', msg, exception, id) - return None + return True def _print_log_msg(prefix = 'LOG', msg = '', exception = None, id = None): @@ -271,6 +361,9 @@ async def get_url_JSON(session: aiohttp.ClientSession, url: str, chk_JSON_func = # debug("Received valid JSON: " + str(json_resp)) return json_resp # Sometimes WG API returns JSON error even a retry gives valid JSON + elif resp.status == 407: + json_resp_err = await resp.json() + error('WG API returned 407: ' + json_resp_err['error']['message']) if retry == max_tries: break debug('Retrying URL [' + str(retry) + '/' + str(max_tries) + ']: ' + url ) @@ -321,7 +414,6 @@ class SlowBar(IncrementalBar): return (self.eta - (self.eta // 3600)*3600) // 60 - ## ----------------------------------------------------------- #### Class StatsNotFound ## ----------------------------------------------------------- @@ -329,6 +421,9 @@ class SlowBar(IncrementalBar): class StatsNotFound(Exception): pass + + + ## ----------------------------------------------------------- #### Class WG ## ----------------------------------------------------------- @@ -456,25 +551,27 @@ class WG: } 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/' + 'eu' : 'https://api.wotblitz.eu/wotb/', + 'ru' : 'https://api.wotblitz.ru/wotb/', + 'na' : 'https://api.wotblitz.com/wotb/', + 'asia' : 'https://api.wotblitz.asia/wotb/', + 'china' : None } ACCOUNT_ID_SERVER= { 'ru' : range(0, int(5e8)), 'eu' : range(int(5e8), int(10e8)), 'na' : range(int(1e9),int(2e9)), - 'asia' : range(int(2e9),int(4e9)) + 'asia' : range(int(2e9),int(3e9)), + 'china' : range(int(3e9),int(4e9)) } - def __init__(self, WG_app_id = None, tankopedia_fn = None, maps_fn = None, stats_cache = False): + def __init__(self, WG_app_id : str = None, tankopedia_fn : str = None, maps_fn : str = None, stats_cache: bool = False, rate_limit: int = 10): self.WG_app_id = WG_app_id self.load_tanks(tankopedia_fn) WG.tanks = self.tanks - + if (maps_fn != None): if os.path.exists(maps_fn) and os.path.isfile(maps_fn): try: @@ -484,41 +581,26 @@ class WG: error('Could not read maps file: ' + maps_fn + '\n' + str(err)) else: verbose('Could not find maps file: ' + maps_fn) + self.rate_limiter = None if self.WG_app_id != None: - self.session = aiohttp.ClientSession() - debug('WG aiohttp session initiated') + self.session = ThrottledClientSession(rate_limit=rate_limit) + debug('WG aiohttp session initiated') else: self.session = None debug('WG aiohttp session NOT initiated') - # cache TBD + # cache self.cache = None self.statsQ = None self.stat_saver_task = None - - - @classmethod - async def create(cls, WG_app_id = None, tankopedia_fn = None, maps_fn = None, stats_cache = False): - """Separete Constuctor method to handle async calls required to initialize the object""" - self = WG(WG_app_id, tankopedia_fn , maps_fn ) - if stats_cache: try: - self.cache = await aiosqlite.connect(WG.CACHE_DB_FILE) - ## Player Tank stats cache table - await self.cache.execute(WG.SQL_TANK_STATS_CREATE_TBL) - - await self.cache.commit() - - async with self.cache.execute(WG.SQL_TANK_STATS_COUNT) as cursor: - debug('Cache contains: ' + str((await cursor.fetchone())[0]) + ' cached player tank stat records' ) - self.statsQ = asyncio.Queue() self.stat_saver_task = asyncio.create_task(self.stat_saver()) except Exception as err: error(exception=err) sys.exit(1) - return self + async def close(self): # close stats queue @@ -529,9 +611,7 @@ class WG: self.stat_saver_task.cancel() debug('statsCacheTask cancelled') await self.stat_saver_task - - - + # close cacheDB if self.cache != None: # prune old cache records @@ -540,7 +620,8 @@ class WG: await self.cache.close() if self.session != None: - await self.session.close() + await self.session.close() + return ## Class methods ---------------------------------------------------------- @@ -649,6 +730,7 @@ class WG: error("JSON format error", err) return False + @classmethod def chk_JSON_get_account_id(cls, json_resp: dict) -> bool: try: @@ -702,6 +784,7 @@ class WG: debug("JSON check failed") return False + @classmethod def chk_JSON_tank_stats(cls, json_resp: dict) -> bool: """"Check String for being a valid Tank list JSON file""" @@ -744,6 +827,7 @@ class WG: error('Invalid tier', err) return None + def get_url_clan_info(self, server: str, clan_id: int) -> str: try: if server == None: @@ -761,7 +845,7 @@ class WG: return self.get_url_player_tanks_stats(account_id, fields='tank_id') - def get_url_player_tanks_stats(self, account_id: int, tank_ids = [], fields = []) -> str: + def get_url_player_tanks_stats(self, account_id: int, tank_ids: list = [], fields: list = []) -> str: server = self.get_server(account_id) if server == None: return None @@ -823,6 +907,7 @@ class WG: if nick == None or server == None: raise ValueError('Invalid nickname given: ' + nickname) url = self.get_url_account_id(nick, server) + json_data = await get_url_JSON(self.session, url, self.chk_JSON_status) for res in json_data['data']: if res['nickname'].lower() == nick.lower(): @@ -852,7 +937,7 @@ class WG: #debug('JSON Response received: ' + str(json_data)) stats = json_data['data'][str(account_id)] if cache: - await self.save_stats('tank_stats', [account_id, tank_ids], stats) + await self.put_2_statsQ('tank_stats', [account_id, tank_ids], stats) return stats except Exception as err: error(exception=err) @@ -884,7 +969,7 @@ class WG: #debug('JSON Response received: ' + str(json_data)) stats = json_data['data'][str(account_id)] if cache: - await self.save_stats('player_stats', [account_id], stats) + await self.put_2_statsQ('player_stats', [account_id], stats) return stats except Exception as err: error(exception=err) @@ -914,12 +999,13 @@ class WG: except KeyError as err: error('Key not found', err) return None - + + def get_tank_tier(self, tank_id: int): return self.get_tank_data(tank_id, 'tier') - async def save_stats(self, statsType: str, key: list, stats: list): + async def put_2_statsQ(self, statsType: str, key: list, stats: list): """Save stats to a async queue to be saved by the stat_saver -task""" if self.statsQ == None: return False @@ -927,12 +1013,26 @@ class WG: await self.statsQ.put([ statsType, key, stats, NOW() ]) return True + async def stat_saver(self): """Async task for saving stats into cache in background""" if self.statsQ == None: error('No statsQ defined') return None + try: + self.cache = await aiosqlite.connect(WG.CACHE_DB_FILE) + ## Create cache tables table + await self.cache.execute(WG.SQL_TANK_STATS_CREATE_TBL) + await self.cache.execute(WG.SQL_PLAYER_STATS_CREATE_TBL) + await self.cache.commit() + + async with self.cache.execute(WG.SQL_TANK_STATS_COUNT) as cursor: + debug('Cache contains: ' + str((await cursor.fetchone())[0]) + ' cached player tank stat records' ) + except Exception as err: + error(exception=err) + sys.exit(1) + while True: try: stats = await self.statsQ.get() @@ -1057,7 +1157,7 @@ class WG: #debug('No cache DB') raise StatsNotFound('No cache DB in use') - async with self.cache.execute(WG.SQL_PLAYER_STATS_CACHED, (account_id, NOW() - WG.CACHE_GRACE_TIME) ) as cursor: + async with self.cache.execute(WG.SQL_PLAYER_STATS_CACHED, [account_id, NOW() - WG.CACHE_GRACE_TIME] ) as cursor: row = await cursor.fetchone() #debug('account_id: ' + str(account_id) + ': 1') if row == None: @@ -1066,18 +1166,18 @@ class WG: raise StatsNotFound('No cached stats found') debug('Cached stats found') - if row[3] == None: + if row[2] == None: # None/null stats found in cache # i.e. stats have been requested, but not returned from WG API return None else: # Return proper stats - return json.loads(row[3]) + return json.loads(row[2]) except StatsNotFound as err: debug(exception=err) raise except Exception as err: - error('Error trying to look for cached stats', err) + error('Error trying to look for cached stats', exception=err) return None @@ -1096,13 +1196,15 @@ class WoTinspector: REPLAY_N = 1 - def __init__(self): - self.session = aiohttp.ClientSession() + def __init__(self, rate_limit: int = 30): + self.session = ThrottledClientSession(rate_limit=rate_limit) + async def close(self): if self.session != None: debug('Closing aiohttp session') - await self.session.close() + await self.session.close() + async def get_tankopedia(self, filename = 'tanks.json'): """Retrieve Tankpedia from WoTinspector.com""" @@ -1143,7 +1245,7 @@ class WoTinspector: async def get_replay_JSON(self, replay_id: str): - json_resp = await get_url_JSON(self.session, self.URL_REPLAY_INFO + replay_id, None) + json_resp = await get_url_JSON(self.session, self.URL_REPLAY_INFO + replay_id, chk_JSON_func=None) try: if self.chk_JSON_replay(json_resp): return json_resp @@ -1196,8 +1298,7 @@ class WoTinspector: if json_resp.get('status', None) == None: error(msg_str +' : ' + title + ' : Received invalid JSON') elif (json_resp['status'] == 'ok'): - debug('Response data read') - verbose_std(msg_str + title + ' posted') + debug('Response data read. Status OK') return json_resp elif (json_resp['status'] == 'error'): error(msg_str + json_resp['error']['message'] + ' : ' + title) @@ -1212,6 +1313,11 @@ class WoTinspector: error(msg_str + ' Could not post replay: ' + title) return None + + async def get_replay_listing(self, page: int = 0) -> aiohttp.ClientResponse: + url = self.get_url_replay_listing(page) + return await self.session.get(url) + @classmethod def get_url_replay_listing(cls, page : int): return cls.URL_REPLAY_LIST + str(page) + '?vt=#filters' @@ -1258,8 +1364,8 @@ class BlitzStars: URL_ACTIVE_PLAYERS = URL_BLITZSTARS + '/api/playerstats/activeinlast30days' - def __init__(self): - self.session = aiohttp.ClientSession() + def __init__(self, rate_limit=30): + self.session = ThrottledClientSession(rate_limit=rate_limit) async def close(self): diff --git a/postwotreplay.py b/postwotreplay.py deleted file mode 100644 index 0eb3058..0000000 --- a/postwotreplay.py +++ /dev/null @@ -1,20 +0,0 @@ -#!/usr/bin/python3 - -import json, requests, sys, base64 - -filename = str(sys.argv[1]) - -#Change this to your own WG id. Find it here: https://api.wotblitz.eu/wotb/account/list/?account_id%2Cnickname -# WG account id of the uploader: -uploader_account_ID = 0 - -# edit this or use arg.parse() to add proper suppod for commandline arguments -title="Some title" -url='https://wotinspector.com/api/replay/upload?title=' + title + '&details=full&uploaded_by=' + str(uploader_account_ID) - -# additional option: 'private=1' to keep the replay way from public listing at WoTinspector.com/replays - -with open(filename,'rb') as f: - payload = {'file' : (filename, base64.b64encode(f.read()))} - r = requests.post(url, data=payload) - print(r.json()) \ No newline at end of file diff --git a/upload_wotb_replays.py b/upload_wotb_replays.py index 088cfa0..5fd5b81 100644 --- a/upload_wotb_replays.py +++ b/upload_wotb_replays.py @@ -11,9 +11,10 @@ logging.getLogger("asyncio").setLevel(logging.DEBUG) FILE_CONFIG = 'blitzstats.ini' # DB_C_REPLAYS = 'Replays' -DEBUG = False +DEBUG = False VERBOSE = False -SLEEP = 2 +SLEEP = 2 +N_WORKERS = 5 MAX_RETRIES = 3 REPLAY_N = 0 SKIPPED_N = 0 @@ -31,29 +32,13 @@ async def main(argv): ## Read config config = configparser.ConfigParser() config.read(FILE_CONFIG) - # configDB = config['DATABASE'] - # DB_SERVER = configDB.get('db_server', 'localhost') - # DB_PORT = configDB.getint('db_port', 27017) - # DB_SSL = configDB.getboolean('db_ssl', False) - # DB_CERT_REQ = configDB.getint('db_ssl_req', ssl.CERT_NONE) - # DB_AUTH = configDB.get('db_auth', 'admin') - # DB_NAME = configDB.get('db_name', 'BlitzStats') - # DB_USER = configDB.get('db_user', 'mongouser') - # DB_PASSWD = configDB.get('db_password', "PASSWORD") - # DB_CERT = configDB.get('db_ssl_cert_file', None) - # DB_CA = configDB.get('db_ssl_ca_file', None) - # bu.debug('DB_SERVER: ' + DB_SERVER) - # bu.debug('DB_PORT: ' + str(DB_PORT)) - # bu.debug('DB_SSL: ' + "True" if DB_SSL else "False") - # bu.debug('DB_AUTH: ' + DB_AUTH) - # bu.debug('DB_NAME: ' + DB_NAME) - - configGeneral = config['GENERAL'] + configWG = config['WG'] # WG account id of the uploader: # # Find it here: https://developers.wargaming.net/reference/all/wotb/account/list/ - WG_ID = configGeneral.getint('wg_id', 0) -# USE_DB = configGeneral.getint('use_DB', False) + WG_ID = configWG.getint('wg_id', None) + ## WG API Rules limit 10 request / sec. Higher rate of requests will return errors ==> extra delay + WG_RATE_LIMIT = configWG.getint('wg_rate_limit', 10) parser = argparse.ArgumentParser(description='Post replays(s) to WoTinspector.com and retrieve battle data') #parser.add_argument('--output', default='file', choices=['file', 'files', 'db'] , help='Select output mode: single/multiple files or database') @@ -61,7 +46,7 @@ async def main(argv): parser.add_argument('-a', '--account', dest='account', type=str, default=None, help='Uploader\'s WG account name. Format: ACCOUNT_NAME@SERVER') parser.add_argument('-t','--title', type=str, default=None, help='Title for replays. Use NN for continous numbering. Default is filename-based numbering') parser.add_argument('-p', '--private', dest="private", action='store_true', default=False, help='Set replays private on WoTinspector.com') - parser.add_argument('--tasks', dest='N_tasks', type=int, default=5, help='Number of worker threads') + parser.add_argument('--tasks', dest='N_tasks', type=int, default=N_WORKERS, help='Number of worker threads') parser.add_argument('--tankopedia', type=str, default='tanks.json', help='JSON file to read Tankopedia from. Default: "tanks.json"') parser.add_argument('--mapfile', type=str, default='maps.json', help='JSON file to read Blitz map names from. Default: "maps.json"') # parser.add_argument('--db', action='store_true', default=USE_DB, help='Use DB - You are unlikely to have it') @@ -75,22 +60,7 @@ async def main(argv): bu.set_log_level(args.silent, args.verbose, args.debug) wg = WG(WG_appID, args.tankopedia, args.mapfile) - wi = WoTinspector() - - #### Connect to MongoDB (TBD) - # client = None - # db = None - # if args.db: - # try: - # client = motor.motor_asyncio.AsyncIOMotorClient(DB_SERVER,DB_PORT, authSource=DB_AUTH, username=DB_USER, password=DB_PASSWD, ssl=DB_SSL, ssl_cert_reqs=DB_CERT_REQ, ssl_certfile=DB_CERT, tlsCAFile=DB_CA) - # db = client[DB_NAME] - # bu.debug('Database connection initiated') - # except Exception as err: - # bu.error("Could no initiate DB connection: Disabling DB", err) - # args.db = False - # pass - # else: - # bu.debug('No DB in use') + wi = WoTinspector(rate_limit=WG_RATE_LIMIT) if args.account != None: args.accountID = await wg.get_account_id(args.account) @@ -170,13 +140,14 @@ async def mkReplayQ(queue : asyncio.Queue, files : list, title : str): bu.debug('Finished') return None + async def mkQueueItem(filename : str, title : str) -> list: """Make an item to replay queue""" global REPLAY_N REPLAY_N +=1 return [filename, REPLAY_N, getTitle(filename, title, REPLAY_N)] -#async def replayWorker(queue: asyncio.Queue, db: motor.motor_asyncio.AsyncIOMotorDatabase, workerID: int, account_id: int, priv = False): + async def replayWorker(queue: asyncio.Queue, workerID: int, account_id: int, priv = False): """Async Worker to process the replay queue""" global SKIPPED_N @@ -199,14 +170,14 @@ async def replayWorker(queue: asyncio.Queue, workerID: int, account_id: int, pri replay_json = json.loads(await fp.read()) #if replay_json['status'] == 'ok': if wi.chk_JSON_replay(replay_json): - bu.verbose(msg_str + title + ' has already been posted. Skipping.' ) - SKIPPED_N += 1 - #await saveReplay2DB(db, replay_json) - queue.task_done() - continue + bu.verbose_std(msg_str + title + ' has already been posted. Skipping.' ) else: os.remove(replay_json_fn) bu.debug("Replay JSON not valid: Deleting " + replay_json_fn) + SKIPPED_N += 1 + queue.task_done() + continue + except asyncio.CancelledError as err: raise err except Exception as err: @@ -218,12 +189,12 @@ async def replayWorker(queue: asyncio.Queue, workerID: int, account_id: int, pri bu.debug(msg_str + 'File: ' + filename) json_resp = await wi.post_replay(await fp.read(), filename, account_id, title, priv, N) if json_resp != None: - if (await bu.save_JSON(replay_json_fn,json_resp)): - bu.debug(msg_str + 'Replay saved OK: ' + filename ) + if (await bu.save_JSON(replay_json_fn,json_resp)): + if not bu.debug(msg_str + 'Replay saved OK: ' + filename): + bu.verbose_std(msg_str + title + ' posted') else: bu.error(msg_str + 'Error saving replay: ' + filename) - ERROR_N += 1 - # await saveReplay2DB(db, json_resp) + ERROR_N += 1 except Exception as err: bu.error(msg_str + 'Unexpected Exception: ' + str(type(err)) + ' : ' + str(err) ) bu.debug(msg_str + 'Marking task done') @@ -232,26 +203,6 @@ async def replayWorker(queue: asyncio.Queue, workerID: int, account_id: int, pri return None -# async def saveReplay2DB(db: motor.motor_asyncio.AsyncIOMotorDatabase, replay: dict) -> bool: -# """Save retriever replay JSON into DB (mongo)""" -# try: -# if db == None: -# return False -# dbc = db[DB_C_REPLAYS] -# replay_link = replay['data']['download_url'] -# replay_id = wi.get_replay_id(replay_link) -# replay['_id'] = replay_id -# await dbc.insert_one(replay) -# bu.debug('Replay added to database') -# except pymongo.errors.DuplicateKeyError as err: -# bu.verbose_std('Replay already in the DB: ' + replay['data']['summary']['title'] + ' : id=' + replay_id) -# return False -# except Exception as err: -# bu.error('Unexpected Exception: ' + str(type(err)) + ' : ' + str(err)) -# return False -# return True - - def getTitle(replayfile: str, title: str, i : int) -> str: global wg