From 037591d5f350cde8b30c598dc2429a07455171bd Mon Sep 17 00:00:00 2001 From: moranfanhua Date: Fri, 14 Aug 2026 19:17:46 +0800 Subject: [PATCH] =?UTF-8?q?feat(tools):=20=E6=96=B0=E5=A2=9E=20MaaFramewor?= =?UTF-8?q?k=20=E6=8E=A7=E5=88=B6=E5=8D=95=E5=85=83=E4=B8=8B=E8=BD=BD?= =?UTF-8?q?=E4=B8=8E=E8=A7=A3=E5=8E=8B=E8=84=9A=E6=9C=AC?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- tools/maafw-extract.py | 70 ++++++++++++++ tools/maafw_download.py | 207 ++++++++++++++++++++++++++++++++++++++++ 2 files changed, 277 insertions(+) create mode 100644 tools/maafw-extract.py create mode 100644 tools/maafw_download.py diff --git a/tools/maafw-extract.py b/tools/maafw-extract.py new file mode 100644 index 0000000..991c942 --- /dev/null +++ b/tools/maafw-extract.py @@ -0,0 +1,70 @@ +#!/usr/bin/env python3 +import sys +from pathlib import Path +import shutil + +basedir = Path(__file__).parent.parent.parent.parent # repository root +target_dir = Path(basedir, "build", "bin", "Release") +archive_dir = Path(basedir, "build", "MaaFramework") + + +def detect_host_platform(): + import platform + + system = platform.system().lower() + machine = platform.machine().lower() + if system == "windows": + if machine in {"amd64", "x86_64"}: + return "win-x86_64" + if machine in {"arm64", "aarch64"}: + return "win-aarch64" + elif system == "linux": + if machine in {"amd64", "x86_64"}: + return "linux-x86_64" + if machine in {"arm64", "aarch64"}: + return "linux-aarch64" + elif system == "darwin": + if machine in {"amd64", "x86_64"}: + return "macos-x86_64" + if machine in {"arm64", "aarch64"}: + return "macos-aarch64" + raise Exception(f"unsupported platform: {system}-{machine}") + + +def main(): + if len(sys.argv) == 2: + platform = sys.argv[1] + else: + platform = detect_host_platform() + + archives = sorted(archive_dir.glob(f"MAA-{platform}-v*.zip")) + if not archives: + print( + f"""Please download the {platform} package from https://github.com/MaaXYZ/MaaFramework/releases/latest, and put it into {archive_dir}""" + ) + archive_dir.mkdir(parents=True, exist_ok=True) + return + + # pick the archive the user most recently put into archive_dir + archive = max(archives, key=lambda p: p.stat().st_mtime) + + print("extracting", archive) + extract_dir = archive_dir / Path(archive.name).stem + if extract_dir.exists(): + shutil.rmtree(extract_dir) + extract_dir.mkdir(parents=True, exist_ok=True) + shutil.unpack_archive(archive, extract_dir) + + bin_dir = extract_dir / "bin" + if not bin_dir.is_dir(): + raise Exception(f"no `bin` directory found in {archive}") + + target_dir.mkdir(parents=True, exist_ok=True) + for src in sorted(bin_dir.glob("*ControlUnit*")): + dst = target_dir / src.name + shutil.copy2(src, dst) + print(f"installed {src.name} -> {dst}") + + +if __name__ == "__main__": + main() diff --git a/tools/maafw_download.py b/tools/maafw_download.py new file mode 100644 index 0000000..52b70ac --- /dev/null +++ b/tools/maafw_download.py @@ -0,0 +1,207 @@ +#!/usr/bin/env python3 +import hashlib +import json +import os +import shutil +import sys +import time +import urllib.error +import urllib.request +from pathlib import Path + +basedir = Path(__file__).parent.parent.parent.parent +target_dir = Path(basedir, "build", "bin", "Release") +download_dir = Path(basedir, "build", "MaaFramework") + + +def detect_host_platform(): + import platform + + system = platform.system().lower() + machine = platform.machine().lower() + if system == "windows": + if machine in {"amd64", "x86_64"}: + return "win-x86_64" + if machine in {"arm64", "aarch64"}: + return "win-aarch64" + elif system == "linux": + if machine in {"amd64", "x86_64"}: + return "linux-x86_64" + if machine in {"arm64", "aarch64"}: + return "linux-aarch64" + elif system == "darwin": + if machine in {"amd64", "x86_64"}: + return "macos-x86_64" + if machine in {"arm64", "aarch64"}: + return "macos-aarch64" + raise Exception(f"unsupported platform: {system}-{machine}") + + +def format_size(num, suffix="B"): + for unit in ["", "Ki", "Mi", "Gi", "Ti", "Pi", "Ei", "Zi"]: + if abs(num) < 1024.0: + return f"{num:3.1f}{unit}{suffix}" + num /= 1024.0 + return f"{num:.1f}Yi{suffix}" + + +class ProgressHook: + def __init__(self): + self.downloaded = 0 + self.last_print = 0 + + def __call__(self, block, chunk, total): + self.downloaded += chunk + t = time.monotonic() + if t - self.last_print >= 0.5 or self.downloaded == total: + self.last_print = t + if total > 0: + print( + f"\r [{self.downloaded / total * 100.0:3.1f}%] {format_size(self.downloaded)} / {format_size(total)} \r", + end="", + ) + if self.downloaded == total: + print("") + + +def sanitize_filename(filename: str): + import platform + + system = platform.system() + if system == "Windows": + filename = filename.translate(str.maketrans('/\\:"?*|\0', "________")).rstrip( + "." + ) + elif system == "Darwin": + filename = filename.translate(str.maketrans("/:\0", "___")) + else: + filename = filename.translate(str.maketrans("/\0", "__")) + return filename + + +def retry_urlopen(*args, **kwargs): + import http.client + + for _ in range(5): + try: + resp: http.client.HTTPResponse = urllib.request.urlopen(*args, **kwargs) + return resp + except urllib.error.HTTPError as e: + if e.status == 403 and e.headers.get("x-ratelimit-remaining") == "0": + # rate limit + t0 = time.time() + reset_time = t0 + 10 + try: + reset_time = int(e.headers.get("x-ratelimit-reset", 0)) + except ValueError: + pass + reset_time = max(reset_time, t0 + 10) + print( + f"rate limit exceeded, retrying after {reset_time - t0:.1f} seconds" + ) + time.sleep(reset_time - t0) + continue + raise + + +def main(platform: str, repo: str, version: str, cache_asset: bool = False): + print("about to download MaaFramework control units for", platform) + if version: + req = urllib.request.Request( + f"https://api.github.com/repos/{repo}/releases/tags/{version}" + ) + else: + req = urllib.request.Request( + f"https://api.github.com/repos/{repo}/releases/latest" + ) + token = os.environ.get("GH_TOKEN", os.environ.get("GITHUB_TOKEN", None)) + if token: + req.add_header("Authorization", f"Bearer {token}") + resp = retry_urlopen(req).read() + release = json.loads(resp) + + asset = None + for candidate in release["assets"]: + name = candidate["name"] + if not name.startswith(f"MAA-{platform}-"): + continue + if name.endswith(".zip"): + asset = candidate + break + if asset is None: + raise Exception( + f"no MAA-{platform}-* archive found in release {release['tag_name']}" + ) + + if cache_asset and check_asset_cache(asset, download_dir): + print("using cached asset", asset["name"]) + return + url = asset["browser_download_url"] + print("downloading from", url) + download_dir.mkdir(parents=True, exist_ok=True) + local_file = download_dir / sanitize_filename(asset["name"]) + if check_local_digest(local_file, asset["digest"]): + print("reusing matched digest", asset["digest"]) + else: + urllib.request.urlretrieve(url, local_file, reporthook=ProgressHook()) + print("extracting", asset["name"]) + extract_dir = download_dir / sanitize_filename(Path(asset["name"]).stem) + if extract_dir.exists(): + shutil.rmtree(extract_dir) + extract_dir.mkdir(parents=True, exist_ok=True) + shutil.unpack_archive(local_file, extract_dir) + bin_dir = extract_dir / "bin" + if not bin_dir.is_dir(): + raise Exception(f"no `bin` directory found in archive {asset['name']}") + target_dir.mkdir(parents=True, exist_ok=True) + for src in sorted(bin_dir.glob("*ControlUnit*")): + dst = target_dir / src.name + shutil.copy2(src, dst) + print(f"installed {src.name} -> {dst}") + if cache_asset: + set_asset_cache(asset, download_dir) + + +def check_local_digest(file: Path, digest: str): + if not file.exists(): + return False + if not digest.startswith("sha256:"): + print("unsupported digest format:", digest) + return False + hasher = hashlib.sha256() + with file.open("rb") as f: + while True: + chunk = f.read(8192) + if not chunk: + break + hasher.update(chunk) + local_digest = hasher.hexdigest() + return local_digest == digest[len("sha256:") :] + + +def check_asset_cache(asset, extract_dir: Path): + name = asset["name"] + digest = asset["digest"] + asset_cache_file = extract_dir / ".cache_digest.json" + if not asset_cache_file.exists(): + return False + try: + with open(asset_cache_file, "r") as f: + asset_caches = json.load(f) + return asset_caches.get(name) == digest + except (json.JSONDecodeError, AttributeError, KeyError): + return False + + +def set_asset_cache(asset, extract_dir: Path): + name = asset["name"] + digest = asset["digest"] + asset_cache_file = extract_dir / ".cache_digest.json" + try: + with open(asset_cache_file, "r") as f: + asset_caches = json.load(f) + asset_caches[name] = digest + except (FileNotFoundError, json.JSONDecodeError, TypeError): + asset_caches = {name: digest} + with open(asset_cache_file, "w") as f: + json.dump(asset_caches, f)