VCV Patch Report

vcv_patch_report.py

#!/usr/bin/env python3
"""Report the VCV Library access status of modules used by a Rack patch."""

from __future__ import annotations

import argparse
import csv
import io
import json
import re
import shutil
import subprocess
import sys
import tarfile
from collections import Counter
from concurrent.futures import ThreadPoolExecutor, as_completed
from dataclasses import asdict, dataclass
from html.parser import HTMLParser
from pathlib import Path
from typing import Any, BinaryIO, Iterable
from urllib.error import HTTPError, URLError
from urllib.parse import quote
from urllib.request import Request, urlopen


API_URL = "https://api.vcvrack.com/library/manifests?version={major}"
LIBRARY_URL = "https://library.vcvrack.com/{plugin}/{model}"
USER_AGENT = "vcv-module-report/1.0 (+https://library.vcvrack.com/)"


class ReportError(Exception):
    """An expected error that should be shown without a traceback."""


@dataclass(frozen=True)
class PatchModule:
    plugin: str
    model: str
    version: str
    count: int


@dataclass
class PageInfo:
    exists: bool = True
    included: bool = False
    addable: bool = False
    unavailable: bool = False
    plus: bool = False
    offers: list[str] | None = None
    title: str = ""
    license: str = ""
    error: str = ""


@dataclass
class Result:
    plugin: str
    module: str
    count: int
    patch_version: str
    access: str
    offer: str
    license: str
    open_source: str
    flag: str
    library_url: str
    note: str


class LibraryPageParser(HTMLParser):
    """Extract stable access markers and offer labels from a Library page."""

    def __init__(self) -> None:
        super().__init__(convert_charrefs=True)
        self.included = False
        self.addable = False
        self.unavailable = False
        self.plus = False
        self._capture: str | None = None
        self._parts: list[str] = []
        self.offers: list[str] = []
        self.title = ""
        self._page_text: list[str] = []

    def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None:
        attr = dict(attrs)
        classes = set((attr.get("class") or "").split())
        if "library-included" in classes:
            self.included = True
        if "library-add" in classes:
            self.addable = True
        if "library-unavailable" in classes:
            self.unavailable = True
        if "button-plus" in classes or "VCV+" in (attr.get("title") or ""):
            self.plus = True
        if tag == "a" and "button-bundle" in classes:
            self._capture = "offer"
            self._parts = []
        elif tag == "title":
            self._capture = "title"
            self._parts = []

    def handle_endtag(self, tag: str) -> None:
        if tag == "a" and self._capture == "offer":
            text = _clean_text(" ".join(self._parts))
            if text:
                self.offers.append(text)
            self._capture = None
            self._parts = []
        elif tag == "title" and self._capture == "title":
            self.title = _clean_text(" ".join(self._parts))
            self._capture = None
            self._parts = []

    def handle_data(self, data: str) -> None:
        if self._capture:
            self._parts.append(data)
        self._page_text.append(data)

    @property
    def page_text(self) -> str:
        return _clean_text(" ".join(self._page_text))


def _clean_text(value: str) -> str:
    return re.sub(r"\s+", " ", value).strip()


def _load_json_bytes(data: bytes, source: str) -> dict[str, Any]:
    try:
        decoded = json.loads(data)
    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
        raise ReportError(f"Could not parse {source} as patch JSON: {exc}") from exc
    if not isinstance(decoded, dict):
        raise ReportError(f"Expected a JSON object in {source}")
    return decoded


def _patch_from_tar(fileobj: BinaryIO, source: str, mode: str = "r:*") -> dict[str, Any]:
    try:
        with tarfile.open(fileobj=fileobj, mode=mode) as archive:
            for member in archive:
                if member.isfile() and Path(member.name).name == "patch.json":
                    extracted = archive.extractfile(member)
                    if extracted is None:
                        break
                    return _load_json_bytes(extracted.read(), f"{source}:{member.name}")
    except (tarfile.TarError, OSError) as exc:
        raise ReportError(f"Could not read the VCV archive {source}: {exc}") from exc
    raise ReportError(f"No patch.json was found in {source}")


