feat: 添加了自动安装插件功能

src/plugins/liteyuki_plugin_npm/common.py

@@ -1 +1,13 @@
+from src.utils.data import Database, LiteModel
+from src.utils.data_manager import plugin_db
+
 LNPM_COMMAND_START = "lnpm"
+
+
+
+
+class InstalledPlugin(LiteModel):
+    module_name: str
+
+
+plugin_db.auto_migrate(InstalledPlugin)

src/plugins/liteyuki_plugin_npm/installer.py

@@ -1,23 +1,27 @@
 import json
 import os.path
 import shutil
+import sys
+from io import StringIO
 from typing import Optional
 
 import nonebot
 from arclet.alconna import Arparma, MultiVar
 from nonebot.permission import SUPERUSER
+from nonebot.utils import run_sync
 from nonebot_plugin_alconna import on_alconna, Alconna, Args, Subcommand
 import pip
 
 import aiohttp, aiofiles
 from typing_extensions import Any
 
-from src.utils.data import LiteModel
 from src.utils.language import get_user_lang
-from src.utils.message import button, send_markdown
+from src.utils.message import Markdown as md, send_markdown
 from src.utils.resource import get_res
 from src.utils.typing import T_Bot, T_MessageEvent
 
+from .common import *
+
 npm_alc = on_alconna(
     Alconna(
         "lnpm",
@@ -28,17 +32,17 @@
         Subcommand(
             "search",
             Args["keywords", MultiVar(str)]["page", int, 1],
-            alias=["s"],
+            alias=["s", "搜索"],
         ),
         Subcommand(
             "install",
             Args["plugin_name", str],
-            alias=["i"],
+            alias=["i", "安装"],
         ),
         Subcommand(
             "remove",
             Args["plugin_name", str],
-            alias=["rm"],
+            alias=["rm", "移除", "卸载"],
         ),
     ),
     permission=SUPERUSER
@@ -82,20 +86,49 @@ async def _(result: Arparma, event: T_MessageEvent, bot: T_Bot):
     elif result.subcommands.get("search"):
         keywords: list[str] = result.subcommands["search"].args.get("keywords")
         rs = await npm_search(keywords)
+        max_show = 20
         if len(rs):
             reply = f"{ulang.get('npm.search_result')} | {ulang.get('npm.total', TOTAL=len(rs))}\n***"
-            for plugin in rs[:min(10, len(rs))]:
-                reply += (f"\n{button(ulang.get('npm.install'), 'lnpm install %s' % plugin.module_name)} | **{plugin.name}**\n"
+            for plugin in rs[:min(max_show, len(rs))]:
+                btn_install = md.button(ulang.get('npm.install'), 'lnpm install %s' % plugin.module_name)
+                link_page = md.link(ulang.get('npm.homepage'), plugin.homepage)
+
+                reply += (f"\n{btn_install} | **{plugin.name}**\n"
                           f"\n > **{plugin.desc}**\n"
-                          f"\n > {ulang.get('npm.author')}: {plugin.author} | [🔗{ulang.get('npm.homepage')}]({plugin.homepage})\n\n***\n")
-            if len(rs) > 10:
-                reply += (f"\n{ulang.get('npm.too_many_results')}"
-                          f"\n{button(ulang.get('npm.prev_page'), 'lnpm search %s %s' % (' '.join(keywords), 2))} | "
-                          f"{button(ulang.get('npm.next_page'), 'lnpm search %s %s' % (' '.join(keywords), 2))}")
+                          f"\n > {ulang.get('npm.author')}: {plugin.author} | {link_page}\n\n***\n")
+            if len(rs) > max_show:
+                reply += f"\n{ulang.get('npm.too_many_results')}"
         else:
             reply = ulang.get("npm.search_no_result")
         await send_markdown(reply, bot, event=event)
 
+    elif result.subcommands.get("install"):
+        plugin_name: str = result.subcommands["install"].args.get("plugin_name")
+        r, log = npm_install(plugin_name)
+        log = log.replace("\\", "/")
+        if r:
+            nonebot.load_plugin(plugin_name)
+            installed_plugin = InstalledPlugin(module_name=plugin_name)
+            store_plugin = await get_store_plugin(plugin_name)
+            plugin_db.save(installed_plugin)
+            await send_markdown(
+                f"**{ulang.get('npm.install_success', NAME=store_plugin.name)}**\n\n"
+                f"```\n{log}\n```",
+                bot,
+                event=event
+            )
+        else:
+            await send_markdown(
+                f"{ulang.get('npm.install_success', NAME=plugin_name)}\n\n"
+                f"```\n{log}\n```",
+                bot,
+                event=event
+            )
+
+    elif result.subcommands.get("remove"):
+        plugin_name: str = result.subcommands["remove"].args.get("plugin_name")
+        await npm_alc.finish(ulang.get("npm.remove_success"))
+
 
 async def npm_update() -> bool:
     """
@@ -107,15 +140,13 @@ async def npm_update() -> bool:
     url_list = [
             "https://registry.nonebot.dev/plugins.json",
     ]
-    # 用aiohttp请求json文件，成功就覆盖本地文件，否则尝试下一个url
     for url in url_list:
         async with aiohttp.ClientSession() as session:
             async with session.get(url) as resp:
                 if resp.status == 200:
                     async with aiofiles.open("data/liteyuki/plugins.json", "wb") as f:
                         data = await resp.read()
                         await f.write(data)
-
                     return True
     return False
 
@@ -150,10 +181,54 @@ async def npm_search(keywords: list[str]) -> list[StorePlugin]:
     return results
 
 
-def install(plugin_name) -> bool:
-    try:
-        pip.main(['install', plugin_name])
-        return True
-    except Exception as e:
-        print(e)
-        return False
+async def get_store_plugin(plugin_module_name: str) -> Optional[StorePlugin]:
+    """
+    获取插件信息
+
+    Args:
+        plugin_module_name (str): 插件模块名
+
+    Returns:
+        Optional[StorePlugin]: 插件信息
+    """
+    async with aiofiles.open("data/liteyuki/plugins.json", "r", encoding="utf-8") as f:
+        plugins: list[StorePlugin] = [StorePlugin(**pobj) for pobj in json.loads(await f.read())]
+    for plugin in plugins:
+        if plugin.module_name == plugin_module_name:
+            return plugin
+    return None
+
+
+def npm_install(plugin_module_name) -> tuple[bool, str]:
+    """
+    Args:
+        plugin_module_name:
+
+    Returns:
+        tuple[bool, str]:
+
+    """
+    buffer = StringIO()
+    sys.stdout = buffer
+    sys.stderr = buffer
+
+    mirrors = [
+            "https://pypi.tuna.tsinghua.edu.cn/simple",
+            "https://pypi.org/simple",
+    ]
+
+    # 使用pip安装包，对每个镜像尝试一次，成功后返回值
+    success = False
+    for mirror in mirrors:
+        try:
+            result = pip.main(['install', plugin_module_name, "-i", mirror])
+            success = result == 0
+            break
+        except Exception as e:
+            success = False
+            continue
+
+    sys.stdout = sys.__stdout__
+    sys.stderr = sys.__stderr__
+
+    return success, buffer.getvalue()

src/plugins/liteyuki_plugin_npm/manager.py

@@ -2,7 +2,7 @@
 from nonebot import on_command
 from nonebot.permission import SUPERUSER
 
-from src.utils.message import button, send_markdown
+from src.utils.message import Markdown as md, send_markdown
 from src.utils.typing import T_Bot, T_MessageEvent
 from src.utils.language import get_user_lang
 
@@ -17,11 +17,11 @@ async def _(event: T_MessageEvent, bot: T_Bot):
     for plugin in nonebot.get_loaded_plugins():
         # 检查是否有 metadata 属性
         if plugin.metadata:
-            reply += (f"\n{button(lang.get('npm.help'), 'help %s' % plugin.name, False, False)} "
+            reply += (f"\n{md.button(lang.get('npm.help'), 'help %s' % plugin.name, False, False)} "
                       f"**{plugin.metadata.name}**\n"
                       f"\n > {plugin.metadata.description}\n\n***\n")
         else:
-            reply += (f"\n{button(lang.get('npm.help'), 'help %s' % plugin.name, False, False)} "
+            reply += (f"\n{md.button(lang.get('npm.help'), 'help %s' % plugin.name, False, False)} "
                       f"**{plugin.name}**\n"
                       f"\n > {lang.get('npm.no_description')}\n\n***\n")
     await send_markdown(reply, bot, event=event)

src/resources/lang/en.lang

@@ -22,7 +22,9 @@ npm.store_update_success=Plugin store data updated successfully
 npm.store_update_failed=Plugin store data update failed
 npm.search_result=Search results
 npm.search_no_result=No result found
-npm.too_many_results=Too many results found
+npm.too_many_results=Too many results found, please refine your search
+npm.install_success={NAME} installed successfully
+npm.install_failed={NAME} installation failed
 npm.author=Author
 npm.homepage=Homepage
 npm.next_page=Next

src/resources/lang/ja.lang

@@ -23,6 +23,8 @@ npm.store_update_failed=プラグインストアのデータの更新に失敗
 npm.search_result=検索結果
 npm.search_no_result=検索結果がありません
 npm.too_many_results=検索結果が多すぎます。ページをめくってください
+npm.install_success={NAME} が正常にインストールされました
+npm.install_failed={NAME} のインストールに失敗しました
 npm.author=著者
 npm.homepage=ホームページ
 npm.next_page=次のページ

src/resources/lang/zh-CN.lang

@@ -22,7 +22,9 @@ npm.store_update_success=插件商店数据更新成功
 npm.store_update_failed=插件商店数据更新失败
 npm.search_result=搜索结果
 npm.search_no_result=无搜索结果
-npm.too_many_results=搜索结果过多，请翻页查看
+npm.too_many_results=搜索结果过多，请限制关键字
+npm.install_success={NAME} 安装成功
+npm.install_failed={NAME} 安装失败
 npm.author=作者
 npm.homepage=主页
 npm.next_page=下一页

src/utils/data_manager.py

@@ -5,6 +5,7 @@
 DATA_PATH = "data/liteyuki"
 
 user_db = DB(os.path.join(DATA_PATH, 'users.ldb'))
+plugin_db = DB(os.path.join(DATA_PATH, 'plugins.ldb'))
 
 
 class User(LiteModel):

src/utils/message.py

@@ -73,16 +73,31 @@ async def send_markdown(markdown: str, bot: T_Bot, *, message_type: str = None,
     return data
 
 
-def button(name: str, cmd: str, reply: bool = False, enter: bool = True) -> str:
-    """生成点击按钮
-    Args:
-        name: 按钮显示内容
-        cmd: 发送的命令，已在函数内url编码，不需要再次编码
-        reply: 是否以回复的方式发送消息
-        enter: 自动发送消息则为True，否则填充到输入框
+class Markdown:
+    @staticmethod
+    def button(name: str, cmd: str, reply: bool = False, enter: bool = True) -> str:
+        """生成点击按钮
+        Args:
+            name: 按钮显示内容
+            cmd: 发送的命令，已在函数内url编码，不需要再次编码
+            reply: 是否以回复的方式发送消息
+            enter: 自动发送消息则为True，否则填充到输入框
 
-    Returns:
-        markdown格式的可点击回调按钮
+        Returns:
+            markdown格式的可点击回调按钮
 
-    """
-    return f"[{name}](mqqapi://aio/inlinecmd?command={encode_url(cmd)}&reply={str(reply).lower()}&enter={str(enter).lower()})"
+        """
+        return f"[{name}](mqqapi://aio/inlinecmd?command={encode_url(cmd)}&reply={str(reply).lower()}&enter={str(enter).lower()})"
+
+    @staticmethod
+    def link(name: str, url: str) -> str:
+        """生成链接
+        Args:
+            name: 链接显示内容
+            url: 链接地址
+
+        Returns:
+            markdown格式的链接
+
+        """
+        return f"[链接{name}]({encode_url(url)})"