modlink_studio.plugin.cli

  1from __future__ import annotations
  2
  3import argparse
  4import json
  5import os
  6import subprocess
  7import sys
  8import tempfile
  9from dataclasses import dataclass
 10from importlib.metadata import PackageNotFoundError, entry_points, version
 11from pathlib import Path
 12from typing import Any
 13from urllib.error import HTTPError, URLError
 14from urllib.request import Request, urlopen
 15
 16from packaging.specifiers import SpecifierSet
 17from packaging.version import InvalidVersion, Version
 18from platformdirs import user_cache_path
 19
 20PLUGIN_ENTRY_POINT_GROUP = "modlink.drivers"
 21DEFAULT_PLUGIN_INDEX_URL = (
 22    "https://modlink-studio.github.io/ModLink-Studio-Plugins/plugins/index.json"
 23)
 24PLUGIN_INDEX_URL_ENV = "MODLINK_PLUGIN_INDEX_URL"
 25USER_AGENT = "modlink-plugin"
 26
 27
 28@dataclass(frozen=True)
 29class PluginRelease:
 30    version: str
 31    host_version_spec: str
 32    wheel_url: str
 33
 34
 35@dataclass(frozen=True)
 36class IndexedPlugin:
 37    plugin_id: str
 38    distribution: str
 39    display_name: str
 40    summary: str
 41    releases: tuple[PluginRelease, ...]
 42
 43
 44@dataclass(frozen=True)
 45class InstalledPlugin:
 46    plugin_id: str
 47    distribution: str
 48    version: str | None
 49
 50
 51def _host_version() -> str:
 52    try:
 53        return version("modlink-studio")
 54    except PackageNotFoundError:
 55        return "0.0.0"
 56
 57
 58def _manifest_url() -> str:
 59    return os.environ.get(PLUGIN_INDEX_URL_ENV, DEFAULT_PLUGIN_INDEX_URL)
 60
 61
 62def _cache_path() -> Path:
 63    return user_cache_path("modlink-studio") / "plugin-index.json"
 64
 65
 66def _fetch_json(url: str) -> dict[str, Any]:
 67    request = Request(
 68        url,
 69        headers={
 70            "Accept": "application/json",
 71            "User-Agent": USER_AGENT,
 72        },
 73    )
 74    with urlopen(request, timeout=10) as response:
 75        payload = json.load(response)
 76    if not isinstance(payload, dict):
 77        raise RuntimeError("Plugin index must be a JSON object.")
 78    return payload
 79
 80
 81def _load_manifest() -> list[IndexedPlugin]:
 82    cache_path = _cache_path()
 83    payload: dict[str, Any] | None = None
 84    fetch_error: str | None = None
 85    try:
 86        payload = _fetch_json(_manifest_url())
 87    except (HTTPError, URLError, TimeoutError, RuntimeError, OSError) as exc:
 88        fetch_error = str(exc)
 89
 90    if payload is not None:
 91        cache_path.parent.mkdir(parents=True, exist_ok=True)
 92        cache_path.write_text(json.dumps(payload, indent=2, ensure_ascii=True), encoding="utf-8")
 93        return _parse_manifest(payload)
 94
 95    if cache_path.exists():
 96        cached_payload = json.loads(cache_path.read_text(encoding="utf-8"))
 97        if not isinstance(cached_payload, dict):
 98            raise RuntimeError("Cached plugin index is invalid.")
 99        return _parse_manifest(cached_payload)
