✨ feat(help): 添加帮助功能的快捷方式支持

- 在帮助插件中新增了对“帮助”命令的快捷方式支持，允许用户通过简化的输入方式获取功能信息。
- 移除了不必要的别名，优化了命令的可用性。

__version__

@@ -1 +1 @@
-__version__: v0.2.4-c7ef6fd
+__version__: v0.2.4-da6d5b4

zhenxun/builtin_plugins/auto_update/__init__.py

@@ -84,13 +84,16 @@ async def _(
 ):
     result = ""
     await MessageUtils.build_message("正在进行检查更新...").send(reply_to=True)
+
+    if not ver_type.available:
+        result += await UpdateManager.check_version()
+        logger.info("查看当前版本...", "检查更新", session=session)
+        await MessageUtils.build_message(result).finish()
+        return
+
     ver_type_str = ver_type.result
     source_str = source.result
     if ver_type_str in {"main", "release"}:
-        if not ver_type.available:
-            result += await UpdateManager.check_version()
-            logger.info("查看当前版本...", "检查更新", session=session)
-            await MessageUtils.build_message(result).finish()
         try:
             result += await UpdateManager.update_zhenxun(
                 bot,

zhenxun/builtin_plugins/auto_update/_data_source.py

@@ -1,37 +1,135 @@
+import asyncio
 from typing import Literal
 
 from nonebot.adapters import Bot
+from packaging.specifiers import SpecifierSet
+from packaging.version import InvalidVersion, Version
 
 from zhenxun.services.log import logger
+from zhenxun.utils.http_utils import AsyncHttpx
 from zhenxun.utils.manager.virtual_env_package_manager import VirtualEnvPackageManager
 from zhenxun.utils.manager.zhenxun_repo_manager import (
     ZhenxunRepoConfig,
     ZhenxunRepoManager,
 )
 from zhenxun.utils.platform import PlatformUtils
+from zhenxun.utils.repo_utils import RepoFileManager
 
 LOG_COMMAND = "AutoUpdate"
 
 
 class UpdateManager:
+    @staticmethod
+    async def _get_latest_commit_date(owner: str, repo: str, path: str) -> str:
+        """获取文件最新 commit 日期"""
+        api_url = f"https://api.github.com/repos/{owner}/{repo}/commits"
+        params = {"path": path, "page": 1, "per_page": 1}
+        try:
+            data = await AsyncHttpx.get_json(api_url, params=params)
+            if data and isinstance(data, list) and data[0]:
+                date_str = data[0]["commit"]["committer"]["date"]
+                return date_str.split("T")[0]
+        except Exception as e:
+            logger.warning(f"获取 {owner}/{repo}/{path} 的 commit 日期失败", e=e)
+        return "获取失败"
+
     @classmethod
     async def check_version(cls) -> str:
-        """检查更新版本
+        """检查真寻和资源的版本"""
+        bot_cur_version = cls.__get_version()
 
-        返回:
+        release_task = ZhenxunRepoManager.zhenxun_get_latest_releases_data()
-            str: 更新信息
+        dev_version_task = RepoFileManager.get_file_content(
-        """
+            ZhenxunRepoConfig.ZHENXUN_BOT_GITHUB_URL, "__version__"
-        cur_version = cls.__get_version()
-        release_data = await ZhenxunRepoManager.zhenxun_get_latest_releases_data()
-        if not release_data:
-            return "检查更新获取版本失败..."
-        return (
-            "检测到当前版本更新\n"
-            f"当前版本：{cur_version}\n"
-            f"最新版本：{release_data.get('name')}\n"
-            f"创建日期：{release_data.get('created_at')}\n"
-            f"更新内容：\n{release_data.get('body')}"
         )
+        bot_commit_date_task = cls._get_latest_commit_date(
+            "HibiKier", "zhenxun_bot", "__version__"
+        )
+        res_commit_date_task = cls._get_latest_commit_date(
+            "zhenxun-org", "zhenxun-bot-resources", "__version__"
+        )
+
+        (
+            release_data,
+            dev_version_text,
+            bot_commit_date,
+            res_commit_date,
+        ) = await asyncio.gather(
+            release_task,
+            dev_version_task,
+            bot_commit_date_task,
+            res_commit_date_task,
+            return_exceptions=True,
+        )
+
+        if isinstance(release_data, dict):
+            bot_release_version = release_data.get("name", "获取失败")
+            bot_release_date = release_data.get("created_at", "").split("T")[0]
+        else:
+            bot_release_version = "获取失败"
+            bot_release_date = "获取失败"
+            logger.warning(f"获取 Bot release 信息失败: {release_data}")
+
+        if isinstance(dev_version_text, str):
+            bot_dev_version = dev_version_text.split(":")[-1].strip()
+        else:
+            bot_dev_version = "获取失败"
+            bot_commit_date = "获取失败"
+            logger.warning(f"获取 Bot dev 版本信息失败: {dev_version_text}")
+
+        bot_update_hint = ""
+        try:
+            cur_base_v = bot_cur_version.split("-")[0].lstrip("v")
+            dev_base_v = bot_dev_version.split("-")[0].lstrip("v")
+
+            if Version(cur_base_v) < Version(dev_base_v):
+                bot_update_hint = "\n-> 发现新开发版本, 可用 `检查更新 main` 更新"
+            elif (
+                Version(cur_base_v) == Version(dev_base_v)
+                and bot_cur_version != bot_dev_version
+            ):
+                bot_update_hint = "\n-> 发现新开发版本, 可用 `检查更新 main` 更新"
+        except (InvalidVersion, TypeError, IndexError):
+            if bot_cur_version != bot_dev_version and bot_dev_version != "获取失败":
+                bot_update_hint = "\n-> 发现新开发版本, 可用 `检查更新 main` 更新"
+
+        bot_update_info = (
+            f"当前版本: {bot_cur_version}\n"
+            f"最新开发版: {bot_dev_version} (更新于: {bot_commit_date})\n"
+            f"最新正式版: {bot_release_version} (发布于: {bot_release_date})"
+            f"{bot_update_hint}"
+        )
+
+        res_version_file = ZhenxunRepoConfig.RESOURCE_PATH / "__version__"
+        res_cur_version = "未找到"
+        if res_version_file.exists():
+            if text := res_version_file.open(encoding="utf8").readline():
+                res_cur_version = text.split(":")[-1].strip()
+
+        res_latest_version = "获取失败"
+        try:
+            res_latest_version_text = await RepoFileManager.get_file_content(
+                ZhenxunRepoConfig.RESOURCE_GITHUB_URL, "__version__"
+            )
+            res_latest_version = res_latest_version_text.split(":")[-1].strip()
+        except Exception as e:
+            res_commit_date = "获取失败"
+            logger.warning(f"获取资源版本信息失败: {e}")
+
+        res_update_hint = ""
+        try:
+            if Version(res_cur_version) < Version(res_latest_version):
+                res_update_hint = "\n-> 发现新资源版本, 可用 `检查更新 resource` 更新"
+        except (InvalidVersion, TypeError):
+            pass
+
+        res_update_info = (
+            f"当前版本: {res_cur_version}\n"
+            f"最新版本: {res_latest_version} (更新于: {res_commit_date})"
+            f"{res_update_hint}"
+        )
+
+        return f"『绪山真寻 Bot』\n{bot_update_info}\n\n『真寻资源』\n{res_update_info}"
 
     @classmethod
     async def update_webui(
@@ -125,6 +223,7 @@ class UpdateManager:
             f"检测真寻已更新，当前版本：{cur_version}\n开始更新...",
             user_id,
         )
+        result_message = ""
         if zip:
             new_version = await ZhenxunRepoManager.zhenxun_zip_update(version_type)
             await PlatformUtils.send_superuser(
@@ -133,7 +232,7 @@ class UpdateManager:
             await VirtualEnvPackageManager.install_requirement(
                 ZhenxunRepoConfig.REQUIREMENTS_FILE
             )
-            return (
+            result_message = (
                 f"版本更新完成！\n版本: {cur_version} -> {new_version}\n"
                 "请重新启动真寻以完成更新!"
             )
@@ -155,13 +254,54 @@ class UpdateManager:
             await VirtualEnvPackageManager.install_requirement(
                 ZhenxunRepoConfig.REQUIREMENTS_FILE
             )
-            return (
+            result_message = (
                 f"版本更新完成！\n"
                 f"版本: {cur_version} -> {result.new_version}\n"
                 f"变更文件个数: {len(result.changed_files)}"
                 f"{'' if source == 'git' else '（阿里云更新不支持查看变更文件）'}\n"
                 "请重新启动真寻以完成更新!"
             )
+        resource_warning = ""
+        if version_type == "main":
+            try:
+                spec_content = await RepoFileManager.get_file_content(
+                    ZhenxunRepoConfig.ZHENXUN_BOT_GITHUB_URL, "resources.spec"
+                )
+                required_spec_str = None
+                for line in spec_content.splitlines():
+                    if line.startswith("require_resources_version:"):
+                        required_spec_str = line.split(":", 1)[1].strip().strip("\"'")
+                        break
+                if required_spec_str:
+                    res_version_file = ZhenxunRepoConfig.RESOURCE_PATH / "__version__"
+                    local_res_version_str = "0.0.0"
+                    if res_version_file.exists():
+                        if text := res_version_file.open(encoding="utf8").readline():
+                            local_res_version_str = text.split(":")[-1].strip()
+
+                    spec = SpecifierSet(required_spec_str)
+                    local_ver = Version(local_res_version_str)
+                    if not spec.contains(local_ver):
+                        warning_header = (
+                            f"⚠️ **资源版本不兼容!**\n"
+                            f"当前代码需要资源版本: `{required_spec_str}`\n"
+                            f"您当前的资源版本是: `{local_res_version_str}`\n"
+                            "**将自动为您更新资源文件...**"
+                        )
+                        await PlatformUtils.send_superuser(bot, warning_header, user_id)
+                        resource_update_source = None if zip else source
+                        resource_update_result = await cls.update_resources(
+                            source=resource_update_source, force=force
+                        )
+                        resource_warning = (
+                            f"\n\n{warning_header}\n{resource_update_result}"
+                        )
+            except Exception as e:
+                logger.warning(f"检查资源版本兼容性时出错: {e}", LOG_COMMAND, e=e)
+                resource_warning = (
+                    "\n\n⚠️ 检查资源版本兼容性时出错，建议手动运行 `检查更新 resource`"
+                )
+        return result_message + resource_warning
 
     @classmethod
     def __get_version(cls) -> str:

zhenxun/builtin_plugins/chat_history/chat_message_handle.py

@@ -19,12 +19,12 @@ from zhenxun.configs.config import Config
 from zhenxun.configs.utils import Command, PluginExtraData, RegisterConfig
 from zhenxun.models.chat_history import ChatHistory
 from zhenxun.models.group_member_info import GroupInfoUser
+from zhenxun.services import avatar_service
 from zhenxun.services.log import logger
 from zhenxun.ui.builders import TableBuilder
 from zhenxun.ui.models import ImageCell, TextCell
 from zhenxun.utils.enum import PluginType
 from zhenxun.utils.message import MessageUtils
-from zhenxun.utils.platform import PlatformUtils
 
 __plugin_meta__ = PluginMetadata(
     name="消息统计",
@@ -147,12 +147,14 @@ async def _(
                 user_in_group.user_name if user_in_group else f"{uid_str}(已退群)"
             )
 
-            avatar_url = PlatformUtils.get_user_avatar_url(uid_str, platform)
+            avatar_path = await avatar_service.get_avatar_path(platform, uid_str)
 
             rows_data.append(
                 [
                     TextCell(content=str(len(rows_data) + 1)),
-                    ImageCell(src=avatar_url or "", shape="circle"),
+                    ImageCell(
+                        src=avatar_path.as_uri() if avatar_path else "", shape="circle"
+                    ),
                     TextCell(content=user_name),
                     TextCell(content=str(num), bold=True),
                 ]

zhenxun/builtin_plugins/help/__init__.py

@@ -78,12 +78,18 @@ _matcher = on_alconna(
         Option("-s|--superuser", action=store_true, help_text="超级用户帮助"),
         Option("-d|--detail", action=store_true, help_text="详细帮助"),
     ),
-    aliases={"help", "帮助", "菜单"},
+    aliases={"help", "菜单"},
     rule=to_me(),
     priority=1,
     block=True,
 )
 
+_matcher.shortcut(
+    r"帮助(?P<name>.*?)",
+    command="功能",
+    arguments=["{name}"],
+    prefix=True,
+)
 
 _matcher.shortcut(
     r"详细帮助",

zhenxun/builtin_plugins/help/_data_source.py

@@ -13,6 +13,7 @@ from zhenxun.models.statistics import Statistics
 from zhenxun.services import (
     LLMException,
     LLMMessage,
+    avatar_service,
     generate,
 )
 from zhenxun.services.log import logger
@@ -105,7 +106,8 @@ async def create_help_img(
 
     platform = PlatformUtils.get_platform(session)
     bot_id = BotConfig.get_qbot_uid(session.self_id) or session.self_id
-    bot_avatar_url = PlatformUtils.get_user_avatar_url(bot_id, platform) or ""
+    bot_avatar_path = await avatar_service.get_avatar_path(platform, bot_id)
+    bot_avatar_url = bot_avatar_path.as_uri() if bot_avatar_path else ""
 
     builder = PluginMenuBuilder(
         bot_name=BotConfig.self_nickname,

zhenxun/builtin_plugins/info/my_info.py

@@ -11,6 +11,7 @@ from zhenxun.models.level_user import LevelUser
 from zhenxun.models.sign_user import SignUser
 from zhenxun.models.statistics import Statistics
 from zhenxun.models.user_console import UserConsole
+from zhenxun.services import avatar_service
 from zhenxun.utils.platform import PlatformUtils
 
 RACE = [
@@ -139,9 +140,8 @@ async def get_user_info(
         bytes: 图片数据
     """
     platform = PlatformUtils.get_platform(session) or "qq"
-    avatar_url = (
+    avatar_path = await avatar_service.get_avatar_path(platform, user_id)
-        PlatformUtils.get_user_avatar_url(user_id, platform, session.self_id) or ""
+    avatar_url = avatar_path.as_uri() if avatar_path else ""
-    )
 
     user = await UserConsole.get_user(user_id, platform)
     permission_level = await LevelUser.get_user_level(user_id, group_id)

zhenxun/builtin_plugins/mahiro_bank/data_source.py

@@ -11,6 +11,7 @@ from zhenxun.models.mahiro_bank import MahiroBank
 from zhenxun.models.mahiro_bank_log import MahiroBankLog
 from zhenxun.models.sign_user import SignUser
 from zhenxun.models.user_console import UserConsole
+from zhenxun.services import avatar_service
 from zhenxun.utils.enum import BankHandleType, GoldHandle
 from zhenxun.utils.platform import PlatformUtils
 
@@ -210,9 +211,8 @@ class BankManager:
             for deposit in user_today_deposit
         ]
         platform = PlatformUtils.get_platform(session)
-        avatar_url = PlatformUtils.get_user_avatar_url(
+        avatar_path = await avatar_service.get_avatar_path(platform, user_id)
-            user_id, platform, session.self_id
+        avatar_url = avatar_path.as_uri() if avatar_path else ""
-        )
         return {
             "name": uname,
             "rank": rank + 1,

zhenxun/builtin_plugins/plugin_store/data_source.py

@@ -1,3 +1,4 @@
+import os
 from pathlib import Path
 import random
 import shutil
@@ -10,6 +11,7 @@ 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.utils.enum import PluginType
 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
@@ -183,6 +185,8 @@ class StoreManager:
             StorePluginInfo: 插件信息
             bool: 是否是外部插件
         """
+        plugin_list: list[StorePluginInfo]
+        extra_plugin_list: list[StorePluginInfo]
         plugin_list, extra_plugin_list = await cls.get_data()
         plugin_info = None
         is_external = False
@@ -206,6 +210,12 @@ class StoreManager:
         if is_remove:
             if plugin_info.module not in modules:
                 raise PluginStoreException(f"插件 {plugin_info.name} 未安装，无法移除")
+            if plugin_obj := await PluginInfo.get_plugin(
+                module=plugin_info.module, plugin_type=PluginType.PARENT
+            ):
+                plugin_info.module_path = plugin_obj.module_path
+            elif plugin_obj := await PluginInfo.get_plugin(module=plugin_info.module):
+                plugin_info.module_path = plugin_obj.module_path
             return plugin_info, is_external
 
         if is_update:
@@ -237,9 +247,7 @@ class StoreManager:
             plugin_info.github_url = f"{github_url_split[0]}/tree/{version_split[1]}"
         logger.info(f"正在安装插件 {plugin_info.name}...", LOG_COMMAND)
         await cls.install_plugin_with_repo(
-            plugin_info.github_url,
+            plugin_info,
-            plugin_info.module_path,
-            plugin_info.is_dir,
             is_external,
             source,
         )
@@ -248,9 +256,7 @@ class StoreManager:
     @classmethod
     async def install_plugin_with_repo(
         cls,
-        github_url: str,
+        plugin_info: StorePluginInfo,
-        module_path: str,
-        is_dir: bool,
         is_external: bool = False,
         source: str | None = None,
     ):
@@ -267,18 +273,26 @@ class StoreManager:
             repo_type = RepoType.ALIYUN
         elif source == "git":
             repo_type = RepoType.GITHUB
-        replace_module_path = module_path.replace(".", "/")
+        module_path = plugin_info.module_path
-        plugin_name = module_path.split(".")[-1]
+        is_dir = plugin_info.is_dir
+        github_url = plugin_info.github_url
+        assert github_url
+        replace_module_path = module_path.replace(".", "/").lstrip("/")
+        plugin_name = module_path.split(".")[-1] or plugin_info.module
         if is_dir:
             files = await RepoFileManager.list_directory_files(
                 github_url, replace_module_path, repo_type=repo_type
             )
         else:
             files = [RepoFileInfo(path=f"{replace_module_path}.py", is_dir=False)]
-        local_path = BASE_PATH / "plugins" if is_external else BASE_PATH
+        if not is_external:
-        target_dir = BASE_PATH / "plugins" / plugin_name
+            target_dir = BASE_PATH
+        elif is_dir and module_path == ".":
+            target_dir = BASE_PATH / "plugins" / plugin_name
+        else:
+            target_dir = BASE_PATH / "plugins"
         files = [file for file in files if not file.is_dir]
-        download_files = [(file.path, local_path / file.path) for file in files]
+        download_files = [(file.path, target_dir / file.path) for file in files]
         result = await RepoFileManager.download_files(
             github_url,
             download_files,
@@ -298,7 +312,7 @@ class StoreManager:
 
         is_install_req = False
         for requirement_path in requirement_paths:
-            requirement_file = local_path / requirement_path.path
+            requirement_file = target_dir / requirement_path.path
             if requirement_file.exists():
                 is_install_req = True
                 await VirtualEnvPackageManager.install_requirement(requirement_file)
@@ -341,13 +355,11 @@ class StoreManager:
             str: 返回消息
         """
         plugin_info, _ = await cls.get_plugin_by_value(index_or_module, is_remove=True)
-        path = BASE_PATH
+        module_path = plugin_info.module_path
-        if plugin_info.github_url:
+        module = module_path.split(".")[-1]
-            path = BASE_PATH / "plugins"
+        path = BASE_PATH.parent / Path(module_path.replace(".", os.sep))
-        for p in plugin_info.module_path.split("."):
-            path = path / p
         if not plugin_info.is_dir:
-            path = Path(f"{path}.py")
+            path = path.parent / f"{module}.py"
         if not path.exists():
             return f"插件 {plugin_info.name} 不存在..."
         logger.debug(f"尝试移除插件 {plugin_info.name} 文件: {path}", LOG_COMMAND)
@@ -356,7 +368,7 @@ class StoreManager:
             shutil.rmtree(path, onerror=win_on_rm_error)
         else:
             path.unlink()
-        await PluginInitManager.remove(f"zhenxun.{plugin_info.module_path}")
+        await PluginInitManager.remove(module_path)
         return f"插件 {plugin_info.name} 移除成功! 重启后生效"
 
     @classmethod
@@ -423,9 +435,7 @@ class StoreManager:
         if plugin_info.github_url is None:
             plugin_info.github_url = DEFAULT_GITHUB_URL
         await cls.install_plugin_with_repo(
-            plugin_info.github_url,
+            plugin_info,
-            plugin_info.module_path,
-            plugin_info.is_dir,
             is_external,
         )
         return f"插件 {plugin_info.name} 更新成功! 重启后生效"
@@ -473,9 +483,7 @@ class StoreManager:
                     plugin_info.github_url = DEFAULT_GITHUB_URL
                     is_external = False
                 await cls.install_plugin_with_repo(
-                    plugin_info.github_url,
+                    plugin_info,
-                    plugin_info.module_path,
-                    plugin_info.is_dir,
                     is_external,
                 )
                 update_success_list.append(plugin_info.name)

zhenxun/builtin_plugins/shop/_data_source.py

@@ -21,6 +21,7 @@ from zhenxun.models.group_member_info import GroupInfoUser
 from zhenxun.models.user_console import UserConsole
 from zhenxun.models.user_gold_log import UserGoldLog
 from zhenxun.models.user_props_log import UserPropsLog
+from zhenxun.services import avatar_service
 from zhenxun.services.log import logger
 from zhenxun.ui.models import ImageCell, TextCell
 from zhenxun.utils.enum import GoldHandle, PropHandle
@@ -123,12 +124,14 @@ async def gold_rank(session: Uninfo, group_id: str | None, num: int) -> bytes |
     data_list = []
     platform = PlatformUtils.get_platform(session)
     for i, user in enumerate(user_list):
-        ava_url = PlatformUtils.get_user_avatar_url(user[0], platform, session.self_id)
+        avatar_path = await avatar_service.get_avatar_path(platform, user[0])
         data_list.append(
             [
                 TextCell(content=f"{i + 1}"),
-                ImageCell(src=ava_url or "", shape="circle")
+                ImageCell(
-                if platform == "qq"
+                    src=avatar_path.as_uri() if avatar_path else "", shape="circle"
+                )
+                if avatar_path
                 else TextCell(content=""),
                 TextCell(content=uid2name.get(user[0]) or user[0]),
                 TextCell(content=str(user[1]), bold=True),

zhenxun/builtin_plugins/sign_in/_data_source.py

@@ -13,6 +13,7 @@ from zhenxun.models.group_member_info import GroupInfoUser
 from zhenxun.models.sign_log import SignLog
 from zhenxun.models.sign_user import SignUser
 from zhenxun.models.user_console import UserConsole
+from zhenxun.services.avatar_service import avatar_service
 from zhenxun.services.log import logger
 from zhenxun.ui.models import ImageCell, TextCell
 from zhenxun.utils.platform import PlatformUtils
@@ -79,14 +80,16 @@ class SignManage:
         data_list = []
         platform = PlatformUtils.get_platform(session)
         for i, user in enumerate(user_list):
-            ava_url = PlatformUtils.get_user_avatar_url(
+            avatar_path = await avatar_service.get_avatar_path(
-                user[0], platform, session.self_id
+                platform=user[3] or "qq", identifier=user[0]
             )
             data_list.append(
                 [
                     TextCell(content=f"{i + 1}"),
-                    ImageCell(src=ava_url or "", shape="circle")
+                    ImageCell(
-                    if user[3] == "qq"
+                        src=avatar_path.as_uri() if avatar_path else "", shape="circle"
+                    )
+                    if avatar_path
                     else TextCell(content=""),
                     TextCell(content=uid2name.get(user[0]) or user[0]),
                     TextCell(content=str(user[1]), bold=True),

zhenxun/builtin_plugins/sign_in/utils.py

@@ -11,6 +11,7 @@ from nonebot_plugin_uninfo import Uninfo
 from zhenxun import ui
 from zhenxun.configs.config import BotConfig, Config
 from zhenxun.models.sign_user import SignUser
+from zhenxun.services import avatar_service
 from zhenxun.utils.manager.priority_manager import PriorityLifecycle
 from zhenxun.utils.platform import PlatformUtils
 
@@ -212,13 +213,13 @@ async def _generate_html_card(
     if len(nickname) > 6:
         font_size = 27
 
+    avatar_path = await avatar_service.get_avatar_path(
+        PlatformUtils.get_platform(session), user.user_id
+    )
     user_info = {
         "nickname": nickname,
         "uid_str": uid_formatted,
-        "avatar_url": PlatformUtils.get_user_avatar_url(
+        "avatar_url": avatar_path.as_uri() if avatar_path else "",
-            user.user_id, PlatformUtils.get_platform(session), session.self_id
-        )
-        or "",
         "sign_count": user.sign_count,
         "font_size": font_size,
     }

zhenxun/models/plugin_info.py

@@ -77,7 +77,7 @@ class PluginInfo(Model):
         返回:
             Self | None: 插件
         """
-        if filter_parent:
+        if not kwargs.get("plugin_type") and filter_parent:
             return await cls.get_or_none(
                 load_status=load_status, plugin_type__not=PluginType.PARENT, **kwargs
             )
@@ -96,7 +96,7 @@ class PluginInfo(Model):
         返回:
             list[Self]: 插件列表
         """
-        if filter_parent:
+        if not kwargs.get("plugin_type") and filter_parent:
             return await cls.filter(
                 load_status=load_status, plugin_type__not=PluginType.PARENT, **kwargs
             ).all()

zhenxun/services/__init__.py

@@ -18,6 +18,7 @@ require("nonebot_plugin_htmlrender")
 require("nonebot_plugin_uninfo")
 require("nonebot_plugin_waiter")
 
+from .avatar_service import avatar_service
 from .db_context import Model, disconnect, with_db_timeout
 from .llm import (
     AI,
@@ -57,6 +58,7 @@ __all__ = [
     "Model",
     "PluginInit",
     "PluginInitManager",
+    "avatar_service",
     "chat",
     "clear_model_cache",
     "code",

zhenxun/services/avatar_service.py

@@ -0,0 +1,141 @@
+"""
+头像缓存服务
+
+提供一个统一的、带缓存的头像获取服务，支持多平台和可配置的过期策略。
+"""
+
+import os
+from pathlib import Path
+import time
+
+from nonebot_plugin_apscheduler import scheduler
+
+from zhenxun.configs.config import Config
+from zhenxun.configs.path_config import DATA_PATH
+from zhenxun.services.log import logger
+from zhenxun.utils.http_utils import AsyncHttpx
+from zhenxun.utils.platform import PlatformUtils
+
+Config.add_plugin_config(
+    "avatar_cache",
+    "ENABLED",
+    True,
+    help="是否启用头像缓存功能",
+    default_value=True,
+    type=bool,
+)
+Config.add_plugin_config(
+    "avatar_cache",
+    "TTL_DAYS",
+    7,
+    help="头像缓存的有效期（天）",
+    default_value=7,
+    type=int,
+)
+Config.add_plugin_config(
+    "avatar_cache",
+    "CLEANUP_INTERVAL_HOURS",
+    24,
+    help="后台清理过期缓存的间隔时间（小时）",
+    default_value=24,
+    type=int,
+)
+
+
+class AvatarService:
+    """
+    一个集中式的头像缓存服务，提供L1（内存）和L2（文件）两级缓存。
+    """
+
+    def __init__(self):
+        self.cache_path = (DATA_PATH / "cache" / "avatars").resolve()
+        self.cache_path.mkdir(parents=True, exist_ok=True)
+        self._memory_cache: dict[str, Path] = {}
+
+    def _get_cache_path(self, platform: str, identifier: str) -> Path:
+        """
+        根据平台和ID生成存储的文件路径。
+        例如: data/cache/avatars/qq/123456789.png
+        """
+        identifier = str(identifier)
+        return self.cache_path / platform / f"{identifier}.png"
+
+    async def get_avatar_path(
+        self, platform: str, identifier: str, force_refresh: bool = False
+    ) -> Path | None:
+        """
+        获取用户或群组的头像本地路径。
+
+        参数:
+            platform: 平台名称 (e.g., 'qq')
+            identifier: 用户ID或群组ID
+            force_refresh: 是否强制刷新缓存
+
+        返回:
+            Path | None: 头像的本地文件路径，如果获取失败则返回None。
+        """
+        if not Config.get_config("avatar_cache", "ENABLED"):
+            return None
+
+        cache_key = f"{platform}-{identifier}"
+        if not force_refresh and cache_key in self._memory_cache:
+            if self._memory_cache[cache_key].exists():
+                return self._memory_cache[cache_key]
+
+        local_path = self._get_cache_path(platform, identifier)
+        ttl_seconds = Config.get_config("avatar_cache", "TTL_DAYS", 7) * 86400
+
+        if not force_refresh and local_path.exists():
+            try:
+                file_mtime = os.path.getmtime(local_path)
+                if time.time() - file_mtime < ttl_seconds:
+                    self._memory_cache[cache_key] = local_path
+                    return local_path
+            except FileNotFoundError:
+                pass
+
+        avatar_url = PlatformUtils.get_user_avatar_url(identifier, platform)
+        if not avatar_url:
+            return None
+
+        local_path.parent.mkdir(parents=True, exist_ok=True)
+
+        if await AsyncHttpx.download_file(avatar_url, local_path):
+            self._memory_cache[cache_key] = local_path
+            return local_path
+        else:
+            logger.warning(f"下载头像失败: {avatar_url}", "AvatarService")
+            return None
+
+    async def _cleanup_cache(self):
+        """后台定时清理过期的缓存文件"""
+        if not Config.get_config("avatar_cache", "ENABLED"):
+            return
+
+        logger.info("开始执行头像缓存清理任务...", "AvatarService")
+        ttl_seconds = Config.get_config("avatar_cache", "TTL_DAYS", 7) * 86400
+        now = time.time()
+        deleted_count = 0
+        for root, _, files in os.walk(self.cache_path):
+            for name in files:
+                file_path = Path(root) / name
+                try:
+                    if now - os.path.getmtime(file_path) > ttl_seconds:
+                        file_path.unlink()
+                        deleted_count += 1
+                except FileNotFoundError:
+                    continue
+
+        logger.info(
+            f"头像缓存清理完成，共删除 {deleted_count} 个过期文件。", "AvatarService"
+        )
+
+
+avatar_service = AvatarService()
+
+
+@scheduler.scheduled_job(
+    "interval", hours=Config.get_config("avatar_cache", "CLEANUP_INTERVAL_HOURS", 24)
+)
+async def _run_avatar_cache_cleanup():
+    await avatar_service._cleanup_cache()

zhenxun/services/plugin_init.py

@@ -87,7 +87,7 @@ class PluginInitManager:
 
     @classmethod
     async def remove(cls, module_path: str):
-        """运行指定插件安装方法"""
+        """运行指定插件移除方法"""
         if model := cls.plugins.get(module_path):
             if model.remove:
                 class_ = model.class_()

zhenxun/utils/browser.py

@@ -4,7 +4,7 @@ from pathlib import Path
 from typing import Any, Literal
 
 from nonebot_plugin_alconna import UniMessage
-from nonebot_plugin_htmlrender import get_browser
+from nonebot_plugin_htmlrender.browser import get_browser
 from playwright.async_api import Page
 
 from zhenxun.utils.message import MessageUtils

zhenxun/utils/platform.py

@@ -247,7 +247,7 @@ class PlatformUtils:
         if platform != "qq":
             return None
         if user_id.isdigit():
-            return f"http://q1.qlogo.cn/g?b=qq&nk={user_id}&s=160"
+            return f"http://q1.qlogo.cn/g?b=qq&nk={user_id}&s=640"
         else:
             return f"https://q.qlogo.cn/qqapp/{appid}/{user_id}/640"
 

zhenxun/utils/repo_utils/file_manager.py

@@ -326,7 +326,7 @@ class RepoFileManager:
 
             # 获取仓库树信息
             strategy = GitHubStrategy()
-            strategy.body = await GitHubStrategy.parse_repo_info(repo_info)
+            strategy.body = await strategy.parse_repo_info(repo_info)
 
             # 处理目录路径，确保格式正确
             if directory_path and not directory_path.endswith("/") and recursive:
@@ -480,7 +480,7 @@ class RepoFileManager:
         target_dir: Path | None = None,
     ) -> FileDownloadResult:
         """
-        下载单个文件
+        下载多个文件
 
         参数:
             repo_url: 仓库URL

zhenxun/utils/repo_utils/utils.py

@@ -7,6 +7,7 @@ import base64
 from pathlib import Path
 import re
 import shutil
+import tempfile
 
 from zhenxun.services.log import logger
 
@@ -145,80 +146,85 @@ async def sparse_checkout_clone(
     target_dir: Path,
 ) -> None:
     """
-    使用 git 稀疏检出克隆指定路径到目标目录（完全独立于主项目 git）。
+    使用 git 稀疏检出克隆指定路径到目标目录（在临时目录中操作）。
 
     关键保障:
-    - 在 target_dir 下检测/初始化 .git，所有 git 操作均以 cwd=target_dir 执行
+    - 在临时目录中执行所有 git 操作，避免影响 target_dir 中的现有内容
-    - 强制拉取与工作区覆盖: fetch --force、checkout -B、reset --hard、clean -xdf
+    - 只操作 target_dir/sparse_path 路径，不影响 target_dir 其他内容
-    - 反复设置 sparse-checkout 路径，确保路径更新生效
     """
     target_dir.mkdir(parents=True, exist_ok=True)
 
     if not await check_git():
         raise GitUnavailableError()
 
-    git_dir = target_dir / ".git"
+    # 在临时目录中进行 git 操作
-    if not git_dir.exists():
+    with tempfile.TemporaryDirectory() as temp_dir:
-        success, out, err = await run_git_command("init", target_dir)
+        temp_path = Path(temp_dir)
+
+        # 初始化临时目录为 git 仓库
+        success, out, err = await run_git_command("init", temp_path)
         if not success:
             raise RuntimeError(f"git init 失败: {err or out}")
         success, out, err = await run_git_command(
-            f"remote add origin {repo_url}", target_dir
+            f"remote add origin {repo_url}", temp_path
         )
         if not success:
             raise RuntimeError(f"添加远程失败: {err or out}")
-    else:
+
+        # 启用稀疏检出（使用 --no-cone 模式以获得更精确的控制）
+        await run_git_command("config core.sparseCheckout true", temp_path)
+        await run_git_command("sparse-checkout init --no-cone", temp_path)
+
+        # 设置需要检出的路径（每次都覆盖配置）
+        if not sparse_path:
+            raise RuntimeError("sparse-checkout 路径不能为空")
+
+        # 使用 --no-cone 模式，直接指定要检出的具体路径
         success, out, err = await run_git_command(
-            f"remote set-url origin {repo_url}", target_dir
+            f"sparse-checkout set {sparse_path}/", temp_path
         )
         if not success:
-            # 兜底尝试添加
+            raise RuntimeError(f"配置稀疏路径失败: {err or out}")
-            await run_git_command(f"remote add origin {repo_url}", target_dir)
 
-    # 启用稀疏检出（使用 --no-cone 模式以获得更精确的控制）
+        # 强制拉取并同步到远端
-    await run_git_command("config core.sparseCheckout true", target_dir)
+        success, out, err = await run_git_command(
-    await run_git_command("sparse-checkout init --no-cone", target_dir)
+            f"fetch --force --depth 1 origin {branch}", temp_path
+        )
+        if not success:
+            raise RuntimeError(f"fetch 失败: {err or out}")
 
-    # 设置需要检出的路径（每次都覆盖配置）
+        # 使用远端强制更新本地分支并覆盖工作区
-    if not sparse_path:
+        success, out, err = await run_git_command(
-        raise RuntimeError("sparse-checkout 路径不能为空")
+            f"checkout -B {branch} origin/{branch}", temp_path
+        )
+        if not success:
+            # 回退方案
+            success2, out2, err2 = await run_git_command(
+                f"checkout {branch}", temp_path
+            )
+            if not success2:
+                raise RuntimeError(f"checkout 失败: {(err or out) or (err2 or out2)}")
 
-    # 使用 --no-cone 模式，直接指定要检出的具体路径
+        # 强制对齐工作区
-    # 例如：sparse_path="plugins/mahiro" -> 只检出 plugins/mahiro/ 下的内容
+        await run_git_command(f"reset --hard origin/{branch}", temp_path)
-    success, out, err = await run_git_command(
+        await run_git_command("clean -xdf", temp_path)
-        f"sparse-checkout set {sparse_path}/", target_dir
-    )
-    if not success:
-        raise RuntimeError(f"配置稀疏路径失败: {err or out}")
 
-    # 强制拉取并同步到远端
+        # 将检出的文件移动到目标位置
-    success, out, err = await run_git_command(
+        source_path = temp_path / sparse_path
-        f"fetch --force --depth 1 origin {branch}", target_dir
+        if source_path.exists():
-    )
+            # 确保目标路径存在
-    if not success:
+            target_path = target_dir / sparse_path
-        raise RuntimeError(f"fetch 失败: {err or out}")
+            target_path.parent.mkdir(parents=True, exist_ok=True)
 
-    # 使用远端强制更新本地分支并覆盖工作区
+            # 如果目标路径已存在，先清理
-    success, out, err = await run_git_command(
+            if target_path.exists():
-        f"checkout -B {branch} origin/{branch}", target_dir
+                if target_path.is_dir():
-    )
+                    shutil.rmtree(target_path)
-    if not success:
+                else:
-        # 回退方案
+                    target_path.unlink()
-        success2, out2, err2 = await run_git_command(f"checkout {branch}", target_dir)
-        if not success2:
-            raise RuntimeError(f"checkout 失败: {(err or out) or (err2 or out2)}")
 
-    # 强制对齐工作区
+            # 移动整个目录结构到目标位置
-    await run_git_command(f"reset --hard origin/{branch}", target_dir)
+            shutil.move(str(source_path), str(target_path))
-    await run_git_command("clean -xdf", target_dir)
-
-    dir_path = target_dir / Path(sparse_path)
-    for f in dir_path.iterdir():
-        shutil.move(f, target_dir / f.name)
-    dir_name = sparse_path.split("/")[0]
-    rm_path = target_dir / dir_name
-    if rm_path.exists():
-        shutil.rmtree(rm_path)
 
 
 def prepare_aliyun_url(repo_url: str) -> str: