Stream XML data to a temp file with retries before decoding it as XML (…

…HearthSim#37)

* Stream XML data to a temp file with retries before decoding it as XML

* Fix locale multiplexing for load_globalstrings

* Add test coverage for new XML download functions

hearthstone/bountyxml.py

@@ -1,7 +1,9 @@
-from typing import Any, Dict, Tuple
+import tempfile
+from typing import Any, Dict, Optional, Tuple
 
 from .enums import Role
 from .utils import ElementTree
+from .xmlutils import download_to_tempfile_retry
 
 
 class BountyXML:
@@ -57,22 +59,44 @@ def region_name(self):
 bounty_cache: Dict[Tuple[str, str], Tuple[Dict[int, BountyXML], Any]] = {}
 
 
+XML_URL = "https://api.hearthstonejson.com/v1/latest/BountyDefs.xml"
+
+
+def _bootstrap_from_web() -> Optional[ElementTree.ElementTree]:
+	with tempfile.TemporaryFile(mode="rb+") as fp:
+		if download_to_tempfile_retry(XML_URL, fp):
+			fp.flush()
+			fp.seek(0)
+
+			return ElementTree.parse(fp)
+		else:
+			return None
+
+
+def _bootstrap_from_library(path=None) -> ElementTree.ElementTree:
+	from hearthstone_data import get_bountydefs_path
+
+	if path is None:
+		path = get_bountydefs_path()
+
+	with open(path, "rb") as f:
+		return ElementTree.parse(f)
+
+
 def load(path=None, locale="enUS"):
 	cache_key = (path, locale)
 	if cache_key not in bounty_cache:
-		from hearthstone_data import get_bountydefs_path
+		xml = _bootstrap_from_web()
 
-		if path is None:
-			path = get_bountydefs_path()
+		if not xml:
+			xml = _bootstrap_from_library(path=path)
 
 		db = {}
 
-		with open(path, "rb") as f:
-			xml = ElementTree.parse(f)
-			for bountydata in xml.findall("Bounty"):
-				bounty = BountyXML.from_xml(bountydata)
-				bounty.locale = locale
-				db[bounty.id] = bounty
+		for bountydata in xml.findall("Bounty"):
+			bounty = BountyXML.from_xml(bountydata)
+			bounty.locale = locale
+			db[bounty.id] = bounty
 
 		bounty_cache[cache_key] = (db, xml)
 

hearthstone/cardxml.py

@@ -1,8 +1,12 @@
+import tempfile
+from typing import Optional
+
 from .enums import (
 	CardClass, CardSet, CardType, Faction, GameTag,
 	MultiClassGroup, PlayReq, Race, Rarity, Role, SpellSchool
 )
 from .utils import ElementTree
+from .xmlutils import download_to_tempfile_retry
 
 
 LOCALIZED_TAGS = [
@@ -386,22 +390,44 @@ def classes(self):
 dbf_cache: dict = {}
 
 
+XML_URL = "https://api.hearthstonejson.com/v1/latest/CardDefs.xml"
+
+
+def _bootstrap_from_web() -> Optional[ElementTree.ElementTree]:
+	with tempfile.TemporaryFile(mode="rb+") as fp:
+		if download_to_tempfile_retry(XML_URL, fp):
+			fp.flush()
+			fp.seek(0)
+
+			return ElementTree.parse(fp)
+		else:
+			return None
+
+
+def _bootstrap_from_library(path=None) -> ElementTree.ElementTree:
+	from hearthstone_data import get_carddefs_path
+
+	if path is None:
+		path = get_carddefs_path()
+
+	with open(path, "rb") as f:
+		return ElementTree.parse(f)
+
+
 def _load(path, locale, cache, attr):
 	cache_key = (path, locale)
 	if cache_key not in cache:
-		from hearthstone_data import get_carddefs_path
+		xml = _bootstrap_from_web()
 
-		if path is None:
-			path = get_carddefs_path()
+		if not xml:
+			xml = _bootstrap_from_library(path=path)
 
 		db = {}
 
-		with open(path, "rb") as f:
-			xml = ElementTree.parse(f)
-			for carddata in xml.findall("Entity"):
-				card = CardXML.from_xml(carddata)
-				card.locale = locale
-				db[getattr(card, attr)] = card
+		for carddata in xml.findall("Entity"):
+			card = CardXML.from_xml(carddata)
+			card.locale = locale
+			db[getattr(card, attr)] = card
 
 		cache[cache_key] = (db, xml)
 

hearthstone/mercenaryxml.py

@@ -1,8 +1,10 @@
-from typing import Any, Dict, Tuple
+import tempfile
+from typing import Any, Dict, Optional, Tuple
 
 from hearthstone.enums import Rarity
 
 from .utils import ElementTree
+from .xmlutils import download_to_tempfile_retry
 
 
 class MercenaryXML:
@@ -186,22 +188,44 @@ def to_xml(self):
 mercenary_cache: Dict[Tuple[str, str], Tuple[Dict[int, MercenaryXML], Any]] = {}
 
 
+XML_URL = "https://api.hearthstonejson.com/v1/latest/MercenaryDefs.xml"
+
+
+def _bootstrap_from_web() -> Optional[ElementTree.ElementTree]:
+	with tempfile.TemporaryFile(mode="rb+") as fp:
+		if download_to_tempfile_retry(XML_URL, fp):
+			fp.flush()
+			fp.seek(0)
+
+			return ElementTree.parse(fp)
+		else:
+			return None
+
+
+def _bootstrap_from_library(path=None) -> ElementTree.ElementTree:
+	from hearthstone_data import get_mercenarydefs_path
+
+	if path is None:
+		path = get_mercenarydefs_path()
+
+	with open(path, "rb") as f:
+		return ElementTree.parse(f)
+
+
 def load(path=None, locale="enUS"):
 	cache_key = (path, locale)
 	if cache_key not in mercenary_cache:
-		from hearthstone_data import get_mercenarydefs_path
+		xml = _bootstrap_from_web()
 
-		if path is None:
-			path = get_mercenarydefs_path()
+		# if not xml:
+		# 	xml = _bootstrap_from_library(path=path)
 
 		db = {}
 
-		with open(path, "rb") as f:
-			xml = ElementTree.parse(f)
-			for mercenarydata in xml.findall("Mercenary"):
-				bounty = MercenaryXML.from_xml(mercenarydata)
-				bounty.locale = locale
-				db[bounty.id] = bounty
+		for mercenarydata in xml.findall("Mercenary"):
+			bounty = MercenaryXML.from_xml(mercenarydata)
+			bounty.locale = locale
+			db[bounty.id] = bounty
 
 		mercenary_cache[cache_key] = (db, xml)
 

hearthstone/stringsfile.py

@@ -5,18 +5,26 @@
 Key is always `TAG`
 """
 import csv
-from typing import Dict
+import json
+import sys
+import tempfile
+from typing import Dict, Optional, Tuple
 
-import hearthstone_data
+from hearthstone.xmlutils import download_to_tempfile_retry
 
 
 StringsRow = Dict[str, str]
 StringsDict = Dict[str, StringsRow]
 
-_cache: Dict[str, StringsDict] = {}
+_cache: Dict[Tuple[str, str], StringsDict] = {}
 
 
-def load(fp) -> StringsDict:
+def load_json(fp) -> StringsDict:
+	hsjson_strings = json.loads(fp.read())
+	return {k: {"TEXT": v} for k, v in hsjson_strings.items()}
+
+
+def load_txt(fp) -> StringsDict:
 	reader = csv.DictReader(
 		filter(lambda row: row.strip() and not row.startswith("#"), fp),
 		delimiter="\t"
@@ -28,19 +36,40 @@ def load(fp) -> StringsDict:
 	}
 
 
+def _load_globalstrings_from_web(locale="enUS") -> Optional[StringsDict]:
+	with tempfile.TemporaryFile() as fp:
+		json_url = "https://api.hearthstonejson.com/v1/strings/%s/GLOBAL.json" % locale
+		if download_to_tempfile_retry(json_url, fp):
+			fp.flush()
+			fp.seek(0)
+
+			return load_json(fp)
+		else:
+			return None
+
+
+def _load_globalstrings_from_library(locale="enUS") -> StringsDict:
+	from hearthstone_data import get_strings_file
+
+	path: str = get_strings_file(locale, filename="GLOBAL.txt")
+	with open(path, "r", encoding="utf-8-sig") as f:
+		return load_txt(f)
+
+
 def load_globalstrings(locale="enUS") -> StringsDict:
-	path: str = hearthstone_data.get_strings_file(locale, filename="GLOBAL.txt")
-	if path not in _cache:
-		with open(path, "r", encoding="utf-8-sig") as f:
-			_cache[path] = load(f)
+	key = (locale, "GLOBAL.txt")
+	if key not in _cache:
+		sd = _load_globalstrings_from_web(locale=locale)
 
-	return _cache[path]
+		if not sd:
+			sd = _load_globalstrings_from_library(locale=locale)
 
+		_cache[key] = sd
 
-if __name__ == "__main__":
-	import json
-	import sys
+	return _cache[key]
 
+
+if __name__ == "__main__":
 	for path in sys.argv[1:]:
 		with open(path, "r") as f:
-			print(json.dumps(load(f)))
+			print(json.dumps(load_txt(f)))

hearthstone/xmlutils.py

@@ -0,0 +1,33 @@
+import requests
+
+
+class RetryException(Exception):
+	pass
+
+
+def download_to_tempfile(url: str, fp) -> bool:
+	try:
+		with requests.get(url, stream=True) as r:
+			if r.ok:
+				for chunk in r.iter_content(chunk_size=8192):
+					fp.write(chunk)
+
+				return True
+			elif 500 <= r.status_code < 600:
+				raise RetryException()
+			else:
+				return False
+	except requests.exceptions.RequestException:
+		raise RetryException()
+
+
+def download_to_tempfile_retry(url: str, fp, retries: int = 3) -> bool:
+	assert retries >= 0
+
+	try:
+		return download_to_tempfile(url, fp)
+	except RetryException:
+		if retries:
+			return download_to_tempfile_retry(url, fp, retries - 1)
+
+		return False

setup.cfg

@@ -20,7 +20,7 @@ packages = find:
 include_package_data = True
 zip_safe = True
 install_requires =
-	hearthstone_data
+	requests
 
 [options.packages.find]
 exclude =

tests/test_enums.py

@@ -1,6 +1,7 @@
 from datetime import datetime
 
 from hearthstone import enums
+from hearthstone.enums import CardClass, Locale, get_localized_name
 
 
 def test_zodiac_dates():
@@ -74,3 +75,27 @@ def test_card_classes(self):
 			enums.CardClass.PALADIN,
 		]
 		assert enums.MultiClassGroup.INVALID.card_classes == []
+
+
+def test_get_localized_name():
+	d = {
+		locale.name: get_localized_name(CardClass.DRUID, locale.name) for locale in Locale
+		if not locale.unused
+	}
+
+	assert d == {
+		"deDE": "Druide",
+		"enUS": "Druid",
+		"esES": "Druida",
+		"esMX": "Druida",
+		"frFR": "Druide",
+		"itIT": "Druido",
+		"jaJP": "ドルイド",
+		"koKR": "드루이드",
+		"plPL": "Druid",
+		"ptBR": "Druida",
+		"ruRU": "Друид",
+		"thTH": "ดรูอิด",
+		"zhCN": "德鲁伊",
+		"zhTW": "德魯伊"
+	}

tests/test_stringsfile.py

@@ -1,6 +1,6 @@
 from io import StringIO
 
-from hearthstone.stringsfile import load
+from hearthstone.stringsfile import load_txt
 
 
 TEST_STRINGS = """TAG	TEXT	COMMENT	AUDIOFILE
@@ -12,7 +12,7 @@
 
 
 def test_load_blank_line():
-	assert load(StringIO(TEST_STRINGS)) == {
+	assert load_txt(StringIO(TEST_STRINGS)) == {
 		"VO_ICC09_Saurfang_Male_Orc_CursedBlade_01": {
 			"TEXT": "Who’s idea was this?"
 		},

tox.ini

@@ -7,6 +7,9 @@ setenv =
 commands = pytest --showlocals {posargs}
 deps =
 	pytest
+	pytest-mock
+	requests-mock
+
 
 [testenv:flake8]
 skip_install = True