100
101    raise RuntimeError(
102        "Failed to load plugin index and no cached copy is available."
103        if fetch_error is None
104        else f"Failed to load plugin index ({fetch_error}) and no cached copy is available."
105    )
106
107
108def _parse_manifest(payload: dict[str, Any]) -> list[IndexedPlugin]:
109    schema_version = payload.get("schema_version")
110    if schema_version != 1:
111        raise RuntimeError(f"Unsupported plugin index schema version: {schema_version!r}")
112
113    plugins = payload.get("plugins")
114    if not isinstance(plugins, list):
115        raise RuntimeError("Plugin index is missing a valid 'plugins' array.")
116
117    parsed: list[IndexedPlugin] = []
118    for item in plugins:
119        if not isinstance(item, dict):
120            raise RuntimeError("Each plugin entry must be an object.")
121        releases_payload = item.get("releases")
122        if not isinstance(releases_payload, list):
123            raise RuntimeError("Each plugin entry must include a 'releases' array.")
124
125        releases: list[PluginRelease] = []
126        for release in releases_payload:
127            if not isinstance(release, dict):
128                raise RuntimeError("Each plugin release entry must be an object.")
129            version_text = release.get("version")
130            host_version_spec = release.get("host_version_spec")
131            wheel_url = release.get("wheel_url")
132            if (
133                not isinstance(version_text, str)
134                or not isinstance(host_version_spec, str)
135                or not isinstance(wheel_url, str)
136            ):
137                raise RuntimeError(
138                    "Plugin release entries must include version, host_version_spec, and wheel_url."
139                )
140            releases.append(
141                PluginRelease(
142                    version=version_text,
143                    host_version_spec=host_version_spec,
144                    wheel_url=wheel_url,
145                )
146            )
147
148        plugin_id = item.get("plugin_id")
149        distribution_name = item.get("distribution")
150        display_name = item.get("display_name")
151        summary = item.get("summary")
152        if (
153            not isinstance(plugin_id, str)
154            or not isinstance(distribution_name, str)
155            or not isinstance(display_name, str)
156            or not isinstance(summary, str)
157        ):
158            raise RuntimeError(
159                "Plugin entries must include plugin_id, distribution, display_name, and summary."
160            )
161        parsed.append(
162            IndexedPlugin(
163                plugin_id=plugin_id,
164                distribution=distribution_name,
165                display_name=display_name,
166                summary=summary,
167                releases=tuple(releases),
168            )
169        )
170
171    return sorted(parsed, key=lambda item: item.plugin_id)
172
173
174def _installed_plugins() -> list[InstalledPlugin]:
175    installed: list[InstalledPlugin] = []
176    for entry_point in sorted(
177        entry_points(group=PLUGIN_ENTRY_POINT_GROUP), key=lambda item: item.name
178    ):
179        dist = getattr(entry_point, "dist", None)
180        distribution_name = getattr(dist, "name", None) or getattr(dist, "metadata", {}).get(
181            "Name", "unknown"
182        )
183        plugin_version = getattr(dist, "version", None)
184        installed.append(
185            InstalledPlugin(
186                plugin_id=entry_point.name,
187                distribution=distribution_name,
188                version=plugin_version,
189            )
190        )
191    return installed
192
193
194def _indexed_plugins_by_id(indexed_plugins: list[IndexedPlugin]) -> dict[str, IndexedPlugin]:
195    return {plugin.plugin_id: plugin for plugin in indexed_plugins}
196
197
198def _installed_plugins_by_id(
199    installed_plugins: list[InstalledPlugin],
200) -> dict[str, InstalledPlugin]:
201    return {plugin.plugin_id: plugin for plugin in installed_plugins}
202
203
204def _distribution_aliases(installed_plugins: list[InstalledPlugin]) -> dict[str, InstalledPlugin]:
205    aliases: dict[str, InstalledPlugin] = {}
206    for plugin in installed_plugins:
207        aliases[plugin.distribution] = plugin
208        aliases[plugin.distribution.replace("-", "_")] = plugin
209    return aliases
210
211
212def _select_release(plugin: IndexedPlugin, host_version: str) -> PluginRelease:
213    try:
214        current = Version(host_version)
215    except InvalidVersion as exc:
216        raise RuntimeError(f"Invalid modlink-studio version: {host_version!r}") from exc
217
218    compatible: list[PluginRelease] = []
219    for release in plugin.releases:
220        if current in SpecifierSet(release.host_version_spec):
221            compatible.append(release)
222    if not compatible:
223        raise RuntimeError(f"No compatible release found for {plugin.plugin_id}.")
224    return max(compatible, key=lambda item: Version(item.version))
225
226
227def _download_wheel(url: str, target_dir: Path) -> Path:
228    wheel_name = url.rsplit("/", 1)[-1]
229    target_path = target_dir / wheel_name
230    request = Request(url, headers={"User-Agent": USER_AGENT})
231    with urlopen(request, timeout=30) as response:
232        target_path.write_bytes(response.read())
233    return target_path
234
235
236def _ensure_pip_available() -> None:
237    probe = subprocess.run(
238        [sys.executable, "-m", "pip", "--version"],
239        stdout=subprocess.DEVNULL,
240        stderr=subprocess.PIPE,
241        text=True,
242        check=False,
243    )
244    if probe.returncode == 0:
245        return
246
247    stderr = (probe.stderr or "").strip()
248    if "No module named pip" not in stderr:
249        raise RuntimeError(
250            "Failed to invoke pip in the current Python environment."
251            if not stderr
252            else f"Failed to invoke pip in the current Python environment: {stderr}"
253        )
254
255    subprocess.run([sys.executable, "-m", "ensurepip", "--upgrade"], check=True)
256
257
258def _run_pip(*args: str) -> None:
259    _ensure_pip_available()
260    subprocess.run([sys.executable, "-m", "pip", *args], check=True)
261
262
263def _format_available(
264    plugin: IndexedPlugin, installed: InstalledPlugin | None, host_version: str
265) -> str:
266    try:
267        release = _select_release(plugin, host_version)
268        availability = f"available={release.version}"
269    except RuntimeError:
270        availability = "available=none"
271    installed_label = installed.version if installed and installed.version else "not-installed"
272    return f"{plugin.plugin_id:<18} installed={installed_label:<14} {availability:<18} {plugin.display_name}"
273
274
275def _format_installed(
276    plugin: InstalledPlugin, indexed_plugin: IndexedPlugin | None, host_version: str
277) -> str:
278    if indexed_plugin is None:
279        compatibility = "unknown"
280        source = "third-party"
281    else:
282        try:
283            _select_release(indexed_plugin, host_version)
284            compatibility = "compatible"
285        except RuntimeError:
286            compatibility = "unsupported"
287        source = "official"
288    version_label = plugin.version if plugin.version is not None else "unknown"
289    return (
290        f"{plugin.plugin_id:<18} version={version_label:<12} "
291        f"source={source:<11} compatible={compatibility:<10} dist={plugin.distribution}"
292    )
293
294
295def _cmd_list(show_installed: bool) -> int:
296    if show_installed:
297        installed_plugins = _installed_plugins()
298        if not installed_plugins:
299            print("- none")
300            return 0
301        for plugin in installed_plugins:
302            print(f"{plugin.plugin_id:<18} {plugin.distribution} {plugin.version or 'unknown'}")
303        return 0
304
305    indexed_plugins = _load_manifest()
306    for plugin in indexed_plugins:
307        print(f"{plugin.plugin_id:<18} {plugin.display_name} - {plugin.summary}")
308    return 0
309
310
311def _cmd_status() -> int:
312    host_version = _host_version()
313    indexed_plugins = _load_manifest()
314    installed_plugins = _installed_plugins()
315    indexed_by_id = _indexed_plugins_by_id(indexed_plugins)
316    installed_by_id = _installed_plugins_by_id(installed_plugins)
317
318    print(f"modlink-studio {host_version}")
319    print()
320    print("Installed")
321    if not installed_plugins:
322        print("- none")
323    else:
324        for plugin in installed_plugins:
325            print(
326                f"- {_format_installed(plugin, indexed_by_id.get(plugin.plugin_id), host_version)}"
327            )
328
329    print()
330    print("Available")
331    for plugin in indexed_plugins:
332        print(f"- {_format_available(plugin, installed_by_id.get(plugin.plugin_id), host_version)}")
333    return 0
334
335
336def _cmd_install(plugin_id: str) -> int:
337    host_version = _host_version()
338    indexed_plugins = _load_manifest()
339    indexed_by_id = _indexed_plugins_by_id(indexed_plugins)
340    plugin = indexed_by_id.get(plugin_id)
341    if plugin is None:
342        raise RuntimeError(f"Plugin '{plugin_id}' is not present in the current plugin index.")
343
344    release = _select_release(plugin, host_version)
345    with tempfile.TemporaryDirectory(prefix="modlink-plugin-") as temp_dir:
346        wheel_path = _download_wheel(release.wheel_url, Path(temp_dir))
347        _run_pip("install", str(wheel_path))
348    print(f"Installed {plugin.display_name} {release.version}.")
349    return 0
350
351
352def _cmd_uninstall(plugin_id: str) -> int:
353    installed_plugins = _installed_plugins()
354    installed_by_id = _installed_plugins_by_id(installed_plugins)
355    aliases = _distribution_aliases(installed_plugins)
356    plugin = installed_by_id.get(plugin_id) or aliases.get(plugin_id)
357    if plugin is None:
358        print(f"Plugin '{plugin_id}' is not installed.")
359        return 0
360
361    _run_pip("uninstall", "-y", plugin.distribution)
362    version_label = plugin.version or "unknown"
363    print(f"Uninstalled {plugin.distribution} {version_label}.")
364    return 0
365
366
367def build_parser() -> argparse.ArgumentParser:
368    parser = argparse.ArgumentParser(
369        prog="modlink-plugin",
370        description="Manage ModLink Studio plugins.",
371    )
372    subparsers = parser.add_subparsers(dest="command", required=True)
373
374    list_parser = subparsers.add_parser("list", help="List available plugins or installed plugins.")
375    list_parser.add_argument(
376        "--installed",
377        action="store_true",
378        help="List plugins currently installed in this environment.",
379    )
380
381    subparsers.add_parser("status", help="Show plugin environment overview.")
382
383    install_parser = subparsers.add_parser(
384        "install", help="Install one plugin from the plugin index."
385    )
386    install_parser.add_argument("plugin_id")
387
388    uninstall_parser = subparsers.add_parser("uninstall", help="Uninstall one installed plugin.")
389    uninstall_parser.add_argument("plugin_id")
390    return parser
391
392
393def main(argv: list[str] | None = None) -> int:
394    parser = build_parser()
395    args = parser.parse_args(argv)
396
397    if args.command == "list":
398        return _cmd_list(args.installed)
399    if args.command == "status":
400        return _cmd_status()
401    if args.command == "install":
402        return _cmd_install(args.plugin_id)
403    if args.command == "uninstall":
404        return _cmd_uninstall(args.plugin_id)
405    parser.error(f"Unknown command: {args.command}")
406    return 2
407
408
409if __name__ == "__main__":
410    raise SystemExit(main())