def _read_with_python_zstd(path: Path) -> dict[str, Any] | None:
    try:
        from compression import zstd  # type: ignore[attr-defined]
    except ImportError:
        return None
    with path.open("rb") as raw:
        with zstd.open(raw, "rb") as decompressed:
            return _patch_from_tar(decompressed, str(path), mode="r|")


def _read_with_third_party_zstd(path: Path) -> dict[str, Any] | None:
    try:
        import zstandard  # type: ignore[import-not-found]
    except ImportError:
        return None
    with path.open("rb") as raw:
        with zstandard.ZstdDecompressor().stream_reader(raw) as decompressed:
            return _patch_from_tar(decompressed, str(path), mode="r|")


def _read_with_system_tar(path: Path) -> dict[str, Any] | None:
    tar = shutil.which("tar")
    if not tar:
        return None
    errors: list[str] = []
    for member in ("./patch.json", "patch.json"):
        proc = subprocess.run(
            [tar, "-xOf", str(path), member],
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            check=False,
        )
        if proc.returncode == 0 and proc.stdout:
            return _load_json_bytes(proc.stdout, f"{path}:{member}")
        errors.append(proc.stderr.decode(errors="replace").strip())
    if any(errors):
        return None
    return None


def read_patch(path: Path) -> dict[str, Any]:
    if not path.is_file():
        raise ReportError(f"Patch file not found: {path}")

    with path.open("rb") as handle:
        prefix = handle.read(64)
        handle.seek(0)
        if prefix.lstrip().startswith((b"{", b"[")):
            return _load_json_bytes(handle.read(), str(path))
        try:
            return _patch_from_tar(handle, str(path))
        except ReportError:
            pass

    for reader in (_read_with_python_zstd, _read_with_third_party_zstd, _read_with_system_tar):
        try:
            patch = reader(path)
        except ReportError:
            raise
        except Exception:
            patch = None
        if patch is not None:
            return patch

    raise ReportError(
        "Could not decompress this .vcv file. Use Python 3.14+, install the "
        "'zstandard' package, or install a tar implementation with Zstandard support."
    )


def unique_modules(patch: dict[str, Any]) -> list[PatchModule]:
    raw_modules = patch.get("modules")
    if not isinstance(raw_modules, list):
        raise ReportError("Patch JSON does not contain a modules array")

    counts: Counter[tuple[str, str]] = Counter()
    versions: dict[tuple[str, str], set[str]] = {}
    for index, item in enumerate(raw_modules):
        if not isinstance(item, dict):
            continue
        plugin = item.get("plugin")
        model = item.get("model")
        version = item.get("version", "")
        if not isinstance(plugin, str) or not isinstance(model, str):
            raise ReportError(f"Module {index} is missing a string plugin or model slug")
        key = (plugin, model)
        counts[key] += 1
        version_text = version if isinstance(version, str) else str(version)
        if version_text:
            versions.setdefault(key, set()).add(version_text)

    return [
        PatchModule(plugin, model, ", ".join(sorted(versions.get((plugin, model), set()))), count)
        for (plugin, model), count in sorted(counts.items())
    ]


def _request(url: str, timeout: float) -> bytes:
    request = Request(url, headers={"User-Agent": USER_AGENT, "Accept": "application/json,text/html"})
    with urlopen(request, timeout=timeout) as response:
        return response.read()


def fetch_manifests(major: str, timeout: float) -> tuple[dict[str, dict[str, Any]], str]:
    url = API_URL.format(major=quote(major, safe=""))
    try:
        payload = json.loads(_request(url, timeout))
        manifests = payload.get("manifests", payload)
        if not isinstance(manifests, dict):
            raise ValueError("response does not contain a manifests object")
        return ({key: value for key, value in manifests.items() if isinstance(value, dict)}, "")
    except (HTTPError, URLError, TimeoutError, json.JSONDecodeError, ValueError) as exc:
        return {}, f"Manifest lookup failed: {exc}"


