refactor(core): 优化权限管理和事件模型

- 重构 AdminManager 和 PermissionManager 以 Redis 为主要数据源
- 为所有事件模型添加 slots=True 提升性能
- 更新文档说明 Mypyc 编译注意事项
- 清理测试和调试文件
- 移动静态资源到 web_static 目录
This commit is contained in:
2026-01-13 08:35:54 +08:00
parent 7f331970dd
commit 3cbf5328bb
25 changed files with 434 additions and 409 deletions

View File

@@ -10,13 +10,14 @@ Bot 核心抽象模块
- 提供高级消息发送功能,如 `send_forwarded_messages`。
- 整合所有细分的 API 调用(消息、群组、好友等)。
"""
from typing import TYPE_CHECKING, Dict, Any, List, Union
from typing import TYPE_CHECKING, Dict, Any, List, Union, Optional
from models.events.base import OneBotEvent
from models.message import MessageSegment
from models.objects import GroupInfo, StrangerInfo
if TYPE_CHECKING:
from .ws import WS
from .utils.executor import CodeExecutor
from .api import MessageAPI, GroupAPI, FriendAPI, AccountAPI, MediaAPI
@@ -37,7 +38,7 @@ class Bot(MessageAPI, GroupAPI, FriendAPI, AccountAPI, MediaAPI):
ws_client (WS): WebSocket 客户端实例,负责底层的 API 请求和响应处理。
"""
super().__init__(ws_client, ws_client.self_id or 0)
self.code_executor = None
self.code_executor: Optional["CodeExecutor"] = None
async def get_group_list(self, no_cache: bool = False) -> List[GroupInfo]:
# GroupAPI.get_group_list 不支持 no_cache 参数,这里忽略它

View File

