LiteyukiBot/liteyuki/plugins/liteyuki_npm/manager.py

import os

import nonebot.plugin
from nonebot import require
from nonebot.exception import FinishedException, IgnoredException
from nonebot.internal.adapter import Event
from nonebot.internal.matcher import Matcher
from nonebot.message import run_preprocessor
from nonebot.permission import SUPERUSER
from nonebot.plugin import Plugin

from liteyuki.utils.data_manager import GlobalPlugin, Group, InstalledPlugin, User, group_db, plugin_db, user_db
from liteyuki.utils.language import get_user_lang
from liteyuki.utils.ly_typing import T_Bot, T_MessageEvent
from liteyuki.utils.message import Markdown as md
from liteyuki.utils.permission import GROUP_ADMIN, GROUP_OWNER
from .common import get_plugin_can_be_toggle, get_plugin_default_enable, get_plugin_global_enable, get_plugin_session_enable
from .installer import get_store_plugin, npm_update
from ...utils.tools import clamp

require("nonebot_plugin_alconna")
from nonebot_plugin_alconna import on_alconna, Alconna, Args, Arparma, Subcommand

list_plugins = on_alconna(
    Alconna(
        "list-plugin",
        Args["page", int, 1]["num", int, 10],
    ),
    aliases={"列出插件", "插件列表"}
)

npm = on_alconna(
    aliases={"插件管理"},
    command=Alconna(
        "npm",
        # Args["plugin_name", str],
        Subcommand(
            "enable",
            Args["plugin_name", str],
            alias=["启用"],

        ),
        Subcommand(
            "disable",
            Args["plugin_name", str],
            alias=["停用"],
        ),
        Subcommand(
            "global-enable",
            Args["plugin_name", str],
            alias=["全局启用"],
        ),
        Subcommand(
            "global-disable",
            Args["plugin_name", str],
            alias=["全局停用"],
        ),
    ),

)


@list_plugins.handle()
async def _(event: T_MessageEvent, bot: T_Bot, result: Arparma):
    ulang = get_user_lang(str(event.user_id))
    if not os.path.exists("data/liteyuki/plugins.json"):
        await npm_update()

    loaded_plugin_list = sorted(nonebot.get_loaded_plugins(), key=lambda x: x.module_name)
    num_per_page = result.args.get("num")
    total = len(loaded_plugin_list) // num_per_page + (1 if len(loaded_plugin_list) % num_per_page else 0)

    page = clamp(result.args.get("page"), 1, total)

    # 已加载插件 | 总计10 | 第1/3页
    reply = (f"# {ulang.get('npm.loaded_plugins')} | "
             f"{ulang.get('npm.total', TOTAL=len(nonebot.get_loaded_plugins()))} | "
             f"{ulang.get('npm.page', PAGE=page, TOTAL=total)} \n***\n")

    for plugin in loaded_plugin_list[(page - 1) * num_per_page: min(page * num_per_page, len(loaded_plugin_list))]:
        # 检查是否有 metadata 属性
        # 添加帮助按钮
        btn_usage = md.btn_cmd(ulang.get("npm.usage"), f"help {plugin.module_name}", False)
        store_plugin = await get_store_plugin(plugin.module_name)
        session_enable = get_plugin_session_enable(event, plugin.module_name)
        if store_plugin:
            btn_homepage = md.btn_link(ulang.get("npm.homepage"), store_plugin.homepage)
            show_name = store_plugin.name
        elif plugin.metadata:
            if plugin.metadata.extra.get("liteyuki"):
                btn_homepage = md.btn_link(ulang.get("npm.homepage"), "https://github.com/snowykami/LiteyukiBot")
            else:
                btn_homepage = ulang.get("npm.homepage")
            show_name = plugin.metadata.name
        else:
            btn_homepage = ulang.get("npm.homepage")
            show_name = plugin.name
            ulang.get("npm.no_description")

        if plugin.metadata:
            reply += f"\n**{md.escape(show_name)}**\n"
        else:
            reply += f"**{md.escape(show_name)}**\n"

        reply += f"\n > {btn_usage}  {btn_homepage}"

        if await GROUP_ADMIN(bot, event) or await GROUP_OWNER(bot, event) or await SUPERUSER(bot, event):
            # 添加启用/停用插件按钮
            cmd_toggle = f"npm {'disable' if session_enable else 'enable'} {plugin.module_name}"
            text_toggle = ulang.get("npm.disable" if session_enable else "npm.enable")
            can_be_toggle = get_plugin_can_be_toggle(plugin.module_name)
            btn_toggle = text_toggle if not can_be_toggle else md.btn_cmd(text_toggle, cmd_toggle)
            reply += f"  {btn_toggle}"

            if await SUPERUSER(bot, event):
                plugin_in_database = plugin_db.first(InstalledPlugin(), "module_name = ?", plugin.module_name)
                # 添加移除插件和全局切换按钮
                global_enable = get_plugin_global_enable(plugin.module_name)
                btn_uninstall = (
                        md.btn_cmd(ulang.get("npm.uninstall"), f'npm uninstall {plugin.module_name}')) if plugin_in_database else ulang.get(
                    'npm.uninstall')
                btn_toggle_global_text = ulang.get("npm.disable_global" if global_enable else "npm.enable_global")
                cmd_toggle_global = f"npm {'global-disable' if global_enable else 'global-enable'} {plugin.module_name}"
                btn_toggle_global = btn_toggle_global_text if not can_be_toggle else md.btn_cmd(btn_toggle_global_text, cmd_toggle_global)

                reply += f"  {btn_uninstall}  {btn_toggle_global}"
        reply += "\n\n***\n"
    await md.send_md(reply, bot, event=event)


