Finish asset gui logic and argument parse module.
This commit is contained in:
@@ -0,0 +1,151 @@
|
||||
import logging
|
||||
import shlex
|
||||
|
||||
from core_lib.game.library import get_natives_platform_info
|
||||
|
||||
logger = logging.getLogger("Launcher.CoreLib")
|
||||
|
||||
def is_argument_allowed(argument: dict | str, allow_features: dict=dict) -> bool:
|
||||
"""
|
||||
Check if argument is allowed or not.
|
||||
:param argument:
|
||||
:param allow_features: Should be like {key: value}
|
||||
:return:
|
||||
allowed: bool
|
||||
"""
|
||||
if isinstance(argument, str):
|
||||
return True
|
||||
|
||||
rules = argument.get('rules', {})
|
||||
|
||||
_, platform_rule_name, _ = get_natives_platform_info()
|
||||
|
||||
allowed = False
|
||||
|
||||
if not rules:
|
||||
return True
|
||||
|
||||
for rule in rules:
|
||||
rule_action = rule.get("action", None)
|
||||
os_section = rule.get("os", {})
|
||||
rule_os = os_section.get("name", None)
|
||||
features_rule = rule.get("features", None)
|
||||
|
||||
if rule_os is not None and platform_rule_name != rule_os:
|
||||
continue
|
||||
|
||||
matched = True
|
||||
|
||||
# Check feature dict
|
||||
if features_rule:
|
||||
for key, expected in features_rule.items():
|
||||
if allow_features.get(key, False) != expected:
|
||||
matched = False
|
||||
break
|
||||
|
||||
if not matched:
|
||||
continue
|
||||
|
||||
|
||||
if rule_action == "allow":
|
||||
allowed = True
|
||||
elif rule_action == "disallow":
|
||||
allowed = False
|
||||
else:
|
||||
logger.warning(f"Unknown action {rule_action}\n")
|
||||
|
||||
return allowed
|
||||
|
||||
def replace_placeholders(argument, placeholders: dict) -> str:
|
||||
if argument:
|
||||
for key, value in placeholders.items():
|
||||
argument = argument.replace("${" + key + "}", "" if value is None else str(value))
|
||||
|
||||
return argument
|
||||
|
||||
class ArgumentMappings:
|
||||
def __init__(self, player_name="Player", version_name="unknown",
|
||||
game_directory=None, assets_root=None, assets_index_name=None,
|
||||
auth_uuid="00000001-0002-0003-0004-000000000005",
|
||||
auth_access_token="", clientid=None, auth_xuid=None,
|
||||
user_type=None, version_type=None, natives_directory=None,
|
||||
launcher_name=None, launcher_version=None,
|
||||
classpath=None, classpath_separator=None, library_directory=None):
|
||||
self.player_name = player_name
|
||||
self.version_name = version_name
|
||||
self.game_directory = game_directory
|
||||
self.assets_root = assets_root
|
||||
self.assets_index_name = assets_index_name
|
||||
self.auth_uuid = auth_uuid
|
||||
self.auth_access_token = auth_access_token
|
||||
self.clientid = clientid
|
||||
self.auth_xuid = auth_xuid
|
||||
self.user_type = user_type
|
||||
self.version_type = version_type
|
||||
self.natives_directory = natives_directory
|
||||
self.launcher_name = launcher_name
|
||||
self.launcher_version = launcher_version
|
||||
self.classpath = classpath
|
||||
self.classpath_separator = classpath_separator
|
||||
self.library_directory = library_directory
|
||||
|
||||
def generate_placeholders(self):
|
||||
return {
|
||||
"player_name": self.player_name,
|
||||
"version_name": self.version_name,
|
||||
"game_directory": self.game_directory,
|
||||
"assets_root": self.assets_root,
|
||||
"assets_index_name": self.assets_index_name,
|
||||
"auth_uuid": self.auth_uuid,
|
||||
"auth_access_token": self.auth_access_token,
|
||||
"auth_player_name": self.player_name,
|
||||
"clientid": self.clientid,
|
||||
"auth_xuid": self.auth_xuid,
|
||||
"user_type": self.user_type,
|
||||
"version_type": self.version_type,
|
||||
"natives_directory": self.natives_directory,
|
||||
"launcher_name": self.launcher_name,
|
||||
"launcher_version": self.launcher_version,
|
||||
"classpath": self.classpath,
|
||||
"classpath_separator": self.classpath_separator,
|
||||
"library_directory": self.library_directory
|
||||
}
|
||||
|
||||
def convert_argument_to_multiple_string(argument: str | dict) -> list[str]:
|
||||
if isinstance(argument, str):
|
||||
return [argument]
|
||||
|
||||
if isinstance(argument, dict):
|
||||
value = argument.get("value", None)
|
||||
|
||||
if isinstance(value, list):
|
||||
return [str(item) for item in value] # ensure all item are string
|
||||
elif isinstance(value, str):
|
||||
return [value]
|
||||
|
||||
raise TypeError("Argument must be of type str or dict")
|
||||
|
||||
def parse_and_generate_arguments(arguments: list | str, mappings: ArgumentMappings, features: dict=dict) -> list[str]:
|
||||
parsed_arguments = []
|
||||
|
||||
if isinstance(arguments, str):
|
||||
# Convert legacy argument into a list type
|
||||
arguments = shlex.split(arguments)
|
||||
|
||||
placeholders = mappings.generate_placeholders()
|
||||
|
||||
for argument in arguments:
|
||||
# Check rules
|
||||
if not is_argument_allowed(argument, allow_features=features):
|
||||
continue
|
||||
|
||||
# Each argument (section) may contain multiple items, convert it to list before replace placeholder
|
||||
converted = convert_argument_to_multiple_string(argument)
|
||||
|
||||
for item in converted:
|
||||
parsed = replace_placeholders(item, placeholders)
|
||||
parsed_arguments.append(parsed)
|
||||
|
||||
return parsed_arguments
|
||||
|
||||
|
||||
+12
-2
@@ -13,6 +13,8 @@ from core_lib.game.version_info import find_useful_part_in_specific_version_mani
|
||||
|
||||
MOJANG_RESOURCE_ENDPOINT = "https://resources.download.minecraft.net/{beginning}/{hash}"
|
||||
|
||||
ASSET_DOWNLOAD_MAX_WORKERS = 20
|
||||
|
||||
logger = logging.getLogger("Launcher.CoreLib")
|
||||
|
||||
def fetch_asset_manifest(target_version_manifest: dict, raw=False) -> tuple[dict | bytes, str]:
|
||||
@@ -84,7 +86,7 @@ def fetch_and_save_asset_manifest(target_version_manifest: dict, destination: Pa
|
||||
with file.path.open("wb") as f:
|
||||
f.write(data)
|
||||
|
||||
if sha1 is not None and not file.verify_sha1(sha1):
|
||||
if sha1 and not file.verify_sha1(sha1):
|
||||
raise HashMismatchException(
|
||||
"sha1",
|
||||
file.sha1,
|
||||
@@ -94,6 +96,7 @@ def fetch_and_save_asset_manifest(target_version_manifest: dict, destination: Pa
|
||||
|
||||
return json.loads(data.decode("utf-8"))
|
||||
except Exception as e:
|
||||
logger.exception(e)
|
||||
raise AssetManifestSaveException(
|
||||
"Error saving version manifest:\n{}".format(e)
|
||||
)
|
||||
@@ -164,7 +167,14 @@ def download_assets_and_copy(assets: list[FileObject], legacy_assets: list[dict]
|
||||
files: list[FileObject] (files that download failed)
|
||||
"""
|
||||
|
||||
fails , _ = download_multiple_files(assets, progress_callback=progress_callback)
|
||||
fails , _ = download_multiple_files(assets, progress_callback=progress_callback,
|
||||
max_workers=ASSET_DOWNLOAD_MAX_WORKERS)
|
||||
|
||||
if fails:
|
||||
logger.warning(
|
||||
"Download failed for assets: \n"
|
||||
+"\n".join([asset.posix_path for asset in fails])
|
||||
)
|
||||
|
||||
for asset in legacy_assets:
|
||||
target = asset.get("target")
|
||||
|
||||
@@ -413,6 +413,7 @@ def fetch_and_save_version_manifest(dest: Path, target_version: str=None, root_m
|
||||
|
||||
return data
|
||||
except Exception as e:
|
||||
logger.exception(e)
|
||||
raise VersionManifestSaveException(
|
||||
"Error saving version manifest:\n{}".format(e)
|
||||
)
|
||||
Reference in New Issue
Block a user