plugin-helper/src/plugin_helper/updates.py

from __future__ import annotations

import fnmatch
import re
from typing import Any, Callable

from .models import Lockfile, Registry


FetchReleases = Callable[[str], list[dict[str, Any]]]


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 _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,
    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}

    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)
        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": [],
        }
        if not repo:
            entry["status"] = "warning"
            entry["messages"].append("missing repository")
            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,
    }