K2CrO4/NeoBot
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

refactor(managers): 重构单例管理器实现并优化代码结构

Browse Source 
feat(ws_pool): 新增 WebSocket 连接池实现

perf(json): 使用 orjson 替代标准 json 库提升性能

style: 清理未使用的导入和冗余代码

docs: 更新架构文档和开发规范

test: 添加 WebSocket 连接池测试用例

fix(plugins): 修复自动审批插件 API 调用参数格式
This commit is contained in:
K2cr2O1
2026-01-22 16:23:03 +08:00
parent d7d732ff4d
commit caf5b06097
42 changed files with 1285 additions and 261 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
core/managers/admin_manager.py
View File

@@ -4,7 +4,7 @@
该模块负责管理机器人的管理员列表。
它现在以 Redis 作为主要数据源，文件仅用作备份。
"""
import json
import orjson
import os
from typing import Set
@@ -66,7 +66,7 @@ class AdminManager(Singleton):
try:
if os.path.exists(self.data_file):
with open(self.data_file, "r", encoding="utf-8") as f:
data = json.load(f)
data = orjson.loads(f.read())
admins = data.get("admins", [])
admins_to_migrate = set(int(admin_id) for admin_id in admins)
@@ -76,7 +76,7 @@ class AdminManager(Singleton):
else:
logger.info("admin.json 文件为空或不存在，无需迁移。")
except (json.JSONDecodeError, ValueError) as e:
except ValueError as e:
logger.error(f"解析 admin.json 失败，无法迁移: {e}")
except Exception as e:
logger.error(f"迁移管理员数据到 Redis 失败: {e}")
@@ -89,7 +89,7 @@ class AdminManager(Singleton):
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)
f.write(orjson.dumps({"admins": admin_list}, indent=2, ensure_ascii=False).decode('utf-8'))
logger.debug(f"管理员列表已备份到 {self.data_file}")
except Exception as e:
logger.error(f"备份管理员列表到 admin.json 失败: {e}")

14
core/managers/browser_manager.py
View File

@@ -7,21 +7,23 @@ import asyncio
from typing import Optional
from playwright.async_api import async_playwright, Browser, Playwright, Page
from ..utils.logger import logger
from ..utils.singleton import Singleton
class BrowserManager:
class BrowserManager(Singleton):
"""
浏览器管理器（异步单例）
"""
_instance = None
_playwright: Optional[Playwright] = None
_browser: Optional[Browser] = None
_page_pool: Optional[asyncio.Queue] = None
_pool_size: int = 3
def __new__(cls):
if cls._instance is None:
cls._instance = super().__new__(cls)
return cls._instance
def __init__(self):
"""
初始化浏览器管理器
"""
# 调用父类 __init__ 确保单例初始化
super().__init__()
async def initialize(self):
"""

3
core/managers/command_manager.py
View File

