Merge pull request #39 from Jylpah/dev-fix-tankopedia

FIX: Make analyze_replays.py to work on Windows
This commit is contained in:
Jylpah authored and GitHub committed 2023-09-11 21:00:06 +03:00
commit 8a6d6176e9
2 files changed
+75 -60

No files matched your search

+26 -11
View File
@@ -5,6 +5,7 @@
import sys, argparse, json, os, concurrent, inspect, aiohttp, asyncio, aiofiles, aioconsole import sys, argparse, json, os, concurrent, inspect, aiohttp, asyncio, aiofiles, aioconsole
import configparser, ssl, re, logging, time, xmltodict, collections, csv import configparser, ssl, re, logging, time, xmltodict, collections, csv
from pathlib import Path
from asyncio import Queue, Task, create_task from asyncio import Queue, Task, create_task
from datetime import datetime, timedelta, date from datetime import datetime, timedelta, date
from motor.motor_asyncio import ( from motor.motor_asyncio import (
@@ -1690,11 +1691,11 @@ class ErrorCatchingArgumentParser(argparse.ArgumentParser):
## main() ------------------------------------------------------------- ## main() -------------------------------------------------------------
async def main(argv): async def main(argv: list[str]):
global wg, wi, WG_APP_ID, OPT_MODE_DEFAULT, OPT_EXPORT_CSV_FILE, OPT_EXPORT_JSON_FILE global wg, wi, WG_APP_ID, OPT_EXPORT_CSV_FILE, OPT_EXPORT_JSON_FILE
# set the directory for the script # set the directory for the script
current_dir = os.getcwd() current_dir = os.getcwd()
os.chdir(os.path.dirname(sys.argv[0])) # os.chdir(os.path.dirname(sys.argv[0]))
## Default options: ## Default options:
OPT_MODE_DEFAULT = "default" OPT_MODE_DEFAULT = "default"
@@ -1728,16 +1729,30 @@ async def main(argv):
try: try:
## Read config ## Read config
if os.path.isfile(FILE_CONFIG):
CONFIG: str = "blitzstats.ini"
CONFIG_FILES: list[Path] = [
Path(".") / CONFIG,
Path(__file__).parent / CONFIG,
Path.home() / f".{CONFIG}",
Path.home() / ".config" / CONFIG,
Path.home() / ".config/blitzstats/config",
]
config_file: Path | None = None
for config_file in CONFIG_FILES:
if config_file.is_file():
break
if config_file is None:
print("no config file found")
else:
config = configparser.ConfigParser() config = configparser.ConfigParser()
config.read(FILE_CONFIG) config.read(config_file)
try: try:
if "ANALYZER" in config.sections(): if "ANALYZER" in config.sections():
configAnalyzer = config["ANALYZER"] configAnalyzer = config["ANALYZER"]
OPT_MODE_DEFAULT = configAnalyzer.getboolean( OPT_MODE_DEFAULT = configAnalyzer.get("mode", OPT_MODE_DEFAULT)
"mode", OPT_MODE_DEFAULT
)
OPT_HIST = configAnalyzer.getboolean("histograms", OPT_HIST) OPT_HIST = configAnalyzer.getboolean("histograms", OPT_HIST)
OPT_STAT_FUNC = configAnalyzer.get( OPT_STAT_FUNC = configAnalyzer.get(
"stat_func", fallback=OPT_STAT_FUNC "stat_func", fallback=OPT_STAT_FUNC
@@ -3022,9 +3037,9 @@ async def mk_replayQ(
scanner: Task = create_task(jsonQ.mk_queue(files=files)) scanner: Task = create_task(jsonQ.mk_queue(files=files))
async for fn in jsonQ: async for fn in jsonQ:
try: try:
replay_json = await bu.open_JSON(fn, wi.chk_JSON_replay) replay_json = await bu.open_JSON(str(fn), wi.chk_JSON_replay)
await replayQ.put(await mk_readerQ_item(replay_json, filename=fn)) await replayQ.put(await mk_readerQ_item(replay_json, filename=fn))
bu.debug("File added to queue: " + fn) bu.debug("File added to queue: " + str(fn))
Nreplays += 1 Nreplays += 1
except Exception as err: except Exception as err:
bu.error(exception=err) bu.error(exception=err)
@@ -3351,4 +3366,4 @@ if __name__ == "__main__":
asyncio.get_event_loop_policy(), WindowsSelectorEventLoopPolicy asyncio.get_event_loop_policy(), WindowsSelectorEventLoopPolicy
): ):
asyncio.set_event_loop_policy(WindowsSelectorEventLoopPolicy()) asyncio.set_event_loop_policy(WindowsSelectorEventLoopPolicy())
asyncio.run(main(sys.argv[1:]), debug=False) asyncio.run(main(sys.argv), debug=False)
+49 -49
View File
@@ -309,60 +309,60 @@ async def replayWorker(
return None return None
def getTitle_old(replayfile: str, title: str, i: int) -> str: # def getTitle_old(replayfile: str, title: str, i: int) -> str:
global wg # global wg
if title is None: # if title is None:
try: # try:
filename = os.path.basename(replayfile) # filename = os.path.basename(replayfile)
bu.debug(filename) # bu.debug(filename)
map_usrStrs = wg.get_map_user_strs() # map_usrStrs = wg.get_map_user_strs()
tank_userStrs = wg.get_tank_user_strs() # tank_userStrs = wg.get_tank_user_strs()
# p = re.compile('\\d{8}_\\d{4}_(.+)_(' + '|'.join(map_usrStrs) + ')(?:-\\d)?\\.wotbreplay$') # # p = re.compile('\\d{8}_\\d{4}_(.+)_(' + '|'.join(map_usrStrs) + ')(?:-\\d)?\\.wotbreplay$')
# update 6.2 changed the file name format. Bug fixed 2019-09-09 Jylpah # # update 6.2 changed the file name format. Bug fixed 2019-09-09 Jylpah
p = re.compile( # p = re.compile(
"\\d{8}_\\d{4}_.*?(" # "\\d{8}_\\d{4}_.*?("
+ "|".join(tank_userStrs) # + "|".join(tank_userStrs)
+ ")_(" # + ")_("
+ "|".join(map_usrStrs) # + "|".join(map_usrStrs)
+ ")(?:-\\d)?\\.wotbreplay$" # + ")(?:-\\d)?\\.wotbreplay$"
) # )
m = p.match(filename) # m = p.match(filename)
if m is not None: # if m is not None:
if wg.tanks is not None: # if wg.tanks is not None:
tank = m.group(1) # tank = m.group(1)
if tank in wg.tanks["userStr"]: # if tank in wg.tanks["userStr"]:
tank = wg.tanks["userStr"][tank] # tank = wg.tanks["userStr"][tank]
else: # else:
bu.error( # bu.error(
'Tank code: "' # 'Tank code: "'
+ tank # + tank
+ '" not found from Tankopedia (tanks.json)' # + '" not found from Tankopedia (tanks.json)'
) # )
else: # else:
tank = m.group(1) # tank = m.group(1)
blitz_map = m.group(2) # blitz_map = m.group(2)
if blitz_map in wg.maps: # if blitz_map in wg.maps:
blitz_map = wg.maps[blitz_map] # blitz_map = wg.maps[blitz_map]
else: # else:
bu.error( # bu.error(
'Mapcode: "' # 'Mapcode: "'
+ blitz_map # + blitz_map
+ '" not found from map database (maps.json)' # + '" not found from map database (maps.json)'
) # )
title = tank + " @ " + blitz_map # title = tank + " @ " + blitz_map
else: # else:
title = re.sub("\\.wotbreplay$", "", filename) # title = re.sub("\\.wotbreplay$", "", filename)
except Exception as err: # except Exception as err:
bu.error(err) # bu.error(err)
else: # else:
title.replace("NN", str(i)) # title.replace("NN", str(i))
return title # return title
def getTitle(replayfile: str, title: str = None, i: int = 0) -> str: def getTitle(replayfile: str, title: str | None = None, i: int = 0) -> str:
global wg global wg
assert wg is not None, "wg must not be None" assert wg is not None, "wg must not be None"