🎉 chore(version): Update version to v0.2.4-c7ef6fd

__version__

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

zhenxun/builtin_plugins/auto_update/__init__.py

@@ -84,16 +84,13 @@ 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,135 +1,37 @@
-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()
+        返回:
-        dev_version_task = RepoFileManager.get_file_content(
+            str: 更新信息
-            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(
@@ -223,7 +125,6 @@ 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(
@@ -232,7 +133,7 @@ class UpdateManager:
             await VirtualEnvPackageManager.install_requirement(
                 ZhenxunRepoConfig.REQUIREMENTS_FILE
             )
-            result_message = (
+            return (
                 f"版本更新完成！\n版本: {cur_version} -> {new_version}\n"
                 "请重新启动真寻以完成更新!"
             )
@@ -254,54 +155,13 @@ class UpdateManager:
             await VirtualEnvPackageManager.install_requirement(
                 ZhenxunRepoConfig.REQUIREMENTS_FILE
             )
-            result_message = (
+            return (
                 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,14 +147,12 @@ async def _(
                 user_in_group.user_name if user_in_group else f"{uid_str}(已退群)"
             )
 
-            avatar_path = await avatar_service.get_avatar_path(platform, uid_str)
+            avatar_url = PlatformUtils.get_user_avatar_url(uid_str, platform)
 
             rows_data.append(
                 [
                     TextCell(content=str(len(rows_data) + 1)),
-                    ImageCell(
+                    ImageCell(src=avatar_url or "", shape="circle"),
-                        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,18 +78,12 @@ _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,7 +13,6 @@ from zhenxun.models.statistics import Statistics
 from zhenxun.services import (
     LLMException,
     LLMMessage,
-    avatar_service,
     generate,
 )
 from zhenxun.services.log import logger
@@ -106,8 +105,7 @@ 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_path = await avatar_service.get_avatar_path(platform, bot_id)
+    bot_avatar_url = PlatformUtils.get_user_avatar_url(bot_id, platform) or ""
-    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,7 +11,6 @@ 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 = [
@@ -140,8 +139,9 @@ async def get_user_info(
         bytes: 图片数据
     """
     platform = PlatformUtils.get_platform(session) or "qq"
-    avatar_path = await avatar_service.get_avatar_path(platform, user_id)
+    avatar_url = (
-    avatar_url = avatar_path.as_uri() if avatar_path else ""
+        PlatformUtils.get_user_avatar_url(user_id, platform, session.self_id) or ""
+    )
 
     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,7 +11,6 @@ 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
 
@@ -211,8 +210,9 @@ class BankManager:
             for deposit in user_today_deposit
         ]
         platform = PlatformUtils.get_platform(session)
-        avatar_path = await avatar_service.get_avatar_path(platform, user_id)
+        avatar_url = PlatformUtils.get_user_avatar_url(
-        avatar_url = avatar_path.as_uri() if avatar_path else ""
+            user_id, platform, session.self_id
+        )
         return {
             "name": uname,
             "rank": rank + 1,

zhenxun/builtin_plugins/plugin_store/data_source.py

@@ -1,4 +1,3 @@
-import os
 from pathlib import Path
 import random
 import shutil
@@ -11,7 +10,6 @@ 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
@@ -185,8 +183,6 @@ 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
@@ -210,12 +206,6 @@ 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:
@@ -247,7 +237,9 @@ 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,
+            plugin_info.github_url,
+            plugin_info.module_path,
+            plugin_info.is_dir,
             is_external,
             source,
         )
@@ -256,7 +248,9 @@ class StoreManager:
     @classmethod
     async def install_plugin_with_repo(
         cls,
-        plugin_info: StorePluginInfo,
+        github_url: str,
+        module_path: str,
+        is_dir: bool,
         is_external: bool = False,
         source: str | None = None,
     ):
@@ -273,26 +267,18 @@ class StoreManager:
             repo_type = RepoType.ALIYUN
         elif source == "git":
             repo_type = RepoType.GITHUB
-        module_path = plugin_info.module_path
+        replace_module_path = module_path.replace(".", "/")
-        is_dir = plugin_info.is_dir
+        plugin_name = module_path.split(".")[-1]
-        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)]
-        if not is_external:
+        local_path = BASE_PATH / "plugins" if is_external else BASE_PATH
-            target_dir = BASE_PATH
+        target_dir = BASE_PATH / "plugins" / plugin_name
-        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, target_dir / file.path) for file in files]
+        download_files = [(file.path, local_path / file.path) for file in files]
         result = await RepoFileManager.download_files(
             github_url,
             download_files,
@@ -312,7 +298,7 @@ class StoreManager:
 
         is_install_req = False
         for requirement_path in requirement_paths:
-            requirement_file = target_dir / requirement_path.path
+            requirement_file = local_path / requirement_path.path
             if requirement_file.exists():
                 is_install_req = True
                 await VirtualEnvPackageManager.install_requirement(requirement_file)
@@ -355,11 +341,13 @@ class StoreManager:
             str: 返回消息
         """
         plugin_info, _ = await cls.get_plugin_by_value(index_or_module, is_remove=True)
-        module_path = plugin_info.module_path
+        path = BASE_PATH
-        module = module_path.split(".")[-1]
+        if plugin_info.github_url:
-        path = BASE_PATH.parent / Path(module_path.replace(".", os.sep))
+            path = BASE_PATH / "plugins"
+        for p in plugin_info.module_path.split("."):
+            path = path / p
         if not plugin_info.is_dir:
-            path = path.parent / f"{module}.py"
+            path = Path(f"{path}.py")
         if not path.exists():
             return f"插件 {plugin_info.name} 不存在..."
         logger.debug(f"尝试移除插件 {plugin_info.name} 文件: {path}", LOG_COMMAND)
@@ -368,7 +356,7 @@ class StoreManager:
             shutil.rmtree(path, onerror=win_on_rm_error)
         else:
             path.unlink()
-        await PluginInitManager.remove(module_path)
+        await PluginInitManager.remove(f"zhenxun.{plugin_info.module_path}")
         return f"插件 {plugin_info.name} 移除成功! 重启后生效"
 
     @classmethod
@@ -435,7 +423,9 @@ class StoreManager:
         if plugin_info.github_url is None:
             plugin_info.github_url = DEFAULT_GITHUB_URL
         await cls.install_plugin_with_repo(
-            plugin_info,
+            plugin_info.github_url,
+            plugin_info.module_path,
+            plugin_info.is_dir,
             is_external,
         )
         return f"插件 {plugin_info.name} 更新成功! 重启后生效"
@@ -483,7 +473,9 @@ class StoreManager:
                     plugin_info.github_url = DEFAULT_GITHUB_URL
                     is_external = False
                 await cls.install_plugin_with_repo(
-                    plugin_info,
+                    plugin_info.github_url,
+                    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,7 +21,6 @@ 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
@@ -124,14 +123,12 @@ 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):
-        avatar_path = await avatar_service.get_avatar_path(platform, user[0])
+        ava_url = PlatformUtils.get_user_avatar_url(user[0], platform, session.self_id)
         data_list.append(
             [
                 TextCell(content=f"{i + 1}"),
-                ImageCell(
+                ImageCell(src=ava_url or "", shape="circle")
-                    src=avatar_path.as_uri() if avatar_path else "", shape="circle"
+                if platform == "qq"
-                )
-                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,7 +13,6 @@ 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
@@ -80,16 +79,14 @@ class SignManage:
         data_list = []
         platform = PlatformUtils.get_platform(session)
         for i, user in enumerate(user_list):
-            avatar_path = await avatar_service.get_avatar_path(
+            ava_url = PlatformUtils.get_user_avatar_url(
-                platform=user[3] or "qq", identifier=user[0]
+                user[0], platform, session.self_id
             )
             data_list.append(
                 [
                     TextCell(content=f"{i + 1}"),
-                    ImageCell(
+                    ImageCell(src=ava_url or "", shape="circle")
-                        src=avatar_path.as_uri() if avatar_path else "", shape="circle"
+                    if user[3] == "qq"
-                    )
-                    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,7 +11,6 @@ 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
 
@@ -213,13 +212,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": avatar_path.as_uri() if avatar_path else "",
+        "avatar_url": PlatformUtils.get_user_avatar_url(
+            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 not kwargs.get("plugin_type") and filter_parent:
+        if 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 not kwargs.get("plugin_type") and filter_parent:
+        if filter_parent:
             return await cls.filter(
                 load_status=load_status, plugin_type__not=PluginType.PARENT, **kwargs
             ).all()

zhenxun/services/__init__.py

@@ -18,7 +18,6 @@ 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,
@@ -58,7 +57,6 @@ __all__ = [
     "Model",
     "PluginInit",
     "PluginInitManager",
-    "avatar_service",
     "chat",
     "clear_model_cache",
     "code",

zhenxun/services/avatar_service.py

@@ -1,141 +0,0 @@
-"""
-头像缓存服务
-
-提供一个统一的、带缓存的头像获取服务，支持多平台和可配置的过期策略。
-"""
-
-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.browser import get_browser
+from nonebot_plugin_htmlrender 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=640"
+            return f"http://q1.qlogo.cn/g?b=qq&nk={user_id}&s=160"
         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 strategy.parse_repo_info(repo_info)
+            strategy.body = await GitHubStrategy.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,7 +7,6 @@ import base64
 from pathlib import Path
 import re
 import shutil
-import tempfile
 
 from zhenxun.services.log import logger
 
@@ -146,85 +145,80 @@ async def sparse_checkout_clone(
     target_dir: Path,
 ) -> None:
     """
-    使用 git 稀疏检出克隆指定路径到目标目录（在临时目录中操作）。
+    使用 git 稀疏检出克隆指定路径到目标目录（完全独立于主项目 git）。
 
     关键保障:
-    - 在临时目录中执行所有 git 操作，避免影响 target_dir 中的现有内容
+    - 在 target_dir 下检测/初始化 .git，所有 git 操作均以 cwd=target_dir 执行
-    - 只操作 target_dir/sparse_path 路径，不影响 target_dir 其他内容
+    - 强制拉取与工作区覆盖: fetch --force、checkout -B、reset --hard、clean -xdf
+    - 反复设置 sparse-checkout 路径，确保路径更新生效
     """
     target_dir.mkdir(parents=True, exist_ok=True)
 
     if not await check_git():
         raise GitUnavailableError()
 
-    # 在临时目录中进行 git 操作
+    git_dir = target_dir / ".git"
-    with tempfile.TemporaryDirectory() as temp_dir:
+    if not git_dir.exists():
-        temp_path = Path(temp_dir)
+        success, out, err = await run_git_command("init", target_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}", temp_path
+            f"remote add origin {repo_url}", target_dir
         )
         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"sparse-checkout set {sparse_path}/", temp_path
+            f"remote set-url origin {repo_url}", target_dir
         )
         if not success:
-            raise RuntimeError(f"配置稀疏路径失败: {err or out}")
+            # 兜底尝试添加
+            await run_git_command(f"remote add origin {repo_url}", target_dir)
 
-        # 强制拉取并同步到远端
+    # 启用稀疏检出（使用 --no-cone 模式以获得更精确的控制）
-        success, out, err = await run_git_command(
+    await run_git_command("config core.sparseCheckout true", target_dir)
-            f"fetch --force --depth 1 origin {branch}", temp_path
+    await run_git_command("sparse-checkout init --no-cone", target_dir)
-        )
-        if not success:
-            raise RuntimeError(f"fetch 失败: {err or out}")
 
-        # 使用远端强制更新本地分支并覆盖工作区
+    # 设置需要检出的路径（每次都覆盖配置）
-        success, out, err = await run_git_command(
+    if not sparse_path:
-            f"checkout -B {branch} origin/{branch}", temp_path
+        raise RuntimeError("sparse-checkout 路径不能为空")
-        )
-        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 模式，直接指定要检出的具体路径
-        await run_git_command(f"reset --hard origin/{branch}", temp_path)
+    # 例如：sparse_path="plugins/mahiro" -> 只检出 plugins/mahiro/ 下的内容
-        await run_git_command("clean -xdf", temp_path)
+    success, out, err = await run_git_command(
+        f"sparse-checkout set {sparse_path}/", target_dir
+    )
+    if not success:
+        raise RuntimeError(f"配置稀疏路径失败: {err or out}")
 
-        # 将检出的文件移动到目标位置
+    # 强制拉取并同步到远端
-        source_path = temp_path / sparse_path
+    success, out, err = await run_git_command(
-        if source_path.exists():
+        f"fetch --force --depth 1 origin {branch}", target_dir
-            # 确保目标路径存在
+    )
-            target_path = target_dir / sparse_path
+    if not success:
-            target_path.parent.mkdir(parents=True, exist_ok=True)
+        raise RuntimeError(f"fetch 失败: {err or out}")
 
-            # 如果目标路径已存在，先清理
+    # 使用远端强制更新本地分支并覆盖工作区
-            if target_path.exists():
+    success, out, err = await run_git_command(
-                if target_path.is_dir():
+        f"checkout -B {branch} origin/{branch}", target_dir
-                    shutil.rmtree(target_path)
+    )
-                else:
+    if not success:
-                    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)}")
 
-            # 移动整个目录结构到目标位置
+    # 强制对齐工作区
-            shutil.move(str(source_path), str(target_path))
+    await run_git_command(f"reset --hard origin/{branch}", target_dir)
+    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: