Add download progress dialog and some mod loader parse function.
This commit is contained in:
@@ -180,6 +180,7 @@ def download_file(file: FileObject, overwrite=True, progress_callback: Callable[
|
||||
|
||||
try:
|
||||
with requests.get(file.url, stream=True) as r:
|
||||
r.raise_for_status()
|
||||
current_chunk = 0
|
||||
total_bytes = r.headers.get("content-length", None)
|
||||
logger.debug("URL: {}".format(file.url))
|
||||
@@ -211,7 +212,7 @@ def download_file(file: FileObject, overwrite=True, progress_callback: Callable[
|
||||
if any([file.expected_sha1, file.expected_md5, file.expected_sha256]):
|
||||
logger.debug("File hash is match as the expected hash.")
|
||||
else:
|
||||
logger.warning("File hash is not checked.")
|
||||
logger.warning("Hash is not checked for file: {}".format(file.posix_path))
|
||||
|
||||
return file
|
||||
|
||||
@@ -299,3 +300,8 @@ def get_platform_info():
|
||||
|
||||
return platform_name, arch, os_version
|
||||
|
||||
def is_valid_zipfile(path: Path) -> bool:
|
||||
try:
|
||||
return zipfile.is_zipfile(path)
|
||||
except OSError:
|
||||
return False
|
||||
+91
-40
@@ -1,12 +1,15 @@
|
||||
import json
|
||||
import logging
|
||||
import re
|
||||
import shlex
|
||||
|
||||
from core_lib.common.common import get_platform_info
|
||||
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 | None=None) -> bool:
|
||||
|
||||
def is_argument_allowed(argument: dict | str, allow_features: dict | None = None) -> bool:
|
||||
"""
|
||||
Check if argument is allowed or not.
|
||||
:param argument:
|
||||
@@ -22,7 +25,8 @@ def is_argument_allowed(argument: dict | str, allow_features: dict | None=None)
|
||||
|
||||
rules = argument.get('rules', {})
|
||||
|
||||
_, platform_rule_name, _ = get_natives_platform_info()
|
||||
_, platform_rule_name, platform_arch_type = get_natives_platform_info()
|
||||
_, _, platform_version = get_platform_info()
|
||||
|
||||
allowed = False
|
||||
|
||||
@@ -30,33 +34,52 @@ def is_argument_allowed(argument: dict | str, allow_features: dict | None=None)
|
||||
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 not isinstance(rule, dict):
|
||||
logger.warning("Ignoring invalid argument rule: {}".format(rule))
|
||||
continue
|
||||
|
||||
rule_action = rule.get("action", None)
|
||||
os_rule = rule.get("os") or {}
|
||||
rule_os = os_rule.get("name", None)
|
||||
feature_rule = rule.get("features") or {}
|
||||
rule_arch = os_rule.get("arch")
|
||||
rule_version = os_rule.get("version")
|
||||
|
||||
# Check platform
|
||||
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:
|
||||
# Check architecture
|
||||
if rule_arch is not None and rule_arch != platform_arch_type:
|
||||
continue
|
||||
|
||||
# Check platform version
|
||||
if rule_version is not None:
|
||||
try:
|
||||
if re.search(
|
||||
rule_version,
|
||||
str(platform_version),
|
||||
) is None:
|
||||
continue
|
||||
except re.error as exc:
|
||||
logger.warning(
|
||||
"Unsupported argument OS version regex %r: %s",
|
||||
rule_version,
|
||||
exc,
|
||||
)
|
||||
continue
|
||||
|
||||
|
||||
# Check feature rule
|
||||
if any(allow_features.get(key, False) != expected for key, expected in feature_rule.items()):
|
||||
continue
|
||||
|
||||
if rule_action == "allow":
|
||||
allowed = True
|
||||
elif rule_action == "disallow":
|
||||
allowed = False
|
||||
else:
|
||||
logger.warning(f"Unknown action {rule_action}\n")
|
||||
logger.warning(f"Unknown argument rule action {rule_action}\n")
|
||||
|
||||
return allowed
|
||||
|
||||
@@ -67,6 +90,7 @@ def replace_placeholders(argument, placeholders: dict) -> str:
|
||||
|
||||
return argument
|
||||
|
||||
|
||||
class ArgumentMappings:
|
||||
def __init__(self, player_name="Player", version_name="unknown",
|
||||
game_directory=None, assets_root=None, legacy_assets_root=None, assets_index_name=None,
|
||||
@@ -87,8 +111,15 @@ class ArgumentMappings:
|
||||
self.auth_xuid = auth_xuid
|
||||
self.user_type = user_type
|
||||
self.user_properties = user_properties
|
||||
if not isinstance(user_properties, dict):
|
||||
self.user_properties = {}
|
||||
if isinstance(user_properties, dict):
|
||||
self.user_properties = json.dumps(
|
||||
user_properties,
|
||||
separators=(",", ":"),
|
||||
)
|
||||
else:
|
||||
if user_properties is not None:
|
||||
logger.warning(f"User properties should be a dictionary.")
|
||||
self.user_properties = "{}"
|
||||
self.version_type = version_type
|
||||
self.natives_directory = natives_directory
|
||||
self.launcher_name = launcher_name
|
||||
@@ -122,6 +153,7 @@ class ArgumentMappings:
|
||||
"library_directory": self.library_directory
|
||||
}
|
||||
|
||||
|
||||
def convert_argument_to_multiple_string(argument: str | dict) -> list[str]:
|
||||
if isinstance(argument, str):
|
||||
return [argument]
|
||||
@@ -130,12 +162,13 @@ def convert_argument_to_multiple_string(argument: str | dict) -> list[str]:
|
||||
value = argument.get("value", None)
|
||||
|
||||
if isinstance(value, list):
|
||||
return [str(item) for item in value] # ensure all item are string
|
||||
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 find_game_and_jvm_args(arguments: dict | str) -> tuple[list, list]:
|
||||
game_args = []
|
||||
jvm_args = []
|
||||
@@ -160,11 +193,13 @@ def find_game_and_jvm_args(arguments: dict | str) -> tuple[list, list]:
|
||||
|
||||
return game_args, jvm_args
|
||||
|
||||
|
||||
REMAPPING_ARGUMENTS = {
|
||||
# For some (I don't know) reason, in 26.2, Mojang include some
|
||||
"-Djava.library.path=${natives_directory}/java": "-Djava.library.path=${natives_directory}"
|
||||
}
|
||||
|
||||
|
||||
def hard_remapping_special_arguments(argument: str) -> str:
|
||||
for arg in REMAPPING_ARGUMENTS:
|
||||
new_value = REMAPPING_ARGUMENTS[arg]
|
||||
@@ -174,7 +209,9 @@ def hard_remapping_special_arguments(argument: str) -> str:
|
||||
|
||||
return argument
|
||||
|
||||
def parse_and_generate_arguments(arguments: list | str, mappings: ArgumentMappings, features: dict | None=None) -> list[str]:
|
||||
|
||||
def parse_and_generate_arguments(arguments: list | str, mappings: ArgumentMappings, features: dict | None = None) -> \
|
||||
list[str]:
|
||||
if features is None:
|
||||
features = {}
|
||||
|
||||
@@ -204,38 +241,59 @@ def parse_and_generate_arguments(arguments: list | str, mappings: ArgumentMappin
|
||||
return parsed_arguments
|
||||
|
||||
|
||||
def append_missing_necessary_jvm_arguments(arguments: list, arg_maps: ArgumentMappings) -> list[str]:
|
||||
result = []
|
||||
has_classpath = any(
|
||||
argument in ("-cp", "-classpath", "--class-path")
|
||||
for argument in arguments
|
||||
)
|
||||
|
||||
has_native_path = any(
|
||||
argument.startswith("-Djava.library.path=")
|
||||
for argument in arguments
|
||||
)
|
||||
|
||||
if not has_native_path:
|
||||
result.append(f"-Djava.library.path={arg_maps.natives_directory}")
|
||||
|
||||
if not has_classpath:
|
||||
result.extend(["-cp", arg_maps.classpath])
|
||||
|
||||
result.extend(arguments)
|
||||
|
||||
return result
|
||||
|
||||
|
||||
def generate_launch_command(arg_maps: ArgumentMappings, main_class: str, java_path: str, full_args: dict | str
|
||||
) -> list[str]:
|
||||
, features: dict | None = None) -> list[str]:
|
||||
cmd = [java_path]
|
||||
game_args, jvm_args = find_game_and_jvm_args(full_args)
|
||||
parsed_jvm_args = parse_and_generate_arguments(
|
||||
jvm_args,
|
||||
arg_maps
|
||||
arg_maps,
|
||||
features=features
|
||||
)
|
||||
parsed_game_args = parse_and_generate_arguments(
|
||||
game_args,
|
||||
arg_maps
|
||||
arg_maps,
|
||||
features=features
|
||||
)
|
||||
|
||||
# Legacy Minecraft don't have jvm arguments exist in the version library. We need to generate by ourselves
|
||||
if len(parsed_jvm_args) == 0:
|
||||
cmd.extend([
|
||||
f"-Djava.library.path={arg_maps.natives_directory}",
|
||||
"-cp",
|
||||
arg_maps.classpath,
|
||||
])
|
||||
else:
|
||||
cmd.extend(parsed_jvm_args)
|
||||
parsed_jvm_args = append_missing_necessary_jvm_arguments(parsed_jvm_args, arg_maps)
|
||||
|
||||
cmd.extend(parsed_jvm_args)
|
||||
cmd.append(main_class)
|
||||
cmd.extend(parsed_game_args)
|
||||
|
||||
return cmd
|
||||
|
||||
|
||||
def is_a_argument(arg: str, startswith="--"):
|
||||
pattern = re.compile(rf"{startswith}(.+)")
|
||||
return pattern.match(arg) is not None
|
||||
|
||||
|
||||
def generate_argument_details(arguments: list[str | dict], startswith="--") -> list[dict]:
|
||||
items = []
|
||||
|
||||
@@ -271,13 +329,6 @@ def generate_argument_details(arguments: list[str | dict], startswith="--") -> l
|
||||
|
||||
return items
|
||||
|
||||
|
||||
def is_same_argument(item: dict, another: dict) -> bool:
|
||||
pass
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
+2
-553
@@ -1,22 +1,17 @@
|
||||
import json
|
||||
import logging
|
||||
import re
|
||||
import zipfile
|
||||
from pathlib import Path
|
||||
from typing import Callable
|
||||
from urllib.parse import urlsplit, unquote
|
||||
|
||||
from ..common.common import get_platform_info, FileObject, download_multiple_files, unzip
|
||||
from ..common.common import get_platform_info
|
||||
from .version_info import (
|
||||
MOJANG_LIBRARIES_ENDPOINT,
|
||||
NATIVE_OS_KEY_MAP,
|
||||
NATIVE_OS_RULE_MAP,
|
||||
NATIVE_ARCH_KEY_MAP,
|
||||
NATIVE_NEW_TO_OLD_ARCH,
|
||||
NATIVE_OLD_TO_NEW_ARCH, find_useful_part_in_specific_version_manifest,
|
||||
NATIVE_OLD_TO_NEW_ARCH
|
||||
)
|
||||
from ..common.exception import UnsupportedPlatformException, NativeResolveException, DependencyException, \
|
||||
VersionDataKeyNotFoundException, DownloadException
|
||||
|
||||
logger = logging.getLogger("Launcher.CoreLib")
|
||||
|
||||
@@ -32,9 +27,6 @@ def parse_maven_coordinate(value: str, default_extension="jar"):
|
||||
group, artifact, version = parts[:3]
|
||||
classifier = parts[3] if len(parts) == 4 else None
|
||||
|
||||
if version.count("-") > 0:
|
||||
version, extra = version.split("-", 1)
|
||||
|
||||
if not group or not artifact or not version:
|
||||
raise ValueError(f"Coordinate key can not be empty: {value!r}")
|
||||
|
||||
@@ -45,7 +37,6 @@ def parse_maven_coordinate(value: str, default_extension="jar"):
|
||||
|
||||
def convert_maven_version(version: str, failback=None) -> tuple[int, int, int]:
|
||||
try:
|
||||
print(version)
|
||||
items = version.split(".")
|
||||
if len(items) >= 3:
|
||||
major, minor, patch = items[0], items[1], items[2]
|
||||
@@ -369,545 +360,3 @@ def generate_formatted_artifact_with_name(lib_name, lib_artifact, libraries_endp
|
||||
artifact_new.update(extra)
|
||||
|
||||
return artifact_new
|
||||
|
||||
|
||||
def collect_library_artifacts(specific_version_manifest):
|
||||
_, _, _, _, libraries, _, _ = find_useful_part_in_specific_version_manifest(specific_version_manifest)
|
||||
|
||||
items = []
|
||||
|
||||
# Same as natives
|
||||
_, _, os_version = get_platform_info()
|
||||
platform_key_name, platform_rule_name, platform_arch_type = get_natives_platform_info()
|
||||
|
||||
for library in libraries:
|
||||
name = library.get("name", "")
|
||||
artifact_raw = library.get("downloads", {}).get("artifact", {})
|
||||
rules = library.get("rules", [])
|
||||
classifiers = library.get("downloads", {}).get("classifiers", {})
|
||||
natives = library.get("natives", {})
|
||||
|
||||
# Ignore natives
|
||||
if (len(natives) > 0 or bool(re.search(r":natives-[^:]+$", name)) or
|
||||
any(native.startswith("natives-") for native in classifiers)):
|
||||
logger.info(f"Skipping {name} because its a native library.")
|
||||
continue
|
||||
|
||||
if not is_library_allowed(
|
||||
rules,
|
||||
platform_rule_name,
|
||||
platform_arch_type,
|
||||
os_version,
|
||||
):
|
||||
logger.debug(f"Library {name} can't pass library rules.")
|
||||
continue
|
||||
|
||||
artifact = generate_formatted_artifact_with_name(
|
||||
name,
|
||||
artifact_raw
|
||||
)
|
||||
|
||||
if artifact:
|
||||
items.append(artifact)
|
||||
|
||||
if not items:
|
||||
raise DependencyException(
|
||||
"Could not find any artifact in any of the libraries."
|
||||
)
|
||||
|
||||
return items
|
||||
|
||||
|
||||
def collect_native_artifacts(specific_version_manifest):
|
||||
"""
|
||||
Collects native artifacts that support current platform
|
||||
:param specific_version_manifest:
|
||||
:return:
|
||||
filtered_natives_artifacts: list[dict]
|
||||
"""
|
||||
filtered = []
|
||||
|
||||
# Convert current platform and architecture type to same formula
|
||||
_, _, os_version = get_platform_info()
|
||||
platform_key_name, platform_rule_name, platform_arch_type = get_natives_platform_info()
|
||||
|
||||
if platform_key_name is None or platform_arch_type is None or platform_rule_name is None:
|
||||
raise UnsupportedPlatformException(
|
||||
f"Unsupported platform: key={platform_key_name}, "
|
||||
f"rule={platform_rule_name}, arch={platform_arch_type}"
|
||||
)
|
||||
|
||||
_, _, _, _, libraries, _, _ = find_useful_part_in_specific_version_manifest(specific_version_manifest)
|
||||
|
||||
for library in libraries:
|
||||
name = library.get("name", "")
|
||||
artifact = library.get("downloads", {}).get("artifact", {})
|
||||
# For newer Minecraft (VER>1.7.10)
|
||||
rules = library.get("rules", [])
|
||||
# For legacy version (VER<1.8)
|
||||
classifiers = library.get("downloads", {}).get("classifiers", {})
|
||||
natives = library.get("natives", {})
|
||||
exclude = library.get("extract", {}).get("exclude", ["META-INF/"]) # For extract use
|
||||
|
||||
is_natives = False
|
||||
|
||||
# Check if library is a native
|
||||
if (len(natives) > 0 or bool(re.search(r":natives-[^:]+$", name)) or
|
||||
any(native.startswith("natives-") for native in classifiers)):
|
||||
is_natives = True
|
||||
|
||||
if not is_natives:
|
||||
continue
|
||||
|
||||
if rules and classifiers:
|
||||
logger.debug(f"Found newer and legacy (classifier) native rules for {name}")
|
||||
elif rules:
|
||||
logger.debug(f"Found newer native rules for {name}")
|
||||
elif classifiers:
|
||||
logger.debug(f"Found legacy native rules (classifier) for {name}")
|
||||
|
||||
# =============== New rules ===============
|
||||
|
||||
# Parse action section
|
||||
if not is_library_allowed(
|
||||
rules,
|
||||
platform_rule_name,
|
||||
platform_arch_type,
|
||||
os_version,
|
||||
):
|
||||
logger.debug(f"Library {name} can't pass library rules.")
|
||||
continue
|
||||
|
||||
if is_native_allowed(
|
||||
name,
|
||||
platform_key_name,
|
||||
platform_arch_type,
|
||||
):
|
||||
result_artifact = generate_formatted_artifact_with_name(name, artifact, extra={"exclude": exclude,
|
||||
"flatten": True})
|
||||
if result_artifact:
|
||||
filtered.append(result_artifact)
|
||||
# For some reason, we still need to continue process because some version use multiple type of
|
||||
# classifier. (Such as 1.14.2)
|
||||
else:
|
||||
logger.debug(f"Library {name} can't pass native rules.")
|
||||
|
||||
# =============== Old rules ===============
|
||||
logger.debug(
|
||||
"Library {} doesn't have native key. This library may use old rule.".format(name)
|
||||
)
|
||||
|
||||
result, result_artifact, classifier = is_legacy_native_allowed(
|
||||
natives,
|
||||
classifiers,
|
||||
platform_rule_name,
|
||||
platform_arch_type,
|
||||
)
|
||||
|
||||
if not result:
|
||||
logger.debug(f"Library {name} can't pass legacy native rules.")
|
||||
continue
|
||||
else:
|
||||
result_artifact = generate_formatted_artifact_with_name(name,
|
||||
result_artifact,
|
||||
extra={"exclude": exclude, "flatten": True},
|
||||
relative_args={
|
||||
"extra": classifier
|
||||
})
|
||||
if result_artifact:
|
||||
filtered.append(result_artifact)
|
||||
|
||||
# Remove duplicate items
|
||||
items = remove_duplicate_artifacts(filtered)
|
||||
|
||||
if not filtered:
|
||||
raise NativeResolveException(
|
||||
"No compatible native artifact found.",
|
||||
user_message="No compatible native library was found for this platform.",
|
||||
)
|
||||
elif not items:
|
||||
raise NativeResolveException(
|
||||
"No compatible native artifact found. Report developer because this may be a issue"
|
||||
" at function \"remove_duplicate_artifacts\"",
|
||||
)
|
||||
|
||||
return items
|
||||
|
||||
|
||||
def download_libraries(artifacts, output_dir, progress_callback: Callable[[str, float], None] = None) \
|
||||
-> tuple[list[FileObject], list[FileObject]]:
|
||||
"""
|
||||
Download libraries from a list that contains artifacts
|
||||
:param artifacts:
|
||||
:param output_dir:
|
||||
:param progress_callback:
|
||||
:return:
|
||||
fails: list (List of artifacts that download failed)
|
||||
successes: list (artifacts that were downloaded)
|
||||
"""
|
||||
files = []
|
||||
successes = []
|
||||
|
||||
for artifact in artifacts:
|
||||
sha1 = artifact.get("sha1")
|
||||
relative_path = artifact.get("path")
|
||||
url = artifact.get("url")
|
||||
|
||||
if not relative_path or not url:
|
||||
logger.warning(
|
||||
"Artifact {} has no path or url. Skipping.".format(artifact["name"])
|
||||
)
|
||||
continue
|
||||
else:
|
||||
full_path = output_dir / Path(relative_path)
|
||||
|
||||
file = FileObject(
|
||||
full_path,
|
||||
url=url,
|
||||
sha1=sha1,
|
||||
)
|
||||
|
||||
if sha1 is None:
|
||||
logger.warning(f"File {full_path} has no SHA1 hash.")
|
||||
|
||||
if file.exists and (sha1 is None or file.verify("sha1", sha1)):
|
||||
logger.debug(f"File {full_path} exists.")
|
||||
successes.append(file)
|
||||
continue
|
||||
|
||||
files.append(file)
|
||||
|
||||
fails, finished = download_multiple_files(files, progress_callback=progress_callback)
|
||||
successes.extend(finished)
|
||||
|
||||
return fails, successes
|
||||
|
||||
|
||||
def unzip_natives(filtered, libraries_dir, destination: Path):
|
||||
"""
|
||||
Unzip native libraries from a list that contains artifacts
|
||||
:param filtered:
|
||||
:param libraries_dir:
|
||||
:param destination:
|
||||
:return:
|
||||
|
||||
IMPORTANT:
|
||||
Please put "exclude" key inside the artifact (from the version data). If not, some unnecessary file will be also
|
||||
included when unzipping the native library.
|
||||
"""
|
||||
destination.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
for result_artifact in filtered:
|
||||
exclude = result_artifact.get("exclude", [])
|
||||
flatten = result_artifact.get("flatten", False)
|
||||
relative_path = result_artifact.get("path")
|
||||
|
||||
if not relative_path:
|
||||
logger.warning(
|
||||
"Artifact {} has no relative path. Skipping.".format(result_artifact["name"])
|
||||
)
|
||||
continue
|
||||
else:
|
||||
full_path = libraries_dir / Path(relative_path)
|
||||
|
||||
unzip(full_path, destination, exclude=exclude, flatten=flatten)
|
||||
|
||||
logger.debug(f"Unzipped {full_path}.")
|
||||
|
||||
|
||||
def check_libraries_exists(manifest: dict, libraries_dir: Path):
|
||||
"""
|
||||
Check if libraries are present in the libraries directory
|
||||
:param manifest:
|
||||
:param libraries_dir:
|
||||
:return:
|
||||
not_found: list
|
||||
"""
|
||||
libraries = manifest.get("libraries", [])
|
||||
|
||||
if not libraries:
|
||||
raise VersionDataKeyNotFoundException(
|
||||
"Libraries not found in manifest.",
|
||||
)
|
||||
|
||||
not_found = []
|
||||
|
||||
# Same as natives
|
||||
_, _, os_version = get_platform_info()
|
||||
platform_key_name, platform_rule_name, platform_arch_type = get_natives_platform_info()
|
||||
|
||||
for library in libraries:
|
||||
name = library.get("name", "")
|
||||
artifact_raw = library.get("downloads", {}).get("artifact", {})
|
||||
rules = library.get("rules", [])
|
||||
classifiers = library.get("downloads", {}).get("classifiers", {})
|
||||
natives = library.get("natives", {})
|
||||
|
||||
# Ignore natives
|
||||
if (len(natives) > 0 or bool(re.search(r":natives-[^:]+$", name)) or
|
||||
any(native.startswith("natives-") for native in classifiers)):
|
||||
logger.info(f"Skipping {name} because its a native library.")
|
||||
continue
|
||||
|
||||
if not is_library_allowed(
|
||||
rules,
|
||||
platform_rule_name,
|
||||
platform_arch_type,
|
||||
os_version,
|
||||
):
|
||||
continue
|
||||
|
||||
artifact = generate_formatted_artifact_with_name(
|
||||
name,
|
||||
artifact_raw
|
||||
)
|
||||
|
||||
if not artifact:
|
||||
logger.warning(f"Artifact {name} are not valid. Skipping...")
|
||||
continue
|
||||
|
||||
relative_path = artifact.get("path")
|
||||
full_path = libraries_dir / Path(relative_path)
|
||||
|
||||
if not full_path.is_file() or not full_path.exists():
|
||||
logger.debug(f"Artifact {name} missing.")
|
||||
not_found.append(full_path)
|
||||
|
||||
return not_found
|
||||
|
||||
|
||||
def check_natives_exists(manifest: dict, libraries_dir: Path, natives_dir: Path):
|
||||
filtered = []
|
||||
|
||||
# Convert current platform and architecture type to same formula
|
||||
_, _, os_version = get_platform_info()
|
||||
platform_key_name, platform_rule_name, platform_arch_type = get_natives_platform_info()
|
||||
|
||||
if platform_key_name is None or platform_arch_type is None or platform_rule_name is None:
|
||||
raise UnsupportedPlatformException(
|
||||
f"Unsupported platform: key={platform_key_name}, "
|
||||
f"rule={platform_rule_name}, arch={platform_arch_type}"
|
||||
)
|
||||
|
||||
_, _, _, _, libraries, _, _ = find_useful_part_in_specific_version_manifest(manifest)
|
||||
|
||||
for library in libraries:
|
||||
name = library.get("name", "")
|
||||
artifact = library.get("downloads", {}).get("artifact", {})
|
||||
# For newer Minecraft (VER>1.7.10)
|
||||
rules = library.get("rules", [])
|
||||
# For legacy version (VER<1.8)
|
||||
classifiers = library.get("downloads", {}).get("classifiers", {})
|
||||
natives = library.get("natives", {})
|
||||
exclude = library.get("extract", {}).get("exclude", ["META-INF/"]) # For extract use
|
||||
|
||||
is_natives = False
|
||||
|
||||
# Check if library is a native
|
||||
if (len(natives) > 0 or bool(re.search(r":natives-[^:]+$", name)) or
|
||||
any(native.startswith("natives-") for native in classifiers)):
|
||||
is_natives = True
|
||||
|
||||
if not is_natives:
|
||||
continue
|
||||
|
||||
# =============== New rules ===============
|
||||
|
||||
# Parse action section
|
||||
if not is_library_allowed(
|
||||
rules,
|
||||
platform_rule_name,
|
||||
platform_arch_type,
|
||||
os_version,
|
||||
):
|
||||
continue
|
||||
|
||||
if is_native_allowed(
|
||||
name,
|
||||
platform_key_name,
|
||||
platform_arch_type,
|
||||
):
|
||||
result_artifact = generate_formatted_artifact_with_name(name, artifact, extra={"exclude": exclude,
|
||||
"flatten": True})
|
||||
if result_artifact:
|
||||
filtered.append(result_artifact)
|
||||
|
||||
# =============== Old rules ===============
|
||||
result, result_artifact, classifier = is_legacy_native_allowed(
|
||||
natives,
|
||||
classifiers,
|
||||
platform_rule_name,
|
||||
platform_arch_type,
|
||||
)
|
||||
|
||||
if not result:
|
||||
continue
|
||||
|
||||
result_artifact = generate_formatted_artifact_with_name(name,
|
||||
result_artifact,
|
||||
extra={"exclude": exclude, "flatten": True},
|
||||
relative_args={
|
||||
"extra": classifier
|
||||
})
|
||||
|
||||
if result_artifact:
|
||||
filtered.append(result_artifact)
|
||||
|
||||
# Remove duplicate items
|
||||
items = remove_duplicate_artifacts(filtered)
|
||||
|
||||
if not filtered:
|
||||
raise NativeResolveException(
|
||||
"No compatible native artifact found.",
|
||||
user_message="No compatible native library was found for this platform.",
|
||||
)
|
||||
elif not items:
|
||||
raise NativeResolveException(
|
||||
"No compatible native artifact found. Report developer because this may be a issue"
|
||||
" at function \"remove_duplicate_artifacts\"",
|
||||
)
|
||||
|
||||
expected = set()
|
||||
|
||||
for result_artifact in items:
|
||||
full_path = libraries_dir / Path(result_artifact["path"])
|
||||
exclude = result_artifact.get("exclude", []) or []
|
||||
flatten = result_artifact.get("flatten", False)
|
||||
|
||||
if not full_path.is_file() or not full_path.exists():
|
||||
return False
|
||||
|
||||
with zipfile.ZipFile(full_path) as zf:
|
||||
for member in zf.namelist():
|
||||
if member.endswith("/"):
|
||||
continue
|
||||
if any(member.startswith(item) for item in exclude):
|
||||
continue
|
||||
|
||||
target = Path(member).name if flatten else member
|
||||
if target:
|
||||
expected.add(target)
|
||||
|
||||
if not expected:
|
||||
return False
|
||||
|
||||
for target in expected:
|
||||
if not (natives_dir / target).exists():
|
||||
return False
|
||||
|
||||
return True
|
||||
|
||||
|
||||
def check_full_libraries_exists(manifest: dict, libraries_dir: Path, natives_dir: Path) -> bool:
|
||||
if not check_natives_exists(manifest, libraries_dir, natives_dir) or not check_libraries_exists(manifest, libraries_dir):
|
||||
return False
|
||||
|
||||
return True
|
||||
|
||||
|
||||
def generate_classpath(manifest: dict, libraries_dir: Path, core_file_path: Path):
|
||||
"""
|
||||
Generate classpath by using specific version's data
|
||||
:param manifest:
|
||||
:param libraries_dir:
|
||||
:param core_file_path:
|
||||
:return:
|
||||
not_found: list
|
||||
"""
|
||||
libraries = manifest.get("libraries", [])
|
||||
|
||||
if not libraries:
|
||||
raise VersionDataKeyNotFoundException(
|
||||
"Libraries not found in manifest.",
|
||||
)
|
||||
|
||||
# Same as natives
|
||||
_, _, os_version = get_platform_info()
|
||||
platform_key_name, platform_rule_name, platform_arch_type = get_natives_platform_info()
|
||||
|
||||
separator = ";" if platform_key_name == "windows" else ":"
|
||||
classes = []
|
||||
|
||||
for library in libraries:
|
||||
name = library.get("name", "")
|
||||
artifact_raw = library.get("downloads", {}).get("artifact", {})
|
||||
rules = library.get("rules", [])
|
||||
classifiers = library.get("downloads", {}).get("classifiers", {})
|
||||
natives = library.get("natives", {})
|
||||
|
||||
# Ignore natives
|
||||
if (len(natives) > 0 or bool(re.search(r":natives-[^:]+$", name)) or
|
||||
any(native.startswith("natives-") for native in classifiers)):
|
||||
logger.info(f"Skipping {name} because its a native library.")
|
||||
continue
|
||||
|
||||
if not is_library_allowed(
|
||||
rules,
|
||||
platform_rule_name,
|
||||
platform_arch_type,
|
||||
os_version,
|
||||
):
|
||||
continue
|
||||
|
||||
artifact = generate_formatted_artifact_with_name(
|
||||
name,
|
||||
artifact_raw
|
||||
)
|
||||
|
||||
if not artifact:
|
||||
logger.warning(f"Artifact {name} are not valid. Skipping...")
|
||||
continue
|
||||
|
||||
relative_path = artifact.get("path")
|
||||
full_path = libraries_dir / Path(relative_path)
|
||||
|
||||
if not full_path.is_file() or not full_path.exists():
|
||||
raise FileNotFoundError(f"Artifact {name} missing.")
|
||||
|
||||
classes.append(full_path.as_posix())
|
||||
|
||||
# Append core file
|
||||
classes.append(core_file_path.as_posix())
|
||||
|
||||
return f"{separator}".join(classes)
|
||||
|
||||
|
||||
def download_full_libraries(manifest: dict, libraries_dir: Path, natives_dir: Path,
|
||||
progress_callback: Callable[[str, float], None] = None,
|
||||
redownload_callback: Callable[
|
||||
[list[FileObject]], tuple[bool, list[FileObject]]] = None) -> None:
|
||||
# libraries
|
||||
libraries = collect_library_artifacts(manifest)
|
||||
|
||||
# natives
|
||||
natives = collect_native_artifacts(manifest)
|
||||
|
||||
# all
|
||||
full = list(libraries)
|
||||
full.extend(natives)
|
||||
|
||||
logger.info(
|
||||
"Supported native count: {}".format(len(natives))
|
||||
)
|
||||
|
||||
# Start download
|
||||
fails, successes = download_libraries(full, libraries_dir, progress_callback=progress_callback)
|
||||
|
||||
download_ok = len(fails) == 0
|
||||
|
||||
if redownload_callback and not download_ok:
|
||||
download_ok, fails = redownload_callback(fails)
|
||||
|
||||
if not download_ok:
|
||||
raise DownloadException(
|
||||
"Unable to continue process due to certain libraries download failed:\n"
|
||||
+ "\n".join(file.posix_path for file in fails)
|
||||
)
|
||||
|
||||
if download_ok:
|
||||
# Start unzip natives
|
||||
unzip_natives(natives, libraries_dir, natives_dir)
|
||||
else:
|
||||
raise DependencyException(
|
||||
"Certain libraries download failed. Unable to continue unzip process.\n"
|
||||
)
|
||||
|
||||
@@ -0,0 +1,589 @@
|
||||
import logging
|
||||
import re
|
||||
import zipfile
|
||||
from pathlib import Path
|
||||
from typing import Callable
|
||||
|
||||
from .library import get_natives_platform_info, is_library_allowed, generate_formatted_artifact_with_name, \
|
||||
is_native_allowed, is_legacy_native_allowed, remove_duplicate_artifacts
|
||||
from .libraryx import normalize_library_to_common_struct
|
||||
from ..common.common import get_platform_info, FileObject, download_multiple_files, unzip, is_valid_zipfile
|
||||
from .version_info import find_useful_part_in_specific_version_manifest
|
||||
from ..common.exception import UnsupportedPlatformException, NativeResolveException, DependencyException, \
|
||||
VersionDataKeyNotFoundException, DownloadException
|
||||
|
||||
logger = logging.getLogger("Launcher.CoreLib")
|
||||
|
||||
def collect_library_artifacts(specific_version_manifest):
|
||||
_, _, _, _, libraries, _, _ = find_useful_part_in_specific_version_manifest(specific_version_manifest)
|
||||
|
||||
items = []
|
||||
|
||||
# Same as natives
|
||||
_, _, os_version = get_platform_info()
|
||||
platform_key_name, platform_rule_name, platform_arch_type = get_natives_platform_info()
|
||||
|
||||
for library_raw in libraries:
|
||||
library = normalize_library_to_common_struct(
|
||||
library_raw
|
||||
)
|
||||
|
||||
name = library.get("name", "")
|
||||
artifact_raw = library.get("downloads", {}).get("artifact", {})
|
||||
rules = library.get("rules", [])
|
||||
classifiers = library.get("downloads", {}).get("classifiers", {})
|
||||
natives = library.get("natives", {})
|
||||
|
||||
# Ignore natives
|
||||
if (len(natives) > 0 or bool(re.search(r":natives-[^:]+$", name)) or
|
||||
any(native.startswith("natives-") for native in classifiers)):
|
||||
logger.info(f"Skipping {name} because its a native library.")
|
||||
continue
|
||||
|
||||
if not is_library_allowed(
|
||||
rules,
|
||||
platform_rule_name,
|
||||
platform_arch_type,
|
||||
os_version,
|
||||
):
|
||||
logger.debug(f"Library {name} can't pass library rules.")
|
||||
continue
|
||||
|
||||
artifact = generate_formatted_artifact_with_name(
|
||||
name,
|
||||
artifact_raw
|
||||
)
|
||||
|
||||
if artifact:
|
||||
items.append(artifact)
|
||||
|
||||
if not items:
|
||||
raise DependencyException(
|
||||
"Could not find any artifact in any of the libraries."
|
||||
)
|
||||
|
||||
return items
|
||||
|
||||
|
||||
def collect_native_artifacts(specific_version_manifest):
|
||||
"""
|
||||
Collects native artifacts that support current platform
|
||||
:param specific_version_manifest:
|
||||
:return:
|
||||
filtered_natives_artifacts: list[dict]
|
||||
"""
|
||||
filtered = []
|
||||
|
||||
# Convert current platform and architecture type to same formula
|
||||
_, _, os_version = get_platform_info()
|
||||
platform_key_name, platform_rule_name, platform_arch_type = get_natives_platform_info()
|
||||
|
||||
if platform_key_name is None or platform_arch_type is None or platform_rule_name is None:
|
||||
raise UnsupportedPlatformException(
|
||||
f"Unsupported platform: key={platform_key_name}, "
|
||||
f"rule={platform_rule_name}, arch={platform_arch_type}"
|
||||
)
|
||||
|
||||
_, _, _, _, libraries, _, _ = find_useful_part_in_specific_version_manifest(specific_version_manifest)
|
||||
|
||||
for library_raw in libraries:
|
||||
library = normalize_library_to_common_struct(
|
||||
library_raw
|
||||
)
|
||||
|
||||
name = library.get("name", "")
|
||||
artifact = library.get("downloads", {}).get("artifact", {})
|
||||
# For newer Minecraft (VER>1.7.10)
|
||||
rules = library.get("rules", [])
|
||||
# For legacy version (VER<1.8)
|
||||
classifiers = library.get("downloads", {}).get("classifiers", {})
|
||||
natives = library.get("natives", {})
|
||||
exclude = library.get("extract", {}).get("exclude", ["META-INF/"]) # For extract use
|
||||
|
||||
is_natives = False
|
||||
|
||||
# Check if library is a native
|
||||
if (len(natives) > 0 or bool(re.search(r":natives-[^:]+$", name)) or
|
||||
any(native.startswith("natives-") for native in classifiers)):
|
||||
is_natives = True
|
||||
|
||||
if not is_natives:
|
||||
continue
|
||||
|
||||
if rules and classifiers:
|
||||
logger.debug(f"Found newer and legacy (classifier) native rules for {name}")
|
||||
elif rules:
|
||||
logger.debug(f"Found newer native rules for {name}")
|
||||
elif classifiers:
|
||||
logger.debug(f"Found legacy native rules (classifier) for {name}")
|
||||
|
||||
# =============== New rules ===============
|
||||
|
||||
# Parse action section
|
||||
if not is_library_allowed(
|
||||
rules,
|
||||
platform_rule_name,
|
||||
platform_arch_type,
|
||||
os_version,
|
||||
):
|
||||
logger.debug(f"Library {name} can't pass library rules.")
|
||||
continue
|
||||
|
||||
if is_native_allowed(
|
||||
name,
|
||||
platform_key_name,
|
||||
platform_arch_type,
|
||||
):
|
||||
result_artifact = generate_formatted_artifact_with_name(name, artifact, extra={"exclude": exclude,
|
||||
"flatten": True})
|
||||
if result_artifact:
|
||||
filtered.append(result_artifact)
|
||||
# For some reason, we still need to continue process because some version use multiple type of
|
||||
# classifier. (Such as 1.14.2)
|
||||
else:
|
||||
logger.debug(f"Library {name} can't pass native rules.")
|
||||
|
||||
# =============== Old rules ===============
|
||||
logger.debug(
|
||||
"Library {} doesn't have native key. This library may use old rule.".format(name)
|
||||
)
|
||||
|
||||
result, result_artifact, classifier = is_legacy_native_allowed(
|
||||
natives,
|
||||
classifiers,
|
||||
platform_rule_name,
|
||||
platform_arch_type,
|
||||
)
|
||||
|
||||
if not result:
|
||||
logger.debug(f"Library {name} can't pass legacy native rules.")
|
||||
continue
|
||||
else:
|
||||
result_artifact = generate_formatted_artifact_with_name(name,
|
||||
result_artifact,
|
||||
extra={"exclude": exclude, "flatten": True},
|
||||
relative_args={
|
||||
"extra": classifier
|
||||
})
|
||||
if result_artifact:
|
||||
filtered.append(result_artifact)
|
||||
|
||||
# Remove duplicate items
|
||||
items = remove_duplicate_artifacts(filtered)
|
||||
|
||||
if not filtered:
|
||||
raise NativeResolveException(
|
||||
"No compatible native artifact found.",
|
||||
user_message="No compatible native library was found for this platform.",
|
||||
)
|
||||
elif not items:
|
||||
raise NativeResolveException(
|
||||
"No compatible native artifact found. Report developer because this may be a issue"
|
||||
" at function \"remove_duplicate_artifacts\"",
|
||||
)
|
||||
|
||||
return items
|
||||
|
||||
|
||||
def download_libraries(artifacts, output_dir, progress_callback: Callable[[str, float], None] = None) \
|
||||
-> tuple[list[FileObject], list[FileObject]]:
|
||||
"""
|
||||
Download libraries from a list that contains artifacts
|
||||
:param artifacts:
|
||||
:param output_dir:
|
||||
:param progress_callback:
|
||||
:return:
|
||||
fails: list (List of artifacts that download failed)
|
||||
successes: list (artifacts that were downloaded)
|
||||
"""
|
||||
files = []
|
||||
successes = []
|
||||
|
||||
for artifact in artifacts:
|
||||
sha1 = artifact.get("sha1")
|
||||
sha256 = artifact.get("sha256")
|
||||
sha512 = artifact.get("sha512")
|
||||
relative_path = artifact.get("path")
|
||||
url = artifact.get("url")
|
||||
|
||||
if not relative_path or not url:
|
||||
logger.warning(
|
||||
"Artifact {} has no path or url. Skipping.".format(artifact["name"])
|
||||
)
|
||||
continue
|
||||
else:
|
||||
full_path = output_dir / Path(relative_path)
|
||||
|
||||
file = FileObject(
|
||||
full_path,
|
||||
url=url,
|
||||
sha1=sha1,
|
||||
)
|
||||
|
||||
if sha1 is None:
|
||||
logger.warning(f"File {full_path} has no SHA1 hash.")
|
||||
|
||||
if file.exists:
|
||||
if sha1 is not None:
|
||||
valid = file.verify("sha1", sha1)
|
||||
elif sha256 is not None:
|
||||
valid = file.verify("sha256", sha256)
|
||||
elif sha512 is not None:
|
||||
valid = file.verify("sha512", sha512)
|
||||
elif full_path.suffix.lower() == ".jar":
|
||||
valid = is_valid_zipfile(file.path)
|
||||
else:
|
||||
valid = full_path.stat().st_size > 0
|
||||
|
||||
if valid:
|
||||
successes.append(file)
|
||||
continue
|
||||
|
||||
files.append(file)
|
||||
|
||||
fails, finished = download_multiple_files(files, progress_callback=progress_callback)
|
||||
successes.extend(finished)
|
||||
|
||||
return fails, successes
|
||||
|
||||
|
||||
def unzip_natives(filtered, libraries_dir, destination: Path):
|
||||
"""
|
||||
Unzip native libraries from a list that contains artifacts
|
||||
:param filtered:
|
||||
:param libraries_dir:
|
||||
:param destination:
|
||||
:return:
|
||||
|
||||
IMPORTANT:
|
||||
Please put "exclude" key inside the artifact (from the version data). If not, some unnecessary file will be also
|
||||
included when unzipping the native library.
|
||||
"""
|
||||
destination.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
for result_artifact in filtered:
|
||||
exclude = result_artifact.get("exclude", [])
|
||||
flatten = result_artifact.get("flatten", False)
|
||||
relative_path = result_artifact.get("path")
|
||||
|
||||
if not relative_path:
|
||||
logger.warning(
|
||||
"Artifact {} has no relative path. Skipping.".format(result_artifact["name"])
|
||||
)
|
||||
continue
|
||||
else:
|
||||
full_path = libraries_dir / Path(relative_path)
|
||||
|
||||
unzip(full_path, destination, exclude=exclude, flatten=flatten)
|
||||
|
||||
logger.debug(f"Unzipped {full_path}.")
|
||||
|
||||
|
||||
def check_libraries_exists(manifest: dict, libraries_dir: Path):
|
||||
"""
|
||||
Check if libraries are present in the libraries directory
|
||||
:param manifest:
|
||||
:param libraries_dir:
|
||||
:return:
|
||||
not_found: list
|
||||
"""
|
||||
libraries = manifest.get("libraries", [])
|
||||
|
||||
if not libraries:
|
||||
raise VersionDataKeyNotFoundException(
|
||||
"Libraries not found in manifest.",
|
||||
)
|
||||
|
||||
not_found = []
|
||||
|
||||
# Same as natives
|
||||
_, _, os_version = get_platform_info()
|
||||
platform_key_name, platform_rule_name, platform_arch_type = get_natives_platform_info()
|
||||
|
||||
for library_raw in libraries:
|
||||
library = normalize_library_to_common_struct(
|
||||
library_raw
|
||||
)
|
||||
|
||||
name = library.get("name", "")
|
||||
artifact_raw = library.get("downloads", {}).get("artifact", {})
|
||||
rules = library.get("rules", [])
|
||||
classifiers = library.get("downloads", {}).get("classifiers", {})
|
||||
natives = library.get("natives", {})
|
||||
|
||||
# Ignore natives
|
||||
if (len(natives) > 0 or bool(re.search(r":natives-[^:]+$", name)) or
|
||||
any(native.startswith("natives-") for native in classifiers)):
|
||||
logger.info(f"Skipping {name} because its a native library.")
|
||||
continue
|
||||
|
||||
if not is_library_allowed(
|
||||
rules,
|
||||
platform_rule_name,
|
||||
platform_arch_type,
|
||||
os_version,
|
||||
):
|
||||
continue
|
||||
|
||||
artifact = generate_formatted_artifact_with_name(
|
||||
name,
|
||||
artifact_raw
|
||||
)
|
||||
|
||||
if not artifact:
|
||||
logger.warning(f"Artifact {name} are not valid. Skipping...")
|
||||
continue
|
||||
|
||||
relative_path = artifact.get("path")
|
||||
full_path = libraries_dir / Path(relative_path)
|
||||
|
||||
if not full_path.is_file() or not full_path.exists():
|
||||
logger.debug(f"Artifact {name} missing.")
|
||||
not_found.append(full_path)
|
||||
|
||||
return not_found
|
||||
|
||||
|
||||
def check_natives_exists(manifest: dict, libraries_dir: Path, natives_dir: Path):
|
||||
filtered = []
|
||||
|
||||
# Convert current platform and architecture type to same formula
|
||||
_, _, os_version = get_platform_info()
|
||||
platform_key_name, platform_rule_name, platform_arch_type = get_natives_platform_info()
|
||||
|
||||
if platform_key_name is None or platform_arch_type is None or platform_rule_name is None:
|
||||
raise UnsupportedPlatformException(
|
||||
f"Unsupported platform: key={platform_key_name}, "
|
||||
f"rule={platform_rule_name}, arch={platform_arch_type}"
|
||||
)
|
||||
|
||||
_, _, _, _, libraries, _, _ = find_useful_part_in_specific_version_manifest(manifest)
|
||||
|
||||
for library_raw in libraries:
|
||||
library = normalize_library_to_common_struct(
|
||||
library_raw
|
||||
)
|
||||
|
||||
name = library.get("name", "")
|
||||
artifact = library.get("downloads", {}).get("artifact", {})
|
||||
# For newer Minecraft (VER>1.7.10)
|
||||
rules = library.get("rules", [])
|
||||
# For legacy version (VER<1.8)
|
||||
classifiers = library.get("downloads", {}).get("classifiers", {})
|
||||
natives = library.get("natives", {})
|
||||
exclude = library.get("extract", {}).get("exclude", ["META-INF/"]) # For extract use
|
||||
|
||||
is_natives = False
|
||||
|
||||
# Check if library is a native
|
||||
if (len(natives) > 0 or bool(re.search(r":natives-[^:]+$", name)) or
|
||||
any(native.startswith("natives-") for native in classifiers)):
|
||||
is_natives = True
|
||||
|
||||
if not is_natives:
|
||||
continue
|
||||
|
||||
# =============== New rules ===============
|
||||
|
||||
# Parse action section
|
||||
if not is_library_allowed(
|
||||
rules,
|
||||
platform_rule_name,
|
||||
platform_arch_type,
|
||||
os_version,
|
||||
):
|
||||
continue
|
||||
|
||||
if is_native_allowed(
|
||||
name,
|
||||
platform_key_name,
|
||||
platform_arch_type,
|
||||
):
|
||||
result_artifact = generate_formatted_artifact_with_name(name, artifact, extra={"exclude": exclude,
|
||||
"flatten": True})
|
||||
if result_artifact:
|
||||
filtered.append(result_artifact)
|
||||
|
||||
# =============== Old rules ===============
|
||||
result, result_artifact, classifier = is_legacy_native_allowed(
|
||||
natives,
|
||||
classifiers,
|
||||
platform_rule_name,
|
||||
platform_arch_type,
|
||||
)
|
||||
|
||||
if not result:
|
||||
continue
|
||||
|
||||
result_artifact = generate_formatted_artifact_with_name(name,
|
||||
result_artifact,
|
||||
extra={"exclude": exclude, "flatten": True},
|
||||
relative_args={
|
||||
"extra": classifier
|
||||
})
|
||||
|
||||
if result_artifact:
|
||||
filtered.append(result_artifact)
|
||||
|
||||
# Remove duplicate items
|
||||
items = remove_duplicate_artifacts(filtered)
|
||||
|
||||
if not filtered:
|
||||
raise NativeResolveException(
|
||||
"No compatible native artifact found.",
|
||||
user_message="No compatible native library was found for this platform.",
|
||||
)
|
||||
elif not items:
|
||||
raise NativeResolveException(
|
||||
"No compatible native artifact found. Report developer because this may be a issue"
|
||||
" at function \"remove_duplicate_artifacts\"",
|
||||
)
|
||||
|
||||
expected = set()
|
||||
|
||||
for result_artifact in items:
|
||||
full_path = libraries_dir / Path(result_artifact["path"])
|
||||
exclude = result_artifact.get("exclude", []) or []
|
||||
flatten = result_artifact.get("flatten", False)
|
||||
|
||||
if not full_path.is_file() or not full_path.exists():
|
||||
return False
|
||||
|
||||
with zipfile.ZipFile(full_path) as zf:
|
||||
for member in zf.namelist():
|
||||
if member.endswith("/"):
|
||||
continue
|
||||
if any(member.startswith(item) for item in exclude):
|
||||
continue
|
||||
|
||||
target = Path(member).name if flatten else member
|
||||
if target:
|
||||
expected.add(target)
|
||||
|
||||
if not expected:
|
||||
return False
|
||||
|
||||
for target in expected:
|
||||
if not (natives_dir / target).exists():
|
||||
return False
|
||||
|
||||
return True
|
||||
|
||||
|
||||
def check_full_libraries_exists(manifest: dict, libraries_dir: Path, natives_dir: Path) -> bool:
|
||||
if not check_natives_exists(manifest, libraries_dir, natives_dir) or not check_libraries_exists(manifest, libraries_dir):
|
||||
return False
|
||||
|
||||
return True
|
||||
|
||||
|
||||
def generate_classpath(manifest: dict, libraries_dir: Path, core_file_path: Path):
|
||||
"""
|
||||
Generate classpath by using specific version's data
|
||||
:param manifest:
|
||||
:param libraries_dir:
|
||||
:param core_file_path:
|
||||
:return:
|
||||
not_found: list
|
||||
"""
|
||||
libraries = manifest.get("libraries", [])
|
||||
|
||||
if not libraries:
|
||||
raise VersionDataKeyNotFoundException(
|
||||
"Libraries not found in manifest.",
|
||||
)
|
||||
|
||||
# Same as natives
|
||||
_, _, os_version = get_platform_info()
|
||||
platform_key_name, platform_rule_name, platform_arch_type = get_natives_platform_info()
|
||||
|
||||
separator = ";" if platform_key_name == "windows" else ":"
|
||||
classes = []
|
||||
|
||||
for library_raw in libraries:
|
||||
library = normalize_library_to_common_struct(
|
||||
library_raw
|
||||
)
|
||||
|
||||
name = library.get("name", "")
|
||||
artifact_raw = library.get("downloads", {}).get("artifact", {})
|
||||
rules = library.get("rules", [])
|
||||
classifiers = library.get("downloads", {}).get("classifiers", {})
|
||||
natives = library.get("natives", {})
|
||||
|
||||
# Ignore natives
|
||||
if (len(natives) > 0 or bool(re.search(r":natives-[^:]+$", name)) or
|
||||
any(native.startswith("natives-") for native in classifiers)):
|
||||
logger.info(f"Skipping {name} because its a native library.")
|
||||
continue
|
||||
|
||||
if not is_library_allowed(
|
||||
rules,
|
||||
platform_rule_name,
|
||||
platform_arch_type,
|
||||
os_version,
|
||||
):
|
||||
continue
|
||||
|
||||
artifact = generate_formatted_artifact_with_name(
|
||||
name,
|
||||
artifact_raw
|
||||
)
|
||||
|
||||
if not artifact:
|
||||
logger.warning(f"Artifact {name} are not valid. Skipping...")
|
||||
continue
|
||||
|
||||
relative_path = artifact.get("path")
|
||||
full_path = libraries_dir / Path(relative_path)
|
||||
|
||||
if not full_path.is_file() or not full_path.exists():
|
||||
raise FileNotFoundError(f"Artifact {name} missing.")
|
||||
|
||||
classes.append(full_path.as_posix())
|
||||
|
||||
# Append core file
|
||||
classes.append(core_file_path.as_posix())
|
||||
|
||||
return f"{separator}".join(classes)
|
||||
|
||||
|
||||
def download_full_libraries(manifest: dict, libraries_dir: Path, natives_dir: Path,
|
||||
progress_callback: Callable[[str, float], None] = None,
|
||||
redownload_callback: Callable[
|
||||
[list[FileObject]], tuple[bool, list[FileObject]]] = None) -> None:
|
||||
# libraries
|
||||
libraries = collect_library_artifacts(manifest)
|
||||
|
||||
# natives
|
||||
natives = collect_native_artifacts(manifest)
|
||||
|
||||
# all
|
||||
full = list(libraries)
|
||||
full.extend(natives)
|
||||
|
||||
logger.info(
|
||||
"Supported native count: {}".format(len(natives))
|
||||
)
|
||||
|
||||
# Start download
|
||||
fails, successes = download_libraries(full, libraries_dir, progress_callback=progress_callback)
|
||||
|
||||
download_ok = len(fails) == 0
|
||||
|
||||
if redownload_callback and not download_ok:
|
||||
download_ok, fails = redownload_callback(fails)
|
||||
|
||||
if not download_ok:
|
||||
raise DownloadException(
|
||||
"Unable to continue process due to certain libraries download failed:\n"
|
||||
+ "\n".join(file.posix_path for file in fails)
|
||||
)
|
||||
|
||||
if download_ok:
|
||||
# Start unzip natives
|
||||
unzip_natives(natives, libraries_dir, natives_dir)
|
||||
else:
|
||||
raise DependencyException(
|
||||
"Certain libraries download failed. Unable to continue unzip process.\n"
|
||||
)
|
||||
@@ -0,0 +1,130 @@
|
||||
"""
|
||||
This module provide some advanced function for library parse/download process use.
|
||||
"""
|
||||
import logging
|
||||
import re
|
||||
from copy import deepcopy
|
||||
|
||||
from .library import (convert_maven_version, parse_maven_coordinate, generate_relative_path_by_maven_coordinate,
|
||||
is_complete_maven_url, generate_repo_url)
|
||||
from .version_info import MOJANG_LIBRARIES_ENDPOINT
|
||||
from ..common.exception import VersionDataKeyNotFoundException, DependencyException
|
||||
|
||||
logger = logging.getLogger("Launcher.CoreLib")
|
||||
|
||||
|
||||
def convert_maven_version_with_extra(version: str, failback=None) -> tuple[tuple[int, int, int] | None, str | None]:
|
||||
match = re.fullmatch(
|
||||
r"(\d+(?:\.\d+){1,2})(.*)",
|
||||
version,
|
||||
)
|
||||
|
||||
if match is None:
|
||||
return failback, None
|
||||
|
||||
base_ver = match.group(1)
|
||||
extra_ver = match.group(2) or None
|
||||
|
||||
base_ver_converted = convert_maven_version(base_ver, failback=None)
|
||||
|
||||
if base_ver_converted is None:
|
||||
return failback, extra_ver
|
||||
|
||||
return base_ver_converted, extra_ver
|
||||
|
||||
def generate_unique_library_key(group: str, artifact: str, classifier: str | None=None, extension: str | None=None, version: str | None=None) -> str:
|
||||
key = f"{group}:{artifact}"
|
||||
if version is not None:
|
||||
key += f":{version}"
|
||||
|
||||
if classifier is not None:
|
||||
key += f":{classifier}"
|
||||
|
||||
if extension is not None:
|
||||
key += f":{extension}"
|
||||
|
||||
return key
|
||||
|
||||
|
||||
def normalize_library_to_common_struct(library: dict, default_maven_url: str = MOJANG_LIBRARIES_ENDPOINT) -> dict:
|
||||
result = deepcopy(library)
|
||||
name: str = library.get("name")
|
||||
|
||||
if name is None:
|
||||
raise VersionDataKeyNotFoundException(
|
||||
"Provided mod loader version manifest library does not contain key \"name\"."
|
||||
)
|
||||
|
||||
group, artifact, version, classifier, extension = parse_maven_coordinate(name)
|
||||
|
||||
downloads_raw = result.setdefault("downloads", {})
|
||||
|
||||
if isinstance(downloads_raw, dict):
|
||||
downloads = downloads_raw
|
||||
else:
|
||||
downloads = {}
|
||||
|
||||
artifact_raw = dict(downloads.get("artifact") or {})
|
||||
|
||||
# Generate new relative path and put back
|
||||
if not artifact_raw.get("path"):
|
||||
path = generate_relative_path_by_maven_coordinate(name)
|
||||
if not path:
|
||||
raise DependencyException(
|
||||
f"Unable to generate artifact path for {name}"
|
||||
)
|
||||
artifact_raw["path"] = path
|
||||
|
||||
if not artifact_raw.get("url"):
|
||||
top_level_url = result.get("url")
|
||||
|
||||
if top_level_url:
|
||||
if is_complete_maven_url(
|
||||
top_level_url,
|
||||
group,
|
||||
artifact,
|
||||
version,
|
||||
classifier,
|
||||
extension,
|
||||
):
|
||||
# Put existing url to downloads.artifact
|
||||
artifact_raw["url"] = top_level_url
|
||||
else:
|
||||
# Replace broken url
|
||||
artifact_raw["url"] = generate_repo_url(
|
||||
top_level_url,
|
||||
group,
|
||||
artifact,
|
||||
version,
|
||||
classifier,
|
||||
extension,
|
||||
)
|
||||
else:
|
||||
# Generate missing url
|
||||
artifact_raw["url"] = generate_repo_url(
|
||||
default_maven_url,
|
||||
group,
|
||||
artifact,
|
||||
version,
|
||||
classifier,
|
||||
extension,
|
||||
)
|
||||
|
||||
# Put hash information to artifact
|
||||
for key in (
|
||||
"sha1",
|
||||
"sha256",
|
||||
"sha512",
|
||||
"md5",
|
||||
"size",
|
||||
):
|
||||
if artifact_raw.get(key) is None:
|
||||
value = result.get(key)
|
||||
if value is not None:
|
||||
artifact_raw[key] = value
|
||||
|
||||
# Put updated data back to result
|
||||
downloads["artifact"] = artifact_raw
|
||||
result["downloads"] = downloads
|
||||
|
||||
return result
|
||||
@@ -9,8 +9,8 @@ from core_lib.common.exception import VersionManifestFetchException, VersionData
|
||||
logger = logging.getLogger("Launcher.CoreLib")
|
||||
|
||||
FABRIC_META_VERSION_ENDPOINT_V2 = "https://meta.fabricmc.net/v2/versions"
|
||||
FABRIC_META_LOADER_VERSION_ENDPOINT_V2 = "https://meta.fabricmc.net/v1/versions/loader/{}/"
|
||||
FABRIC_META_LOADER_MANIFEST_ENDPOINT_V2 = "https://meta.fabricmc.net//v2/versions/loader/{}/{}/profile/json"
|
||||
FABRIC_META_LOADER_VERSION_ENDPOINT_V2 = "https://meta.fabricmc.net/v2/versions/loader/{}/"
|
||||
FABRIC_META_LOADER_MANIFEST_ENDPOINT_V2 = "https://meta.fabricmc.net/v2/versions/loader/{}/{}/profile/json"
|
||||
FABRIC_MAVEN_URL = "https://maven.fabricmc.net/"
|
||||
|
||||
def fetch_fabric_version_list(version_manifest_url: str=FABRIC_META_VERSION_ENDPOINT_V2, timeout: int=5) -> list[dict]:
|
||||
|
||||
@@ -3,8 +3,10 @@ from copy import deepcopy
|
||||
import shlex
|
||||
|
||||
from core_lib.common.exception import VersionDataKeyNotFoundException, DependencyException
|
||||
from core_lib.game.library import parse_maven_coordinate, convert_maven_version, generate_repo_url, \
|
||||
from core_lib.game.library import parse_maven_coordinate, generate_repo_url, \
|
||||
is_complete_maven_url
|
||||
from core_lib.game.libraryx import convert_maven_version_with_extra, generate_unique_library_key, \
|
||||
normalize_library_to_common_struct
|
||||
from core_lib.game.version_info import MOJANG_LIBRARIES_ENDPOINT
|
||||
|
||||
logger = logging.getLogger("Launcher.CoreLib")
|
||||
@@ -58,10 +60,12 @@ def find_useful_part_in_mod_version_manifest(mod_version_manifest: dict):
|
||||
"mainClass": main_class,
|
||||
}
|
||||
|
||||
|
||||
def merge_necessary_dependencies(inherits_libraries: list, mod_libraries: list, mod_maven_url=None,
|
||||
inherits_maven_url=MOJANG_LIBRARIES_ENDPOINT, default_extension="jar") -> list[dict]:
|
||||
|
||||
inherits = {}
|
||||
result_pre = []
|
||||
result = []
|
||||
|
||||
# ensure below step won't modify original data
|
||||
@@ -73,7 +77,7 @@ def merge_necessary_dependencies(inherits_libraries: list, mod_libraries: list,
|
||||
source_from = lib["sourceFrom"]
|
||||
lib_group=lib["parsed"]["group"]
|
||||
lib_artifact=lib["parsed"]["artifact"]
|
||||
lib_version=lib["parsed"]["version"]
|
||||
lib_version=lib["parsed"]["versionRaw"]
|
||||
lib_classifier=lib["parsed"]["classifier"]
|
||||
lib_extension=lib["parsed"]["extension"]
|
||||
|
||||
@@ -111,73 +115,119 @@ def merge_necessary_dependencies(inherits_libraries: list, mod_libraries: list,
|
||||
"Provided inherited library does not contain key \"name\"."
|
||||
)
|
||||
|
||||
group, artifact, version, classifier, extension = parse_maven_coordinate(name,
|
||||
default_extension=default_extension)
|
||||
group, artifact, version_raw, classifier, extension = parse_maven_coordinate(name, default_extension=default_extension)
|
||||
ver_converted, extra = convert_maven_version_with_extra(version_raw)
|
||||
|
||||
if extra:
|
||||
logger.debug("Found extra version \"{}\" for library {}".format(extra, name))
|
||||
|
||||
if not ver_converted:
|
||||
raise DependencyException(
|
||||
f"Unable to convert provided inherited library name {name}'s version {version_raw} to tuple."
|
||||
)
|
||||
|
||||
lib_key = generate_unique_library_key(
|
||||
group=group,
|
||||
artifact=artifact,
|
||||
classifier=classifier,
|
||||
extension=extension,
|
||||
)
|
||||
|
||||
# Save into dict for next step to check if library is existing
|
||||
if not f"{group}:{artifact}" in inherits:
|
||||
inherits[f"{group}:{artifact}"] = library
|
||||
inherits[f"{group}:{artifact}"]["parsed"] = {
|
||||
if not lib_key in inherits:
|
||||
inherits[lib_key] = library
|
||||
inherits[lib_key]["parsed"] = {
|
||||
"group": group,
|
||||
"artifact": artifact,
|
||||
"version": version,
|
||||
"version": ver_converted,
|
||||
"versionRaw": version_raw,
|
||||
"versionExtra": extra,
|
||||
"classifier": classifier,
|
||||
"extension": extension,
|
||||
"endpoint": inherits_maven_url
|
||||
}
|
||||
inherits[f"{group}:{artifact}"]["sourceFrom"] = "official"
|
||||
inherits[lib_key]["sourceFrom"] = "official"
|
||||
else:
|
||||
logger.warning("Duplicate library: {}".format(name))
|
||||
|
||||
for mod_library in mod_libraries:
|
||||
name = mod_library.get("name", None)
|
||||
group, artifact, version, classifier, extension = parse_maven_coordinate(name, default_extension=default_extension)
|
||||
|
||||
ver_converted = convert_maven_version(version)
|
||||
|
||||
if not name:
|
||||
raise DependencyException(
|
||||
"Provided mod library does not contain key \"name\"."
|
||||
)
|
||||
|
||||
group, artifact, version_raw, classifier, extension = parse_maven_coordinate(name, default_extension=default_extension)
|
||||
|
||||
ver_converted, extra = convert_maven_version_with_extra(version_raw)
|
||||
|
||||
if extra:
|
||||
logger.debug("Found extra version \"{}\" for library {}".format(extra, name))
|
||||
|
||||
if ver_converted is None:
|
||||
raise DependencyException(
|
||||
"Unable to convert inherits library version {} to tuple".format(version),
|
||||
"Unable to convert mod library version {} to tuple".format(version_raw),
|
||||
)
|
||||
|
||||
group, artifact, version, classifier, extension = parse_maven_coordinate(name, default_extension=default_extension)
|
||||
group, artifact, version_raw, classifier, extension = parse_maven_coordinate(name, default_extension=default_extension)
|
||||
lib_key = generate_unique_library_key(
|
||||
group=group,
|
||||
artifact=artifact,
|
||||
classifier=classifier,
|
||||
extension=extension,
|
||||
)
|
||||
|
||||
mod_library["sourceFrom"] = "mod"
|
||||
mod_library["parsed"] = {
|
||||
"group": group,
|
||||
"artifact": artifact,
|
||||
"version": version,
|
||||
# "version": {"raw": version, "converted": ver_converted, "extra": extra},
|
||||
"version": ver_converted,
|
||||
"versionRaw": version_raw,
|
||||
"versionExtra": extra,
|
||||
"classifier": classifier,
|
||||
"extension": extension,
|
||||
"endpoint": mod_maven_url,
|
||||
}
|
||||
|
||||
if f"{group}:{artifact}" in inherits:
|
||||
found_lib = inherits.pop(f"{group}:{artifact}")
|
||||
found_lib_ver = convert_maven_version(found_lib["parsed"]["version"], None)
|
||||
if lib_key in inherits:
|
||||
found_lib = inherits.pop(lib_key)
|
||||
found_lib_ver = found_lib["parsed"]["version"]
|
||||
found_lib_ver_extra = found_lib["parsed"]["versionExtra"]
|
||||
|
||||
if found_lib_ver is None:
|
||||
raise DependencyException(
|
||||
"Unable to convert mod library version {} to tuple".format(found_lib["version"])
|
||||
"Unable to convert mod library version {} to tuple".format(found_lib["versionRaw"])
|
||||
)
|
||||
|
||||
if found_lib_ver > ver_converted or found_lib_ver == ver_converted:
|
||||
result.append(found_lib)
|
||||
if found_lib_ver > ver_converted:
|
||||
result_pre.append(found_lib)
|
||||
elif found_lib_ver < ver_converted:
|
||||
result.append(mod_library)
|
||||
result_pre.append(mod_library)
|
||||
else:
|
||||
# Use mod loader library if extra version is existing
|
||||
if found_lib_ver_extra == mod_library["parsed"]["versionExtra"]:
|
||||
result_pre.append(found_lib)
|
||||
else:
|
||||
result_pre.append(mod_library)
|
||||
else:
|
||||
result.append(mod_library)
|
||||
result_pre.append(mod_library)
|
||||
|
||||
# Add the remaining dependencies back to result
|
||||
for key in inherits:
|
||||
value = inherits[key]
|
||||
result.append(value)
|
||||
result_pre.append(value)
|
||||
|
||||
for library in result:
|
||||
for library in result_pre:
|
||||
# Cleanup
|
||||
check_lib(library)
|
||||
library_normalized = normalize_library_to_common_struct(library, default_maven_url=library["parsed"]["endpoint"])
|
||||
|
||||
library_normalized.pop("parsed", None)
|
||||
library_normalized.pop("sourceFrom", None)
|
||||
library_normalized.pop("endpoint", None)
|
||||
result.append(library_normalized)
|
||||
|
||||
return result
|
||||
|
||||
@@ -240,4 +290,14 @@ def merge_manifest(inherits_version_manifest: dict, mod_version_manifest: dict,
|
||||
|
||||
merged_manifest["arguments"] = merge_game_and_jvm_args(inherits_arguments, mod_arguments)
|
||||
|
||||
# Update version number
|
||||
mod_id = mod_version_manifest.get("id")
|
||||
inherits_id = inherits_version_manifest.get("id")
|
||||
|
||||
if mod_id:
|
||||
merged_manifest["id"] = mod_id
|
||||
|
||||
if inherits_id:
|
||||
merged_manifest["inheritsFrom"] = inherits_id
|
||||
|
||||
return merged_manifest
|
||||
@@ -0,0 +1,168 @@
|
||||
from PySide6.QtCore import Qt, Slot, QTimer
|
||||
from PySide6.QtWidgets import (
|
||||
QDialog,
|
||||
QLabel,
|
||||
QProgressBar,
|
||||
QPushButton,
|
||||
QScrollArea,
|
||||
QSizePolicy,
|
||||
QVBoxLayout,
|
||||
QWidget,
|
||||
)
|
||||
|
||||
|
||||
class ProgressRow(QWidget):
|
||||
def __init__(self, name: str, parent=None):
|
||||
super().__init__(parent)
|
||||
|
||||
self.setSizePolicy(
|
||||
QSizePolicy.Policy.Expanding,
|
||||
QSizePolicy.Policy.Fixed,
|
||||
)
|
||||
|
||||
self.name_label = QLabel(name, self)
|
||||
self.name_label.setWordWrap(True)
|
||||
self.name_label.setTextInteractionFlags(
|
||||
Qt.TextInteractionFlag.TextSelectableByMouse
|
||||
)
|
||||
|
||||
self.status_label = QLabel("Preparing...", self)
|
||||
self.status_label.setWordWrap(True)
|
||||
self.status_label.setTextInteractionFlags(
|
||||
Qt.TextInteractionFlag.TextSelectableByMouse
|
||||
)
|
||||
|
||||
self.progress_bar = QProgressBar(self)
|
||||
self.progress_bar.setRange(0, 100)
|
||||
self.progress_bar.setValue(0)
|
||||
|
||||
# Flags
|
||||
self.finished = False
|
||||
|
||||
layout = QVBoxLayout(self)
|
||||
layout.setContentsMargins(8, 8, 8, 8)
|
||||
layout.setSpacing(4)
|
||||
layout.addWidget(self.name_label)
|
||||
layout.addWidget(self.status_label)
|
||||
layout.addWidget(self.progress_bar)
|
||||
|
||||
|
||||
class ProgressDialog(QDialog):
|
||||
def __init__(self, parent=None):
|
||||
super().__init__(parent)
|
||||
|
||||
self.setWindowTitle("Downloads")
|
||||
self.resize(500, 300)
|
||||
self.setMinimumSize(360, 220)
|
||||
|
||||
self.rows: dict[str, ProgressRow] = {}
|
||||
|
||||
self.scroll_area = QScrollArea(self)
|
||||
self.download_widget = QWidget(self.scroll_area)
|
||||
self.download_layout = QVBoxLayout(self.download_widget)
|
||||
self.download_layout.setContentsMargins(4, 4, 4, 4)
|
||||
self.download_layout.setSpacing(8)
|
||||
self.download_layout.setAlignment(Qt.AlignmentFlag.AlignTop)
|
||||
|
||||
self.scroll_area.setWidget(self.download_widget)
|
||||
self.scroll_area.setWidgetResizable(True)
|
||||
self.scroll_area.setHorizontalScrollBarPolicy(
|
||||
Qt.ScrollBarPolicy.ScrollBarAlwaysOff
|
||||
)
|
||||
|
||||
self.close_button = QPushButton("Close", self)
|
||||
self.close_button.clicked.connect(self.hide)
|
||||
|
||||
layout = QVBoxLayout(self)
|
||||
layout.setContentsMargins(8, 8, 8, 8)
|
||||
layout.setSpacing(8)
|
||||
layout.addWidget(self.scroll_area, 1)
|
||||
layout.addWidget(
|
||||
self.close_button,
|
||||
0,
|
||||
Qt.AlignmentFlag.AlignRight,
|
||||
)
|
||||
|
||||
self.close_timer = QTimer(self)
|
||||
self.close_timer.setSingleShot(True)
|
||||
self.close_timer.timeout.connect(self.cleanup)
|
||||
|
||||
@Slot(str, str)
|
||||
def add_task(self, task_id: str, name: str):
|
||||
if task_id in self.rows:
|
||||
return
|
||||
|
||||
row = ProgressRow(name, self.download_widget)
|
||||
self.rows[task_id] = row
|
||||
|
||||
self.download_layout.addWidget(row)
|
||||
|
||||
if not self.isVisible():
|
||||
self.show()
|
||||
|
||||
@Slot(str, str, float)
|
||||
def update_task(
|
||||
self,
|
||||
task_id: str,
|
||||
status: str,
|
||||
percent: float,
|
||||
):
|
||||
row = self.rows.get(task_id)
|
||||
if row is None:
|
||||
# Prevent started、progress missing due to timing issues
|
||||
self.add_task(task_id, task_id)
|
||||
row = self.rows[task_id]
|
||||
|
||||
value = max(0, min(100, round(percent)))
|
||||
row.status_label.setText(status)
|
||||
row.progress_bar.setValue(value)
|
||||
|
||||
@Slot(str)
|
||||
def finish_task(self, task_id: str):
|
||||
row = self.rows.get(task_id)
|
||||
if row is None:
|
||||
return
|
||||
|
||||
row.status_label.setText("Completed")
|
||||
row.progress_bar.setValue(100)
|
||||
row.finished = True
|
||||
|
||||
if self.all_tasks_finished():
|
||||
self.close_timer.start(2000)
|
||||
|
||||
def all_tasks_finished(self):
|
||||
return all(row.finished == True for task_id, row in self.rows.items())
|
||||
|
||||
def cleanup(self):
|
||||
"""
|
||||
If you want to use this function with a timer. use self.close_timer.start(TIMEOUT)
|
||||
:return:
|
||||
"""
|
||||
QTimer.singleShot(2000, self.cleanup)
|
||||
for task_id in list(self.rows):
|
||||
self.remove_task(task_id)
|
||||
|
||||
self.hide()
|
||||
|
||||
@Slot(str, str)
|
||||
def fail_task(self, task_id: str, error: str):
|
||||
row = self.rows.get(task_id)
|
||||
if row is None:
|
||||
self.add_task(task_id, task_id)
|
||||
row = self.rows[task_id]
|
||||
|
||||
row.status_label.setText(f"Failed: {error}")
|
||||
row.progress_bar.setStyleSheet(
|
||||
"QProgressBar::chunk { background-color: #c62828; }"
|
||||
)
|
||||
# Move to top
|
||||
self.download_layout.removeWidget(row)
|
||||
self.download_layout.insertWidget(0, row)
|
||||
|
||||
@Slot(str)
|
||||
def remove_task(self, task_id: str):
|
||||
row = self.rows.pop(task_id, None)
|
||||
if row is not None:
|
||||
self.download_layout.removeWidget(row)
|
||||
row.setParent(None)
|
||||
row.deleteLater()
|
||||
Reference in New Issue
Block a user