From a63f26c3b6301e95bb763f6ad88bbce7d31f6b4f Mon Sep 17 00:00:00 2001 From: HibiKier <45528451+HibiKier@users.noreply.github.com> Date: Thu, 21 Aug 2025 11:08:34 +0800 Subject: [PATCH] =?UTF-8?q?=E2=9C=A8=20=E5=A2=9E=E5=BC=BA=E6=8F=92?= =?UTF-8?q?=E4=BB=B6=E5=95=86=E5=BA=97=E5=8A=9F=E8=83=BD=EF=BC=8C=E6=94=AF?= =?UTF-8?q?=E6=8C=81=E6=B7=BB=E5=8A=A0=E6=8F=92=E4=BB=B6=E6=97=B6=E6=8C=87?= =?UTF-8?q?=E5=AE=9A=E6=BA=90=E7=B1=BB=E5=9E=8B=E3=80=82=20(#2028)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../builtin_plugins/plugin_store/__init__.py | 39 ++- .../plugin_store/data_source.py | 236 +++++++----------- 2 files changed, 112 insertions(+), 163 deletions(-) diff --git a/zhenxun/builtin_plugins/plugin_store/__init__.py b/zhenxun/builtin_plugins/plugin_store/__init__.py index e65273e8..f7998310 100644 --- a/zhenxun/builtin_plugins/plugin_store/__init__.py +++ b/zhenxun/builtin_plugins/plugin_store/__init__.py @@ -1,6 +1,6 @@ from nonebot.permission import SUPERUSER from nonebot.plugin import PluginMetadata -from nonebot_plugin_alconna import Alconna, Args, Subcommand, on_alconna +from nonebot_plugin_alconna import Alconna, Args, Match, Option, Subcommand, on_alconna from nonebot_plugin_session import EventSession from zhenxun.configs.utils import PluginExtraData @@ -16,11 +16,16 @@ __plugin_meta__ = PluginMetadata( description="插件商店", usage=""" 插件商店 : 查看当前的插件商店 - 添加插件 id or module : 添加插件 - 移除插件 id or module : 移除插件 - 搜索插件 name or author : 搜索插件 - 更新插件 id or module : 更新插件 + 添加插件 id或module或插件名称 ?[-s [git, ali]]: 添加插件 + 使用-s时指定源，git为github，ali为阿里云 + 移除插件 id或module: 移除插件 + 搜索插件 name或author: 搜索插件 + 更新插件 id或module: 更新插件 更新全部插件 : 更新全部插件 + + 示例： + 添加插件 pix + 添加插件 真寻日报 -s git """.strip(), extra=PluginExtraData( author="HibiKier", @@ -32,7 +37,11 @@ __plugin_meta__ = PluginMetadata( _matcher = on_alconna( Alconna( "插件商店", - Subcommand("add", Args["plugin_id", str]), + Subcommand( + "add", + Args["plugin_id", str], + Option("-s", Args["source", str]), + ), Subcommand("remove", Args["plugin_id", str]), Subcommand("search", Args["plugin_name_or_author", str]), Subcommand("update", Args["plugin_id", str]), @@ -84,7 +93,6 @@ async def _(session: EventSession): try: result = await StoreManager.get_plugins_info() logger.info("查看插件列表", "插件商店", session=session) - await MessageUtils.build_message([*result]).send() except Exception as e: logger.error(f"查看插件列表失败 e: {e}", "插件商店", session=session, e=e) @@ -92,13 +100,18 @@ async def _(session: EventSession): @_matcher.assign("add") -async def _(session: EventSession, plugin_id: str): +async def _(session: EventSession, plugin_id: str, source: Match[str]): + if is_number(plugin_id): + await MessageUtils.build_message(f"正在添加插件 Id: {plugin_id}").send() + else: + await MessageUtils.build_message(f"正在添加插件 Module: {plugin_id}").send() + source_str = source.result if source.available else None + if source_str and source_str not in ["ali", "git"]: + await MessageUtils.build_message( + f"源类型错误: {source_str} 请使用 ali 或 git" + ).finish() try: - if is_number(plugin_id): - await MessageUtils.build_message(f"正在添加插件 Id: {plugin_id}").send() - else: - await MessageUtils.build_message(f"正在添加插件 Module: {plugin_id}").send() - result = await StoreManager.add_plugin(plugin_id) + result = await StoreManager.add_plugin(plugin_id, source_str) except Exception as e: logger.error(f"添加插件 Id: {plugin_id}失败", "插件商店", session=session, e=e) await MessageUtils.build_message( diff --git a/zhenxun/builtin_plugins/plugin_store/data_source.py b/zhenxun/builtin_plugins/plugin_store/data_source.py index 82b21e6f..a7332f4d 100644 --- a/zhenxun/builtin_plugins/plugin_store/data_source.py +++ b/zhenxun/builtin_plugins/plugin_store/data_source.py @@ -5,14 +5,12 @@ import shutil from aiocache import cached import ujson as json -from zhenxun import ui from zhenxun.builtin_plugins.plugin_store.models import StorePluginInfo from zhenxun.configs.path_config import TEMP_PATH from zhenxun.models.plugin_info import PluginInfo from zhenxun.services.log import logger from zhenxun.services.plugin_init import PluginInitManager -from zhenxun.ui.builders import TableBuilder -from zhenxun.ui.models import StatusBadgeCell, TextCell +from zhenxun.utils.image_utils import BuildImage, ImageTemplate, RowStyle from zhenxun.utils.manager.virtual_env_package_manager import VirtualEnvPackageManager from zhenxun.utils.repo_utils import RepoFileManager from zhenxun.utils.repo_utils.models import RepoFileInfo, RepoType @@ -27,6 +25,22 @@ from .config import ( from .exceptions import PluginStoreException +def row_style(column: str, text: str) -> RowStyle: + """被动技能文本风格 + + 参数: + column: 表头 + text: 文本内容 + + 返回: + RowStyle: RowStyle + """ + style = RowStyle() + if column == "-" and text == "已安装": + style.font_color = "#67C23A" + return style + + class StoreManager: @classmethod @cached(60) @@ -91,123 +105,61 @@ class StoreManager: return await PluginInfo.filter(load_status=True).values_list(*args) @classmethod - async def get_plugins_info(cls) -> list[bytes] | str: + async def get_plugins_info(cls) -> list[BuildImage] | str: """插件列表 返回: - bytes | str: 返回消息 + BuildImage | str: 返回消息 """ plugin_list, extra_plugin_list = await cls.get_data() column_name = ["-", "ID", "名称", "简介", "作者", "版本", "类型"] db_plugin_list = await cls.get_loaded_plugins("module", "version") suc_plugin = {p[0]: (p[1] or "0.1") for p in db_plugin_list} - - HIGHLIGHT_COLOR = "#E6A23C" - - structured_native_list = [] - structured_extra_list = [] index = 0 - + data_list = [] + extra_data_list = [] for plugin_info in plugin_list: - is_new = cls.check_version_is_new(plugin_info, suc_plugin) - structured_native_list.append( - { - "is_installed": plugin_info.module in suc_plugin, - "id": index, - "name": plugin_info.name, - "description": plugin_info.description, - "author": plugin_info.author, - "version_str": cls.version_check(plugin_info, suc_plugin), - "type_name": plugin_info.plugin_type_name, - "has_update": not is_new and plugin_info.module in suc_plugin, - } + data_list.append( + [ + "已安装" if plugin_info.module in suc_plugin else "", + index, + plugin_info.name, + plugin_info.description, + plugin_info.author, + cls.version_check(plugin_info, suc_plugin), + plugin_info.plugin_type_name, + ] ) index += 1 - for plugin_info in extra_plugin_list: - is_new = cls.check_version_is_new(plugin_info, suc_plugin) - structured_extra_list.append( - { - "is_installed": plugin_info.module in suc_plugin, - "id": index, - "name": plugin_info.name, - "description": plugin_info.description, - "author": plugin_info.author, - "version_str": cls.version_check(plugin_info, suc_plugin), - "type_name": plugin_info.plugin_type_name, - "has_update": not is_new and plugin_info.module in suc_plugin, - } + extra_data_list.append( + [ + "已安装" if plugin_info.module in suc_plugin else "", + index, + plugin_info.name, + plugin_info.description, + plugin_info.author, + cls.version_check(plugin_info, suc_plugin), + plugin_info.plugin_type_name, + ] ) index += 1 - - native_table_builder = TableBuilder( - title="原生插件列表", tip="通过添加/移除插件 ID 来管理插件" - ).set_headers(column_name) - - native_rows_data = [] - for row_data in structured_native_list: - row_color = HIGHLIGHT_COLOR if row_data["has_update"] else None - status_cell = ( - StatusBadgeCell(text="已安装", status_type="ok") - if row_data["is_installed"] - else TextCell(content="") - ) - native_rows_data.append( - [ - status_cell, - TextCell(content=str(row_data["id"]), color=row_color), - TextCell(content=row_data["name"], color=row_color), - TextCell(content=row_data["description"], color=row_color), - TextCell(content=row_data["author"], color=row_color), - TextCell( - content=row_data["version_str"], - color=row_color, - bold=bool(row_color), - ), - TextCell(content=row_data["type_name"], color=row_color), - ] - ) - native_table_builder.add_rows(native_rows_data) - native_table_bytes = await ui.render( - native_table_builder.build(), - viewport={"width": 1400, "height": 10}, - device_scale_factor=2, - ) - extra_table_builder = TableBuilder( - title="第三方插件列表", tip="通过添加/移除插件 ID 来管理插件" - ).set_headers(column_name) - - extra_rows_data = [] - for row_data in structured_extra_list: - row_color = HIGHLIGHT_COLOR if row_data["has_update"] else None - status_cell = ( - StatusBadgeCell(text="已安装", status_type="ok") - if row_data["is_installed"] - else TextCell(content="") - ) - extra_rows_data.append( - [ - status_cell, - TextCell(content=str(row_data["id"]), color=row_color), - TextCell(content=row_data["name"], color=row_color), - TextCell(content=row_data["description"], color=row_color), - TextCell(content=row_data["author"], color=row_color), - TextCell( - content=row_data["version_str"], - color=row_color, - bold=bool(row_color), - ), - TextCell(content=row_data["type_name"], color=row_color), - ] - ) - extra_table_builder.add_rows(extra_rows_data) - extra_table_bytes = await ui.render( - extra_table_builder.build(), - viewport={"width": 1400, "height": 10}, - device_scale_factor=2, - ) - - return [native_table_bytes, extra_table_bytes] + return [ + await ImageTemplate.table_page( + "原生插件列表", + "通过添加/移除插件 ID 来管理插件", + column_name, + data_list, + text_style=row_style, + ), + await ImageTemplate.table_page( + "第三方插件列表", + "通过添加/移除插件 ID 来管理插件", + column_name, + extra_data_list, + text_style=row_style, + ), + ] @classmethod async def get_plugin_by_value( @@ -267,7 +219,7 @@ class StoreManager: return plugin_info, is_external @classmethod - async def add_plugin(cls, index_or_module: str) -> str: + async def add_plugin(cls, index_or_module: str, source: str | None) -> str: """添加插件 参数: @@ -289,6 +241,7 @@ class StoreManager: plugin_info.module_path, plugin_info.is_dir, is_external, + source, ) return f"插件 {plugin_info.name} 安装成功! 重启后生效" @@ -299,6 +252,7 @@ class StoreManager: module_path: str, is_dir: bool, is_external: bool = False, + source: str | None = None, ): """安装插件 @@ -309,6 +263,10 @@ class StoreManager: is_external: 是否是外部仓库 """ repo_type = RepoType.GITHUB if is_external else None + if source == "ali": + repo_type = RepoType.ALIYUN + elif source == "git": + repo_type = RepoType.GITHUB replace_module_path = module_path.replace(".", "/") if is_dir: files = await RepoFileManager.list_directory_files( @@ -338,6 +296,7 @@ class StoreManager: await VirtualEnvPackageManager.install_requirement(requirement_file) if not is_install_req: + # 从仓库根目录查找文件 rand = random.randint(1, 10000) requirement_path = TEMP_PATH / f"plugin_store_{rand}_req.txt" requirements_path = TEMP_PATH / f"plugin_store_{rand}_reqs.txt" @@ -392,20 +351,19 @@ class StoreManager: return f"插件 {plugin_info.name} 移除成功! 重启后生效" @classmethod - async def search_plugin(cls, plugin_name_or_author: str) -> bytes | str: + async def search_plugin(cls, plugin_name_or_author: str) -> BuildImage | str: """搜索插件 参数: plugin_name_or_author: 插件名称或作者 返回: - bytes | str: 返回消息 + BuildImage | str: 返回消息 """ plugin_list, extra_plugin_list = await cls.get_data() all_plugin_list = plugin_list + extra_plugin_list db_plugin_list = await cls.get_loaded_plugins("module", "version") suc_plugin = {p[0]: (p[1] or "Unknown") for p in db_plugin_list} - filtered_data = [ (id, plugin_info) for id, plugin_info in enumerate(all_plugin_list) @@ -413,50 +371,28 @@ class StoreManager: or plugin_name_or_author.lower() in plugin_info.author.lower() ] - if not filtered_data: + data_list = [ + [ + "已安装" if plugin_info.module in suc_plugin else "", + id, + plugin_info.name, + plugin_info.description, + plugin_info.author, + cls.version_check(plugin_info, suc_plugin), + plugin_info.plugin_type_name, + ] + for id, plugin_info in filtered_data + ] + if not data_list: return "未找到相关插件..." - - HIGHLIGHT_COLOR = "#E6A23C" column_name = ["-", "ID", "名称", "简介", "作者", "版本", "类型"] - - builder = TableBuilder( - title=f"插件搜索结果: '{plugin_name_or_author}'", - tip="通过添加/移除插件 ID 来管理插件", + return await ImageTemplate.table_page( + "商店插件列表", + "通过添加/移除插件 ID 来管理插件", + column_name, + data_list, + text_style=row_style, ) - builder.set_headers(column_name) - - rows_to_add = [] - for id, plugin_info in filtered_data: - is_new = cls.check_version_is_new(plugin_info, suc_plugin) - has_update = not is_new and plugin_info.module in suc_plugin - row_color = HIGHLIGHT_COLOR if has_update else None - - status_cell = ( - StatusBadgeCell(text="已安装", status_type="ok") - if plugin_info.module in suc_plugin - else TextCell(content="") - ) - - rows_to_add.append( - [ - status_cell, - TextCell(content=str(id), color=row_color), - TextCell(content=plugin_info.name, color=row_color), - TextCell(content=plugin_info.description, color=row_color), - TextCell(content=plugin_info.author, color=row_color), - TextCell( - content=cls.version_check(plugin_info, suc_plugin), - color=row_color, - bold=has_update, - ), - TextCell(content=plugin_info.plugin_type_name, color=row_color), - ] - ) - - builder.add_rows(rows_to_add) - - render_viewport = {"width": 1400, "height": 10} - return await ui.render(builder.build(), viewport=render_viewport) @classmethod async def update_plugin(cls, index_or_module: str) -> str: