remove legacy scripts, use Jylpah/blitz-replays

extract_tankopedia.py

@@ -1,248 +0,0 @@
-#!/usr/bin/env python3
-## PYTHON VERSION MUST BE 3.7 OR HIGHER
-
-# Extract tankopedia data in WG API JSON format from Blitz app files (Android APK unzipped)
-
-from os import path
-import aiofiles 				# type: ignore
-import xmltodict				# type: ignore
-from asyncio import set_event_loop_policy, run, create_task, get_event_loop_policy
-import sys, argparse, json, os, inspect, asyncio, re, logging, time, configparser
-from collections import OrderedDict
-import blitzutils as bu
-from blitzutils import WG
-
-logging.getLogger("asyncio").setLevel(logging.DEBUG)
-
-FILE_CONFIG 	= 'blitzstats.ini'
-
-# BLITZAPP_STRINGS='/assets/Data/Strings/en.yaml'
-# BLITZAPP_VEHICLES_DIR='/assets/Data/XML/item_defs/vehicles/'
-BLITZAPP_STRINGS='Data/Strings/en.yaml'
-BLITZAPP_VEHICLES_DIR='Data/XML/item_defs/vehicles/'
-BLITZAPP_VEHICLE_FILE='list.xml'
-
-wg : WG | None = None
-
-## main() -------------------------------------------------------------
-async def main() -> None:
-	global wg
-	# set the directory for the script
-	os.chdir(os.path.dirname(sys.argv[0]))
-
-	## Read config
-	BLITZAPP_FOLDER = '.'
-	try:
-		if os.path.isfile(FILE_CONFIG):
-			config = configparser.ConfigParser()
-			config.read(FILE_CONFIG)
-
-			configOptions 	= config['EXTRACT_TANKOPEDIA']
-			BLITZAPP_FOLDER = configOptions.get('blitz_app_dir', BLITZAPP_FOLDER)
-	except:
-		pass
-	
-	parser = argparse.ArgumentParser(description='Extract Tankopedia data from Blitz game files')
-	parser.add_argument('blitz_app_base', type=str, nargs='?', metavar="BLITZAPP_FOLDER", default=BLITZAPP_FOLDER, help='Base dir of the Blitz App files')
-	parser.add_argument('tanks', type=str, default='tanks.json', nargs='?', metavar="TANKS_FILE", help='File to write Tankopedia')
-	parser.add_argument('maps', type=str, default='maps.json', nargs='?', metavar='MAPS_FILE', help='File to write map names')
-	arggroup = parser.add_mutually_exclusive_group()
-	arggroup.add_argument('-d', '--debug', action='store_true', default=False, help='Debug mode')
-	arggroup.add_argument('-v', '--verbose', action='store_true', default=False, help='Verbose mode')
-	arggroup.add_argument('-s', '--silent', action='store_true', default=False, help='Silent mode')
-		
-	args = parser.parse_args()
-	bu.set_log_level(args.silent, args.verbose, args.debug)
-
-	wg = WG()
-	
-	tasks = []
-	for nation in wg.NATION:
-		tasks.append(asyncio.create_task(extract_tanks(args.blitz_app_base, nation)))
-
-	tanklist = []
-	for tanklist_tmp in await asyncio.gather(*tasks):
-		tanklist.extend(tanklist_tmp)
-	
-	tank_strs, map_strs = await read_user_strs(args.blitz_app_base)
-
-	json_data = None
-	userStrs = {}
-	tanks = {}
-	if os.path.exists(args.tanks):
-		try:
-			async with aiofiles.open(args.tanks) as infile:
-				json_data = json.loads(await infile.read())
-				userStrs = json_data['userStr']
-				tanks = json_data['data']
-		except Exception as err:
-			bu.error('Unexpected error when reading file: ' + args.tanks + ' : ' + str(err))
-
-	async with aiofiles.open(args.tanks, 'w', encoding="utf8") as outfile:
-		new_tanks, new_userStrs = await convert_tank_names(tanklist, tank_strs)
-		# merge old and new tankopedia
-		tanks.update(new_tanks)
-		userStrs.update(new_userStrs) 
-		tankopedia : OrderedDict[str, str| int | dict ]= OrderedDict()
-		tankopedia['status'] = 'ok'
-		tankopedia['meta'] = { "count":  len(tanks) }
-		tankopedia['data'] = bu.sort_dict(tanks, number=True)
-		tankopedia['userStr'] = bu.sort_dict(userStrs)
-		bu.verbose_std('New tankopedia \'' + args.tanks + '\' contains ' + str(len(tanks)) + ' tanks')
-		bu.verbose_std('New tankopedia \'' + args.tanks + '\' contains ' + str(len(userStrs)) + ' tank strings')        
-		await outfile.write(json.dumps(tankopedia, ensure_ascii=False, indent=4, sort_keys=False))
-	
-	if args.maps is not None:
-		maps = {}
-		if os.path.exists(args.maps):
-			try:
-				async with aiofiles.open(args.maps) as infile:
-					maps = json.loads(await infile.read())
-			except Exception as err:
-				bu.error('Unexpected error when reading file: ' + args.maps + ' : ' + str(err))
-		# merge old and new map data
-		maps.update(map_strs)
-		async with aiofiles.open(args.maps, 'w', encoding="utf8") as outfile:
-			bu.verbose_std('New maps file \'' + args.maps + '\' contains ' + str(len(maps)) + ' maps')
-			await outfile.write(json.dumps(maps, ensure_ascii=False, indent=4, sort_keys=True))
-
-	return None
-	
-async def extract_tanks(blitz_app_base : str, nation: str):
-
-	tanks = []
-	# WG has changed the location of Data directory - at least in steam client
-	if path.isdir(path.join(blitz_app_base, 'assets')):
-		blitz_app_base = path.join(blitz_app_base, 'assets')
-	list_xml_file = path.join(blitz_app_base, BLITZAPP_VEHICLES_DIR ,nation ,BLITZAPP_VEHICLE_FILE)
-	if not path.isfile(list_xml_file): 
-		print('ERROR: cannot open ' + list_xml_file)
-		return None
-	bu.debug('Opening file: ' + list_xml_file + ' (Nation: ' + nation + ')')
-	async with aiofiles.open(list_xml_file, 'r', encoding="utf8") as f:
-		try: 
-			tankList = xmltodict.parse(await f.read())
-			for data in tankList['root'].keys():
-				tank_xml = tankList['root'][data]
-				tank : dict[str, bool | int | str]= dict()
-				tank['is_premium'] = issubclass(type(tank_xml['price']), dict)
-				tank['nation']  = nation
-				tank['tank_id'] = get_tank_id(nation, int(tank_xml['id']))
-				tank['tier']    = int(tank_xml['level'])
-				tank['type']    = get_tank_type(tank_xml['tags'])
-				tank['userStr'] = tank_xml['userString']
-				#bu.debug('Reading tank string: ' + tank['userStr'], force=True)
-				tanks.append(tank)
-		except Exception as err:
-			bu.error(err)
-			sys.exit(2)
-	return tanks
-
-async def read_user_strs(blitz_app_base : str) -> tuple[dict, dict]:
-	"""Read user strings to convert map and tank names"""
-	tank_strs = {}
-	map_strs = {}
-	filename = path.join(blitz_app_base, BLITZAPP_STRINGS)
-	bu.debug('Opening file: ' + filename + ' for reading UserStrings')
-	try:
-		async with aiofiles.open(filename, 'r', encoding="utf8") as f:
-			p_tank = re.compile('^"(#\\w+?_vehicles:.+?)": "(.+)"$')
-			p_map = re.compile('^"#maps:(.+?):.+?: "(.+?)"$')
-			
-			async for l in f:
-				m = p_tank.match(l)
-				if m is not None: 
-					tank_strs[m.group(1)] = m.group(2)
-				
-				m = p_map.match(l)
-				if m is not None and m.group(2) != 'Macragge':                    
-					map_strs[m.group(1)] = m.group(2)   
-	
-	except Exception as err:
-		bu.error(err)
-		sys.exit(1)
-
-	return tank_strs, map_strs
-	
-async def convert_tank_names(tanklist : list, tank_strs: dict) -> tuple[dict, dict]:
-	"""Convert tank names for Tankopedia"""
-	tankopedia = {}
-	userStrs = {}
-
-	bu.debug(f'tank_strs:')
-	for key, value in tank_strs.items():
-		bu.debug(f'{key}: {value}')
-	bu.debug('---------')
-	try:
-		for tank in tanklist:
-			try:
-				bu.debug(f'tank: {tank}')
-				if tank['userStr'] in tank_strs:
-					tank['name'] = tank_strs[tank['userStr']]
-				else:
-					tank['name'] = tank['userStr'].split(':')[1]
-				tank.pop('userStr', None)
-				tank_tmp = OrderedDict()
-				for key in sorted(tank.keys()):
-					tank_tmp[key] = tank[key]
-				tankopedia[str(tank['tank_id'])] = tank_tmp
-			except:
-				bu.error(f'Could not process tank: {tank}')
-
-		for tank_str in tank_strs:
-			skip = False
-			key = tank_str.split(':')[1]
-			bu.debug('Tank string: ' + key + ' = ' + tank_strs[tank_str])
-			re_strs = [r'^Chassis_', r'^Turret_', r'^_', r'_short$' ]
-			for re_str in re_strs:
-				p = re.compile(re_str)
-				if p.match(key):
-					skip = True
-					break
-			if skip:
-				continue
-			
-			userStrs[key] = tank_strs[tank_str]
-
-		# sorting
-		tankopedia_sorted = OrderedDict()
-		for tank_id in sorted(tankopedia.keys(), key=int):
-			tankopedia_sorted[str(tank_id)] = tankopedia[str(tank_id)]
-
-		userStrs_sorted = OrderedDict()
-		for userStr in sorted(userStrs.keys()):
-			userStrs_sorted[userStr] = userStrs[userStr]
-		# bu.debug('Tank strings: ' + str(len(userStrs_sorted)))
-
-	except Exception as err:
-		bu.error(err)
-		sys.exit(1)
-
-	return tankopedia_sorted, userStrs_sorted
-
-
-def get_tank_id(nation: str, tank_id : int) -> int:
-	assert wg is not None, "wg has to be set"
-	return (tank_id << 8) + (wg.NATION_ID[nation] << 4) + 1 
-
-
-def get_tank_type(tagstr : str):
-	assert wg is not None, "wg has to be set"
-	tags = tagstr.split(' ')
-	for t_type in wg.TANK_TYPE:
-		if tags[0] == t_type:
-			return t_type
-	return None
-
-### main()
-if __name__ == "__main__":
-	# To avoid 'Event loop is closed' RuntimeError due to compatibility issue with aiohttp
-	if sys.platform.startswith("win") and sys.version_info >= (3, 8):
-		try:
-			from asyncio import WindowsSelectorEventLoopPolicy
-		except ImportError:
-			pass
-		else:
-			if not isinstance(get_event_loop_policy(), WindowsSelectorEventLoopPolicy):
-				set_event_loop_policy(WindowsSelectorEventLoopPolicy())
-	run(main())