def fetch_page(module: PatchModule, timeout: float) -> tuple[str, PageInfo]:
    url = LIBRARY_URL.format(
        plugin=quote(module.plugin, safe=""), model=quote(module.model, safe="")
    )
    try:
        html = _request(url, timeout).decode("utf-8", errors="replace")
    except HTTPError as exc:
        if exc.code == 404:
            return url, PageInfo(exists=False)
        return url, PageInfo(error=f"HTTP {exc.code}")
    except (URLError, TimeoutError) as exc:
        return url, PageInfo(error=str(exc.reason) if isinstance(exc, URLError) else str(exc))

    parser = LibraryPageParser()
    try:
        parser.feed(html)
    except Exception as exc:
        return url, PageInfo(error=f"HTML parse error: {exc}")

    license_match = re.search(
        r"\bLicense:\s*([^|]+?)(?:Last updated:|Created:|Popularity:|Plugin:|Browse all modules|VCV Library Instructions)",
        parser.page_text,
        re.IGNORECASE,
    )
    unavailable = parser.unavailable or bool(
        re.search(r"\bUnavailable\b", parser.page_text[:500], re.IGNORECASE)
    )
    return url, PageInfo(
        included=parser.included,
        addable=parser.addable,
        unavailable=unavailable,
        plus=parser.plus,
        offers=parser.offers,
        title=parser.title.removeprefix("VCV Library - ").strip(),
        license=_clean_text(license_match.group(1)) if license_match else "",
    )


def _open_source(manifest: dict[str, Any] | None, license_name: str) -> str:
    if manifest and manifest.get("openSource") is True:
        return "yes"
    if manifest and manifest.get("openSource") is False:
        return "no"
    lowered = license_name.lower()
    if "eula" in lowered or "proprietary" in lowered:
        return "no"
    return "unknown"


def classify(
    module: PatchModule,
    page_url: str,
    page: PageInfo,
    manifest: dict[str, Any] | None,
) -> Result:
    offers = list(page.offers or [])
    if page.plus:
        offers.append("VCV+")

    license_name = str(manifest.get("license", "")) if manifest else page.license
    open_source = _open_source(manifest, license_name)
    manifest_status = str(manifest.get("status", "")).lower() if manifest else ""
    manifest_unavailable = manifest is not None and (
        manifest.get("available") is False
        or manifest_status in {"unavailable", "hidden", "removed"}
    )

    if page.included or (manifest and manifest.get("builtIn") is True):
        access = "included"
    elif page.unavailable or manifest_unavailable:
        access = "unavailable"
    elif any(re.search(r"\$\s*\d", offer) for offer in offers):
        access = "paid"
    elif page.plus and not page.addable:
        access = "subscription"
    elif page.addable:
        access = "free"
    elif not page.exists:
        access = "unlisted" if manifest is None else "module missing"
    elif page.error:
        access = "unknown"
    elif manifest is not None and manifest.get("available") is True:
        access = "free?"
    else:
        access = "unknown"

    flags: list[str] = []
    flag_for_access = {
        "paid": "PAID",
        "subscription": "SUBSCRIPTION",
        "unavailable": "UNAVAILABLE",
        "unlisted": "UNLISTED",
        "module missing": "MODULE MISSING",
        "unknown": "CHECK",
        "free?": "CHECK",
    }
    if access in flag_for_access:
        flags.append(flag_for_access[access])
    if open_source == "no":
        flags.append("PROPRIETARY")
    if not flags:
        flags.append("OK")

    notes: list[str] = []
    if page.error:
        notes.append(page.error)
    if not page.exists and manifest is not None:
        notes.append("plugin is listed, but module page returned 404")
    if access == "unlisted":
        notes.append("not found in the VCV Library")

    return Result(
        plugin=module.plugin,
        module=module.model,
        count=module.count,
        patch_version=module.version,
        access=access,
        offer="; ".join(dict.fromkeys(offers)) or "—",
        license=license_name or "unknown",
        open_source=open_source,
        flag=", ".join(flags),
        library_url=page_url,
        note="; ".join(notes),
    )


def build_results(
    modules: list[PatchModule],
    manifests: dict[str, dict[str, Any]],
    timeout: float,
    workers: int,
) -> list[Result]:
    pages: dict[tuple[str, str], tuple[str, PageInfo]] = {}
    with ThreadPoolExecutor(max_workers=workers) as executor:
        futures = {executor.submit(fetch_page, module, timeout): module for module in modules}
        for future in as_completed(futures):
            module = futures[future]
            try:
                pages[(module.plugin, module.model)] = future.result()
            except Exception as exc:
                url = LIBRARY_URL.format(
                    plugin=quote(module.plugin, safe=""), model=quote(module.model, safe="")
                )
                pages[(module.plugin, module.model)] = (url, PageInfo(error=str(exc)))

    return [
        classify(
            module,
            *pages[(module.plugin, module.model)],
            manifests.get(module.plugin),
        )
        for module in modules
    ]


def _truncate(value: object, maximum: int) -> str:
    text = str(value)
    if len(text) <= maximum:
        return text
    return text[: maximum - 1] + "…"


def print_table(results: list[Result]) -> None:
    columns: list[tuple[str, str, int]] = [
        ("Plugin", "plugin", 24),
        ("Module", "module", 28),
        ("#", "count", 3),
        ("Access", "access", 14),
        ("Offer", "offer", 40),
        ("License", "license", 24),
        ("Flag", "flag", 28),
    ]
    rows = [[_truncate(getattr(result, key), maximum) for _, key, maximum in columns] for result in results]
    widths = [
        max(len(heading), *(len(row[index]) for row in rows))
        for index, (heading, _, _) in enumerate(columns)
    ]

    def line(parts: Iterable[str], separator: str = " | ") -> str:
        return separator.join(part.ljust(widths[index]) for index, part in enumerate(parts))

    headings = [heading for heading, _, _ in columns]
    print(line(headings))
    print(line(("-" * width for width in widths), "-+-"))
    for row in rows:
        print(line(row))

    counts = Counter(result.access for result in results)
    summary = ", ".join(f"{name}: {count}" for name, count in sorted(counts.items()))
    flagged = sum(result.flag != "OK" for result in results)
    print(f"\n{len(results)} unique modules ({sum(result.count for result in results)} instances); {summary}")
    print(f"{flagged} module(s) flagged for paid, restricted, missing, or uncertain status.")


def print_csv(results: list[Result]) -> None:
    fieldnames = list(Result.__dataclass_fields__)
    writer = csv.DictWriter(sys.stdout, fieldnames=fieldnames)
    writer.writeheader()
    writer.writerows(asdict(result) for result in results)


def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="List unique modules in a .vcv patch and check their VCV Library access status."
    )
    parser.add_argument("patch", type=Path, help="VCV Rack .vcv patch (archive or legacy JSON)")
    parser.add_argument(
        "--format",
        choices=("table", "csv", "json"),
        default="table",
        help="output format (default: table)",
    )
    parser.add_argument("--timeout", type=float, default=15.0, help="HTTP timeout in seconds")
    parser.add_argument("--workers", type=int, default=8, help="parallel Library page requests")
    parser.add_argument(
        "--fail-on-flagged",
        action="store_true",
        help="exit with status 3 when any row is not OK",
    )
    return parser.parse_args(argv)


def main(argv: list[str] | None = None) -> int:
    args = parse_args(argv)
    if args.timeout <= 0:
        print("error: --timeout must be greater than zero", file=sys.stderr)
        return 2
    if args.workers <= 0:
        print("error: --workers must be greater than zero", file=sys.stderr)
        return 2

    try:
        patch = read_patch(args.patch)
        modules = unique_modules(patch)
    except ReportError as exc:
        print(f"error: {exc}", file=sys.stderr)
        return 2

    if not modules:
        print("No modules found in patch.")
        return 0

    patch_version = str(patch.get("version", "2"))
    major_match = re.match(r"\d+", patch_version)
    major = major_match.group(0) if major_match else "2"
    manifests, manifest_warning = fetch_manifests(major, args.timeout)
    if manifest_warning:
        print(f"warning: {manifest_warning}; continuing with module pages", file=sys.stderr)

    results = build_results(modules, manifests, args.timeout, args.workers)
    if args.format == "json":
        json.dump([asdict(result) for result in results], sys.stdout, indent=2)
        print()
    elif args.format == "csv":
        print_csv(results)
    else:
        print_table(results)

    if args.fail_on_flagged and any(result.flag != "OK" for result in results):
        return 3
    return 0


if __name__ == "__main__":
    raise SystemExit(main())