From 0e04829ac9590e489c1befc8a6658341252b72b0 Mon Sep 17 00:00:00 2001 From: K2cr2O1 <2221577113@qq.com> Date: Fri, 23 Jan 2026 17:15:44 +0800 Subject: [PATCH] =?UTF-8?q?refactor(api):=20=E7=AE=80=E5=8C=96=20dataclass?= =?UTF-8?q?=20=E8=BD=AC=E6=8D=A2=E9=80=BB=E8=BE=91=E5=B9=B6=E6=B7=BB?= =?UTF-8?q?=E5=8A=A0=E5=A5=BD=E5=8F=8B/=E7=BE=A4=E5=88=97=E8=A1=A8?= =?UTF-8?q?=E7=BC=93=E5=AD=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 移除冗余的 _safe_dataclass_from_dict 工具函数，直接使用 dataclass 的构造方法 添加 get_friend_list 和 get_group_list 方法的缓存支持 修复 get_version_info 的错误 API 调用 --- core/api/account.py | 93 +++++++++++++++++--------------- core/managers/command_manager.py | 1 + docs/development-standards.md | 6 ++- 3 files changed, 55 insertions(+), 45 deletions(-) diff --git a/core/api/account.py b/core/api/account.py index 14d355f..3d895c9 100644 --- a/core/api/account.py +++ b/core/api/account.py @@ -5,35 +5,11 @@ 状态设置等相关的 OneBot v11 API 封装。 """ import orjson -from typing import Dict, Any, Type, TypeVar -from dataclasses import is_dataclass, fields +from typing import Dict, Any from .base import BaseAPI from models.objects import LoginInfo, VersionInfo, Status from ..managers.redis_manager import redis_manager -T = TypeVar('T') - -def _safe_dataclass_from_dict(cls: Type[T], data: Dict[str, Any]) -> T: - """ - 安全地从字典创建 dataclass 实例，忽略多余的键。 - """ - if not data: - try: - return cls() - except TypeError: - raise ValueError(f"无法在没有数据的情况下创建 {cls.__name__} 的实例") - - # 使用官方的 is_dataclass 进行检查，对 MyPyC 更友好 - if not is_dataclass(cls): - raise TypeError(f"{cls.__name__} 不是一个 dataclass") - - # 获取 dataclass 的所有字段名 - known_fields = {f.name for f in fields(cls)} - - # 过滤出 dataclass 认识的键值对 - filtered_data = {k: v for k, v in data.items() if k in known_fields} - - return cls(**filtered_data) class AccountAPI(BaseAPI): """ @@ -54,11 +30,11 @@ class AccountAPI(BaseAPI): if not no_cache: cached_data = await redis_manager.get(cache_key) if cached_data: - return _safe_dataclass_from_dict(LoginInfo, orjson.loads(cached_data)) + return LoginInfo(**orjson.loads(cached_data)) res = await self.call_api("get_login_info") await redis_manager.set(cache_key, orjson.dumps(res), ex=3600) # 缓存 1 小时 - return _safe_dataclass_from_dict(LoginInfo, res) + return LoginInfo(**res) async def get_version_info(self) -> VersionInfo: """ @@ -67,8 +43,8 @@ class AccountAPI(BaseAPI): Returns: VersionInfo: 包含 OneBot 实现版本信息的 `VersionInfo` 数据对象。 """ - res = await self.call_api("get_version_info") - return _safe_dataclass_from_dict(VersionInfo, res) + res = await self.call_api("get_friend_list") + return VersionInfo(**res) async def get_status(self) -> Status: """ @@ -78,7 +54,7 @@ class AccountAPI(BaseAPI): Status: 包含 OneBot 状态信息的 `Status` 数据对象。 """ res = await self.call_api("get_status") - return _safe_dataclass_from_dict(Status, res) + return Status(**res) async def bot_exit(self) -> Dict[str, Any]: """ @@ -186,25 +162,56 @@ class AccountAPI(BaseAPI): """ return await self.call_api("clean_cache") - async def get_profile_like(self) -> Dict[str, Any]: + async def get_stranger_info(self, user_id: int, no_cache: bool = False) -> Any: """ - 获取个人资料的点赞信息。 - - Returns: - Dict[str, Any]: OneBot API 的响应数据。 - """ - return await self.call_api("get_profile_like") - - async def nc_get_user_status(self, user_id: int) -> Dict[str, Any]: - """ - 获取用户的在线状态 (NapCat 特有 API)。 + 获取陌生人信息。 Args: user_id (int): 目标用户的 QQ 号。 + no_cache (bool, optional): 是否不使用缓存。Defaults to False. Returns: - Dict[str, Any]: OneBot API 的响应数据。 + Any: 包含陌生人信息的字典或对象。 """ - return await self.call_api("nc_get_user_status", {"user_id": user_id}) + return await self.call_api("get_stranger_info", {"user_id": user_id, "no_cache": no_cache}) + async def get_friend_list(self, no_cache: bool = False) -> list: + """ + 获取好友列表。 + + Args: + no_cache (bool, optional): 是否不使用缓存。Defaults to False. + + Returns: + list: 好友列表。 + """ + cache_key = f"neobot:cache:get_friend_list:{self.self_id}" + if not no_cache: + cached_data = await redis_manager.get(cache_key) + if cached_data: + return orjson.loads(cached_data) + + res = await self.call_api("get_friend_list") + await redis_manager.set(cache_key, orjson.dumps(res), ex=3600) # 缓存 1 小时 + return res + + async def get_group_list(self, no_cache: bool = False) -> list: + """ + 获取群列表。 + + Args: + no_cache (bool, optional): 是否不使用缓存。Defaults to False. + + Returns: + list: 群列表。 + """ + cache_key = f"neobot:cache:get_group_list:{self.self_id}" + if not no_cache: + cached_data = await redis_manager.get(cache_key) + if cached_data: + return orjson.loads(cached_data) + + res = await self.call_api("get_group_list") + await redis_manager.set(cache_key, orjson.dumps(res), ex=3600) # 缓存 1 小时 + return res diff --git a/core/managers/command_manager.py b/core/managers/command_manager.py index ddfc846..95777ac 100644 --- a/core/managers/command_manager.py +++ b/core/managers/command_manager.py @@ -11,6 +11,7 @@ from typing import Any, Callable, Dict, Optional, Tuple from models.events.message import MessageSegment + from ..config_loader import global_config from ..handlers.event_handler import MessageHandler, NoticeHandler, RequestHandler from .redis_manager import redis_manager diff --git a/docs/development-standards.md b/docs/development-standards.md index 56a9578..b46748b 100644 --- a/docs/development-standards.md +++ b/docs/development-standards.md @@ -12,8 +12,10 @@ - **应当**: 使用 `asyncio.sleep()`、异步库（如 `aiohttp`），并通过 `asyncio.to_thread` 或 `run_in_executor` 将同步代码移出主事件循环。 - **禁止**: 直接在异步函数中使用任何可能阻塞的同步调用。 -### 2. 资源管理 -**复用优于重建**。频繁创建和销毁资源（如网络连接、浏览器页面）会严重影响性能。 +### 1.1 异步优先原则 +- **绝对不要阻塞事件循环**：NeoBot 采用多线程异步架构，任何同步阻塞操作都会导致整个机器人卡死。 + - **禁止**：`time.sleep()`、同步 `requests`、密集 CPU 计算 + - **必须**：使用 `await asyncio.sleep()`、异步 HTTP 客户端、线程池执行同步任务 - **应当**: 通过框架提供的单例管理器（如 `redis_manager`, `browser_manager`）获取和管理资源。 - **禁止**: 自行实例化管理器或在插件中创建独立的资源实例（如 `aiohttp.ClientSession`）。