198 lines
7.5 KiB
Python
198 lines
7.5 KiB
Python
from __future__ import annotations
|
|
|
|
import fnmatch
|
|
import json
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Any
|
|
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,
|
|
)
|
|
|
|
|
|
ALLOWED_BSIPA_TOP_LEVEL = {"IPA", "Libs", "Plugins"}
|
|
|
|
|
|
def _now_slug() -> str:
|
|
return datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
|
|
|
|
|
|
def _find_asset(asset: str, state_root: Path, instance: str, repo_root: Path, plugin_id: str | None = None) -> Path | None:
|
|
candidates = [
|
|
Path(asset).expanduser(),
|
|
repo_root / "assets" / asset,
|
|
repo_root / "locks" / "assets" / asset,
|
|
]
|
|
if plugin_id:
|
|
candidates.insert(1, plugin_downloads_dir(state_root, instance, plugin_id) / asset)
|
|
candidates.insert(2 if plugin_id else 1, downloads_dir(state_root, instance) / asset)
|
|
for candidate in candidates:
|
|
if candidate.exists() and candidate.is_file():
|
|
return candidate
|
|
return None
|
|
|
|
|
|
def _zip_members(asset_path: Path) -> list[tuple[str, int, str]]:
|
|
result: list[tuple[str, int, str]] = []
|
|
with ZipFile(asset_path) as archive:
|
|
for info in archive.infolist():
|
|
if info.is_dir():
|
|
continue
|
|
rel = ensure_relative(info.filename).as_posix()
|
|
data = archive.read(info)
|
|
result.append((rel, len(data), sha256_bytes(data)))
|
|
return result
|
|
|
|
|
|
def _target_for_member(strategy: str, member: str) -> str:
|
|
rel = ensure_relative(member).as_posix()
|
|
if strategy == "zip-to-pending":
|
|
return f"IPA/Pending/{rel}"
|
|
if strategy == "root-zip":
|
|
return rel
|
|
if strategy == "bsipa-zip":
|
|
top = rel.split("/", 1)[0]
|
|
if top not in ALLOWED_BSIPA_TOP_LEVEL:
|
|
raise ValueError(f"bsipa-zip member has unsupported top-level path: {rel}")
|
|
return rel
|
|
raise ValueError(f"unsupported zip strategy: {strategy}")
|
|
|
|
|
|
def _asset_matches_patterns(name: str, patterns: tuple[str, ...]) -> bool:
|
|
return not patterns or any(fnmatch.fnmatch(name, pattern) for pattern in patterns)
|
|
|
|
|
|
def _target_matches_live(instance_path: Path, change: dict[str, Any]) -> bool:
|
|
target = instance_path / ensure_relative(change["target"])
|
|
return target.is_file() and sha256_file(target) == change["sha256"]
|
|
|
|
|
|
def _expand_required_dependencies(selected_ids: set[str], registry: Registry, lockfile: Lockfile) -> set[str]:
|
|
locked_ids = {plugin.id for plugin in lockfile.plugins}
|
|
expanded = set(selected_ids)
|
|
pending = list(selected_ids)
|
|
while pending:
|
|
plugin_id = pending.pop()
|
|
registry_plugin = registry.get(plugin_id)
|
|
if not registry_plugin:
|
|
continue
|
|
for dependency in registry_plugin.dependencies:
|
|
if not dependency.required:
|
|
continue
|
|
if dependency.id not in locked_ids:
|
|
raise ValueError(f"{plugin_id}: required dependency is not locked for this instance: {dependency.id}")
|
|
if dependency.id not in expanded:
|
|
expanded.add(dependency.id)
|
|
pending.append(dependency.id)
|
|
return expanded
|
|
|
|
|
|
def create_plan(
|
|
*,
|
|
instance: str,
|
|
instance_path: Path,
|
|
beat_saber_version: str,
|
|
registry: Registry,
|
|
lockfile: Lockfile,
|
|
state_root: Path,
|
|
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)
|
|
if selected is not None
|
|
else {plugin.id for plugin in lockfile.plugins}
|
|
)
|
|
changes: list[dict[str, Any]] = []
|
|
warnings: list[str] = []
|
|
|
|
if require_bootstrap and planning_requires_bootstrap(lockfile.plugins, selected_ids):
|
|
health = check_bsipa_health(instance_path, state_root, instance, install_id=install_id)
|
|
if not health["ok"]:
|
|
raise ValueError(bootstrap_health_error(health))
|
|
|
|
for locked in lockfile.plugins:
|
|
if locked.id not in selected_ids:
|
|
continue
|
|
registry_plugin = registry.get(locked.id)
|
|
strategy = locked.install_strategy or (registry_plugin.install_strategy if registry_plugin else "manual")
|
|
if strategy not in VALID_STRATEGIES:
|
|
raise ValueError(f"{locked.id}: invalid install strategy: {strategy}")
|
|
if strategy == "manual":
|
|
raise ValueError(f"{locked.id}: install_strategy is manual; add a concrete catalog rule first")
|
|
if not locked.asset:
|
|
raise ValueError(f"{locked.id}: lock entry has no asset")
|
|
if registry_plugin and not _asset_matches_patterns(Path(locked.asset).name, registry_plugin.asset_patterns):
|
|
warnings.append(f"{locked.id}: asset does not match catalog patterns")
|
|
|
|
asset_path = _find_asset(locked.asset, state_root, instance, repo_root, locked.id)
|
|
if not asset_path:
|
|
raise FileNotFoundError(
|
|
f"{locked.id}: asset not found: {locked.asset}; put it in "
|
|
f"{plugin_downloads_dir(state_root, instance, locked.id)}"
|
|
)
|
|
asset_sha = sha256_file(asset_path)
|
|
if locked.sha256 and locked.sha256 != asset_sha:
|
|
raise ValueError(f"{locked.id}: asset sha256 mismatch for {asset_path}")
|
|
|
|
if strategy == "dll-to-plugins":
|
|
if asset_path.suffix.lower() != ".dll":
|
|
raise ValueError(f"{locked.id}: dll-to-plugins expects a .dll asset")
|
|
changes.append(
|
|
{
|
|
"plugin": locked.id,
|
|
"action": "copy",
|
|
"source": str(asset_path),
|
|
"sourceSha256": asset_sha,
|
|
"target": f"Plugins/{asset_path.name}",
|
|
"size": asset_path.stat().st_size,
|
|
"sha256": asset_sha,
|
|
}
|
|
)
|
|
continue
|
|
|
|
if asset_path.suffix.lower() != ".zip":
|
|
raise ValueError(f"{locked.id}: {strategy} expects a .zip asset")
|
|
for member, size, member_sha in _zip_members(asset_path):
|
|
changes.append(
|
|
{
|
|
"plugin": locked.id,
|
|
"action": "extract",
|
|
"source": str(asset_path),
|
|
"sourceSha256": asset_sha,
|
|
"archiveMember": member,
|
|
"target": _target_for_member(strategy, member),
|
|
"size": size,
|
|
"sha256": member_sha,
|
|
}
|
|
)
|
|
|
|
unchanged = [change for change in changes if _target_matches_live(instance_path, change)]
|
|
changes = [change for change in changes if change not in unchanged]
|
|
plan = {
|
|
"schemaVersion": 1,
|
|
"createdAt": datetime.now(timezone.utc).isoformat().replace("+00:00", "Z"),
|
|
"instance": instance,
|
|
"instancePath": str(instance_path),
|
|
"beatSaberVersion": beat_saber_version,
|
|
"warnings": warnings,
|
|
"changes": changes,
|
|
"unchanged": unchanged,
|
|
}
|
|
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")
|
|
return plan, plan_path
|