Feature: 添加获取当前已安装适配器和插件的功能支持 (#171)

* ✨ add functions to list installed adapters and plugins

* 🐛 fix ambiguous error message for `find_exact_package` when `packages` is empty

* ♻️ re-export functions for listing installed adapters and plugins

* ✨ integrate list_installed_adapters/plugins into cli

* 🌐 add translations for new messages

Author: NCBM

nb_cli/cli/commands/adapter.py

@@ -6,6 +6,7 @@
 
 from nb_cli import _
 from nb_cli.config import GLOBAL_CONFIG
+from nb_cli.exceptions import NoSelectablePackageError
 from nb_cli.cli.utils import find_exact_package, format_package_results
 from nb_cli.cli import (
     CLI_DEFAULT_STYLE,
@@ -21,6 +22,7 @@
     call_pip_update,
     call_pip_install,
     call_pip_uninstall,
+    list_installed_adapters,
 )
 
 
@@ -80,6 +82,13 @@ async def store():
 @adapter.command(
     name="list", help=_("List nonebot adapters published on nonebot homepage.")
 )
+@click.option(
+    "--installed",
+    is_flag=True,
+    default=False,
+    flag_value=True,
+    help=_("Whether to list installed adapters only in current project."),
+)
 @click.option(
     "--include-unpublished",
     is_flag=True,
@@ -88,8 +97,12 @@ async def store():
     help=_("Whether to include unpublished adapters."),
 )
 @run_async
-async def list_(include_unpublished: bool = False):
-    adapters = await list_adapters(include_unpublished=include_unpublished)
+async def list_(installed: bool = False, include_unpublished: bool = False):
+    adapters = (
+        await list_installed_adapters()
+        if installed
+        else await list_adapters(include_unpublished=include_unpublished)
+    )
     if include_unpublished:
         click.secho(_("WARNING: Unpublished adapters may be included."), fg="yellow")
     click.echo(format_package_results(adapters))
@@ -143,13 +156,23 @@ async def install(
     include_unpublished: bool = False,
 ):
     try:
+        _installed = {
+            (a.project_link, a.module_name) for a in await list_installed_adapters()
+        }
         adapter = await find_exact_package(
             _("Adapter name to install:"),
             name,
-            await list_adapters(include_unpublished=include_unpublished),
+            [
+                a
+                for a in await list_adapters(include_unpublished=include_unpublished)
+                if (a.project_link, a.module_name) not in _installed
+            ],
         )
     except CancelledError:
         return
+    except NoSelectablePackageError:
+        click.echo(_("No available adapter found to install."))
+        return
 
     if include_unpublished:
         click.secho(
@@ -216,10 +239,13 @@ async def update(
         adapter = await find_exact_package(
             _("Adapter name to update:"),
             name,
-            await list_adapters(include_unpublished=include_unpublished),
+            await list_installed_adapters(),
         )
     except CancelledError:
         return
+    except NoSelectablePackageError:
+        click.echo(_("No installed adapter found to update."))
+        return
 
     if include_unpublished:
         click.secho(
@@ -263,12 +289,13 @@ async def uninstall(name: str | None, pip_args: list[str] | None):
         adapter = await find_exact_package(
             _("Adapter name to uninstall:"),
             name,
-            await list_adapters(
-                include_unpublished=True  # unpublished modules are always removable
-            ),
+            await list_installed_adapters(),
         )
     except CancelledError:
         return
+    except NoSelectablePackageError:
+        click.echo(_("No installed adapter found to uninstall."))
+        return
 
     try:
         can_uninstall = GLOBAL_CONFIG.remove_adapter(adapter)

nb_cli/cli/commands/plugin.py

@@ -6,6 +6,7 @@
 
 from nb_cli import _
 from nb_cli.config import GLOBAL_CONFIG
+from nb_cli.exceptions import NoSelectablePackageError
 from nb_cli.cli.utils import find_exact_package, format_package_results
 from nb_cli.cli import (
     CLI_DEFAULT_STYLE,
@@ -21,6 +22,7 @@
     call_pip_update,
     call_pip_install,
     call_pip_uninstall,
+    list_installed_plugins,
 )
 
 
@@ -80,6 +82,13 @@ async def store():
 @plugin.command(
     name="list", help=_("List nonebot plugins published on nonebot homepage.")
 )
+@click.option(
+    "--installed",
+    is_flag=True,
+    default=False,
+    flag_value=True,
+    help=_("Whether to list installed plugins only in current project."),
+)
 @click.option(
     "--include-unpublished",
     is_flag=True,
@@ -88,8 +97,12 @@ async def store():
     help=_("Whether to include unpublished plugins."),
 )
 @run_async
-async def list_(include_unpublished: bool = False):
-    plugins = await list_plugins(include_unpublished=include_unpublished)
+async def list_(installed: bool = False, include_unpublished: bool = False):
+    plugins = (
+        await list_installed_plugins()
+        if installed
+        else await list_plugins(include_unpublished=include_unpublished)
+    )
     if include_unpublished:
         click.secho(_("WARNING: Unpublished plugins may be included."), fg="yellow")
     click.echo(format_package_results(plugins))
@@ -143,13 +156,23 @@ async def install(
     include_unpublished: bool = False,
 ):
     try:
+        _installed = {
+            (p.project_link, p.module_name) for p in await list_installed_plugins()
+        }
         plugin = await find_exact_package(
             _("Plugin name to install:"),
             name,
-            await list_plugins(include_unpublished=include_unpublished),
+            [
+                p
+                for p in await list_plugins(include_unpublished=include_unpublished)
+                if (p.project_link, p.module_name) not in _installed
+            ],
         )
     except CancelledError:
         return
+    except NoSelectablePackageError:
+        click.echo(_("No available plugin found to install."))
+        return
 
     if include_unpublished:
         click.secho(
@@ -216,10 +239,13 @@ async def update(
         plugin = await find_exact_package(
             _("Plugin name to update:"),
             name,
-            await list_plugins(include_unpublished=include_unpublished),
+            await list_installed_plugins(),
         )
     except CancelledError:
         return
+    except NoSelectablePackageError:
+        click.echo(_("No installed plugin found to update."))
+        return
 
     if include_unpublished:
         click.secho(
@@ -263,12 +289,13 @@ async def uninstall(name: str | None, pip_args: list[str] | None):
         plugin = await find_exact_package(
             _("Plugin name to uninstall:"),
             name,
-            await list_plugins(
-                include_unpublished=True  # unpublished modules are always removable
-            ),
+            await list_installed_plugins(),
         )
     except CancelledError:
         return
+    except NoSelectablePackageError:
+        click.echo(_("No installed plugin found to uninstall."))
+        return
 
     try:
         can_uninstall = GLOBAL_CONFIG.remove_plugin(plugin)

nb_cli/cli/utils.py

@@ -14,6 +14,7 @@
 
 from nb_cli import _
 from nb_cli.config import Driver, Plugin, Adapter
+from nb_cli.exceptions import NoSelectablePackageError
 
 T = TypeVar("T", Adapter, Plugin, Driver)
 P = ParamSpec("P")
@@ -59,6 +60,8 @@ def __call__(self, x: Adapter | Plugin | Driver, *, value: str) -> bool: ...
 
 async def find_exact_package(question: str, name: str | None, packages: list[T]) -> T:
     if name is None:
+        if not packages:
+            raise NoSelectablePackageError("No packages available to select.")
         return (
             await ListPrompt(
                 question,

nb_cli/exceptions.py

@@ -24,3 +24,7 @@ class ProjectInvalidError(RuntimeError):
 
 class LocalCacheExpired(RuntimeError):
     """Raised when local metadata cache is outdated."""
+
+
+class NoSelectablePackageError(RuntimeError):
+    """Raised when there is no selectable package."""

nb_cli/handlers/__init__.py

@@ -77,12 +77,14 @@
 from .plugin import list_plugins as list_plugins
 from .plugin import create_plugin as create_plugin
 from .plugin import list_builtin_plugins as list_builtin_plugins
+from .plugin import list_installed_plugins as list_installed_plugins
 
 # isort: split
 
 # adapter
 from .adapter import list_adapters as list_adapters
 from .adapter import create_adapter as create_adapter
+from .adapter import list_installed_adapters as list_installed_adapters
 
 # isort: split
 

nb_cli/handlers/adapter.py

@@ -2,9 +2,11 @@
 
 from cookiecutter.main import cookiecutter
 
-from nb_cli.config import Adapter
 from nb_cli.compat import model_dump
+from nb_cli.exceptions import ProjectInvalidError
+from nb_cli.config import Adapter, NoneBotConfig, LegacyNoneBotConfig
 
+from .meta import get_nonebot_config, requires_project_root
 from .store import load_module_data, load_unpublished_modules
 
 TEMPLATE_ROOT = Path(__file__).parent.parent / "template" / "adapter"
@@ -42,3 +44,34 @@ async def list_adapters(
             ).values()
         )
     ]
+
+
+@requires_project_root
+async def list_installed_adapters(*, cwd: Path | None = None) -> list[Adapter]:
+    config_data = get_nonebot_config(cwd)
+    adapters = await load_module_data("adapter") + await load_unpublished_modules(
+        Adapter
+    )
+
+    result: list[Adapter] = []
+
+    if isinstance(config_data, NoneBotConfig):
+        adapter_info = config_data.adapters
+        allowed_pairs = {
+            (pkg_name, m.module_name)
+            for pkg_name, modules in adapter_info.items()
+            for m in modules
+        }
+        for adapter in adapters:
+            if (adapter.project_link, adapter.module_name) in allowed_pairs:
+                result.append(adapter)
+    elif isinstance(config_data, LegacyNoneBotConfig):
+        adapter_info = config_data.adapters
+        allowed_pairs = {m.module_name for m in adapter_info}
+        for adapter in adapters:
+            if adapter.module_name in allowed_pairs:
+                result.append(adapter)
+    else:
+        raise ProjectInvalidError("Invalid config data type")
+
+    return result

nb_cli/handlers/plugin.py

@@ -4,13 +4,19 @@
 
 from cookiecutter.main import cookiecutter
 
-from nb_cli.config import Plugin
 from nb_cli.compat import model_dump
+from nb_cli.exceptions import ProjectInvalidError
+from nb_cli.config import Plugin, NoneBotConfig, LegacyNoneBotConfig
 
 from . import templates
 from .process import create_process
-from .meta import requires_nonebot, get_default_python
 from .store import load_module_data, load_unpublished_modules
+from .meta import (
+    requires_nonebot,
+    get_default_python,
+    get_nonebot_config,
+    requires_project_root,
+)
 
 TEMPLATE_ROOT = Path(__file__).parent.parent / "template" / "plugin"
 
@@ -67,3 +73,32 @@ async def list_plugins(
             ).values()
         )
     ]
+
+
+@requires_project_root
+async def list_installed_plugins(*, cwd: Path | None = None) -> list[Plugin]:
+    config_data = get_nonebot_config(cwd)
+    plugins = await load_module_data("plugin") + await load_unpublished_modules(Plugin)
+
+    result: list[Plugin] = []
+
+    if isinstance(config_data, NoneBotConfig):
+        plugin_info = config_data.plugins
+        allowed_plugins = {
+            (pkg_name, module_name)
+            for pkg_name, module_names in plugin_info.items()
+            for module_name in module_names
+        }
+        for plugin in plugins:
+            if (plugin.project_link, plugin.module_name) in allowed_plugins:
+                result.append(plugin)
+    elif isinstance(config_data, LegacyNoneBotConfig):
+        plugin_info = config_data.plugins
+        allowed_plugins = set(plugin_info)
+        for plugin in plugins:
+            if plugin.module_name in allowed_plugins:
+                result.append(plugin)
+    else:
+        raise ProjectInvalidError("Invalid config data type")
+
+    return result