pleb/plugin-helper
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'main' of gitea.satstack.dev:pleb/beatsaber-plugin-helper

Browse Source
This commit is contained in:
pleb
2026-07-11 15:14:17 -07:00
parent 8ff20bd4e4 538feae811
commit 77a15daeda
35 changed files with 2242 additions and 214 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/plugin_helper/beatmods.py
+21
View File
@@ -1,7 +1,10 @@
from __future__ import annotations
import json
from dataclasses import dataclass
from typing import Any
from urllib.parse import urlencode
from urllib.request import Request, urlopen
@dataclass(frozen=True)
@@ -49,6 +52,24 @@ def normalize_mods(payload: Any) -> list[BeatModsEntry]:
return [normalize_entry(entry) for entry in extract_mods(payload)]
def fetch_verified_mods(game_version: str) -> list[dict[str, Any]]:
query = urlencode(
{
"status": "verified",
"gameVersion": game_version,
"gameName": "BeatSaber",
"platform": "steampc",
}
)
request = Request(
f"https://beatmods.com/api/mods?{query}",
headers={"User-Agent": "Mozilla/5.0 plugin-helper"},
)
with urlopen(request, timeout=20) as response:
data = json.loads(response.read().decode("utf-8"))
return extract_mods(data)
def by_version_id(entries: list[BeatModsEntry]) -> dict[int, BeatModsEntry]:
return {entry.version_id: entry for entry in entries if entry.version_id is not None}
src/plugin_helper/bootstrap.py
+12 -6
View File
@@ -215,6 +215,7 @@ def run_bootstrap(
native: bool | None = None,
progress: Callable[[str], None] | None = None,
ipa_timeout_seconds: int = DEFAULT_IPA_TIMEOUT_SECONDS,
install_id: str | None = None,
) -> dict[str, Any]:
tell = progress or (lambda _message: None)
use_native = is_windows() if native is None else native
@@ -240,6 +241,7 @@ def run_bootstrap(
repo_root=repo_root,
selected={BSIPA_PLUGIN_ID},
require_bootstrap=False,
install_id=install_id,
)
if not plan["changes"]:
raise ValueError("BSIPA bootstrap plan has no changes")
@@ -292,12 +294,14 @@ def run_bootstrap(
"delta": delta,
"health": {},
}
if install_id:
state["installId"] = install_id
if not use_native:
state["proton"] = str(proton or _default_proton())
save_bootstrap_state(state_root, instance, state)
state["health"] = check_bsipa_health(instance_path, state_root, instance)
save_bootstrap_state(state_root, instance, state)
state["statePath"] = str(bootstrap_state_path(state_root, instance))
save_bootstrap_state(state_root, instance, state, install_id=install_id)
state["health"] = check_bsipa_health(instance_path, state_root, instance, install_id=install_id)
save_bootstrap_state(state_root, instance, state, install_id=install_id)
state["statePath"] = str(bootstrap_state_path(state_root, instance, install_id=install_id))
if completed["timedOut"]:
raise TimeoutError(f"IPA.exe -n timed out after {ipa_timeout_seconds}s; state written to {state['statePath']}")
if completed["returncode"] != 0:
@@ -319,11 +323,12 @@ def ensure_healthy_bootstrap(
native: bool | None = None,
progress: Callable[[str], None] | None = None,
ipa_timeout_seconds: int = DEFAULT_IPA_TIMEOUT_SECONDS,
install_id: str | None = None,
) -> None:
if not planning_requires_bootstrap(lockfile.plugins, selected_ids):
return
health = check_bsipa_health(instance_path, state_root, instance)
health = check_bsipa_health(instance_path, state_root, instance, install_id=install_id)
if health["ok"]:
return
@@ -341,8 +346,9 @@ def ensure_healthy_bootstrap(
native=native,
progress=progress,
ipa_timeout_seconds=ipa_timeout_seconds,
install_id=install_id,
)
health = check_bsipa_health(instance_path, state_root, instance)
health = check_bsipa_health(instance_path, state_root, instance, install_id=install_id)
if not health["ok"]:
raise ValueError(bootstrap_health_error(health))
src/plugin_helper/bsipa.py
+10 -4
View File
@@ -33,8 +33,14 @@ def _native_bootstrap_satisfied(state: dict[str, Any]) -> bool:
)
def check_bsipa_health(instance_path: Path, state_root: Path, instance: str) -> dict[str, Any]:
state = load_bootstrap_state(state_root, instance)
def check_bsipa_health(
instance_path: Path,
state_root: Path,
instance: str,
*,
install_id: str | None = None,
) -> dict[str, Any]:
state = load_bootstrap_state(state_root, instance, install_id=install_id)
messages: list[str] = []
required = ["IPA.exe", "winhttp.dll"]
@@ -57,12 +63,12 @@ def check_bsipa_health(instance_path: Path, state_root: Path, instance: str) ->
messages.append("Logs/_latest.log does not show BSIPA startup")
if not state:
messages.append(f"missing bootstrap state: {bootstrap_state_path(state_root, instance)}")
messages.append(f"missing bootstrap state: {bootstrap_state_path(state_root, instance, install_id=install_id)}")
return {
"ok": not messages,
"messages": messages,
"statePath": str(bootstrap_state_path(state_root, instance)),
"statePath": str(bootstrap_state_path(state_root, instance, install_id=install_id)),
"logPath": str(log_path),
"logSha256": sha256_file(log_path) if log_path.is_file() else None,
"bootstrapRecordedAt": state.get("updatedAt"),
src/plugin_helper/cli.py
+106 -18
View File
@@ -7,6 +7,7 @@ from pathlib import Path
from typing import Any
from .config import repo_root, resolve_runtime_config
from .beatmods import fetch_verified_mods
from .bootstrap import run_bootstrap
from .bsipa import check_bsipa_health
from .checker import check_lock
@@ -18,7 +19,8 @@ from .operations import enable_disabled_plugin, restore_known_good_set, save_kno
from .planner import create_plan
from .reports import installed_plugins_report, print_installed_plugins
from .scanner import scan_instance
from .state import load_installed_state
from .state import installation_id, load_installed_state
from .update_runner import run_update
from .updates import check_updates
from .userdata import restore_windows_data_repo, sync_windows_data_repo
@@ -33,12 +35,14 @@ def print_updates(report: dict[str, Any]) -> None:
print(
f"{report['instance']} updates: "
f"{summary['updates']} available, {summary['current']} current, "
f"{summary.get('reviews', 0)} review, {summary.get('skipped', 0)} skipped, "
f"{summary['warnings']} warnings, {summary['errors']} errors"
)
if not plugins:
return
headers = ("Plugin", "Current", "Latest", "Asset", "Status")
headers = ("Plugin", "Current", "Latest", "Asset", "Status", "Notes")
rows = [
(
f"{plugin['name']} ({plugin['id']})",
@@ -46,6 +50,7 @@ def print_updates(report: dict[str, Any]) -> None:
plugin.get("latestTag") or "(unknown)",
plugin.get("latestAsset") or plugin.get("currentAsset") or "(unknown)",
plugin["status"],
", ".join(plugin.get("reviewReasons") or plugin.get("messages") or ()),
)
for plugin in plugins
]
@@ -59,6 +64,20 @@ def print_updates(report: dict[str, Any]) -> None:
print(" ".join(value.ljust(widths[index]) for index, value in enumerate(row)))
def _installation_id_for(runtime: Any, instance_name: str, instance_path: Path) -> str:
return installation_id(
profile_id=getattr(runtime, "profile_id", None),
root=instance_path.parent,
instance=instance_name,
instance_path=instance_path,
)
def _get_installation(runtime: Any, instance_name: str) -> tuple[Any, str]:
instance = get_instance(runtime.instances_roots, instance_name)
return instance, _installation_id_for(runtime, instance_name, instance.path)
def _add_common(parser: argparse.ArgumentParser, *, suppress_default: bool = False) -> None:
default = argparse.SUPPRESS if suppress_default else None
parser.add_argument("--config", default=default, help="plugin-helper config TOML path")
@@ -142,7 +161,7 @@ def build_parser() -> argparse.ArgumentParser:
updates = subcommands.add_parser(
"updates",
help="Check GitHub for newer matching releases for locked plugins",
help="Check GitHub and BeatMods for newer matching releases for locked plugins",
parents=[_common_parent()],
)
updates.add_argument("--instance", required=True)
@@ -152,6 +171,21 @@ def build_parser() -> argparse.ArgumentParser:
updates.add_argument("--include-prerelease", action="store_true", help="Include prerelease GitHub releases")
updates.add_argument("--json", action="store_true", help="Print full JSON update output")
update = subcommands.add_parser(
"update",
help="Download selected update candidates, update the lock, and create an install plan",
parents=[_common_parent()],
)
update.add_argument("--instance", required=True)
update.add_argument("--registry", default="registry/plugins")
update.add_argument("--lockfile")
update.add_argument("--plugin", action="append", required=True, help="Update this locked plugin id; repeatable")
update.add_argument("--include-prerelease", action="store_true", help="Include prerelease GitHub releases")
update.add_argument("--json", action="store_true", help="Print full JSON update output")
update_mode = update.add_mutually_exclusive_group()
update_mode.add_argument("--apply", action="store_true", help="Apply the generated update plan")
update_mode.add_argument("--dry-run", action="store_true", help="Show selected updates without downloading or writing")
plan = subcommands.add_parser(
"plan",
help="Create a dry-run install plan from the catalog and version lock",
@@ -258,12 +292,13 @@ def _run_menu(
return 1
choices: list[InstallationChoice] = []
for index, root in enumerate(runtime.instances_roots, start=1):
for root in runtime.instances_roots:
install_label = str(root) if len(runtime.instances_roots) > 1 else "Default"
for instance in list_instances(root):
install_id = _installation_id_for(runtime, instance.name, instance.path)
choices.append(
InstallationChoice(
install_id=f"root-{index}",
install_id=install_id,
install_label=install_label,
instance_name=instance.name,
instance_path=instance.path,
@@ -347,17 +382,19 @@ def run(argv: list[str] | None = None) -> int:
return 0
if args.command == "state":
_json(load_installed_state(st_root, args.instance))
_instance, install_id = _get_installation(runtime, args.instance)
_json(load_installed_state(st_root, args.instance, install_id=install_id))
return 0
if args.command == "installed":
instance, install_id = _get_installation(runtime, args.instance)
root = repo_root()
registry_path = (root / args.registry).resolve() if not Path(args.registry).is_absolute() else Path(args.registry)
lock_path = Path(args.lockfile) if args.lockfile else root / "locks" / f"{args.instance}.lock.toml"
if not lock_path.is_absolute():
lock_path = (root / lock_path).resolve()
result = installed_plugins_report(
installed_state=load_installed_state(st_root, args.instance),
installed_state=load_installed_state(st_root, instance.name, install_id=install_id),
registry=load_registry(registry_path),
lockfile=load_lockfile(lock_path),
)
@@ -406,6 +443,7 @@ def run(argv: list[str] | None = None) -> int:
registry=load_registry(registry_path),
lockfile=load_lockfile(lock_path),
fetch_releases=fetch_releases,
fetch_beatmods=fetch_verified_mods,
selected=set(args.plugin) if args.plugin else None,
include_prerelease=args.include_prerelease,
)
@@ -415,8 +453,53 @@ def run(argv: list[str] | None = None) -> int:
print_updates(result)
return 2 if result["summary"]["errors"] else 0
if args.command == "update":
instance, install_id = _get_installation(runtime, args.instance)
root = repo_root()
registry_path = (root / args.registry).resolve() if not Path(args.registry).is_absolute() else Path(args.registry)
lock_path = Path(args.lockfile) if args.lockfile else root / "locks" / f"{args.instance}.lock.toml"
if not lock_path.is_absolute():
lock_path = (root / lock_path).resolve()
result = run_update(
instance=args.instance,
instance_path=instance.path,
registry=load_registry(registry_path),
lockfile=load_lockfile(lock_path),
lock_path=lock_path,
state_root=st_root,
repo_root=root,
selected=set(args.plugin),
fetch_releases=fetch_releases,
fetch_beatmods=fetch_verified_mods,
include_prerelease=args.include_prerelease,
dry_run=args.dry_run,
apply=args.apply,
install_id=install_id,
)
if args.json:
_json(result)
else:
action = "Would update" if result["dryRun"] else "Updated"
print(f"{action}: {len(result['updated'])}")
for item in result["updated"]:
print(f" {item['plugin']}: {item.get('fromTag')} -> {item.get('toTag')} ({item.get('asset')})")
if result["refused"]:
print("Refused:")
for item in result["refused"]:
notes = ", ".join(item.get("reviewReasons") or item.get("messages") or ())
suffix = f": {notes}" if notes else ""
print(f" {item['plugin']}: {item['status']}{suffix}")
if result.get("planPath"):
print(f"Plan: {result['planPath']}")
if not result["dryRun"] and result["updated"]:
print(f"Lockfile: {result['lockfile']}")
if args.apply:
print(f"Applied: {len(result['applied'])}")
print(f"State: {result.get('statePath')}")
return 0 if not result["refused"] else 2
if args.command == "bootstrap":
instance = get_instance(inst_roots, args.instance)
instance, install_id = _get_installation(runtime, args.instance)
root = repo_root()
registry_path = (root / args.registry).resolve() if not Path(args.registry).is_absolute() else Path(args.registry)
lock_path = Path(args.lockfile) if args.lockfile else root / "locks" / f"{args.instance}.lock.toml"
@@ -434,6 +517,7 @@ def run(argv: list[str] | None = None) -> int:
proton=Path(args.proton).expanduser() if args.proton else None,
native=True if args.native else None,
progress=lambda message: print(f" {message}", flush=True),
install_id=install_id,
)
if args.json:
_json(result)
@@ -453,8 +537,8 @@ def run(argv: list[str] | None = None) -> int:
return 0 if result["health"]["ok"] else 2
if args.command == "bootstrap-check":
instance = get_instance(inst_roots, args.instance)
result = check_bsipa_health(instance.path, st_root, args.instance)
instance, install_id = _get_installation(runtime, args.instance)
result = check_bsipa_health(instance.path, st_root, args.instance, install_id=install_id)
if args.json:
_json(result)
else:
@@ -466,7 +550,7 @@ def run(argv: list[str] | None = None) -> int:
return 0 if result["ok"] else 2
if args.command == "plan":
instance = get_instance(inst_roots, args.instance)
instance, install_id = _get_installation(runtime, args.instance)
root = repo_root()
registry_path = (root / args.registry).resolve() if not Path(args.registry).is_absolute() else Path(args.registry)
lock_path = Path(args.lockfile) if args.lockfile else root / "locks" / f"{args.instance}.lock.toml"
@@ -484,6 +568,7 @@ def run(argv: list[str] | None = None) -> int:
state_root=st_root,
repo_root=root,
selected=selected,
install_id=install_id,
)
print(f"Wrote plan: {path}")
print(f"Changes: {len(plan['changes'])}")
@@ -500,8 +585,8 @@ def run(argv: list[str] | None = None) -> int:
return 0
if args.command == "uninstall":
instance = get_instance(inst_roots, args.instance)
result = uninstall_plugin(args.instance, instance.path, st_root, args.plugin, force=args.force)
instance, install_id = _get_installation(runtime, args.instance)
result = uninstall_plugin(args.instance, instance.path, st_root, args.plugin, force=args.force, install_id=install_id)
print(f"Removed: {len(result['removed'])}")
if result["skipped"]:
print("Skipped:")
@@ -510,8 +595,8 @@ def run(argv: list[str] | None = None) -> int:
return 0 if result["stateUpdated"] else 2
if args.command == "disable":
instance = get_instance(inst_roots, args.instance)
result = disable_plugin(args.instance, instance.path, st_root, args.plugin, force=args.force)
instance, install_id = _get_installation(runtime, args.instance)
result = disable_plugin(args.instance, instance.path, st_root, args.plugin, force=args.force, install_id=install_id)
print(f"Disabled: {args.plugin}")
print(f"Removed: {len(result['removed'])}")
if result["skipped"]:
@@ -521,7 +606,7 @@ def run(argv: list[str] | None = None) -> int:
return 0 if result["stateUpdated"] else 2
if args.command == "enable":
instance = get_instance(inst_roots, args.instance)
instance, install_id = _get_installation(runtime, args.instance)
progress = (lambda message: print(f" {message}", flush=True))
result = enable_disabled_plugin(
instance=args.instance,
@@ -531,6 +616,7 @@ def run(argv: list[str] | None = None) -> int:
registry=args.registry,
lockfile=args.lockfile,
progress=progress,
install_id=install_id,
)
print(f"Enabled: {args.plugin}")
print(f"Plan: {result['planPath']}")
@@ -539,7 +625,8 @@ def run(argv: list[str] | None = None) -> int:
return 0
if args.command == "save-known-good":
result = save_known_good_set(instance=args.instance, state_root=st_root)
_instance, install_id = _get_installation(runtime, args.instance)
result = save_known_good_set(instance=args.instance, state_root=st_root, install_id=install_id)
if args.json:
_json(result)
else:
@@ -549,7 +636,7 @@ def run(argv: list[str] | None = None) -> int:
return 0
if args.command == "restore-known-good":
instance = get_instance(inst_roots, args.instance)
instance, install_id = _get_installation(runtime, args.instance)
progress = (lambda message: print(f" {message}", flush=True))
result = restore_known_good_set(
instance=args.instance,
@@ -558,6 +645,7 @@ def run(argv: list[str] | None = None) -> int:
registry=args.registry,
lockfile=args.lockfile,
progress=progress,
install_id=install_id,
)
if args.json:
_json(result)
src/plugin_helper/config.py
+11 -1
View File
@@ -189,7 +189,17 @@ def resolve_runtime_config(
) -> RuntimeConfig:
repo = root or repo_root()
local_config, loaded_path, loaded = load_local_config(config_path_value, root=repo)
profile = _select_profile(local_config, profile_id)
explicit_runtime_paths = bool(
instances_root_value
or state_dir_value
or _env_instances_roots(repo)
or _env_state_root(repo)
)
profile = (
_select_profile(local_config, profile_id)
if profile_id or not explicit_runtime_paths
else None
)
resolved_instances = (
_resolve_path_list(instances_root_value, repo)
src/plugin_helper/installer.py
+28 -11
View File
@@ -7,7 +7,7 @@ from typing import Any
from zipfile import ZipFile
from .fsutil import ensure_inside, ensure_relative, sha256_bytes, sha256_file
from .state import backups_dir, load_installed_state, save_installed_state
from .state import backups_dir, installed_state_path, load_installed_state, save_installed_state
def _timestamp() -> str:
@@ -26,12 +26,13 @@ def _backup_existing(instance_path: Path, backup_root: Path, rel_target: str) ->
def apply_plan(plan: dict[str, Any], state_root: Path) -> dict[str, Any]:
instance = plan["instance"]
install_id = plan.get("installId")
instance_path = Path(plan["instancePath"])
if not instance_path.is_dir():
raise FileNotFoundError(f"instance path does not exist: {instance_path}")
backup_root = backups_dir(state_root, instance) / f"apply-{_timestamp()}"
installed_state = load_installed_state(state_root, instance)
backup_root = backups_dir(state_root, instance, install_id=install_id) / f"apply-{_timestamp()}"
installed_state = load_installed_state(state_root, instance, install_id=install_id)
installed_state.setdefault("beatSaberVersion", plan.get("beatSaberVersion"))
installed_state.setdefault("plugins", {})
@@ -80,12 +81,20 @@ def apply_plan(plan: dict[str, Any], state_root: Path) -> dict[str, Any]:
installed_state.setdefault("disabledPlugins", {}).pop(change["plugin"], None)
applied.append({"path": rel_target, "plugin": change["plugin"], "backup": backup})
save_installed_state(state_root, instance, installed_state)
return {"applied": applied, "statePath": str(state_root / "instances" / instance / "installed.json")}
save_installed_state(state_root, instance, installed_state, install_id=install_id)
return {"applied": applied, "statePath": str(installed_state_path(state_root, instance, install_id=install_id))}
def uninstall_plugin(instance: str, instance_path: Path, state_root: Path, plugin_id: str, force: bool = False) -> dict[str, Any]:
installed_state = load_installed_state(state_root, instance)
def uninstall_plugin(
instance: str,
instance_path: Path,
state_root: Path,
plugin_id: str,
force: bool = False,
*,
install_id: str | None = None,
) -> dict[str, Any]:
installed_state = load_installed_state(state_root, instance, install_id=install_id)
plugin_state = installed_state.get("plugins", {}).get(plugin_id)
if not plugin_state:
raise KeyError(f"plugin is not recorded in managed install state: {plugin_id}")
@@ -109,12 +118,20 @@ def uninstall_plugin(instance: str, instance_path: Path, state_root: Path, plugi
return {"removed": removed, "skipped": skipped, "stateUpdated": False}
installed_state.get("plugins", {}).pop(plugin_id, None)
save_installed_state(state_root, instance, installed_state)
save_installed_state(state_root, instance, installed_state, install_id=install_id)
return {"removed": removed, "skipped": skipped, "stateUpdated": True}
def disable_plugin(instance: str, instance_path: Path, state_root: Path, plugin_id: str, force: bool = False) -> dict[str, Any]:
installed_state = load_installed_state(state_root, instance)
def disable_plugin(
instance: str,
instance_path: Path,
state_root: Path,
plugin_id: str,
force: bool = False,
*,
install_id: str | None = None,
) -> dict[str, Any]:
installed_state = load_installed_state(state_root, instance, install_id=install_id)
plugin_state = installed_state.get("plugins", {}).get(plugin_id)
if not plugin_state:
if plugin_id in installed_state.get("disabledPlugins", {}):
@@ -143,5 +160,5 @@ def disable_plugin(instance: str, instance_path: Path, state_root: Path, plugin_
disabled_state["disabledAt"] = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
installed_state.setdefault("disabledPlugins", {})[plugin_id] = disabled_state
installed_state.get("plugins", {}).pop(plugin_id, None)
save_installed_state(state_root, instance, installed_state)
save_installed_state(state_root, instance, installed_state, install_id=install_id)
return {"removed": removed, "skipped": skipped, "stateUpdated": True}
src/plugin_helper/models.py
+46 -1
View File
@@ -1,7 +1,7 @@
from __future__ import annotations
import tomllib
from dataclasses import dataclass, field
from dataclasses import dataclass, field, replace
from pathlib import Path
from typing import Any
@@ -42,6 +42,7 @@ class LockedPlugin:
tag: str | None
asset: str | None
sha256: str | None
download_url: str | None = None
install_strategy: str | None = None
reason: str | None = None
@@ -53,6 +54,49 @@ class Lockfile:
plugins: tuple[LockedPlugin, ...]
def _quote_toml_string(value: str) -> str:
return '"' + value.replace("\\", "\\\\").replace('"', '\\"') + '"'
def lockfile_to_toml(lockfile: Lockfile) -> str:
lines = [
f"beat_saber_version = {_quote_toml_string(lockfile.beat_saber_version)}",
f"instance = {_quote_toml_string(lockfile.instance)}",
"",
]
for plugin in lockfile.plugins:
lines.append("[[plugins]]")
lines.append(f"id = {_quote_toml_string(plugin.id)}")
if plugin.repo is not None:
lines.append(f"repo = {_quote_toml_string(plugin.repo)}")
if plugin.tag is not None:
lines.append(f"tag = {_quote_toml_string(plugin.tag)}")
if plugin.asset is not None:
lines.append(f"asset = {_quote_toml_string(plugin.asset)}")
if plugin.download_url is not None:
lines.append(f"download_url = {_quote_toml_string(plugin.download_url)}")
if plugin.sha256 is not None:
lines.append(f"sha256 = {_quote_toml_string(plugin.sha256)}")
if plugin.install_strategy is not None:
lines.append(f"install_strategy = {_quote_toml_string(plugin.install_strategy)}")
if plugin.reason is not None:
lines.append(f"reason = {_quote_toml_string(plugin.reason)}")
lines.append("")
return "\n".join(lines)
def write_lockfile(path: Path, lockfile: Lockfile) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(lockfile_to_toml(lockfile), encoding="utf-8")
def replace_locked_plugins(lockfile: Lockfile, replacements: dict[str, LockedPlugin]) -> Lockfile:
return replace(
lockfile,
plugins=tuple(replacements.get(plugin.id, plugin) for plugin in lockfile.plugins),
)
def _load_toml(path: Path) -> dict[str, Any]:
with path.open("rb") as handle:
return tomllib.load(handle)
@@ -115,6 +159,7 @@ def load_lockfile(path: Path) -> Lockfile:
tag=item.get("tag"),
asset=item.get("asset"),
sha256=item.get("sha256"),
download_url=item.get("download_url"),
install_strategy=item.get("install_strategy"),
reason=item.get("reason"),
)
src/plugin_helper/operations.py
+19 -9
View File
@@ -39,8 +39,9 @@ def enable_disabled_plugin(
lockfile: str | None = None,
repo: Path | None = None,
progress: Callable[[str], None] | None = None,
install_id: str | None = None,
) -> dict[str, Any]:
installed_state = load_installed_state(state_root, instance)
installed_state = load_installed_state(state_root, instance, install_id=install_id)
if plugin_id in installed_state.get("plugins", {}):
raise KeyError(f"plugin is already enabled: {plugin_id}")
@@ -65,6 +66,7 @@ def enable_disabled_plugin(
repo_root=root,
selected={plugin_id},
require_bootstrap=False,
install_id=install_id,
)
ensure_healthy_bootstrap(
@@ -77,6 +79,7 @@ def enable_disabled_plugin(
repo_root=root,
selected_ids={plugin_id},
progress=progress,
install_id=install_id,
)
tell(f"Applying {plugin_id}")
@@ -94,6 +97,7 @@ def enable_disabled_plugins(
lockfile: str | None = None,
repo: Path | None = None,
progress: Callable[[str], None] | None = None,
install_id: str | None = None,
) -> dict[str, Any]:
if not plugin_ids:
return {"enabled": [], "errors": []}
@@ -104,7 +108,7 @@ def enable_disabled_plugins(
lockfile=lockfile,
repo=repo,
)
installed_state = load_installed_state(state_root, instance)
installed_state = load_installed_state(state_root, instance, install_id=install_id)
enabled_plugins = installed_state.get("plugins", {})
locked_ids = {plugin.id for plugin in loaded_lockfile.plugins}
@@ -136,6 +140,7 @@ def enable_disabled_plugins(
repo_root=root,
selected=selected_ids,
require_bootstrap=False,
install_id=install_id,
)
ensure_healthy_bootstrap(
@@ -148,6 +153,7 @@ def enable_disabled_plugins(
repo_root=root,
selected_ids=selected_ids,
progress=progress,
install_id=install_id,
)
tell(f"Applying {len(plannable_ids)} plugins")
@@ -163,16 +169,18 @@ def enable_disabled_plugins(
return {"enabled": enabled, "errors": errors}
def save_known_good_set(*, instance: str, state_root: Path) -> dict[str, Any]:
def save_known_good_set(*, instance: str, state_root: Path, install_id: str | None = None) -> dict[str, Any]:
"""Record the currently enabled plugin ids as the known-good set for this instance."""
installed_state = load_installed_state(state_root, instance)
installed_state = load_installed_state(state_root, instance, install_id=install_id)
plugin_ids = sorted(installed_state.get("plugins", {}))
known_good = {
"instance": instance,
"beatSaberVersion": installed_state.get("beatSaberVersion"),
"pluginIds": plugin_ids,
}
save_known_good_state(state_root, instance, known_good)
if install_id:
known_good["installId"] = install_id
save_known_good_state(state_root, instance, known_good, install_id=install_id)
return known_good
@@ -185,14 +193,15 @@ def restore_known_good_set(
lockfile: str | None = None,
repo: Path | None = None,
progress: Callable[[str], None] | None = None,
install_id: str | None = None,
) -> dict[str, Any]:
"""Disable plugins outside the saved known-good set and re-enable the ones that are missing."""
known_good = load_known_good_state(state_root, instance)
known_good = load_known_good_state(state_root, instance, install_id=install_id)
known_good_ids = set(known_good.get("pluginIds", []))
if not known_good:
raise KeyError(f"no known-good set is saved for this instance: {instance}")
installed_state = load_installed_state(state_root, instance)
installed_state = load_installed_state(state_root, instance, install_id=install_id)
currently_enabled = set(installed_state.get("plugins", {}))
to_disable = sorted(currently_enabled - known_good_ids)
@@ -200,7 +209,7 @@ def restore_known_good_set(
disable_errors: list[dict[str, str]] = []
for plugin_id in to_disable:
try:
result = disable_plugin(instance, instance_path, state_root, plugin_id, False)
result = disable_plugin(instance, instance_path, state_root, plugin_id, False, install_id=install_id)
if result["stateUpdated"]:
disabled.append(plugin_id)
else:
@@ -209,7 +218,7 @@ def restore_known_good_set(
except Exception as exc:
disable_errors.append({"plugin": plugin_id, "error": str(exc)})
installed_state = load_installed_state(state_root, instance)
installed_state = load_installed_state(state_root, instance, install_id=install_id)
currently_enabled = set(installed_state.get("plugins", {}))
to_enable = sorted(known_good_ids - currently_enabled)
enable_result = enable_disabled_plugins(
@@ -221,6 +230,7 @@ def restore_known_good_set(
lockfile=lockfile,
repo=repo,
progress=progress,
install_id=install_id,
)
return {
src/plugin_helper/planner.py
+10 -3
View File
@@ -10,7 +10,11 @@ from zipfile import ZipFile
from .fsutil import ensure_relative, sha256_bytes, sha256_file
from .models import Lockfile, Registry, VALID_STRATEGIES
from .bsipa import bootstrap_health_error, check_bsipa_health, planning_requires_bootstrap
from .state import downloads_dir, plans_dir, plugin_downloads_dir
from .state import (
downloads_dir,
plans_dir,
plugin_downloads_dir,
)
ALLOWED_BSIPA_TOP_LEVEL = {"IPA", "Libs", "Plugins"}
@@ -96,6 +100,7 @@ def create_plan(
repo_root: Path,
selected: set[str] | None = None,
require_bootstrap: bool = True,
install_id: str | None = None,
) -> tuple[dict[str, Any], Path]:
selected_ids = (
_expand_required_dependencies(selected, registry, lockfile)
@@ -106,7 +111,7 @@ def create_plan(
warnings: list[str] = []
if require_bootstrap and planning_requires_bootstrap(lockfile.plugins, selected_ids):
health = check_bsipa_health(instance_path, state_root, instance)
health = check_bsipa_health(instance_path, state_root, instance, install_id=install_id)
if not health["ok"]:
raise ValueError(bootstrap_health_error(health))
@@ -175,7 +180,9 @@ def create_plan(
"warnings": warnings,
"changes": changes,
}
plan_path = plans_dir(state_root, instance) / f"plan-{_now_slug()}.json"
if install_id:
plan["installId"] = install_id
plan_path = plans_dir(state_root, instance, install_id=install_id) / f"plan-{_now_slug()}.json"
with plan_path.open("w", encoding="utf-8") as handle:
json.dump(plan, handle, indent=2, sort_keys=True)
handle.write("\n")
src/plugin_helper/recipes/__init__.py
+1
View File
@@ -0,0 +1 @@
"""Asset reconstruction recipes for helper-managed composite packages."""
src/plugin_helper/recipes/cinema_tools.py
+170
View File
@@ -0,0 +1,170 @@
from __future__ import annotations
import argparse
import hashlib
import os
import shutil
import sys
import tempfile
from dataclasses import dataclass
from pathlib import Path
from urllib.request import Request, urlopen
from zipfile import ZIP_STORED, ZipFile, ZipInfo
PLUGIN_ID = "cinema-tools"
DEFAULT_INSTANCE = "1.44.1"
YT_DLP_VERSION = "2026.07.04"
YT_DLP_URL = f"https://github.com/yt-dlp/yt-dlp/releases/download/{YT_DLP_VERSION}/yt-dlp.exe"
YT_DLP_SHA256 = "52fe3c26dcf71fbdc85b528589020bb0b8e383155cfa81b64dd447bbe35e24b8"
FFMPEG_VERSION = "8.1.2"
FFMPEG_URL = (
f"https://github.com/GyanD/codexffmpeg/releases/download/{FFMPEG_VERSION}/"
f"ffmpeg-{FFMPEG_VERSION}-essentials_build.zip"
)
FFMPEG_ZIP_SHA256 = "db580001caa24ac104c8cb856cd113a87b0a443f7bdf47d8c12b1d740584a2ec"
FFMPEG_EXE_SHA256 = "1326dde4c84ff1f96fe6b8916c5bed29e163e9b5dccf995f6f3db069d143ec5e"
ASSET_NAME = f"CinemaTools-yt-dlp-{YT_DLP_VERSION}-ffmpeg-{FFMPEG_VERSION}.zip"
BUNDLE_SHA256 = "4d299e40f747abb9777838542ca68ec450e0b86e6eb11af341e6bd568f47edf2"
ZIP_TIMESTAMP = (1980, 1, 1, 0, 0, 0)
@dataclass(frozen=True)
class BuiltAsset:
path: Path
sha256: str
size: int
def _default_state_root() -> Path:
configured = os.environ.get("PLUGIN_HELPER_STATE_DIR")
if configured:
return Path(configured).expanduser()
xdg_state_home = os.environ.get("XDG_STATE_HOME")
if xdg_state_home:
return Path(xdg_state_home).expanduser() / "plugin-helper"
return Path("~/.local/state/plugin-helper").expanduser()
def _sha256_file(path: Path) -> str:
digest = hashlib.sha256()
with path.open("rb") as handle:
for chunk in iter(lambda: handle.read(1024 * 1024), b""):
digest.update(chunk)
return digest.hexdigest()
def _download(url: str, destination: Path) -> None:
request = Request(url, headers={"User-Agent": "plugin-helper"})
destination.parent.mkdir(parents=True, exist_ok=True)
with urlopen(request, timeout=120) as response, destination.open("wb") as handle:
shutil.copyfileobj(response, handle)
def _verify(path: Path, expected_sha256: str, label: str) -> None:
actual = _sha256_file(path)
if actual != expected_sha256:
raise ValueError(f"{label} sha256 mismatch: expected {expected_sha256}, got {actual}")
def _read_ffmpeg_exe(archive_path: Path) -> bytes:
with ZipFile(archive_path) as archive:
matches = [
info
for info in archive.infolist()
if not info.is_dir() and info.filename.replace("\\", "/").endswith("/bin/ffmpeg.exe")
]
if len(matches) != 1:
raise ValueError(f"expected exactly one ffmpeg.exe under */bin/, found {len(matches)}")
data = archive.read(matches[0])
actual = hashlib.sha256(data).hexdigest()
if actual != FFMPEG_EXE_SHA256:
raise ValueError(f"ffmpeg.exe sha256 mismatch: expected {FFMPEG_EXE_SHA256}, got {actual}")
return data
def _write_member(archive: ZipFile, name: str, data: bytes) -> None:
info = ZipInfo(name, date_time=ZIP_TIMESTAMP)
info.compress_type = ZIP_STORED
info.external_attr = 0o644 << 16
archive.writestr(info, data)
def build_asset(output: Path, *, work_dir: Path | None = None) -> BuiltAsset:
output = output.expanduser()
output.parent.mkdir(parents=True, exist_ok=True)
with tempfile.TemporaryDirectory(dir=work_dir) as tmp:
tmp_path = Path(tmp)
yt_dlp_path = tmp_path / "yt-dlp.exe"
ffmpeg_zip_path = tmp_path / f"ffmpeg-{FFMPEG_VERSION}-essentials_build.zip"
staged_output = tmp_path / ASSET_NAME
_download(YT_DLP_URL, yt_dlp_path)
_verify(yt_dlp_path, YT_DLP_SHA256, "yt-dlp.exe")
_download(FFMPEG_URL, ffmpeg_zip_path)
_verify(ffmpeg_zip_path, FFMPEG_ZIP_SHA256, "ffmpeg essentials zip")
yt_dlp_data = yt_dlp_path.read_bytes()
ffmpeg_data = _read_ffmpeg_exe(ffmpeg_zip_path)
with ZipFile(staged_output, "w") as archive:
_write_member(archive, "Libs/yt-dlp.exe", yt_dlp_data)
_write_member(archive, "Libs/ffmpeg.exe", ffmpeg_data)
actual_bundle_sha256 = _sha256_file(staged_output)
if actual_bundle_sha256 != BUNDLE_SHA256:
raise ValueError(
f"{ASSET_NAME} sha256 mismatch: expected {BUNDLE_SHA256}, got {actual_bundle_sha256}"
)
shutil.move(str(staged_output), output)
return BuiltAsset(path=output, sha256=_sha256_file(output), size=output.stat().st_size)
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(description="Rebuild the helper-managed Cinema tools bundle.")
parser.add_argument("--instance", default=DEFAULT_INSTANCE, help="Beat Saber instance/version cache key")
parser.add_argument(
"--state-dir",
type=Path,
default=_default_state_root(),
help="plugin-helper state root; defaults to PLUGIN_HELPER_STATE_DIR, XDG_STATE_HOME, or ~/.local/state/plugin-helper",
)
parser.add_argument(
"--output",
type=Path,
help="Exact output path; defaults to <state-dir>/cache/downloads/<instance>/cinema-tools/<asset>",
)
parser.add_argument("--work-dir", type=Path, help="Parent directory for temporary downloads")
return parser
def run(argv: list[str] | None = None) -> int:
parser = build_parser()
args = parser.parse_args(argv)
output = args.output or args.state_dir / "cache" / "downloads" / args.instance / PLUGIN_ID / ASSET_NAME
try:
built = build_asset(output, work_dir=args.work_dir)
except Exception as exc:
print(f"error: {exc}", file=sys.stderr)
return 2
print(f"Wrote: {built.path}")
print(f"Size: {built.size}")
print(f"SHA-256: {built.sha256}")
print("Members:")
print(f" Libs/yt-dlp.exe {YT_DLP_SHA256}")
print(f" Libs/ffmpeg.exe {FFMPEG_EXE_SHA256}")
return 0
def main() -> None:
raise SystemExit(run())
if __name__ == "__main__":
main()
src/plugin_helper/state.py
+87 -33
View File
@@ -1,5 +1,7 @@
from __future__ import annotations
import hashlib
import re
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
@@ -7,49 +9,97 @@ from typing import Any
from .fsutil import atomic_write_json, read_json
def instance_state_dir(state_root: Path, instance: str) -> Path:
return state_root / "instances" / instance
def _slug(value: str) -> str:
slug = re.sub(r"[^A-Za-z0-9_.-]+", "-", value.strip()).strip("-._")
return slug[:48] or "install"
def installed_state_path(state_root: Path, instance: str) -> Path:
return instance_state_dir(state_root, instance) / "installed.json"
def bootstrap_state_path(state_root: Path, instance: str) -> Path:
return instance_state_dir(state_root, instance) / "bootstrap.json"
def load_installed_state(state_root: Path, instance: str) -> dict[str, Any]:
return read_json(
installed_state_path(state_root, instance),
{"instance": instance, "plugins": {}},
def installation_id(
*,
profile_id: str | None,
root: Path | None,
instance: str,
instance_path: Path,
) -> str:
resolved_root = (
root.expanduser().resolve(strict=False)
if root
else instance_path.parent.expanduser().resolve(strict=False)
)
resolved_path = instance_path.expanduser().resolve(strict=False)
profile = profile_id or "default"
digest = hashlib.sha256(
f"{profile}\0{resolved_root}\0{resolved_path}".encode("utf-8")
).hexdigest()[:12]
return f"{_slug(profile)}-{_slug(instance)}-{digest}"
def load_bootstrap_state(state_root: Path, instance: str) -> dict[str, Any]:
return read_json(bootstrap_state_path(state_root, instance), {})
def instance_state_dir(state_root: Path, instance: str, *, install_id: str | None = None) -> Path:
if install_id:
return state_root / "installs" / install_id
return state_root / "installs" / _slug(instance)
def save_bootstrap_state(state_root: Path, instance: str, state: dict[str, Any]) -> None:
def installed_state_path(state_root: Path, instance: str, *, install_id: str | None = None) -> Path:
return instance_state_dir(state_root, instance, install_id=install_id) / "installed.json"
def bootstrap_state_path(state_root: Path, instance: str, *, install_id: str | None = None) -> Path:
return instance_state_dir(state_root, instance, install_id=install_id) / "bootstrap.json"
def known_good_state_path(state_root: Path, instance: str, *, install_id: str | None = None) -> Path:
return instance_state_dir(state_root, instance, install_id=install_id) / "known-good.json"
def load_installed_state(state_root: Path, instance: str, *, install_id: str | None = None) -> dict[str, Any]:
default = {"instance": instance, "plugins": {}}
state = read_json(installed_state_path(state_root, instance, install_id=install_id), default)
if install_id:
state.setdefault("installId", install_id)
return state
def load_bootstrap_state(state_root: Path, instance: str, *, install_id: str | None = None) -> dict[str, Any]:
return read_json(bootstrap_state_path(state_root, instance, install_id=install_id), {})
def save_bootstrap_state(
state_root: Path,
instance: str,
state: dict[str, Any],
*,
install_id: str | None = None,
) -> None:
state.setdefault("instance", instance)
if install_id:
state.setdefault("installId", install_id)
state["updatedAt"] = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
atomic_write_json(bootstrap_state_path(state_root, instance), state)
atomic_write_json(bootstrap_state_path(state_root, instance, install_id=install_id), state)
def save_installed_state(state_root: Path, instance: str, state: dict[str, Any]) -> None:
def save_installed_state(
state_root: Path,
instance: str,
state: dict[str, Any],
*,
install_id: str | None = None,
) -> None:
state.setdefault("instance", instance)
if install_id:
state.setdefault("installId", install_id)
state["updatedAt"] = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
atomic_write_json(installed_state_path(state_root, instance), state)
atomic_write_json(installed_state_path(state_root, instance, install_id=install_id), state)
def plans_dir(state_root: Path, instance: str) -> Path:
path = instance_state_dir(state_root, instance) / "plans"
def plans_dir(state_root: Path, instance: str, *, install_id: str | None = None) -> Path:
path = instance_state_dir(state_root, instance, install_id=install_id) / "plans"
path.mkdir(parents=True, exist_ok=True)
return path
def downloads_dir(state_root: Path, instance: str) -> Path:
path = instance_state_dir(state_root, instance) / "downloads"
path = state_root / "cache" / "downloads" / instance
path.mkdir(parents=True, exist_ok=True)
return path
@@ -60,21 +110,25 @@ def plugin_downloads_dir(state_root: Path, instance: str, plugin_id: str) -> Pat
return path
def backups_dir(state_root: Path, instance: str) -> Path:
path = instance_state_dir(state_root, instance) / "backups"
def backups_dir(state_root: Path, instance: str, *, install_id: str | None = None) -> Path:
path = instance_state_dir(state_root, instance, install_id=install_id) / "backups"
path.mkdir(parents=True, exist_ok=True)
return path
def known_good_state_path(state_root: Path, instance: str) -> Path:
return instance_state_dir(state_root, instance) / "known-good.json"
def load_known_good_state(state_root: Path, instance: str, *, install_id: str | None = None) -> dict[str, Any]:
return read_json(known_good_state_path(state_root, instance, install_id=install_id), {})
def load_known_good_state(state_root: Path, instance: str) -> dict[str, Any]:
return read_json(known_good_state_path(state_root, instance), {})
def save_known_good_state(state_root: Path, instance: str, state: dict[str, Any]) -> None:
def save_known_good_state(
state_root: Path,
instance: str,
state: dict[str, Any],
*,
install_id: str | None = None,
) -> None:
state.setdefault("instance", instance)
if install_id:
state.setdefault("installId", install_id)
state["savedAt"] = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
atomic_write_json(known_good_state_path(state_root, instance), state)
atomic_write_json(known_good_state_path(state_root, instance, install_id=install_id), state)
src/plugin_helper/tui.py
+14 -5
View File
@@ -20,7 +20,7 @@ from .operations import (
save_known_good_set,
)
from .reports import installed_plugins_report
from .state import load_installed_state, load_known_good_state
from .state import instance_state_dir, load_installed_state, load_known_good_state
@dataclass(frozen=True)
@@ -133,6 +133,7 @@ class PluginHelperTui(App[int]):
target.state_root,
plugin_id,
False,
install_id=target.install_id,
)
if not result["stateUpdated"]:
self._set_status(f"Could not disable {plugin_id}: {self._format_skipped(result['skipped'])}")
@@ -148,6 +149,7 @@ class PluginHelperTui(App[int]):
plugin_id=plugin_id,
repo=self.repo_root,
progress=self._operation_progress,
install_id=target.install_id,
)
self._set_status(f"Enabled {plugin_id}; applied {len(result['applied'])} files.")
else:
@@ -190,6 +192,7 @@ class PluginHelperTui(App[int]):
target.state_root,
plugin_id,
False,
install_id=target.install_id,
)
if result["stateUpdated"]:
changed += 1
@@ -225,6 +228,7 @@ class PluginHelperTui(App[int]):
plugin_ids=plugin_ids,
repo=self.repo_root,
progress=self._operation_progress,
install_id=target.install_id,
)
except Exception as exc:
self._set_status(f"Could not enable plugins: {exc}")
@@ -240,7 +244,11 @@ class PluginHelperTui(App[int]):
if self._busy or self.mode != "plugins" or self.selected_installation is None:
return
target = self.selected_installation
known_good = save_known_good_set(instance=target.instance_name, state_root=target.state_root)
known_good = save_known_good_set(
instance=target.instance_name,
state_root=target.state_root,
install_id=target.install_id,
)
count = len(known_good["pluginIds"])
self._set_status(f"Saved known-good set: {count} enabled plugins.")
@@ -248,7 +256,7 @@ class PluginHelperTui(App[int]):
if self._busy or self.mode != "plugins" or self.selected_installation is None:
return
target = self.selected_installation
if not load_known_good_state(target.state_root, target.instance_name):
if not load_known_good_state(target.state_root, target.instance_name, install_id=target.install_id):
self._set_status("No known-good set saved yet. Press s to save the current selection.")
return
self._busy = True
@@ -261,6 +269,7 @@ class PluginHelperTui(App[int]):
state_root=target.state_root,
repo=self.repo_root,
progress=self._operation_progress,
install_id=target.install_id,
)
except Exception as exc:
self._set_status(f"Could not restore known-good set: {exc}")
@@ -298,7 +307,7 @@ class PluginHelperTui(App[int]):
choice.install_label,
choice.instance_name,
str(choice.instance_path),
str(choice.state_root),
str(instance_state_dir(choice.state_root, choice.instance_name, install_id=choice.install_id)),
)
if self.setup_hint:
self._set_status(self.setup_hint)
@@ -322,7 +331,7 @@ class PluginHelperTui(App[int]):
try:
lockfile = load_lockfile(self.repo_root / "locks" / f"{target.instance_name}.lock.toml")
report = installed_plugins_report(
installed_state=load_installed_state(target.state_root, target.instance_name),
installed_state=load_installed_state(target.state_root, target.instance_name, install_id=target.install_id),
registry=load_registry(self.repo_root / "registry" / "plugins"),
lockfile=lockfile,
)
src/plugin_helper/update_runner.py
+239
View File
@@ -0,0 +1,239 @@
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
src/plugin_helper/updates.py
+181 -1
View File
@@ -4,10 +4,28 @@ import fnmatch
import re
from typing import Any, Callable
from .beatmods import BeatModsEntry, normalize_mods
from .models import Lockfile, Registry
FetchReleases = Callable[[str], list[dict[str, Any]]]
FetchBeatMods = Callable[[str], list[dict[str, Any]]]
PRIVATE_SOURCE_PREFIXES = ("discord/", "patreon/")
REVIEW_PATTERNS = (
"local build",
"localbuild",
"github pr",
"/pull/",
" pr ",
"manual",
"compatibility trial",
"failed compatibility",
"currently failed",
"smoke blocked",
"resmoke pending",
"do not reinstall",
)
def _release_tag(release: dict[str, Any]) -> str:
@@ -23,6 +41,71 @@ def _asset_name(asset: dict[str, Any]) -> str:
return str(asset.get("name") or "")
def _source_kind(repo: str | None) -> str:
if not repo:
return "missing"
lowered = repo.lower()
if lowered.startswith(PRIVATE_SOURCE_PREFIXES):
return "private"
if re.fullmatch(r"[^/\s]+/[^/\s]+", repo):
return "github"
return "unknown"
def _is_beatmods_locked(tag: str | None, reason: str | None) -> bool:
return str(tag or "").startswith("beatmods-") or "beatmods" in str(reason or "").lower()
def _current_beatmods_version(tag: str | None, reason: str | None, asset: str | None) -> str | None:
if tag and tag.startswith("beatmods-"):
return tag.removeprefix("beatmods-")
for text in (reason, asset):
if not text:
continue
match = re.search(r"\bBeatMods(?:\s+verified)?\s+[^0-9]*(\d+(?:\.\d+){1,4})\b", text, re.I)
if match:
return match.group(1)
match = re.search(r"-(\d+(?:\.\d+){1,4})(?:\.zip|\+|-|$)", text)
if match:
return match.group(1)
return None
def _normalized_name(text: str) -> str:
return re.sub(r"[^a-z0-9]+", "", text.lower())
def _beatmods_match(
entries: list[BeatModsEntry],
*,
plugin_id: str,
plugin_name: str,
) -> BeatModsEntry | None:
wanted = {_normalized_name(plugin_id), _normalized_name(plugin_name)}
for entry in entries:
if _normalized_name(entry.name) in wanted:
return entry
return None
def _review_reasons(tag: str | None, reason: str | None) -> list[str]:
text = f"{tag or ''} {reason or ''}".lower()
reasons: list[str] = []
if any(pattern in text for pattern in ("local build", "localbuild")):
reasons.append("local build")
if any(pattern in text for pattern in ("github pr", "/pull/", " pr ")) or str(tag or "").startswith("pr-"):
reasons.append("pull request build")
if "failed" in text or "compatibility trial" in text:
reasons.append("compatibility failure history")
if "smoke blocked" in text or "resmoke pending" in text or "currently failed" in text:
reasons.append("verification follow-up pending")
if "manual" in text:
reasons.append("manual install notes")
if not reasons and any(pattern in text for pattern in REVIEW_PATTERNS):
reasons.append("special handling noted")
return reasons
def _semver_key(tag: str) -> tuple[int, tuple[int, ...], str]:
match = re.search(r"(\d+(?:\.\d+){0,3})", tag)
if not match:
@@ -99,18 +182,22 @@ def check_updates(
registry: Registry,
lockfile: Lockfile,
fetch_releases: FetchReleases,
fetch_beatmods: FetchBeatMods | None = None,
selected: set[str] | None = None,
include_prerelease: bool = False,
) -> dict[str, Any]:
selected_ids = selected or {plugin.id for plugin in lockfile.plugins}
plugins: list[dict[str, Any]] = []
summary = {"current": 0, "updates": 0, "warnings": 0, "errors": 0}
summary = {"current": 0, "updates": 0, "warnings": 0, "errors": 0, "skipped": 0, "reviews": 0}
beatmods_entries: list[BeatModsEntry] | None = None
beatmods_error: Exception | None = None
for locked in lockfile.plugins:
if locked.id not in selected_ids:
continue
registry_plugin = registry.get(locked.id)
repo = locked.repo or (registry_plugin.repo if registry_plugin else None)
review_reasons = _review_reasons(locked.tag, locked.reason)
entry: dict[str, Any] = {
"id": locked.id,
"name": registry_plugin.name if registry_plugin else locked.id,
@@ -123,13 +210,106 @@ def check_updates(
"latestAssetSha256": None,
"status": "unknown",
"messages": [],
"review": bool(review_reasons),
"reviewReasons": review_reasons,
}
if review_reasons:
summary["reviews"] += 1
source_kind = _source_kind(repo)
if source_kind == "private":
entry["status"] = "skipped"
entry["messages"].append("paid/private source; check manually outside public update APIs")
summary["skipped"] += 1
plugins.append(entry)
continue
needs_manual_review = any(
reason in review_reasons
for reason in (
"local build",
"pull request build",
"verification follow-up pending",
"manual install notes",
)
)
if review_reasons and (needs_manual_review or not _is_beatmods_locked(locked.tag, locked.reason)):
entry["status"] = "review"
entry["messages"].append("special install or compatibility history; review notes before updating")
plugins.append(entry)
continue
if _is_beatmods_locked(locked.tag, locked.reason):
if fetch_beatmods is None:
entry["status"] = "warning"
entry["messages"].append("BeatMods check unavailable")
summary["warnings"] += 1
plugins.append(entry)
continue
if beatmods_entries is None and beatmods_error is None:
try:
beatmods_entries = normalize_mods(fetch_beatmods(lockfile.beat_saber_version))
except Exception as exc:
beatmods_error = exc
if beatmods_error is not None:
entry["status"] = "error"
entry["messages"].append(f"BeatMods: {beatmods_error}")
summary["errors"] += 1
plugins.append(entry)
continue
beatmods_match = _beatmods_match(
beatmods_entries or [],
plugin_id=locked.id,
plugin_name=entry["name"],
)
if beatmods_match is not None and beatmods_match.mod_version:
current_version = _current_beatmods_version(locked.tag, locked.reason, locked.asset)
entry["latestTag"] = f"beatmods-{beatmods_match.mod_version}"
entry["latestAsset"] = (
f"{beatmods_match.name}-{beatmods_match.mod_version}.zip"
if beatmods_match.name
else None
)
entry["latestBeatModsVersionId"] = beatmods_match.version_id
entry["latestBeatModsZipHash"] = beatmods_match.zip_hash
entry["latestAssetUrl"] = (
f"https://beatmods.com/cdn/mod/{beatmods_match.zip_hash}.zip"
if beatmods_match.zip_hash
else None
)
if current_version == beatmods_match.mod_version:
entry["status"] = "current"
summary["current"] += 1
else:
entry["status"] = "update"
entry["messages"].append(
f"BeatMods verified {beatmods_match.mod_version} for Beat Saber {lockfile.beat_saber_version}"
)
summary["updates"] += 1
plugins.append(entry)
continue
if source_kind != "github":
entry["status"] = "warning"
entry["messages"].append("no matching BeatMods verified entry found")
summary["warnings"] += 1
plugins.append(entry)
continue
entry["messages"].append("no matching BeatMods verified entry found; falling back to GitHub")
if not repo:
entry["status"] = "warning"
entry["messages"].append("missing repository")
summary["warnings"] += 1
plugins.append(entry)
continue
if source_kind != "github":
entry["status"] = "warning"
entry["messages"].append("unsupported repository source")
summary["warnings"] += 1
plugins.append(entry)
continue
try:
releases = fetch_releases(repo)
src/plugin_helper/userdata.py
+10 -2
View File
@@ -36,13 +36,19 @@ DEFAULT_BACKUP_EXCLUDES = (
)
def backup_userdata(instance: str, instance_path: Path, state_root: Path) -> dict[str, Any]:
def backup_userdata(
instance: str,
instance_path: Path,
state_root: Path,
*,
install_id: str | None = None,
) -> dict[str, Any]:
source = instance_path / "UserData"
if not source.is_dir():
raise FileNotFoundError(f"UserData directory not found: {source}")
created_at = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
destination = backups_dir(state_root, instance) / f"userdata-{created_at}.tar.gz"
destination = backups_dir(state_root, instance, install_id=install_id) / f"userdata-{created_at}.tar.gz"
files: list[dict[str, Any]] = []
total_size = 0
for path in sorted(item for item in source.rglob("*") if item.is_file()):
@@ -59,6 +65,8 @@ def backup_userdata(instance: str, instance_path: Path, state_root: Path) -> dic
"totalSize": total_size,
"files": files,
}
if install_id:
manifest["installId"] = install_id
destination.parent.mkdir(parents=True, exist_ok=True)
manifest_path = destination.parent / f".{destination.name}.manifest.json"