pleb
360 lines
13 KiB
Python
360 lines
13 KiB
Python
from __future__ import annotations
|
|
|
|
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:
|
|
return str(release.get("tag_name") or "")
|
|
|
|
|
|
def _release_assets(release: dict[str, Any]) -> list[dict[str, Any]]:
|
|
assets = release.get("assets") or []
|
|
return [asset for asset in assets if isinstance(asset, dict)]
|
|
|
|
|
|
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:
|
|
return (0, (), tag)
|
|
return (1, tuple(int(part) for part in match.group(1).split(".")), tag)
|
|
|
|
|
|
def _sorted_releases(releases: list[dict[str, Any]], include_prerelease: bool) -> list[dict[str, Any]]:
|
|
candidates = [
|
|
release
|
|
for release in releases
|
|
if not release.get("draft") and (include_prerelease or not release.get("prerelease"))
|
|
]
|
|
return sorted(
|
|
candidates,
|
|
key=lambda release: (
|
|
str(release.get("published_at") or release.get("created_at") or ""),
|
|
_semver_key(_release_tag(release)),
|
|
),
|
|
reverse=True,
|
|
)
|
|
|
|
|
|
def _matching_assets(
|
|
release: dict[str, Any],
|
|
*,
|
|
asset_patterns: tuple[str, ...],
|
|
current_asset: str | None,
|
|
beat_saber_version: str,
|
|
) -> list[dict[str, Any]]:
|
|
assets = _release_assets(release)
|
|
if asset_patterns:
|
|
assets = [
|
|
asset
|
|
for asset in assets
|
|
if any(fnmatch.fnmatch(_asset_name(asset), pattern) for pattern in asset_patterns)
|
|
]
|
|
if not assets:
|
|
return []
|
|
|
|
exact_version = [asset for asset in assets if _asset_name(asset) == f"{beat_saber_version}.zip"]
|
|
if exact_version:
|
|
return exact_version
|
|
if current_asset:
|
|
same_name = [asset for asset in assets if _asset_name(asset) == current_asset]
|
|
if same_name:
|
|
return same_name
|
|
contains_version = [asset for asset in assets if beat_saber_version in _asset_name(asset)]
|
|
return contains_version or assets
|
|
|
|
|
|
def _find_latest_matching_release(
|
|
releases: list[dict[str, Any]],
|
|
*,
|
|
asset_patterns: tuple[str, ...],
|
|
current_asset: str | None,
|
|
beat_saber_version: str,
|
|
include_prerelease: bool,
|
|
) -> tuple[dict[str, Any], dict[str, Any]] | None:
|
|
for release in _sorted_releases(releases, include_prerelease):
|
|
assets = _matching_assets(
|
|
release,
|
|
asset_patterns=asset_patterns,
|
|
current_asset=current_asset,
|
|
beat_saber_version=beat_saber_version,
|
|
)
|
|
if assets:
|
|
return release, assets[0]
|
|
return None
|
|
|
|
|
|
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, "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,
|
|
"repo": repo,
|
|
"currentTag": locked.tag,
|
|
"currentAsset": locked.asset,
|
|
"currentSha256": locked.sha256,
|
|
"latestTag": None,
|
|
"latestAsset": None,
|
|
"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)
|
|
except Exception as exc:
|
|
entry["status"] = "error"
|
|
entry["messages"].append(str(exc))
|
|
summary["errors"] += 1
|
|
plugins.append(entry)
|
|
continue
|
|
|
|
match = _find_latest_matching_release(
|
|
releases,
|
|
asset_patterns=registry_plugin.asset_patterns if registry_plugin else (),
|
|
current_asset=locked.asset,
|
|
beat_saber_version=lockfile.beat_saber_version,
|
|
include_prerelease=include_prerelease,
|
|
)
|
|
if not match:
|
|
entry["status"] = "warning"
|
|
entry["messages"].append("no matching release asset found")
|
|
summary["warnings"] += 1
|
|
plugins.append(entry)
|
|
continue
|
|
|
|
latest_release, latest_asset = match
|
|
entry["latestTag"] = _release_tag(latest_release)
|
|
entry["latestAsset"] = _asset_name(latest_asset)
|
|
entry["latestAssetSha256"] = str(latest_asset.get("digest") or "").removeprefix("sha256:") or None
|
|
entry["latestUrl"] = latest_release.get("html_url")
|
|
entry["latestAssetUrl"] = latest_asset.get("browser_download_url")
|
|
same_release = locked.tag == entry["latestTag"] and locked.asset == entry["latestAsset"]
|
|
same_hash = not entry["latestAssetSha256"] or locked.sha256 == entry["latestAssetSha256"]
|
|
if same_release and same_hash:
|
|
entry["status"] = "current"
|
|
summary["current"] += 1
|
|
else:
|
|
entry["status"] = "update"
|
|
summary["updates"] += 1
|
|
plugins.append(entry)
|
|
|
|
return {
|
|
"instance": lockfile.instance,
|
|
"beatSaberVersion": lockfile.beat_saber_version,
|
|
"includePrerelease": include_prerelease,
|
|
"summary": summary,
|
|
"plugins": plugins,
|
|
}
|