pleb
243 lines
8.6 KiB
Python
243 lines
8.6 KiB
Python
from __future__ import annotations
|
|
|
|
from collections.abc import Callable
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from .bootstrap import ensure_healthy_bootstrap
|
|
from .config import repo_root
|
|
from .installer import apply_plan, disable_plugin
|
|
from .models import Lockfile, Registry, load_lockfile, load_registry
|
|
from .planner import create_plan
|
|
from .state import load_installed_state, load_known_good_state, save_known_good_state
|
|
|
|
|
|
def _resolve_paths(
|
|
*,
|
|
instance: str,
|
|
registry: str,
|
|
lockfile: str | None,
|
|
repo: Path | None,
|
|
) -> tuple[Path, Path, Path, Lockfile, Registry]:
|
|
root = repo or repo_root()
|
|
registry_path = (root / registry).resolve() if not Path(registry).is_absolute() else Path(registry)
|
|
lock_path = Path(lockfile) if lockfile else root / "locks" / f"{instance}.lock.toml"
|
|
if not lock_path.is_absolute():
|
|
lock_path = (root / lock_path).resolve()
|
|
loaded_lockfile = load_lockfile(lock_path)
|
|
loaded_registry = load_registry(registry_path)
|
|
return root, registry_path, lock_path, loaded_lockfile, loaded_registry
|
|
|
|
|
|
def enable_disabled_plugin(
|
|
*,
|
|
instance: str,
|
|
instance_path: Path,
|
|
state_root: Path,
|
|
plugin_id: str,
|
|
registry: str = "registry/plugins",
|
|
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, install_id=install_id)
|
|
if plugin_id in installed_state.get("plugins", {}):
|
|
raise KeyError(f"plugin is already enabled: {plugin_id}")
|
|
|
|
root, registry_path, _lock_path, loaded_lockfile, loaded_registry = _resolve_paths(
|
|
instance=instance,
|
|
registry=registry,
|
|
lockfile=lockfile,
|
|
repo=repo,
|
|
)
|
|
if not any(plugin.id == plugin_id for plugin in loaded_lockfile.plugins):
|
|
raise KeyError(f"plugin is not locked for this instance: {plugin_id}")
|
|
|
|
tell = progress or (lambda _message: None)
|
|
tell(f"Planning {plugin_id} from local assets")
|
|
plan, path = create_plan(
|
|
instance=instance,
|
|
instance_path=instance_path,
|
|
beat_saber_version=loaded_lockfile.beat_saber_version,
|
|
registry=loaded_registry,
|
|
lockfile=loaded_lockfile,
|
|
state_root=state_root,
|
|
repo_root=root,
|
|
selected={plugin_id},
|
|
require_bootstrap=False,
|
|
install_id=install_id,
|
|
)
|
|
|
|
ensure_healthy_bootstrap(
|
|
instance=instance,
|
|
instance_path=instance_path,
|
|
beat_saber_version=loaded_lockfile.beat_saber_version,
|
|
registry=loaded_registry,
|
|
lockfile=loaded_lockfile,
|
|
state_root=state_root,
|
|
repo_root=root,
|
|
selected_ids={plugin_id},
|
|
progress=progress,
|
|
install_id=install_id,
|
|
)
|
|
|
|
tell(f"Applying {plugin_id}")
|
|
result = apply_plan(plan, state_root)
|
|
return {"planPath": str(path), **result}
|
|
|
|
|
|
def enable_disabled_plugins(
|
|
*,
|
|
instance: str,
|
|
instance_path: Path,
|
|
state_root: Path,
|
|
plugin_ids: list[str],
|
|
registry: str = "registry/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": []}
|
|
|
|
root, _registry_path, _lock_path, loaded_lockfile, loaded_registry = _resolve_paths(
|
|
instance=instance,
|
|
registry=registry,
|
|
lockfile=lockfile,
|
|
repo=repo,
|
|
)
|
|
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}
|
|
|
|
enabled: list[dict[str, Any]] = []
|
|
errors: list[dict[str, str]] = []
|
|
plannable_ids: list[str] = []
|
|
for plugin_id in plugin_ids:
|
|
if plugin_id in enabled_plugins:
|
|
errors.append({"plugin": plugin_id, "error": f"plugin is already enabled: {plugin_id}"})
|
|
continue
|
|
if plugin_id not in locked_ids:
|
|
errors.append({"plugin": plugin_id, "error": f"plugin is not locked for this instance: {plugin_id}"})
|
|
continue
|
|
plannable_ids.append(plugin_id)
|
|
|
|
if not plannable_ids:
|
|
return {"enabled": enabled, "errors": errors}
|
|
|
|
selected_ids = set(plannable_ids)
|
|
tell = progress or (lambda _message: None)
|
|
tell(f"Planning {len(plannable_ids)} plugins from local assets")
|
|
batch_plan, batch_path = create_plan(
|
|
instance=instance,
|
|
instance_path=instance_path,
|
|
beat_saber_version=loaded_lockfile.beat_saber_version,
|
|
registry=loaded_registry,
|
|
lockfile=loaded_lockfile,
|
|
state_root=state_root,
|
|
repo_root=root,
|
|
selected=selected_ids,
|
|
require_bootstrap=False,
|
|
install_id=install_id,
|
|
)
|
|
|
|
ensure_healthy_bootstrap(
|
|
instance=instance,
|
|
instance_path=instance_path,
|
|
beat_saber_version=loaded_lockfile.beat_saber_version,
|
|
registry=loaded_registry,
|
|
lockfile=loaded_lockfile,
|
|
state_root=state_root,
|
|
repo_root=root,
|
|
selected_ids=selected_ids,
|
|
progress=progress,
|
|
install_id=install_id,
|
|
)
|
|
|
|
tell(f"Applying {len(plannable_ids)} plugins")
|
|
result = apply_plan(batch_plan, state_root)
|
|
applied_by_plugin: dict[str, int] = {}
|
|
for item in result["applied"]:
|
|
plugin_id = item["plugin"]
|
|
applied_by_plugin[plugin_id] = applied_by_plugin.get(plugin_id, 0) + 1
|
|
|
|
for plugin_id in plannable_ids:
|
|
enabled.append({"plugin": plugin_id, "planPath": str(batch_path), "applied": applied_by_plugin.get(plugin_id, 0)})
|
|
|
|
return {"enabled": enabled, "errors": errors}
|
|
|
|
|
|
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, install_id=install_id)
|
|
plugin_ids = sorted(installed_state.get("plugins", {}))
|
|
known_good = {
|
|
"instance": instance,
|
|
"beatSaberVersion": installed_state.get("beatSaberVersion"),
|
|
"pluginIds": plugin_ids,
|
|
}
|
|
if install_id:
|
|
known_good["installId"] = install_id
|
|
save_known_good_state(state_root, instance, known_good, install_id=install_id)
|
|
return known_good
|
|
|
|
|
|
def restore_known_good_set(
|
|
*,
|
|
instance: str,
|
|
instance_path: Path,
|
|
state_root: Path,
|
|
registry: str = "registry/plugins",
|
|
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, 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, install_id=install_id)
|
|
currently_enabled = set(installed_state.get("plugins", {}))
|
|
|
|
to_disable = sorted(currently_enabled - known_good_ids)
|
|
disabled: list[str] = []
|
|
disable_errors: list[dict[str, str]] = []
|
|
for plugin_id in to_disable:
|
|
try:
|
|
result = disable_plugin(instance, instance_path, state_root, plugin_id, False, install_id=install_id)
|
|
if result["stateUpdated"]:
|
|
disabled.append(plugin_id)
|
|
else:
|
|
skipped = "; ".join(f"{item['path']} {item['reason']}" for item in result["skipped"])
|
|
disable_errors.append({"plugin": plugin_id, "error": skipped or "no files changed"})
|
|
except Exception as exc:
|
|
disable_errors.append({"plugin": plugin_id, "error": str(exc)})
|
|
|
|
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(
|
|
instance=instance,
|
|
instance_path=instance_path,
|
|
state_root=state_root,
|
|
plugin_ids=to_enable,
|
|
registry=registry,
|
|
lockfile=lockfile,
|
|
repo=repo,
|
|
progress=progress,
|
|
install_id=install_id,
|
|
)
|
|
|
|
return {
|
|
"knownGood": known_good,
|
|
"disabled": disabled,
|
|
"disableErrors": disable_errors,
|
|
"enabled": [item["plugin"] for item in enable_result["enabled"]],
|
|
"enableErrors": enable_result["errors"],
|
|
}
|