@@ -7,12 +7,9 @@
"""
from typing import Any, Callable, Dict, Optional, Tuple
import os
import base64
from models.events.message import MessageSegment
from models.events.message import MessageSegment
from ..config_loader import global_config
from ..handlers.event_handler import MessageHandler, NoticeHandler, RequestHandler

16
core/managers/image_manager.py
View File

@@ -10,19 +10,21 @@ from jinja2 import Template
from .browser_manager import browser_manager
from ..utils.logger import logger
from ..utils.singleton import Singleton
class ImageManager:
class ImageManager(Singleton):
"""
图片生成管理器（单例）
"""
_instance = None
def __new__(cls):
if cls._instance is None:
cls._instance = super().__new__(cls)
return cls._instance
def __init__(self):
"""
初始化图片生成管理器
"""
# 检查是否已经初始化
if hasattr(self, 'template_dir'):
return
# 模板目录
self.template_dir = os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(__file__))), "templates")
# 临时文件目录

8
core/managers/permission_manager.py
View File

@@ -4,7 +4,7 @@
该模块负责管理用户权限，支持 admin、op、user 三个权限级别。
以 Redis Hash 作为主要数据源，文件仅用作备份和首次数据迁移。
"""
import json
import orjson
import os
from typing import Dict
@@ -71,7 +71,7 @@ class PermissionManager(Singleton):
try:
if os.path.exists(self.data_file):
with open(self.data_file, "r", encoding="utf-8") as f:
data = json.load(f)
data = orjson.loads(f.read())
perms_to_migrate = data.get("users", {})
if perms_to_migrate:
@@ -84,7 +84,7 @@ class PermissionManager(Singleton):
else:
logger.info("permissions.json 文件为空或不存在，无需迁移。")
except (json.JSONDecodeError, ValueError) as e:
except ValueError as e:
logger.error(f"解析 permissions.json 失败，无法迁移: {e}")
except Exception as e:
logger.error(f"迁移权限数据到 Redis 失败: {e}")
@@ -98,7 +98,7 @@ class PermissionManager(Singleton):
# 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({"users": users_data}, f, indent=2, ensure_ascii=False)
f.write(orjson.dumps({"users": users_data}, indent=2, ensure_ascii=False).decode('utf-8'))
logger.debug(f"权限数据已备份到 {self.data_file}")
except Exception as e:
logger.error(f"备份权限数据到 permissions.json 失败: {e}")

35
core/managers/plugin_manager.py
View File

@@ -10,28 +10,41 @@ import sys
from typing import Set
from .command_manager import CommandManager
from ..utils.exceptions import SyncHandlerError, PluginError, PluginLoadError, PluginReloadError, PluginNotFoundError
from ..utils.exceptions import SyncHandlerError, PluginLoadError, PluginReloadError, PluginNotFoundError
from ..utils.logger import logger, ModuleLogger
from ..utils.error_codes import ErrorCode, create_error_response
from ..utils.singleton import Singleton
# 确保logger在模块级别可见
__all__ = ['PluginManager', 'logger']
class PluginManager:
class PluginManager(Singleton):
"""
插件管理器类
"""
def __init__(self, command_manager: "CommandManager") -> None:
def __init__(self, command_manager: "CommandManager" | None = None) -> None:
"""
初始化插件管理器
:param command_manager: CommandManager的实例
"""
self.command_manager = command_manager
self.loaded_plugins: Set[str] = set()
# 创建模块专用日志记录器
self.logger = ModuleLogger("PluginManager")
# 检查是否已经初始化
if hasattr(self, '_command_manager'):
return
# 只有首次初始化时才执行
if command_manager:
self._command_manager = command_manager
self.loaded_plugins: Set[str] = set()
# 创建模块专用日志记录器
self.logger = ModuleLogger("PluginManager")
@property
def command_manager(self):
"""
获取命令管理器实例
"""
return self._command_manager
def load_all_plugins(self) -> None:
"""
@@ -99,12 +112,12 @@ class PluginManager:
self.logger.warning(f"尝试重载一个未被加载的插件: {full_module_name}，将按首次加载处理。")
if full_module_name not in sys.modules:
error = PluginNotFoundError(
reload_error = PluginNotFoundError(
plugin_name=full_module_name,
message="模块未在sys.modules中找到"
)
self.logger.error(f"重载失败: {error.message}")
self.logger.log_custom_exception(error)
self.logger.error(f"重载失败: {reload_error.message}")
self.logger.log_custom_exception(reload_error)
return
try:

14
core/managers/redis_manager.py
View File

@@ -1,18 +1,20 @@
import redis.asyncio as redis
from ..config_loader import global_config as config
from ..utils.logger import logger
from ..utils.singleton import Singleton
class RedisManager:
class RedisManager(Singleton):
"""
Redis 连接管理器（异步单例）
"""
_instance = None
_redis = None
def __new__(cls):
if cls._instance is None:
cls._instance = super().__new__(cls)
return cls._instance
def __init__(self):
"""
初始化 Redis 管理器
"""
# 调用父类 __init__ 确保单例初始化
super().__init__()
async def initialize(self):
"""