pleb
240 lines
8.0 KiB
Python
240 lines
8.0 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
from collections.abc import Callable
|
|
from dataclasses import replace
|
|
from pathlib import Path
|
|
from typing import Any
|
|
from urllib.request import Request, urlopen
|
|
|
|
from .fsutil import sha256_file
|
|
from .installer import apply_plan
|
|
from .models import LockedPlugin, Lockfile, Registry, replace_locked_plugins, write_lockfile
|
|
from .planner import create_plan
|
|
from .state import plugin_downloads_dir
|
|
from .updates import FetchBeatMods, FetchReleases, check_updates
|
|
|
|
|
|
DownloadAsset = Callable[[str, Path], dict[str, Any]]
|
|
ApplyPlan = Callable[[dict[str, Any], Path], dict[str, Any]]
|
|
|
|
|
|
def _download_asset(url: str, destination: Path) -> dict[str, Any]:
|
|
destination.parent.mkdir(parents=True, exist_ok=True)
|
|
headers = {"User-Agent": "plugin-helper"}
|
|
token = os.environ.get("GITHUB_TOKEN")
|
|
if token:
|
|
headers["Authorization"] = f"Bearer {token}"
|
|
request = Request(url, headers=headers)
|
|
with urlopen(request, timeout=60) as response:
|
|
destination.write_bytes(response.read())
|
|
return {"url": url, "path": str(destination), "sha256": sha256_file(destination)}
|
|
|
|
|
|
def _reason_for_update(plugin: dict[str, Any], *, actual_sha256: str, beat_saber_version: str) -> str:
|
|
if plugin.get("latestBeatModsVersionId") is not None:
|
|
return (
|
|
f"BeatMods verified {plugin['name']} "
|
|
f"{str(plugin.get('latestTag') or '').removeprefix('beatmods-')} "
|
|
f"for Beat Saber {beat_saber_version} as version id {plugin['latestBeatModsVersionId']}, "
|
|
f"zipHash {plugin.get('latestBeatModsZipHash')}. "
|
|
f"Downloaded from BeatMods CDN and verified SHA-256 {actual_sha256}."
|
|
)
|
|
|
|
digest = plugin.get("latestAssetSha256")
|
|
digest_note = (
|
|
f"GitHub release digest matched {digest}."
|
|
if digest
|
|
else f"GitHub release did not publish a digest; computed SHA-256 {actual_sha256}."
|
|
)
|
|
release = f" release {plugin.get('latestUrl')}" if plugin.get("latestUrl") else ""
|
|
return (
|
|
f"Updated from GitHub {plugin.get('repo')} tag {plugin.get('latestTag')}{release}, "
|
|
f"asset {plugin.get('latestAsset')}. {digest_note}"
|
|
)
|
|
|
|
|
|
def _updated_locked_plugin(
|
|
locked: LockedPlugin,
|
|
plugin: dict[str, Any],
|
|
*,
|
|
actual_sha256: str,
|
|
beat_saber_version: str,
|
|
) -> LockedPlugin:
|
|
return replace(
|
|
locked,
|
|
tag=plugin.get("latestTag"),
|
|
asset=plugin.get("latestAsset"),
|
|
sha256=actual_sha256,
|
|
download_url=plugin.get("latestAssetUrl"),
|
|
reason=_reason_for_update(
|
|
plugin,
|
|
actual_sha256=actual_sha256,
|
|
beat_saber_version=beat_saber_version,
|
|
),
|
|
)
|
|
|
|
|
|
def run_update(
|
|
*,
|
|
instance: str,
|
|
instance_path: Path,
|
|
registry: Registry,
|
|
lockfile: Lockfile,
|
|
lock_path: Path,
|
|
state_root: Path,
|
|
repo_root: Path,
|
|
selected: set[str],
|
|
fetch_releases: FetchReleases,
|
|
fetch_beatmods: FetchBeatMods | None = None,
|
|
include_prerelease: bool = False,
|
|
dry_run: bool = False,
|
|
apply: bool = False,
|
|
download_asset: DownloadAsset = _download_asset,
|
|
apply_plan_func: ApplyPlan = apply_plan,
|
|
install_id: str | None = None,
|
|
) -> dict[str, Any]:
|
|
if not selected:
|
|
raise ValueError("update requires at least one --plugin")
|
|
|
|
report = check_updates(
|
|
registry=registry,
|
|
lockfile=lockfile,
|
|
fetch_releases=fetch_releases,
|
|
fetch_beatmods=fetch_beatmods,
|
|
selected=selected,
|
|
include_prerelease=include_prerelease,
|
|
)
|
|
|
|
locked_by_id = {plugin.id: plugin for plugin in lockfile.plugins}
|
|
reported_ids = {plugin["id"] for plugin in report["plugins"]}
|
|
updated: list[dict[str, Any]] = []
|
|
refused: list[dict[str, Any]] = []
|
|
downloads: list[dict[str, Any]] = []
|
|
replacements: dict[str, LockedPlugin] = {}
|
|
|
|
for missing_id in sorted(selected - reported_ids):
|
|
refused.append(
|
|
{
|
|
"plugin": missing_id,
|
|
"status": "error",
|
|
"messages": ["plugin is not locked for this instance"],
|
|
"reviewReasons": [],
|
|
}
|
|
)
|
|
|
|
for plugin in report["plugins"]:
|
|
plugin_id = plugin["id"]
|
|
if plugin["status"] != "update":
|
|
refused.append(
|
|
{
|
|
"plugin": plugin_id,
|
|
"status": plugin["status"],
|
|
"messages": plugin.get("messages", []),
|
|
"reviewReasons": plugin.get("reviewReasons", []),
|
|
}
|
|
)
|
|
continue
|
|
if not plugin.get("latestAsset") or not plugin.get("latestAssetUrl") or not plugin.get("latestTag"):
|
|
refused.append(
|
|
{
|
|
"plugin": plugin_id,
|
|
"status": "error",
|
|
"messages": ["update candidate is missing tag, asset, or download URL"],
|
|
"reviewReasons": [],
|
|
}
|
|
)
|
|
continue
|
|
if plugin_id not in locked_by_id:
|
|
refused.append(
|
|
{
|
|
"plugin": plugin_id,
|
|
"status": "error",
|
|
"messages": ["plugin is not locked for this instance"],
|
|
"reviewReasons": [],
|
|
}
|
|
)
|
|
continue
|
|
|
|
if dry_run:
|
|
updated.append(
|
|
{
|
|
"plugin": plugin_id,
|
|
"fromTag": plugin.get("currentTag"),
|
|
"toTag": plugin.get("latestTag"),
|
|
"asset": plugin.get("latestAsset"),
|
|
"url": plugin.get("latestAssetUrl"),
|
|
"dryRun": True,
|
|
}
|
|
)
|
|
continue
|
|
|
|
destination = plugin_downloads_dir(state_root, instance, plugin_id) / plugin["latestAsset"]
|
|
downloaded = download_asset(plugin["latestAssetUrl"], destination)
|
|
actual_sha = str(downloaded.get("sha256") or sha256_file(destination))
|
|
expected_sha = plugin.get("latestAssetSha256")
|
|
if expected_sha and actual_sha != expected_sha:
|
|
destination.unlink(missing_ok=True)
|
|
raise ValueError(f"{plugin_id}: downloaded asset sha256 mismatch")
|
|
|
|
downloads.append(
|
|
{
|
|
"plugin": plugin_id,
|
|
"asset": plugin["latestAsset"],
|
|
"path": str(destination),
|
|
"url": plugin["latestAssetUrl"],
|
|
"sha256": actual_sha,
|
|
}
|
|
)
|
|
replacements[plugin_id] = _updated_locked_plugin(
|
|
locked_by_id[plugin_id],
|
|
plugin,
|
|
actual_sha256=actual_sha,
|
|
beat_saber_version=report["beatSaberVersion"],
|
|
)
|
|
updated.append(
|
|
{
|
|
"plugin": plugin_id,
|
|
"fromTag": plugin.get("currentTag"),
|
|
"toTag": plugin.get("latestTag"),
|
|
"asset": plugin.get("latestAsset"),
|
|
"sha256": actual_sha,
|
|
}
|
|
)
|
|
|
|
result: dict[str, Any] = {
|
|
"instance": report["instance"],
|
|
"beatSaberVersion": report["beatSaberVersion"],
|
|
"lockfile": str(lock_path),
|
|
"updated": updated,
|
|
"refused": refused,
|
|
"downloads": downloads,
|
|
"planPath": None,
|
|
"applied": [],
|
|
"dryRun": dry_run,
|
|
}
|
|
if dry_run or not replacements:
|
|
return result
|
|
|
|
updated_lockfile = replace_locked_plugins(lockfile, replacements)
|
|
plan, plan_path = create_plan(
|
|
instance=instance,
|
|
instance_path=instance_path,
|
|
beat_saber_version=updated_lockfile.beat_saber_version,
|
|
registry=registry,
|
|
lockfile=updated_lockfile,
|
|
state_root=state_root,
|
|
repo_root=repo_root,
|
|
selected=set(replacements),
|
|
install_id=install_id,
|
|
)
|
|
result["planPath"] = str(plan_path)
|
|
|
|
if apply:
|
|
apply_result = apply_plan_func(plan, state_root)
|
|
result["applied"] = apply_result["applied"]
|
|
result["statePath"] = apply_result["statePath"]
|
|
|
|
write_lockfile(lock_path, updated_lockfile)
|
|
return result
|