@@ -2,8 +2,7 @@
管理员管理器模块
该模块负责管理机器人的管理员列表。
实现了文件和 Redis 缓存之间的数据同步,并提供了一套清晰的 API
供其他模块调用。
现在以 Redis 作为主要数据源,文件仅用作备份。
"""
import json
import os
@@ -18,10 +17,11 @@ class AdminManager(Singleton):
"""
管理员管理器类
负责加载、缓存和管理管理员列表
使用单例模式,确保全局只有一个实例
以 Redis Set 作为管理员列表的唯一真实来源,提供高速的读写能力
文件 (admin.json) 仅用于首次启动时的数据迁移和作为灾备
"""
_REDIS_KEY = "neobot:admins" # 用于存储管理员集合的 Redis 键
def __init__(self):
"""
初始化 AdminManager
@@ -29,7 +29,7 @@ class AdminManager(Singleton):
if hasattr(self, '_initialized') and self._initialized:
return
# 管理员数据文件路径
# 管理员数据文件路径,主要用于备份和首次迁移
self.data_file = os.path.join(
os.path.dirname(os.path.abspath(__file__)),
"..",
@@ -37,124 +37,113 @@ class AdminManager(Singleton):
"admin.json"
)
self._admins: Set[int] = set()
# 确保数据目录存在
os.makedirs(os.path.dirname(self.data_file), exist_ok=True)
logger.info("管理员管理器初始化完成")
super().__init__()
async def initialize(self):
"""
异步初始化,加载数据并同步到 Redis
异步初始化,检查 Redis 数据,如果为空则尝试从文件迁移
"""
await self._load_from_file()
await self._sync_to_redis()
logger.info("管理员数据加载并同步到 Redis 完成")
try:
# 检查 Redis 中是否已存在数据
if await redis_manager.redis.exists(self._REDIS_KEY):
admin_count = await redis_manager.redis.scard(self._REDIS_KEY)
logger.info(f"Redis 中已存在管理员数据,共 {admin_count} 位。")
else:
# Redis 为空,尝试从文件迁移
logger.info("Redis 中未找到管理员数据,尝试从 admin.json 文件迁移...")
await self._migrate_from_file_to_redis()
except Exception as e:
logger.error(f"初始化管理员数据时发生错误: {e}")
async def _load_from_file(self):
async def _migrate_from_file_to_redis(self):
"""
从 admin.json 加载管理员列表
从 admin.json 加载管理员列表并存入 Redis
这通常只在首次启动或 Redis 数据丢失时执行一次
"""
admins_to_migrate = set()
try:
if os.path.exists(self.data_file):
with open(self.data_file, "r", encoding="utf-8") as f:
data = json.load(f)
admins = data.get("admins", [])
self._admins = set(int(admin_id) for admin_id in admins)
logger.debug(f"{self.data_file} 加载了 {len(self._admins)} 位管理员")
admins_to_migrate = set(int(admin_id) for admin_id in admins)
if admins_to_migrate:
await redis_manager.redis.sadd(self._REDIS_KEY, *admins_to_migrate)
logger.success(f"成功从文件迁移 {len(admins_to_migrate)} 位管理员到 Redis。")
else:
# 如果文件不存在,创建一个空的
self._admins = set()
await self._save_to_file()
except (json.JSONDecodeError, ValueError) as e:
logger.error(f"加载或解析 admin.json 失败: {e}")
self._admins = set()
logger.info("admin.json 文件为空或不存在,无需迁移。")
async def _save_to_file(self):
except (json.JSONDecodeError, ValueError) as e:
logger.error(f"解析 admin.json 失败,无法迁移: {e}")
except Exception as e:
logger.error(f"迁移管理员数据到 Redis 失败: {e}")
async def _save_to_file_backup(self):
"""
当前管理员列表保存回 admin.json
Redis 中的管理员列表备份到 admin.json
"""
try:
# 确保目录存在
os.makedirs(os.path.dirname(self.data_file), exist_ok=True)
# 将 set 转换为 list 以便 JSON 序列化
admin_list = [str(admin_id) for admin_id in self._admins]
admins = await self.get_all_admins()
admin_list = [str(admin_id) for admin_id in admins]
with open(self.data_file, "w", encoding="utf-8") as f:
json.dump({"admins": admin_list}, f, indent=2, ensure_ascii=False)
logger.debug(f"管理员列表已保存{self.data_file}")
logger.debug(f"管理员列表已备份{self.data_file}")
except Exception as e:
logger.error(f"保存 admin.json 失败: {e}")
async def _sync_to_redis(self):
"""
将内存中的管理员集合同步到 Redis
"""
from core.managers.redis_manager import redis_manager
try:
# 首先清空旧的集合
await redis_manager.redis.delete(self._REDIS_KEY)
if self._admins:
# 将所有管理员ID添加到集合中
await redis_manager.redis.sadd(self._REDIS_KEY, *self._admins)
logger.debug(f"已将 {len(self._admins)} 位管理员同步到 Redis")
except Exception as e:
logger.error(f"同步管理员到 Redis 失败: {e}")
logger.error(f"备份管理员列表到 admin.json 失败: {e}")
async def is_admin(self, user_id: int) -> bool:
"""
检查用户是否为管理员(从 Redis 缓存读取)
检查用户是否为管理员(直接从 Redis 读取)
"""
try:
return await redis_manager.redis.sismember(self._REDIS_KEY, user_id)
except Exception as e:
logger.error(f"从 Redis 检查管理员权限失败: {e}")
# Redis 失败时,回退到内存检查
return user_id in self._admins
return False
async def add_admin(self, user_id: int) -> bool:
"""
添加管理员,并同步到文件和 Redis
添加管理员到 Redis并更新文件备份
"""
from .redis_manager import redis_manager
if user_id in self._admins:
return False # 用户已经是管理员
self._admins.add(user_id)
await self._save_to_file()
try:
await redis_manager.redis.sadd(self._REDIS_KEY, user_id)
logger.info(f"已添加新管理员 {user_id} 并更新缓存")
return True
# sadd 返回成功添加的成员数量1 表示成功0 表示已存在
if await redis_manager.redis.sadd(self._REDIS_KEY, user_id) == 1:
logger.info(f"已添加新管理员 {user_id} 到 Redis")
await self._save_to_file_backup() # 更新备份
return True
return False # 用户已经是管理员
except Exception as e:
logger.error(f"添加管理员 {user_id} 到 Redis 失败: {e}")
return False
async def remove_admin(self, user_id: int) -> bool:
"""
移除管理员,并同步到文件和 Redis
从 Redis 移除管理员,并更新文件备份
"""
from .redis_manager import redis_manager
if user_id not in self._admins:
return False # 用户不是管理员
self._admins.remove(user_id)
await self._save_to_file()
try:
await redis_manager.redis.srem(self._REDIS_KEY, user_id)
logger.info(f"已移除管理员 {user_id} 并更新缓存")
return True
# srem 返回成功移除的成员数量1 表示成功0 表示不存在
if await redis_manager.redis.srem(self._REDIS_KEY, user_id) == 1:
logger.info(f"已从 Redis 移除管理员 {user_id}")
await self._save_to_file_backup() # 更新备份
return True
return False # 用户不是管理员
except Exception as e:
logger.error(f"从 Redis 移除管理员 {user_id} 失败: {e}")
return False
async def get_all_admins(self) -> Set[int]:
"""
获取所有管理员的集合
从 Redis 获取所有管理员的集合
"""
return self._admins.copy()
try:
admins = await redis_manager.redis.smembers(self._REDIS_KEY)
return {int(admin_id) for admin_id in admins}
except Exception as e:
logger.error(f"从 Redis 获取所有管理员失败: {e}")
return set()
# 全局 AdminManager 实例

View File

@@ -2,14 +2,7 @@
权限管理器模块
该模块负责管理用户权限,支持 admin、op、user 三个权限级别。
权限数据存储在 `permissions.json` 文件中,格式为:
{
"users": {
"123456": "admin",
"789012": "op",
"345678": "user"
}
}
以 Redis Hash 作为主要数据源,文件仅用作备份和首次数据迁移。
"""
import json
import os
@@ -18,6 +11,7 @@ from typing import Dict
from ..utils.logger import logger
from ..utils.singleton import Singleton
from .admin_manager import admin_manager
from .redis_manager import redis_manager
from ..permission import Permission
@@ -31,176 +25,167 @@ class PermissionManager(Singleton):
"""
权限管理器类
负责加载、保存和查询用户权限数据
使用单例模式,确保全局只有一个权限管理器实例
以 Redis Hash 作为权限数据的唯一真实来源,提供高速的读写能力
文件 (permissions.json) 仅用于首次启动时的数据迁移和作为灾备
"""
_REDIS_KEY = "neobot:permissions" # 用于存储用户权限的 Redis Hash 键
def __init__(self):
"""
初始化权限管理器
如果已经初始化过,则直接返回。
"""
if hasattr(self, '_initialized') and self._initialized:
return
# 权限数据文件路径
# 权限数据文件路径,主要用于备份和首次迁移
self.data_file = os.path.join(
os.path.dirname(os.path.abspath(__file__)),
"..",
"data",
"permissions.json"
)
# 确保数据目录存在
data_dir = os.path.dirname(self.data_file)
os.makedirs(data_dir, exist_ok=True)
# 权限数据存储结构:{"users": {"user_id": "level_name"}}
self._data: Dict[str, Dict[str, str]] = {"users": {}}
# 加载现有数据
self.load()
os.makedirs(os.path.dirname(self.data_file), exist_ok=True)
logger.info("权限管理器初始化完成")
super().__init__()
def load(self) -> None:
async def initialize(self):
"""
从文件加载权限数据
异步初始化,检查 Redis 数据,如果为空则尝试从文件迁移
"""
try:
if not await redis_manager.redis.exists(self._REDIS_KEY):
logger.info("Redis 中未找到权限数据,尝试从 permissions.json 文件迁移...")
await self._migrate_from_file_to_redis()
else:
perm_count = await redis_manager.redis.hlen(self._REDIS_KEY)
logger.info(f"Redis 中已存在权限数据,共 {perm_count} 条。")
except Exception as e:
logger.error(f"初始化权限数据时发生错误: {e}")
如果文件不存在,则创建空文件并初始化默认数据结构。
async def _migrate_from_file_to_redis(self):
"""
从 permissions.json 加载权限数据并存入 Redis Hash
"""
perms_to_migrate = {}
try:
if os.path.exists(self.data_file):
with open(self.data_file, "r", encoding="utf-8") as f:
data = json.load(f)
# 兼容旧格式
if "users" in data:
self._data["users"] = data["users"]
else:
self._data["users"] = {}
logger.debug(f"权限数据已从 {self.data_file} 加载")
perms_to_migrate = data.get("users", {})
if perms_to_migrate:
# 使用 pipeline 批量写入,提高效率
async with redis_manager.redis.pipeline(transaction=True) as pipe:
for user_id, level_name in perms_to_migrate.items():
pipe.hset(self._REDIS_KEY, user_id, level_name)
await pipe.execute()
logger.success(f"成功从文件迁移 {len(perms_to_migrate)} 条权限数据到 Redis。")
else:
# 文件不存在,创建空文件
self.save()
logger.debug(f"创建空的权限数据文件: {self.data_file}")
except json.JSONDecodeError as e:
logger.error(f"权限数据文件格式错误: {e}")
# 文件损坏,重置为空数据
self._data["users"] = {}
self.save()
except Exception as e:
logger.error(f"加载权限数据失败: {e}")
self._data["users"] = {}
logger.info("permissions.json 文件为空或不存在,无需迁移。")
def save(self) -> None:
except (json.JSONDecodeError, ValueError) as e:
logger.error(f"解析 permissions.json 失败,无法迁移: {e}")
except Exception as e:
logger.error(f"迁移权限数据到 Redis 失败: {e}")
async def _save_to_file_backup(self):
"""
权限数据保存到文件
Redis 中的权限数据完整备份到 permissions.json
"""
try:
all_perms = await redis_manager.redis.hgetall(self._REDIS_KEY)
# Redis 返回的是 bytes需要解码
users_data = {k.decode('utf-8'): v.decode('utf-8') for k, v in all_perms.items()}
with open(self.data_file, "w", encoding="utf-8") as f:
json.dump(self._data, f, indent=2, ensure_ascii=False)
logger.debug(f"权限数据已保存{self.data_file}")
json.dump({"users": users_data}, f, indent=2, ensure_ascii=False)
logger.debug(f"权限数据已备份{self.data_file}")
except Exception as e:
logger.error(f"保存权限数据失败: {e}")
logger.error(f"备份权限数据到 permissions.json 失败: {e}")
async def get_user_permission(self, user_id: int) -> Permission:
"""
获取指定用户的权限对象
Args:
user_id (int): 用户 QQ 号
Returns:
Permission: 用户的权限对象,如果用户不存在则返回默认级别 USER
优先检查是否为机器人管理员,然后从 Redis 查询。
"""
# 首先,通过 AdminManager 检查是否为管理员
if await admin_manager.is_admin(user_id):
return Permission.ADMIN
# 如果不是管理员,则从 permissions.json 中查找
user_id_str = str(user_id)
level_name = self._data["users"].get(user_id_str, Permission.USER.value)
return _PERMISSIONS.get(level_name, Permission.USER)
try:
level_name_bytes = await redis_manager.redis.hget(self._REDIS_KEY, str(user_id))
if level_name_bytes:
level_name = level_name_bytes.decode('utf-8')
return _PERMISSIONS.get(level_name, Permission.USER)
except Exception as e:
logger.error(f"从 Redis 获取用户 {user_id} 权限失败: {e}")
return Permission.USER
def set_user_permission(self, user_id: int, permission: Permission) -> None:
async def set_user_permission(self, user_id: int, permission: Permission) -> None:
"""
设置指定用户的权限级别
Args:
user_id (int): 用户 QQ 号
permission (Permission): 权限对象
Raises:
ValueError: 如果权限对象无效
在 Redis 中设置指定用户的权限级别,并更新文件备份
"""
if not isinstance(permission, Permission):
raise ValueError(f"无效的权限对象: {permission}")
user_id_str = str(user_id)
self._data["users"][user_id_str] = permission.value
self.save()
logger.info(f"设置用户 {user_id} 的权限级别{permission.value}")
try:
await redis_manager.redis.hset(self._REDIS_KEY, str(user_id), permission.value)
await self._save_to_file_backup()
logger.info(f"已在 Redis 中设置用户 {user_id} 的权限为 {permission.value}")
except Exception as e:
logger.error(f"在 Redis 中设置用户 {user_id} 权限失败: {e}")
def remove_user(self, user_id: int) -> None:
async def remove_user(self, user_id: int) -> None:
"""
移除指定用户的权限设置,恢复为默认级别
Args:
user_id (int): 用户 QQ 号
从 Redis 中移除指定用户的权限设置,并更新文件备份
"""
user_id_str = str(user_id)
if user_id_str in self._data["users"]:
del self._data["users"][user_id_str]
self.save()
logger.info(f"移除用户 {user_id} 的权限设置")
try:
if await redis_manager.redis.hdel(self._REDIS_KEY, str(user_id)):
await self._save_to_file_backup()
logger.info(f"已从 Redis 中移除用户 {user_id} 的权限设置")
except Exception as e:
logger.error(f"从 Redis 移除用户 {user_id} 权限失败: {e}")
async def check_permission(self, user_id: int, required_permission: Permission) -> bool:
"""
检查用户是否具有指定权限级别
Args:
user_id (int): 用户 QQ 号
required_permission (Permission): 所需的权限对象
Returns:
bool: 如果用户权限 >= 所需权限,返回 True否则返回 False
"""
user_permission = await self.get_user_permission(user_id)
return user_permission >= required_permission
async def get_all_user_permissions(self) -> Dict[str, str]:
"""
获取所有已配置的用户权限(包括 AdminManager 中的管理员
:return: 一个包含所有用户权限的字典
获取所有已配置的用户权限(合并 Redis 和 AdminManager
"""
permissions = self._data["users"].copy()
# 合并 AdminManager 中的管理员
admins = await admin_manager.get_all_admins()
for admin_id in admins:
permissions[str(admin_id)] = Permission.ADMIN.value
permissions = {}
try:
# 从 Redis 获取基础权限
all_perms = await redis_manager.redis.hgetall(self._REDIS_KEY)
permissions = {k.decode('utf-8'): v.decode('utf-8') for k, v in all_perms.items()}
except Exception as e:
logger.error(f"从 Redis 获取所有权限失败: {e}")
# 合并 AdminManager 中的管理员ADMIN 权限覆盖一切
try:
admins = await admin_manager.get_all_admins()
for admin_id in admins:
permissions[str(admin_id)] = Permission.ADMIN.value
except Exception as e:
logger.error(f"获取管理员列表以合并权限时失败: {e}")
return permissions
def get_all_users(self) -> Dict[str, str]:
async def clear_all(self) -> None:
"""
获取所有设置了权限的用户及其级别名称
Returns:
Dict[str, str]: 用户ID到权限级别名称的映射
清空 Redis 中的所有权限设置,并更新备份文件
"""
return self._data["users"].copy()
def clear_all(self) -> None:
"""
清空所有权限设置
"""
self._data["users"].clear()
self.save()
logger.info("已清空所有权限设置")
try:
await redis_manager.redis.delete(self._REDIS_KEY)
await self._save_to_file_backup()
logger.info("已清空 Redis 中的所有权限设置")
except Exception as e:
logger.error(f"清空 Redis 权限数据失败: {e}")
def require_admin(func):

View File

@@ -8,6 +8,7 @@ import os
import pkgutil
import sys
from typing import Set
from .command_manager import CommandManager
from ..utils.exceptions import SyncHandlerError
from ..utils.logger import logger
@@ -20,7 +21,7 @@ class PluginManager:
"""
插件管理器类
"""
def __init__(self, command_manager):
def __init__(self, command_manager: "CommandManager") -> None:
"""
初始化插件管理器
@@ -29,7 +30,7 @@ class PluginManager:
self.command_manager = command_manager
self.loaded_plugins: Set[str] = set()
def load_all_plugins(self):
def load_all_plugins(self) -> None:
"""
扫描并加载 `plugins` 目录下的所有插件。
"""
@@ -77,7 +78,7 @@ class PluginManager:
f" 加载插件 {module_name} 失败: {e}"
)
def reload_plugin(self, full_module_name: str):
def reload_plugin(self, full_module_name: str) -> None:
"""
精确重载单个插件。
"""

View File

@@ -1,6 +1,9 @@
"""
通用单例模式基类
"""
from typing import Any, Optional, Type, TypeVar
T = TypeVar('T')
class Singleton:
"""
@@ -10,18 +13,25 @@ class Singleton:
它通过重写 __new__ 方法来确保每个类只有一个实例。
同时,它处理了重复初始化的问题,确保 __init__ 方法只在第一次实例化时被调用。
"""
_instance = None
_initialized = False
_instance: Optional[Any] = None
_initialized: bool = False
def __new__(cls, *args, **kwargs):
def __new__(cls: Type[T], *args: Any, **kwargs: Any) -> T:
"""
创建或返回现有的实例
Args:
*args: 传递给构造函数的位置参数
**kwargs: 传递给构造函数的关键字参数
Returns:
T: 单例实例
"""
if cls._instance is None:
cls._instance = super().__new__(cls)
return cls._instance
def __init__(self):
def __init__(self) -> None:
"""
确保初始化逻辑只执行一次
"""

View File

@@ -13,16 +13,18 @@ WebSocket 连接。它是整个机器人框架的底层通信基础。
"""
import asyncio
import json
from typing import Any, Dict, Optional
from typing import Any, Dict, Optional, cast
import uuid
import websockets
from websockets.legacy.client import WebSocketClientProtocol
from models.events.factory import EventFactory
from .bot import Bot
from .config_loader import global_config
from .managers.command_manager import matcher
from .utils.executor import CodeExecutor
from .utils.logger import logger
@@ -31,7 +33,7 @@ class WS:
WebSocket 客户端,负责与 OneBot v11 实现进行底层通信。
"""
def __init__(self, code_executor=None):
def __init__(self, code_executor: Optional[CodeExecutor] = None) -> None:
"""
初始化 WebSocket 客户端。
@@ -43,13 +45,13 @@ class WS:
self.token = cfg.token
self.reconnect_interval = cfg.reconnect_interval
self.ws = None
self._pending_requests = {}
self.ws: Optional[WebSocketClientProtocol] = None
self._pending_requests: Dict[str, asyncio.Future] = {}
self.bot: Bot | None = None
self.self_id: int | None = None
self.code_executor = code_executor
async def connect(self):
async def connect(self) -> None:
"""
启动并管理 WebSocket 连接。
@@ -63,7 +65,8 @@ class WS:
logger.info(f"正在尝试连接至 NapCat: {self.url}")
async with websockets.connect(
self.url, additional_headers=headers
) as websocket:
) as websocket_raw:
websocket = cast(WebSocketClientProtocol, websocket_raw)
self.ws = websocket
logger.success("连接成功!")
await self._listen_loop(websocket)
@@ -79,7 +82,7 @@ class WS:
logger.info(f"{self.reconnect_interval}秒后尝试重连...")
await asyncio.sleep(self.reconnect_interval)
async def _listen_loop(self, websocket_connection):
async def _listen_loop(self, websocket_connection: WebSocketClientProtocol) -> None:
"""
核心监听循环,处理所有接收到的 WebSocket 消息。
@@ -111,7 +114,7 @@ class WS:
except Exception as e:
logger.exception(f"解析消息异常: {e}")
async def on_event(self, event_data: dict):
async def on_event(self, event_data: Dict[str, Any]) -> None:
"""
事件处理和分发层。