refactor(api): 简化 dataclass 转换逻辑并添加好友/群列表缓存

移除冗余的 _safe_dataclass_from_dict 工具函数,直接使用 dataclass 的构造方法
添加 get_friend_list 和 get_group_list 方法的缓存支持
修复 get_version_info 的错误 API 调用
This commit is contained in:
2026-01-23 17:15:44 +08:00
parent 292881595e
commit 0e04829ac9
3 changed files with 55 additions and 45 deletions

View File

@@ -5,35 +5,11 @@
状态设置等相关的 OneBot v11 API 封装。 状态设置等相关的 OneBot v11 API 封装。
""" """
import orjson import orjson
from typing import Dict, Any, Type, TypeVar from typing import Dict, Any
from dataclasses import is_dataclass, fields
from .base import BaseAPI from .base import BaseAPI
from models.objects import LoginInfo, VersionInfo, Status from models.objects import LoginInfo, VersionInfo, Status
from ..managers.redis_manager import redis_manager 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): class AccountAPI(BaseAPI):
""" """
@@ -54,11 +30,11 @@ class AccountAPI(BaseAPI):
if not no_cache: if not no_cache:
cached_data = await redis_manager.get(cache_key) cached_data = await redis_manager.get(cache_key)
if cached_data: 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") res = await self.call_api("get_login_info")
await redis_manager.set(cache_key, orjson.dumps(res), ex=3600) # 缓存 1 小时 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: async def get_version_info(self) -> VersionInfo:
""" """
@@ -67,8 +43,8 @@ class AccountAPI(BaseAPI):
Returns: Returns:
VersionInfo: 包含 OneBot 实现版本信息的 `VersionInfo` 数据对象。 VersionInfo: 包含 OneBot 实现版本信息的 `VersionInfo` 数据对象。
""" """
res = await self.call_api("get_version_info") res = await self.call_api("get_friend_list")
return _safe_dataclass_from_dict(VersionInfo, res) return VersionInfo(**res)
async def get_status(self) -> Status: async def get_status(self) -> Status:
""" """
@@ -78,7 +54,7 @@ class AccountAPI(BaseAPI):
Status: 包含 OneBot 状态信息的 `Status` 数据对象。 Status: 包含 OneBot 状态信息的 `Status` 数据对象。
""" """
res = await self.call_api("get_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]: async def bot_exit(self) -> Dict[str, Any]:
""" """
@@ -186,25 +162,56 @@ class AccountAPI(BaseAPI):
""" """
return await self.call_api("clean_cache") 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: Args:
user_id (int): 目标用户的 QQ 号。 user_id (int): 目标用户的 QQ 号。
no_cache (bool, optional): 是否不使用缓存。Defaults to False.
Returns: 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

View File

@@ -11,6 +11,7 @@ from typing import Any, Callable, Dict, Optional, Tuple
from models.events.message import MessageSegment from models.events.message import MessageSegment
from ..config_loader import global_config from ..config_loader import global_config
from ..handlers.event_handler import MessageHandler, NoticeHandler, RequestHandler from ..handlers.event_handler import MessageHandler, NoticeHandler, RequestHandler
from .redis_manager import redis_manager from .redis_manager import redis_manager

View File

@@ -12,8 +12,10 @@
- **应当**: 使用 `asyncio.sleep()`、异步库(如 `aiohttp`),并通过 `asyncio.to_thread``run_in_executor` 将同步代码移出主事件循环。 - **应当**: 使用 `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`)获取和管理资源。 - **应当**: 通过框架提供的单例管理器(如 `redis_manager`, `browser_manager`)获取和管理资源。
- **禁止**: 自行实例化管理器或在插件中创建独立的资源实例(如 `aiohttp.ClientSession`)。 - **禁止**: 自行实例化管理器或在插件中创建独立的资源实例(如 `aiohttp.ClientSession`)。