Merge 32c322b1d5 into 3cc882b116

- 将插件版本从0.1更新至0.2
- 新增安全获取ARM设备CPU频率的函数get_arm_cpu_freq_safe
- 优化CPU信息采集逻辑，提高在ARM架构下的兼容性

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/check/__init__.py

@@ -26,7 +26,7 @@ __plugin_meta__ = PluginMetadata(
     """.strip(),
     extra=PluginExtraData(
         author="HibiKier",
-        version="0.1",
+        version="0.2",
         plugin_type=PluginType.SUPERUSER,
         configs=[
             RegisterConfig(

zhenxun/builtin_plugins/check/data_source.py

@@ -1,3 +1,4 @@
+import contextlib
 from dataclasses import dataclass
 import os
 from pathlib import Path
@@ -18,7 +19,47 @@ BAIDU_URL = "https://www.baidu.com/"
 GOOGLE_URL = "https://www.google.com/"
 
 VERSION_FILE = Path() / "__version__"
-ARM_KEY = "aarch64"
+
+
+def get_arm_cpu_freq_safe():
+    """获取ARM设备CPU频率"""
+    # 方法1: 优先从系统频率文件读取
+    freq_files = [
+        "/sys/devices/system/cpu/cpu0/cpufreq/cpuinfo_max_freq",
+        "/sys/devices/system/cpu/cpu0/cpufreq/scaling_max_freq",
+        "/sys/devices/system/cpu/cpu0/cpufreq/cpuinfo_cur_freq",
+        "/sys/devices/system/cpu/cpu0/cpufreq/scaling_cur_freq",
+    ]
+
+    for freq_file in freq_files:
+        try:
+            with open(freq_file) as f:
+                frequency = int(f.read().strip())
+                return round(frequency / 1000000, 2)  # 转换为GHz
+        except (OSError, ValueError):
+            continue
+
+    # 方法2: 解析/proc/cpuinfo
+    with contextlib.suppress(OSError, FileNotFoundError, ValueError, PermissionError):
+        with open("/proc/cpuinfo") as f:
+            for line in f:
+                if "CPU MHz" in line:
+                    freq = float(line.split(":")[1].strip())
+                    return round(freq / 1000, 2)  # 转换为GHz
+    # 方法3: 使用lscpu命令
+    with contextlib.suppress(OSError, subprocess.SubprocessError, ValueError):
+        env = os.environ.copy()
+        env["LC_ALL"] = "C"
+        result = subprocess.run(
+            ["lscpu"], capture_output=True, text=True, env=env, timeout=10
+        )
+
+        if result.returncode == 0:
+            for line in result.stdout.split("\n"):
+                if "CPU max MHz" in line or "CPU MHz" in line:
+                    freq = float(line.split(":")[1].strip())
+                    return round(freq / 1000, 2)  # 转换为GHz
+    return 0  # 如果所有方法都失败，返回0
 
 
 @dataclass
@@ -37,7 +78,7 @@ class CPUInfo:
         if _cpu_freq := psutil.cpu_freq():
             cpu_freq = round(_cpu_freq.current / 1000, 2)
         else:
-            cpu_freq = 0
+            cpu_freq = get_arm_cpu_freq_safe()
         return CPUInfo(core=cpu_core, usage=cpu_usage, freq=cpu_freq)
 
 
@@ -160,44 +201,13 @@ def __get_version() -> str | None:
     return None
 
 
-def __get_arm_cpu():
-    env = os.environ.copy()
-    env["LC_ALL"] = "en_US.UTF-8"
-    cpu_info = subprocess.check_output(["lscpu"], env=env).decode()
-    model_name = ""
-    cpu_freq = 0
-    for line in cpu_info.splitlines():
-        if "Model name" in line:
-            model_name = line.split(":")[1].strip()
-        if "CPU MHz" in line:
-            cpu_freq = float(line.split(":")[1].strip())
-    return model_name, cpu_freq
-
-
-def __get_arm_oracle_cpu_freq():
-    cpu_freq = subprocess.check_output(
-        ["dmidecode", "-s", "processor-frequency"]
-    ).decode()
-    return round(float(cpu_freq.split()[0]) / 1000, 2)
-
-
 async def get_status_info() -> dict:
     """获取信息"""
     data = await __build_status()
 
     system = platform.uname()
-    if system.machine == ARM_KEY and not (
+    data = data.get_system_info()
-        cpuinfo.get_cpu_info().get("brand_raw") and data.cpu.freq
+    data["brand_raw"] = cpuinfo.get_cpu_info().get("brand_raw", "Unknown")
-    ):
-        model_name, cpu_freq = __get_arm_cpu()
-        if not data.cpu.freq:
-            data.cpu.freq = cpu_freq or __get_arm_oracle_cpu_freq()
-        data = data.get_system_info()
-        data["brand_raw"] = model_name
-    else:
-        data = data.get_system_info()
-        data["brand_raw"] = cpuinfo.get_cpu_info().get("brand_raw", "Unknown")
-
     baidu, google = await __get_network_info()
     data["baidu"] = "#8CC265" if baidu else "red"
     data["google"] = "#8CC265" if google else "red"

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

@@ -287,7 +287,7 @@ class StoreManager:
             files = [RepoFileInfo(path=f"{replace_module_path}.py", is_dir=False)]
         if not is_external:
             target_dir = BASE_PATH
-        elif is_dir and module_path == ".":
+        elif is_dir:
             target_dir = BASE_PATH / "plugins" / plugin_name
         else:
             target_dir = BASE_PATH / "plugins"

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/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/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"