✨ 完善内置插件

gsuid_core/bot.py

@@ -20,7 +20,7 @@ class _Bot:
 
     async def target_send(
         self,
-        message: Union[Message, List[Message], str, bytes],
+        message: Union[Message, List[Message], List[str], str, bytes],
         target_type: Literal['group', 'direct', 'channel', 'sub_channel'],
         target_id: Optional[str],
         bot_id: str,
@@ -38,6 +38,8 @@ class _Bot:
                 message = [MessageSegment.text(message)]
         elif isinstance(message, bytes):
             message = [MessageSegment.image(message)]
+        elif isinstance(message, List):
+            message = [MessageSegment.node(message)]
 
         if at_sender and sender_id:
             message.append(MessageSegment.at(sender_id))
@@ -70,7 +72,7 @@ class Bot:
 
     async def send(
         self,
-        message: Union[Message, List[Message], str, bytes],
+        message: Union[Message, List[Message], str, bytes, List[str]],
         at_sender: bool = False,
     ):
         return await self.bot.target_send(
@@ -86,7 +88,7 @@ class Bot:
 
     async def target_send(
         self,
-        message: Union[Message, List[Message], str, bytes],
+        message: Union[Message, List[Message], str, bytes, List[str]],
         target_type: Literal['group', 'direct', 'channel', 'sub_channel'],
         target_id: Optional[str],
         at_sender: bool = False,

gsuid_core/plugins/core_command/install_plugins/__init__.py

@@ -2,19 +2,24 @@ from gsuid_core.sv import SV
 from gsuid_core.bot import Bot
 from gsuid_core.models import Event
 
-from ._plugins import refresh_list, get_plugins_url, install_plugins
+from ._plugins import (
+    refresh_list,
+    update_plugins,
+    get_plugins_url,
+    install_plugins,
+)
 
 sv_core_install_plugins = SV('core管理插件', pm=1)
 
 
 @sv_core_install_plugins.on_prefix(('core安装插件'))
 async def send_plugins_install(bot: Bot, ev: Event):
-    plugins_url = await get_plugins_url(ev.text.strip().lower())
+    plugins = await get_plugins_url(ev.text.strip().lower())
-    if not plugins_url:
+    if not plugins:
         return await bot.send('不存在该插件...可以使用[core刷新插件列表]获取最新列表!')
 
     await bot.send('开始安装...请稍等一段时间...')
-    im = install_plugins(plugins_url)
+    im = install_plugins(plugins)
     await bot.send(im)
 
 
@@ -26,3 +31,10 @@ async def refresh_plugins_list(bot: Bot, ev: Event):
     else:
         im = f'刷新成功! 已刷新{len(_list)}个插件!'
     await bot.send(im)
+
+
+@sv_core_install_plugins.on_prefix(('core更新插件'))
+async def send_update_msg(bot: Bot, ev: Event):
+    await bot.send('开始更新...请稍等一段时间...')
+    _list = update_plugins(ev.text)
+    await bot.send(_list)

gsuid_core/plugins/core_command/install_plugins/_plugins.py

@@ -1,14 +1,14 @@
-import re
+from typing import Dict, List, Union, Optional
-from typing import Dict, List, Tuple, Optional
 
 import aiohttp
 from git.repo import Repo
+from git.exc import GitCommandError
 
 from gsuid_core.logger import logger
 
 from .api import PLUGINS_PATH, proxy_url, plugins_lib
 
-plugins_list: Dict[str, str] = {}
+plugins_list: Dict[str, Dict[str, str]] = {}
 
 
 async def refresh_list() -> List[str]:
@@ -16,19 +16,18 @@ async def refresh_list() -> List[str]:
     async with aiohttp.ClientSession() as session:
         logger.info(f'稍等...开始刷新插件列表, 地址: {plugins_lib}')
         async with session.get(plugins_lib) as resp:
-            content = await resp.text()
+            _plugins_list: Dict[
-            _plugins_list: List[Tuple[str, str]] = re.findall(
+                str, Dict[str, Dict[str, str]]
-                r'\[([^]]+)\]\(([^)]+)\)', content
+            ] = await resp.json()
-            )
+            for i in _plugins_list['plugins']:
-            for i in _plugins_list:
+                if i.lower() not in plugins_list:
-                if i[0].lower() not in plugins_list:
+                    refresh_list.append(i)
-                    refresh_list.append(i[0])
+                    logger.info(f'[刷新插件列表] 列表新增插件 {i}')
-                    logger.info(f'[刷新插件列表] 列表新增插件 {i[0]}')
+                plugins_list[i.lower()] = _plugins_list['plugins'][i]
-                plugins_list[i[0].lower()] = i[1]
     return refresh_list
 
 
-async def get_plugins_url(name: str) -> Optional[str]:
+async def get_plugins_url(name: str) -> Optional[Dict[str, str]]:
     if not plugins_list:
         await refresh_list()
 
@@ -43,12 +42,63 @@ async def get_plugins_url(name: str) -> Optional[str]:
             return None
 
 
-def install_plugins(url: str) -> str:
+def install_plugins(plugins: Dict[str, str]) -> str:
-    plugin_name = url.split('/')[-1]
+    plugin_name = plugins['link'].split('/')[-1]
-    git_path = f'{proxy_url}{url}.git'
+    git_path = f'{proxy_url}{plugins["link"]}.git'
     logger.info(f'稍等...开始安装插件, 地址: {git_path}')
     path = PLUGINS_PATH / plugin_name
     if path.exists():
         return '该插件已经安装过了!'
     Repo.clone_from(git_path, path, single_branch=True, depth=1)
     return f'插件{plugin_name}安装成功!发送[gs重启]以应用!'
+
+
+def update_plugins(
+    plugin_name: str,
+    level: int = 0,
+    log_key: List[str] = [],
+    log_limit: int = 10,
+) -> Union[str, List]:
+    for _n in PLUGINS_PATH.iterdir():
+        _name = _n.name
+        sim = len(set(_name.lower()) & set(plugin_name.lower()))
+        if sim >= 0.5 * len(_name):
+            plugin_name = _name
+            break
+
+    path = PLUGINS_PATH / plugin_name
+    if not path.exists():
+        return '更新失败, 不存在该插件!'
+    repo = Repo(path)  # type: ignore
+    o = repo.remotes.origin
+
+    if level >= 2:
+        logger.warning(f'[更新][{plugin_name}] 正在执行 git clean --xdf')
+        repo.git.clean('-xdf')
+    # 还原上次更改
+    if level >= 1:
+        logger.warning(f'[更新][{plugin_name}] 正在执行 git reset --hard')
+        repo.git.reset('--hard')
+
+    try:
+        pull_log = o.pull()
+        logger.info(f'[更新][{plugin_name}] {pull_log}')
+    except GitCommandError as e:
+        logger.warning(e)
+        return '更新失败, 请检查控制台...'
+
+    commits = list(repo.iter_commits(max_count=40))
+    log_list = []
+    for commit in commits:
+        if isinstance(commit.message, str):
+            if log_key:
+                for key in log_key:
+                    if key in commit.message:
+                        log_list.append(commit.message.replace('\n', ''))
+                        if len(log_list) >= log_limit:
+                            break
+            else:
+                log_list.append(commit.message.replace('\n', ''))
+                if len(log_list) >= log_limit:
+                    break
+    return log_list

gsuid_core/plugins/core_command/install_plugins/api.py

@@ -3,8 +3,8 @@ from pathlib import Path
 # raw_url = 'https://raw.githubusercontent.com'
 # repo_title = 'Genshin-bots/GenshinUID-docs/master/docs'
 # plugins_lib = f'{raw_url}/{repo_title}/PluginsList.md'
-plugins_lib = 'https://docs.gsuid.gbots.work/PluginsList.md'
+plugins_lib = 'https://docs.gsuid.gbots.work/plugin_list.json'
 
 proxy_url = 'https://ghproxy.com/'
 
-PLUGINS_PATH = Path(__file__).parents[1]
+PLUGINS_PATH = Path(__file__).parents[2]