From 9b563421dd2054b2c6b2875fe86fd231ac3f7f87 Mon Sep 17 00:00:00 2001 From: Dusk-92 Date: Wed, 2 Sep 2026 14:23:35 +0200 Subject: [PATCH] Add lightweight package revision tracking --- remote_packages.py | 193 +++++++++++++++++++++++++++++++++++++++++++-- 1 file changed, 188 insertions(+), 5 deletions(-) diff --git a/remote_packages.py b/remote_packages.py index 7504265..e7d02c1 100644 --- a/remote_packages.py +++ b/remote_packages.py @@ -5,6 +5,7 @@ import shutil import stat import struct import tempfile +import time import urllib.error import urllib.request import uuid @@ -40,13 +41,25 @@ def _request(url, accept=None): return urllib.request.Request(url, headers=headers) +JSON_CACHE_TTL = 300 +_JSON_CACHE = {} + + def _get_json(url): + now = time.monotonic() + cached = _JSON_CACHE.get(url) + if cached and now - cached[0] < JSON_CACHE_TTL: + return cached[1] + try: with urllib.request.urlopen(_request(url, "application/vnd.github+json"), timeout=NETWORK_TIMEOUT) as response: - return json.loads(response.read().decode("utf-8")) + data = json.loads(response.read().decode("utf-8")) except (urllib.error.URLError, urllib.error.HTTPError, TimeoutError, json.JSONDecodeError) as exc: raise RemotePackageError(f"GitHub request failed: {exc}") from exc + _JSON_CACHE[url] = (now, data) + return data + def _latest_release(repo): data = _get_json(f"{GITHUB_API}/repos/{repo}/releases/latest") @@ -55,6 +68,24 @@ def _latest_release(repo): return data +def _branch_head_sha(repo, branch): + data = _get_json(f"{GITHUB_API}/repos/{repo}/commits/{branch}") + sha = data.get("sha") if isinstance(data, dict) else None + if not isinstance(sha, str) or len(sha) < 7: + raise RemotePackageError(f"{repo}@{branch}: could not resolve branch revision.") + return sha + + +def _release_revision(release): + tag = release.get("tag_name") if isinstance(release, dict) else None + if isinstance(tag, str) and tag.strip(): + return tag.strip() + release_id = release.get("id") if isinstance(release, dict) else None + if release_id is not None: + return f"release:{release_id}" + return "latest" + + def _find_asset(release, exact_name=None, predicate=None): for asset in release.get("assets", []): name = asset.get("name", "") @@ -469,9 +500,7 @@ def _find_directory_with_file(root, filename): return None -def prepare_vanilla_tweaks(progress=None): - """Download and extract the latest stable tubtubs vanilla-tweaks Windows build.""" - _emit_progress(progress, "Checking vanilla-tweaks release...", None, None) +def vanilla_tweaks_release_info(): release = _latest_release("tubtubs/vanilla-tweaks") asset = _find_asset( release, @@ -481,6 +510,20 @@ def prepare_vanilla_tweaks(progress=None): and not name.lower().endswith(".sha256sum") ), ) + return { + "release": release, + "asset": asset, + "revision": _release_revision(release), + "version": release.get("name") or _release_revision(release), + } + + +def prepare_vanilla_tweaks(progress=None, release_info=None): + """Download and extract the latest stable tubtubs vanilla-tweaks Windows build.""" + _emit_progress(progress, "Checking vanilla-tweaks release...", None, None) + info = release_info or vanilla_tweaks_release_info() + release = info["release"] + asset = info["asset"] zip_path = _download_asset( asset, progress=progress, @@ -500,7 +543,7 @@ def prepare_vanilla_tweaks(progress=None): except OSError: pass - return exe_path, extract_root, release.get("name") or release.get("tag_name", "latest") + return exe_path, extract_root, info["version"], info["revision"] def _write_text_if_missing(path, text): @@ -880,6 +923,146 @@ def _file_sha256(path): return digest.hexdigest().lower() +PACKAGE_STATE_DIR = "package_state" + + +def _package_state_path(target_dir, package_id): + safe_id = "".join( + ch if ch.isalnum() or ch in ("-", "_", ".") else "_" + for ch in str(package_id) + ) + return os.path.join( + target_dir, + MANAGED_ROOT, + PACKAGE_STATE_DIR, + safe_id + ".json", + ) + + +def _hash_directory(path): + digest = hashlib.sha256() + if not os.path.isdir(path): + raise OSError(f"Directory does not exist: {path}") + + found_file = False + for current_root, dirs, files in os.walk(path): + dirs.sort(key=str.casefold) + files.sort(key=str.casefold) + for filename in files: + found_file = True + full_path = os.path.join(current_root, filename) + rel = os.path.relpath(full_path, path).replace(os.sep, "/") + digest.update(b"F\0") + digest.update(rel.encode("utf-8", "surrogatepass")) + digest.update(b"\0") + digest.update(_file_sha256(full_path).encode("ascii")) + digest.update(b"\0") + + if not found_file: + digest.update(b"EMPTY\0") + return digest.hexdigest().lower() + + +def _snapshot_package_paths(target_dir, relative_paths): + entries = {} + for relative in relative_paths: + rel = _safe_relative_path(relative) + full_path = os.path.join(target_dir, rel) + key = rel.replace(os.sep, "/") + + if os.path.isfile(full_path): + entries[key] = { + "type": "file", + "sha256": _file_sha256(full_path), + } + elif os.path.isdir(full_path): + entries[key] = { + "type": "dir", + "sha256": _hash_directory(full_path), + } + else: + raise OSError(f"Package path is missing: {full_path}") + return entries + + +def _load_package_state(target_dir, package_id): + path = _package_state_path(target_dir, package_id) + if not os.path.isfile(path): + return {} + try: + with open(path, "r", encoding="utf-8") as handle: + data = json.load(handle) + return data if isinstance(data, dict) else {} + except (OSError, json.JSONDecodeError, ValueError, TypeError): + return {} + + +def _record_package_state(target_dir, package_id, revision, relative_paths): + entries = _snapshot_package_paths(target_dir, relative_paths) + path = _package_state_path(target_dir, package_id) + os.makedirs(os.path.dirname(path), exist_ok=True) + temp_path = path + ".new" + payload = { + "schema": 1, + "package_id": str(package_id), + "revision": str(revision), + "paths": entries, + } + try: + with open(temp_path, "w", encoding="utf-8") as handle: + json.dump(payload, handle, indent=2, sort_keys=True) + os.replace(temp_path, path) + finally: + if os.path.exists(temp_path): + try: + os.remove(temp_path) + except OSError: + pass + return payload + + +def _package_state_is_current(target_dir, package_id, revision): + data = _load_package_state(target_dir, package_id) + if str(data.get("revision")) != str(revision): + return False + + paths = data.get("paths") + if not isinstance(paths, dict) or not paths: + return False + + try: + current = _snapshot_package_paths(target_dir, paths.keys()) + except OSError: + return False + return current == paths + + +def _record_release_asset_state_if_matching( + target_dir, + package_id, + revision, + relative_path, + asset, +): + """Migrate an existing direct release asset without downloading it again.""" + expected_sha = _asset_sha256(asset) + if not expected_sha: + return False + + rel = _safe_relative_path(relative_path) + path = os.path.join(target_dir, rel) + if not os.path.isfile(path): + return False + + try: + if _file_sha256(path) != expected_sha: + return False + _record_package_state(target_dir, package_id, revision, [rel]) + return True + except OSError: + return False + + def _installed_asset_is_current(path, asset, label): """Validate an installed release asset without trusting existence alone.""" if not os.path.isfile(path):