@npm.handle()
async def _(result: Arparma, event: T_MessageEvent, bot: T_Bot):
    if not os.path.exists("data/liteyuki/plugins.json"):
        await npm_update()
    # 判断会话类型
    ulang = get_user_lang(str(event.user_id))
    plugin_module_name = result.args.get("plugin_name")
    # 支持对自定义command_start的判断
    if result.subcommands.get("enable") or result.subcommands.get("disable"):

        toggle = result.subcommands.get("enable") is not None

        session_enable = get_plugin_session_enable(event, plugin_module_name)  # 获取插件当前状态

        default_enable = get_plugin_default_enable(plugin_module_name)  # 获取插件默认状态

        can_be_toggled = get_plugin_can_be_toggle(plugin_module_name)  # 获取插件是否可以被启用/停用

        if not can_be_toggled:
            await npm.finish(ulang.get("npm.plugin_cannot_be_toggled", NAME=plugin_module_name))

        if session_enable == toggle:
            await npm.finish(
                ulang.get("npm.plugin_already", NAME=plugin_module_name, STATUS=ulang.get("npm.enable") if toggle else ulang.get("npm.disable")))

        if event.message_type == "private":
            session = user_db.first(User(), "user_id = ?", event.user_id, default=User(user_id=event.user_id))
        else:
            if await GROUP_ADMIN(bot, event) or await GROUP_OWNER(bot, event) or await SUPERUSER(bot, event):
                session = group_db.first(Group(), "group_id = ?", event.group_id, default=Group(group_id=str(event.group_id)))
            else:
                raise FinishedException(ulang.get("Permission Denied"))
        try:
            if toggle:
                if default_enable:
                    session.disabled_plugins.remove(plugin_module_name)
                else:
                    session.enabled_plugins.append(plugin_module_name)
            else:
                if default_enable:
                    session.disabled_plugins.append(plugin_module_name)
                else:
                    session.enabled_plugins.remove(plugin_module_name)
            if event.message_type == "private":
                user_db.upsert(session)
            else:
                group_db.upsert(session)
        except Exception as e:
            print(e)
            await npm.finish(
                ulang.get(
                    "npm.toggle_failed",
                    NAME=plugin_module_name,
                    STATUS=ulang.get("npm.enable") if toggle else ulang.get("npm.disable"),
                    ERROR=str(e))
            )

        await npm.finish(
            ulang.get(
                "npm.toggle_success",
                NAME=plugin_module_name,
                STATUS=ulang.get("npm.enable") if toggle else ulang.get("npm.disable"))
        )
    elif result.subcommands.get("global-enable") or result.subcommands.get("global-disable") and await SUPERUSER(bot, event):
        toggle = result.subcommands.get("global-enable") is not None
        can_be_toggled = get_plugin_can_be_toggle(plugin_module_name)
        if not can_be_toggled:
            await npm.finish(ulang.get("npm.plugin_cannot_be_toggled", NAME=plugin_module_name))

        global_enable = get_plugin_global_enable(plugin_module_name)
        if global_enable == toggle:
            await npm.finish(
                ulang.get("npm.plugin_already", NAME=plugin_module_name, STATUS=ulang.get("npm.enable") if toggle else ulang.get("npm.disable")))

        try:
            plugin = plugin_db.first(GlobalPlugin(), "module_name = ?", plugin_module_name, default=GlobalPlugin(module_name=plugin_module_name))
            if toggle:
                plugin.enabled = True
            else:
                plugin.enabled = False
            plugin_db.upsert(plugin)
        except Exception as e:
            print(e)
            await npm.finish(
                ulang.get(
                    "npm.toggle_failed",
                    NAME=plugin_module_name,
                    STATUS=ulang.get("npm.enable") if toggle else ulang.get("npm.disable"),
                    ERROR=str(e))
            )

        await npm.finish(
            ulang.get(
                "npm.toggle_success",
                NAME=plugin_module_name,
                STATUS=ulang.get("npm.enable") if toggle else ulang.get("npm.disable"))
        )


@run_preprocessor
async def pre_handle(event: Event, matcher: Matcher):
    plugin: Plugin = matcher.plugin
    plugin_global_enable = get_plugin_global_enable(plugin.module_name)
    if not plugin_global_enable:
        raise IgnoredException("Plugin disabled globally")
    if event.get_type() == "message":
        plugin_session_enable = get_plugin_session_enable(event, plugin.module_name)
        if not plugin_session_enable:
            raise IgnoredException("Plugin disabled in session")

# @Bot.on_calling_api
# async def _(bot: Bot, api: str, data: dict[str, any]):
#     nonebot.logger.info(f"Plugin Callapi: {api}: {data}")