get_tankopedia.py

@@ -1,91 +0,0 @@
-#!/usr/bin/env python3
-
-# Get Tankpedia from WoTinspector.com
-
-import sys
-import argparse
-import json
-import os
-import inspect
-import aiohttp
-import asyncio
-import re
-import logging
-from pathlib import Path
-import blitzutils as bu
-from blitzutils import WG
-from blitzutils import WoTinspector
-
-logging.getLogger("asyncio").setLevel(logging.DEBUG)
-
-VERBOSE = False
-DEBUG = False
-
-tank_type = ['lightTank', 'mediumTank', 'heavyTank', 'AT-SPG']
-
-# @asyncio.coroutine
-async def get_tankopedia(filename: str):
-	url = "https://wotinspector.com/static/armorinspector/tank_db_blitz.js"
-	async with aiohttp.ClientSession() as client:
-		async with client.get(url) as r:
-			if r.status == 200:
-				WI_tank_db = await r.text()
-				WI_tank_db = WI_tank_db.split("\n")
-			else:
-				print('Error: Could not get valid HTTPS response. HTTP: ' + str(r.status) )  
-				sys.exit(1) 
-			tanks = {}
-			n = 0
-			p = re.compile('\\s*(\\d+):\\s{"en":"([^"]+)",.*?"tier":(\\d+), "type":(\\d), "premium":(\\d).*')
-			for line in WI_tank_db[1:-1]:
-				m = p.match(line)
-				tank = {}
-				tank['tank_id'] = int(m.group(1))
-				tank['name'] = m.group(2)
-				tank['tier'] = int(m.group(3))
-				tank['type'] = tank_type[int(m.group(4))]
-				tank['is_premium'] = (int(m.group(5)) == 1)
-				tanks[str(m.group(1))] = tank
-				n += 1
-
-			tankopedia = {}
-			tankopedia['status'] = "ok"
-			tankopedia['meta'] = {"count" : n}
-			tankopedia['data'] = tanks
-			print("Tankopedia has " + str(n) + " tanks in: " + filename)
-			with open(filename,'w') as outfile:
-				outfile.write(json.dumps(tankopedia, ensure_ascii=False, indent=4, sort_keys=True))
-			return None
-
-
-async def main(argv):
-	# set the directory for the script
-	os.chdir(os.path.dirname(sys.argv[0]))
-
-	parser = argparse.ArgumentParser(description='Retrieve Tankopedia from WoTinspector.com')
-	parser.add_argument('--file', dest="outfile", help='Write Tankopedia to file')
-	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')
-	parser.add_argument('-s', '--silent', action='store_true', default=False, help='Silent mode')
-
-	args = parser.parse_args()
-	bu.set_log_level(args.silent, args.verbose, args.debug)
-
-	wi = WoTinspector()	
-	await wi.get_tankopedia(args.outfile)
-
-	await wi.close()
-    
-
-# main()
-if __name__ == "__main__":
-	# To avoid 'Event loop is closed' RuntimeError due to compatibility issue with aiohttp
-	if sys.platform.startswith("win") and sys.version_info >= (3, 8):
-		try:
-			from asyncio import WindowsSelectorEventLoopPolicy
-		except ImportError:
-			pass
-		else:
-			if not isinstance(asyncio.get_event_loop_policy(), WindowsSelectorEventLoopPolicy):
-				asyncio.set_event_loop_policy(WindowsSelectorEventLoopPolicy())
-	asyncio.run(main(sys.argv[1:]), debug=False)