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

Merge branch 'main' into dev

Browse Source
This commit is contained in:
镀铬酸钾
2026-01-23 17:16:58 +08:00
committed by GitHub
parent 0e04829ac9 11867820b3
commit 7302419ba7
9 changed files with 169 additions and 697 deletions
Download Patch File Download Diff File Expand all files Collapse all files

93
core/api/account.py
View File

@@ -5,11 +5,35 @@
状态设置等相关的 OneBot v11 API 封装。
"""
import orjson
from typing import Dict, Any
from typing import Dict, Any, Type, TypeVar
from dataclasses import is_dataclass, fields
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):
"""
@@ -30,11 +54,11 @@ class AccountAPI(BaseAPI):
if not no_cache:
cached_data = await redis_manager.get(cache_key)
if cached_data:
return LoginInfo(**orjson.loads(cached_data))
return _safe_dataclass_from_dict(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 LoginInfo(**res)
return _safe_dataclass_from_dict(LoginInfo, res)
async def get_version_info(self) -> VersionInfo:
"""
@@ -43,8 +67,8 @@ class AccountAPI(BaseAPI):
Returns:
VersionInfo: 包含 OneBot 实现版本信息的 `VersionInfo` 数据对象。
"""
res = await self.call_api("get_friend_list")
return VersionInfo(**res)
res = await self.call_api("get_version_info")
return _safe_dataclass_from_dict(VersionInfo, res)
async def get_status(self) -> Status:
"""
@@ -54,7 +78,7 @@ class AccountAPI(BaseAPI):
Status: 包含 OneBot 状态信息的 `Status` 数据对象。
"""
res = await self.call_api("get_status")
return Status(**res)
return _safe_dataclass_from_dict(Status, res)
async def bot_exit(self) -> Dict[str, Any]:
"""
@@ -162,56 +186,25 @@ class AccountAPI(BaseAPI):
"""
return await self.call_api("clean_cache")
async def get_stranger_info(self, user_id: int, no_cache: bool = False) -> Any:
async def get_profile_like(self) -> Dict[str, 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:
Any: 包含陌生人信息的字典或对象。
Dict[str, Any]: OneBot API 的响应数据。
"""
return await self.call_api("get_stranger_info", {"user_id": user_id, "no_cache": no_cache})
return await self.call_api("nc_get_user_status", {"user_id": user_id})
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

8
core/managers/image_manager.py
View File

@@ -122,7 +122,13 @@ class ImageManager(Singleton):
content = f.read()
mime_type = "image/jpeg" if image_type == "jpeg" else "image/png"
return f"data:{mime_type};base64," + base64.b64encode(content).decode("utf-8")
base64_str = base64.b64encode(content).decode("utf-8")
# 记录摘要日志，避免刷屏
log_message = f"Base64 图片已生成 (MIME: {mime_type}, Size: {len(base64_str)/1024:.2f} KB, Preview: {base64_str[:30]}...{base64_str[-30:]})"
logger.debug(log_message)
return f"data:{mime_type};base64," + base64_str
except Exception as e:
logger.error(f"读取图片文件失败: {e}")
return None

2
core/managers/permission_manager.py
View File

@@ -412,7 +412,7 @@ class PermissionManager(Singleton):
"""
try:
# 创建空的权限数据
empty_data = {"users": {}}
empty_data: Dict[str, Dict] = {"users": {}}
# 原子性写入文件
temp_file = self.data_file + ".tmp"

18
docs/core-concepts/performance.md
View File

@@ -119,4 +119,22 @@ python setup_mypyc.py
- **当前状态**：为了确保稳定性，`setup_mypyc.py` 脚本**默认不编译** `models/events/` 目录下的事件模型文件。这些文件虽然也被频繁使用，但它们的结构相对复杂，与 `Mypyc` 的兼容性问题仍在探索中。
- **未来展望**：我们会持续关注 `Mypyc` 的更新，当其对 `dataclass` 的支持得到改善后，会重新尝试将事件模型加入编译列表，以实现极致的性能。
## 7. 健壮的 WebSocket 连接池
### 痛点
在高并发或网络不稳定的情况下，单个 WebSocket 连接可能会因为各种原因（如超时、服务器重启、网络波动）而中断或变得不可靠。如果框架依赖于单一的、不稳定的连接，会导致 API 调用频繁失败，甚至整个机器人无响应。
### 解决方案
`NeoBot` 实现了一个健壮的 `WebSocket 连接池` (`core/ws_pool.py`)，它不仅管理多个连接，还具备智能的健康检查和恢复机制。
- **多连接管理**: 启动时会建立一个包含多个 WebSocket 连接的池，API 调用会被分发到这些连接上，实现负载均衡。
- **自动健康检查**: 连接池会定期对池中的每个连接进行健康检查（发送 `get_status` 心跳包）。如果一个连接在规定时间内没有响应，它会被标记为“不健康”。
- **故障转移与恢复**: 当一个 API 调用需要使用连接时，连接池会自动选择一个“健康”的连接。如果所有连接都不健康，它会尝试重新建立新的连接，直到成功为止。
- **无感切换**: 对于上层调用者（如插件开发者）来说，这一切都是透明的。你只需要正常调用 `bot.call_api()`，连接池会在底层处理好所有的连接问题。
### 收益
- **高可用性**: 即使部分连接失效，机器人依然可以通过健康的连接继续提供服务，大大减少了因网络问题导致的停机时间。
- **高并发性能**: 通过连接池，多个 API 请求可以并行地通过不同的连接发送，提高了在高并发场景下的吞吐量。
- **自动恢复**: 无需手动重启机器人，连接池能够自动从网络故障中恢复，增强了系统的稳定性和无人值守能力。
通过这种方式，我们在保证核心模块性能的同时，也维持了项目的稳定性和可维护性。

2
docs/getting-started.md
View File

@@ -44,6 +44,8 @@ source venv/bin/activate
pip install -r requirements.txt
```
这个文件里包含了所有需要的 Python 库，比如 `aiohttp` (HTTP 请求), `orjson` (JSON 解析), `jinja2` (模板渲染), `psutil` (系统监控) 等等。
### d. 安装 Playwright 依赖
我们用 Playwright 来截图画画，它需要一个浏览器核心。

6
docs/plugin-development/index.md
View File

@@ -72,3 +72,9 @@ Bot 应该会回复你：“你好，[你的昵称]！”
* `args: str`: 如果命令有参数（比如 `/echo hello world`），`args` 就是 `hello world` 这部分字符串。
就这么简单，一个最基础的插件就写完了。
## 进阶阅读
- [指令处理](./command-handling.md): 了解如何处理参数、获取用户输入。
- [最佳实践](./best-practices.md): 学习如何编写更健壮、更高效的插件。
- [插件详解：/status 状态监控](./status-plugin.md): 深入了解内置的状态监控插件是如何实现的。

1
requirements.txt
View File

@@ -42,6 +42,7 @@ platformdirs==4.5.1
playwright==1.57.0
pluggy==1.6.0
propcache==0.4.1
psutil==5.9.8
pycparser==2.23
pydantic==2.12.5
pydantic_core==2.41.5

662
scripts/compile_machine_code.py
View File

@@ -1,40 +1,38 @@
#!/usr/bin/env python3
"""
优化版跨平台 Python 模块编译脚本
自动化跨平台 Python 模块编译脚本 (v2.2)
将核心 Python 模块编译为机器码（.pyd 或 .so）以提升性能。
此版本基于对项目结构的深入分析，包含了更多高频使用的模块。
将核心 Python 模块编译为机器码 (.pyd 或 .so) 以提升性能。
此版本实现了全自动清理和保守的编译范围，以确保稳定性。
支持的平台：
- Windows: 生成 .pyd 文件
- Linux: 生成 .so 文件
使用方法:
python compile_machine_code.py [options]
python scripts/compile_machine_code.py
选项:
--compile, -c 编译指定的模块（默认）
--list, -l 列出已编译的模块
--clean, -k 清理编译生成的文件
--help, -h 显示帮助信息
脚本会自动执行以下步骤:
1. 清理所有旧的编译文件 (.pyd, .so) 和 build 目录。
2. 编译模块列表中所有兼容的模块。
3. 列出所有成功编译的模块。
注意：
1. 需要安装 C 编译器 (Windows 上需要 Visual Studio Build Tools, Linux 上需要 GCC)
2. 需要安装 mypyc: pip install mypyc
3. 编译后的文件是平台相关的，不能跨平台复制
4. 建议在部署的目标环境上运行此脚本
5. Mypyc 不支持动态特性，如 eval/exec/getattr/setattr 等
3. 编译后的文件是平台相关的，不能跨平台复制。
"""
import os
import sys
import glob
import subprocess
import shutil
import argparse
# --- 配置区 ---
# 检测当前平台和 Python 版本
PLATFORM = sys.platform
PYTHON_VERSION = f"{sys.version_info.major}{sys.version_info.minor}" # 例如 "314"
PYTHON_VERSION = f"{sys.version_info.major}{sys.version_info.minor}"
if PLATFORM.startswith('win'):
EXTENSION = '.pyd'
@@ -48,637 +46,159 @@ else:
print(f"不支持的平台: {PLATFORM}")
sys.exit(1)
# 根据项目分析，优化要编译的模块列表
# 这些是项目中使用频率最高的模块，编译后能显著提升性能
# 经过测试的稳定编译模块列表
MODULES = [
# 工具模块 - 高频使用
'core/utils/json_utils.py', # JSON 处理 - 高频使用
'core/utils/executor.py', # 代码执行引擎 - 高频使用
'core/utils/exceptions.py', # 自定义异常 - 基础组件
'core/utils/performance.py', # 性能监控工具 - 重要组件
'core/utils/logger.py', # 日志模块 - 高频使用
'core/utils/singleton.py', # 单例模式 - 基础组件
# 工具模块
'core/utils/executor.py',
'core/utils/exceptions.py',
'core/utils/logger.py',
# 核心管理模块 - 高频使用
# 'core/managers/command_manager.py', # 指令匹配和分发 - 包含动态特性，不适合编译
# 'core/managers/admin_manager.py', # 管理员管理 - 包含动态特性，不适合编译
# 'core/managers/permission_manager.py', # 权限管理 - 包含动态特性，不适合编译
# 'core/managers/plugin_manager.py', # 插件管理器 - 包含动态特性，不适合编译
# 'core/managers/redis_manager.py', # Redis 管理器 - 包含动态特性，不适合编译
# 'core/managers/image_manager.py', # 图片管理器 - 包含动态特性，不适合编译
# 核心基础模块
'core/ws.py',
'core/config_loader.py',
# 核心基础模块 - 高频使用
'core/ws.py', # WebSocket 核心 - 核心通信，被10个文件引用
# 'core/bot.py', # Bot 核心抽象 - 使用多重继承，不适合编译
'core/config_loader.py', # 配置加载 - 启动必需，被7个文件引用
# 'core/config_models.py', # 配置模型 - 包含复杂类型定义，不适合编译
# 'core/permission.py', # 权限枚举 - 包含动态属性，不适合编译
# 数据模型 - 高频使用
'models/message.py', # 消息段模型 - 高频消息处理
'models/sender.py', # 发送者模型 - 高频消息处理
'models/objects.py', # API 响应数据模型 - 高频数据处理
# 事件处理相关 - 高频使用
'core/handlers/event_handler.py', # 事件处理器 - 核心事件处理
# 事件模型 - 高频使用，但包含dataclass，可能有编译问题，暂时排除
# 'models/events/message.py', # 消息事件 - 最高频事件类型
# 'models/events/notice.py', # 通知事件 - 高频事件类型
# 'models/events/request.py', # 请求事件 - 高频事件类型
# 'models/events/meta.py', # 元事件 - 高频事件类型
# 注意：以下文件不适合编译
# - 主程序文件（main.py）
# - 测试文件（tests/目录）
# - 插件文件（plugins/目录）
# - 编译（脚本compile_machine_code.py等）
# - 包含复杂动态特性的文件
# - API 基础类（由于多重继承问题）
# 数据模型 (仅包含安全的、无复杂元类的部分)
'models/message.py',
'models/sender.py',
'models/objects.py',
]
# --- 功能函数 ---
def list_compiled_modules():
"""列出已编译的模块"""
print(f"\n已编译的 {PLATFORM} 模块:")
print("=" * 50)
# 查找所有编译后的文件
compiled_files = []
for ext in [EXTENSION, f'__mypyc{EXTENSION}']:
compiled_files.extend(glob.glob(f'**/*{ext}', recursive=True))
# 过滤掉虚拟环境中的文件
compiled_files = glob.glob(f'**/*{EXTENSION}', recursive=True)
compiled_files = [f for f in compiled_files if 'venv' not in f and '.venv' not in f]
if compiled_files:
for f in sorted(compiled_files):
size = os.path.getsize(f) // 1024 # KB
print(f"{f} ({size} KB)")
try:
size = os.path.getsize(f) // 1024
print(f"{f} ({size} KB)")
except FileNotFoundError:
continue
else:
print(f"未找到已编译的 {EXTENSION} 文件")
print(f"\n总计: {len(compiled_files)} 个文件")
def clean_compiled_files():
"""清理编译生成的文件"""
print(f"\n清理编译生成的 {EXTENSION} 文件...")
"""清理所有编译生成的文件和目录"""
print("--- 步骤 1: 清理旧的编译文件 ---")
# 查找所有编译后的文件
compiled_files = []
for ext in [EXTENSION, f'__mypyc{EXTENSION}']:
compiled_files.extend(glob.glob(f'**/*{ext}', recursive=True))
# 过滤掉虚拟环境中的文件
compiled_files = glob.glob(f'**/*{EXTENSION}', recursive=True)
compiled_files = [f for f in compiled_files if 'venv' not in f and '.venv' not in f]
if compiled_files:
for f in sorted(compiled_files):
try:
os.remove(f)
print(f"已删除: {f}")
print(f" 已删除: {f}")
except Exception as e:
print(f"删除失败 {f}: {e}")
# 清理 build 目录
if os.path.exists('build'):
try:
shutil.rmtree('build')
print("已删除 build 目录")
except Exception as e:
print(f"删除 build 目录失败: {e}")
print(f" 删除失败 {f}: {e}")
else:
print(f"没有可清理的 {EXTENSION} 文件")
def get_platform_specific_module_name(module_path):
"""获取平台特定的模块文件名"""
module_name = module_path.replace('.py', '')
return f"{module_name}.{BUILD_PREFIX}{EXTENSION}"
print(" 没有可清理的文件")
if os.path.exists('build'):
try:
shutil.rmtree('build')
print(" 已删除 build 目录")
except Exception as e:
print(f" 删除 build 目录失败: {e}")
print("-" * 35)
def compile_module(module_path):
"""编译单个模块"""
print(f"\n编译: {module_path}")
try:
# 直接调用 mypyc 命令行工具
# 使用二进制模式捕获输出以避免编码问题
result = subprocess.run(
[sys.executable, '-m', 'mypyc', module_path],
capture_output=True,
check=True
check=True,
text=True,
encoding='utf-8',
errors='replace'
)
# 解码输出时处理可能的编码错误
try:
stdout_text = result.stdout.decode('utf-8', errors='replace')
stderr_text = result.stderr.decode('utf-8', errors='replace')
except AttributeError:
# 如果已经是字符串（Python 3.7+），则直接使用
stdout_text = result.stdout
stderr_text = result.stderr
# 智能查找编译产物
base_name = os.path.basename(module_path).replace('.py', '')
search_pattern = os.path.join(BUILD_PATH, '**', f'{base_name}.*{EXTENSION}')
# 获取平台特定的模块名
platform_module = get_platform_specific_module_name(module_path)
mypyc_platform_module = platform_module.replace(EXTENSION, f'__mypyc{EXTENSION}')
found_files = glob.glob(search_pattern, recursive=True)
# 检查编译产物是否在当前目录
if os.path.exists(platform_module):
print(f" ✓ 编译成功: {platform_module}")
if found_files:
build_module_path = found_files[0]
dest_path = os.path.join(os.path.dirname(module_path), os.path.basename(build_module_path))
os.makedirs(os.path.dirname(dest_path), exist_ok=True)
shutil.copy2(build_module_path, dest_path)
print(f" ✓ 编译成功: {dest_path}")
return True
else:
# 检查 build 目录中是否有编译产物
build_module_path = os.path.join(BUILD_PATH, platform_module)
build_mypyc_path = os.path.join(BUILD_PATH, mypyc_platform_module)
print(f" ✗ 编译失败：在 {BUILD_PATH} 中找不到编译产物")
if result.stdout: print(f" 输出: {result.stdout[:500]}...")
if result.stderr: print(f" 错误: {result.stderr[:500]}...")
return False
if os.path.exists(build_module_path):
# 如果在 build 目录中，复制到正确位置
os.makedirs(os.path.dirname(platform_module), exist_ok=True)
shutil.copy2(build_module_path, platform_module)
if os.path.exists(build_mypyc_path):
shutil.copy2(build_mypyc_path, mypyc_platform_module)
print(f" ✓ 编译成功（已从 build 目录复制）: {platform_module}")
return True
else:
print(" ✗ 编译失败：找不到编译产物")
if result.stdout:
print(f" 编译输出：{stdout_text[:500]}...")
if result.stderr:
print(f" 错误信息：{stderr_text[:500]}...")
return False
except subprocess.CalledProcessError as e:
print(f" ✗ 编译失败，退出码: {e.returncode}")
if hasattr(e, 'stdout') and e.stdout:
try:
stdout_text = e.stdout.decode('utf-8', errors='replace') if isinstance(e.stdout, bytes) else e.stdout
print(f" 编译输出：{stdout_text[:500]}...")
except Exception:
print(f" 编译输出：{str(e.stdout)[:500]}...")
if hasattr(e, 'stderr') and e.stderr:
try:
stderr_text = e.stderr.decode('utf-8', errors='replace') if isinstance(e.stderr, bytes) else e.stderr
print(f" 错误信息：{stderr_text[:500]}...")
except Exception:
print(f" 错误信息：{str(e.stderr)[:500]}...")
print(f" ✗ 编译失败 (Exit Code: {e.returncode})")
if e.stdout: print(f" 输出: {e.stdout[:500]}...")
if e.stderr: print(f" 错误: {e.stderr[:500]}...")
return False
except Exception as e:
print(f" ✗ 编译失败，意外错误: {e}")
import traceback
traceback.print_exc()
print(f" ✗ 编译时发生意外错误: {e}")
return False
def should_skip_module(module_path):
"""检查模块是否应该被跳过编译"""
try:
with open(module_path, 'r', encoding='utf-8') as f:
content = f.read()
# 检查是否包含抽象基类相关代码
if 'from abc import ABC' in content or 'from abc import abstractmethod' in content:
return True, "包含抽象基类，不适合编译"
# 检查是否包含危险的动态特性
# 注意：我们允许基本的动态特性，如getattr，但对于eval、exec等危险操作仍然阻止
if ('eval(' in content or 'exec(' in content or
'compile(' in content):
return True, "包含危险动态特性，不适合编译"
# 检查是否包含复杂的动态属性访问
if ('__dict__' in content or '__class__' in content or
'__module__' in content or '__bases__' in content):
return True, "包含复杂动态特性，不适合编译"
# 检查是否包含复杂的动态属性访问
if '.__dict__' in content or '.__class__' in content:
return True, "包含复杂动态特性，不适合编译"
return False, ""
except Exception as e:
return True, f"读取文件时出错: {e}"
def compile_all_modules():
"""编译所有指定的模块"""
print(f"\n开始编译 {len(MODULES)} 个模块 (平台: {PLATFORM})")
print("=" * 60)
print("\n--- 步骤 2: 开始编译模块 ---")
# 验证模块文件是否存在并检查是否适合编译
valid_modules = []
skipped_modules = []
for module_path in MODULES:
if os.path.exists(module_path):
should_skip, reason = should_skip_module(module_path)
if should_skip:
print(f"跳过: {module_path} ({reason})")
skipped_modules.append((module_path, reason))
else:
valid_modules.append(module_path)
for module in MODULES:
if os.path.exists(module):
valid_modules.append(module)
else:
print(f"警告: 模块 {module_path} 不存在，将被跳过")
print(f"警告: 模块 {module} 不存在，已跳过")
print(f"\n找到 {len(valid_modules)} 个有效模块进行编译。")
print(f"\n有效模块: {len(valid_modules)}, 跳过模块: {len(skipped_modules)}")
if not valid_modules:
print("错误: 没有有效的模块可编译")
return False
# 编译模块
success_count = 0
failed_modules = []
for module_path in valid_modules:
if compile_module(module_path):
for module in valid_modules:
if compile_module(module):
success_count += 1
else:
failed_modules.append(module_path)
failed_modules.append(module)
print("\n" + "=" * 60)
print("\n" + "=" * 50)
print(f"编译完成: {success_count}/{len(valid_modules)} 个模块成功")
if failed_modules:
print(f"失败模块: {failed_modules}")
if success_count == len(valid_modules):
print("✓ 所有模块编译成功")
return True
else:
print("✗ 部分模块编译失败")
return False
else:
print("✓ 所有模块编译成功")
print("=" * 50)
def main():
"""主函数"""
# 检查 Python 版本
"""主函数：执行清理、编译和列出结果的全过程"""
if not (sys.version_info.major == 3 and sys.version_info.minor >= 8):
print("警告: 推荐使用 Python 3.8+ 以获得最佳性能")
print(f"当前版本: {sys.version_info.major}.{sys.version_info.minor}.{sys.version_info.micro}")
print("继续编译可能导致兼容性问题")
print()
print("警告: 推荐使用 Python 3.8+ 以获得最佳性能。")
parser = argparse.ArgumentParser(description='优化版跨平台 Python 模块编译脚本')
group = parser.add_mutually_exclusive_group()
group.add_argument('--compile', '-c', action='store_true', default=True,
help='编译指定的模块 (默认)')
group.add_argument('--list', '-l', action='store_true',
help='列出已编译的模块')
group.add_argument('--clean', '-k', action='store_true',
help='清理编译生成的文件')
args = parser.parse_args()
# 检查是否安装了 mypyc
try:
import mypyc
except ImportError:
print("错误: 未安装 mypyc，请先安装: pip install mypyc")
print("错误: 未安装 mypyc，请运行: pip install mypyc")
sys.exit(1)
if args.list:
list_compiled_modules()
elif args.clean:
clean_compiled_files()
else:
compile_all_modules()
print("\n使用 --list 选项查看已编译的模块")
print("使用 --clean 选项清理编译文件")
clean_compiled_files()
compile_all_modules()
list_compiled_modules()
if __name__ == '__main__':
main()#!/usr/bin/env python3
"""
跨平台 Python 模块编译脚本
将核心 Python 模块编译为机器码（.pyd 或 .so）以提升性能。
支持的平台：
- Windows: 生成 .pyd 文件
- Linux: 生成 .so 文件
使用方法:
python compile_machine_code.py [options]
选项:
--compile, -c 编译指定的模块（默认）
--list, -l 列出已编译的模块
--clean, -k 清理编译生成的文件
--help, -h 显示帮助信息
注意：
1. 需要安装 C 编译器 (Windows 上需要 Visual Studio Build Tools, Linux 上需要 GCC)
2. 需要安装 mypyc: pip install mypyc
3. 编译后的文件是平台相关的，不能跨平台复制
4. 建议在部署的目标环境上运行此脚本
"""
import os
import sys
# 检测当前平台
PLATFORM = sys.platform
if PLATFORM.startswith('win'):
EXTENSION = '.pyd'
BUILD_PREFIX = 'cp314-win_amd64'
BUILD_PATH = os.path.join('build', 'lib.win-amd64-cpython-314')
elif PLATFORM.startswith('linux'):
EXTENSION = '.so'
BUILD_PREFIX = 'cp314-x86_64-linux-gnu'
BUILD_PATH = os.path.join('build', 'lib.linux-x86_64-cpython-314')
else:
print(f"不支持的平台: {PLATFORM}")
sys.exit(1)
# 要编译的模块列表
# 注意：Mypyc 对动态特性支持有限，只选择计算密集或类型明确的模块
MODULES = [
# 工具模块 - 高频使用
'core/utils/json_utils.py', # JSON 处理 - 高频使用
'core/utils/executor.py', # 代码执行引擎 - 高频使用
'core/utils/exceptions.py', # 自定义异常 - 基础组件
'core/utils/performance.py', # 性能监控工具 - 重要组件
'core/utils/logger.py', # 日志模块 - 高频使用
'core/utils/singleton.py', # 单例模式 - 基础组件
# 核心管理模块 - 高频使用
# 'core/managers/command_manager.py', # 指令匹配和分发 - 包含动态特性，不适合编译
# 'core/managers/admin_manager.py', # 管理员管理 - 包含动态特性，不适合编译
# 'core/managers/permission_manager.py', # 权限管理 - 包含动态特性，不适合编译
# 'core/managers/plugin_manager.py', # 插件管理器 - 包含动态特性，不适合编译
# 'core/managers/redis_manager.py', # Redis 管理器 - 包含动态特性，不适合编译
# 'core/managers/image_manager.py', # 图片管理器 - 包含动态特性，不适合编译
# 核心基础模块 - 高频使用
'core/ws.py', # WebSocket 核心 - 核心通信，被10个文件引用
# 'core/bot.py', # Bot 核心抽象 - 使用多重继承，不适合编译
'core/config_loader.py', # 配置加载 - 启动必需，被7个文件引用
# 'core/config_models.py', # 配置模型 - 包含复杂类型定义，不适合编译
# 'core/permission.py', # 权限枚举 - 包含动态属性，不适合编译
# 数据模型 - 高频使用
'models/message.py', # 消息段模型 - 高频消息处理
'models/sender.py', # 发送者模型 - 高频消息处理
'models/objects.py', # API 响应数据模型 - 高频数据处理
# 事件处理相关 - 高频使用
'core/handlers/event_handler.py', # 事件处理器 - 核心事件处理
# 事件模型 - 高频使用，但包含dataclass，可能有编译问题，暂时排除
# 'models/events/message.py', # 消息事件 - 最高频事件类型
# 'models/events/notice.py', # 通知事件 - 高频事件类型
# 'models/events/request.py', # 请求事件 - 高频事件类型
# 'models/events/meta.py', # 元事件 - 高频事件类型
# 注意：以下文件不适合编译
# - 主程序文件（main.py）
# - 测试文件（tests/目录）
# - 插件文件（plugins/目录）
# - 编译（脚本compile_machine_code.py等）
# - 包含复杂动态特性的文件
# - API 基础类（由于多重继承问题）
]
def list_compiled_modules():
"""列出已编译的模块"""
print(f"\n已编译的 {PLATFORM} 模块:")
print("=" * 50)
# 查找所有编译后的文件
compiled_files = []
for ext in [EXTENSION, f'__mypyc{EXTENSION}']:
compiled_files.extend(glob.glob(f'**/*{ext}', recursive=True))
# 过滤掉虚拟环境中的文件
compiled_files = [f for f in compiled_files if 'venv' not in f and '.venv' not in f]
if compiled_files:
for f in sorted(compiled_files):
size = os.path.getsize(f) // 1024 # KB
print(f"{f} ({size} KB)")
else:
print(f"未找到已编译的 {EXTENSION} 文件")
print(f"\n总计: {len(compiled_files)} 个文件")
def clean_compiled_files():
"""清理编译生成的文件"""
print(f"\n清理编译生成的 {EXTENSION} 文件...")
# 查找所有编译后的文件
compiled_files = []
for ext in [EXTENSION, f'__mypyc{EXTENSION}']:
compiled_files.extend(glob.glob(f'**/*{ext}', recursive=True))
# 过滤掉虚拟环境中的文件
compiled_files = [f for f in compiled_files if 'venv' not in f and '.venv' not in f]
if compiled_files:
for f in sorted(compiled_files):
try:
os.remove(f)
print(f"已删除: {f}")
except Exception as e:
print(f"删除失败 {f}: {e}")
# 清理 build 目录
if os.path.exists('build'):
try:
shutil.rmtree('build')
print("已删除 build 目录")
except Exception as e:
print(f"删除 build 目录失败: {e}")
else:
print(f"没有可清理的 {EXTENSION} 文件")
def get_platform_specific_module_name(module_path):
"""获取平台特定的模块文件名"""
# 只获取模块名，不包含路径
module_name = os.path.basename(module_path).replace('.py', '')
return f"{module_name}.{BUILD_PREFIX}{EXTENSION}"
def compile_module(module_path):
"""编译单个模块"""
print(f"\n编译: {module_path}")
try:
# 直接调用 mypyc 命令行工具
# 使用二进制模式捕获输出以避免编码问题
result = subprocess.run(
[sys.executable, '-m', 'mypyc', module_path],
capture_output=True,
check=True
)
# 解码输出时处理可能的编码错误
try:
stdout_text = result.stdout.decode('utf-8', errors='replace')
stderr_text = result.stderr.decode('utf-8', errors='replace')
except AttributeError:
# 如果已经是字符串（Python 3.7+），则直接使用
stdout_text = result.stdout
stderr_text = result.stderr
# 获取平台特定的模块名
# 获取模块名和目录
module_dir = os.path.dirname(module_path)
module_basename = os.path.basename(module_path).replace('.py', '')
# 生成平台特定的模块文件名（仅文件名，不含路径）
platform_module_name = f"{module_basename}.{BUILD_PREFIX}{EXTENSION}"
mypyc_platform_module_name = f"{module_basename}__mypyc.{BUILD_PREFIX}{EXTENSION}"
# 完整路径构造
platform_module = os.path.join(module_dir, platform_module_name)
mypyc_platform_module = os.path.join(module_dir, mypyc_platform_module_name)
# 检查编译产物是否在当前目录
if os.path.exists(platform_module):
print(f" ✓ 编译成功: {platform_module}")
return True
else:
# 检查 build 目录中是否有编译产物
build_module_path = os.path.join(BUILD_PATH, module_dir, platform_module_name)
build_mypyc_path = os.path.join(BUILD_PATH, module_dir, mypyc_platform_module_name)
if os.path.exists(build_module_path):
# 如果在 build 目录中，复制到正确位置
os.makedirs(module_dir, exist_ok=True)
shutil.copy2(build_module_path, platform_module)
if os.path.exists(build_mypyc_path):
shutil.copy2(build_mypyc_path, mypyc_platform_module)
print(f" ✓ 编译成功（已从 build 目录复制）: {platform_module}")
return True
else:
print(" ✗ 编译失败：找不到编译产物")
if result.stdout:
print(f" 编译输出：{stdout_text[:500]}...")
if result.stderr:
print(f" 错误信息：{stderr_text[:500]}...")
return False
except subprocess.CalledProcessError as e:
print(f" ✗ 编译失败，退出码: {e.returncode}")
if hasattr(e, 'stdout') and e.stdout:
try:
stdout_text = e.stdout.decode('utf-8', errors='replace') if isinstance(e.stdout, bytes) else e.stdout
print(f" 编译输出：{stdout_text[:500]}...")
except Exception:
print(f" 编译输出：{str(e.stdout)[:500]}...")
if hasattr(e, 'stderr') and e.stderr:
try:
stderr_text = e.stderr.decode('utf-8', errors='replace') if isinstance(e.stderr, bytes) else e.stderr
print(f" 错误信息：{stderr_text[:500]}...")
except Exception:
print(f" 错误信息：{str(e.stderr)[:500]}...")
return False
except Exception as e:
print(f" ✗ 编译失败，意外错误: {e}")
import traceback
traceback.print_exc()
return False
def should_skip_module(module_path):
"""检查模块是否应该被跳过编译"""
try:
with open(module_path, 'r', encoding='utf-8') as f:
content = f.read()
# 检查是否包含抽象基类相关代码
if 'from abc import ABC' in content or 'from abc import abstractmethod' in content:
return True, "包含抽象基类，不适合编译"
# 检查是否包含危险的动态特性
# 注意：我们允许基本的动态特性，如getattr，但对于eval、exec等危险操作仍然阻止
if ('eval(' in content or 'exec(' in content or
'compile(' in content):
return True, "包含危险动态特性，不适合编译"
# 检查是否包含复杂的动态属性访问
if ('__dict__' in content or '__class__' in content or
'__module__' in content or '__bases__' in content):
return True, "包含复杂动态特性，不适合编译"
# 检查是否包含复杂的动态属性访问
if '.__dict__' in content or '.__class__' in content:
return True, "包含复杂动态特性，不适合编译"
return False, ""
except Exception as e:
return True, f"读取文件时出错: {e}"
def compile_all_modules():
"""编译所有指定的模块"""
print(f"\n开始编译 {len(MODULES)} 个模块 (平台: {PLATFORM})")
print("=" * 60)
# 验证模块文件是否存在并检查是否适合编译
valid_modules = []
skipped_modules = []
for module_path in MODULES:
if os.path.exists(module_path):
should_skip, reason = should_skip_module(module_path)
if should_skip:
print(f"跳过: {module_path} ({reason})")
skipped_modules.append((module_path, reason))
else:
valid_modules.append(module_path)
else:
print(f"警告: 模块 {module_path} 不存在，将被跳过")
print(f"\n有效模块: {len(valid_modules)}, 跳过模块: {len(skipped_modules)}")
if not valid_modules:
print("错误: 没有有效的模块可编译")
return False
# 编译模块
success_count = 0
failed_modules = []
for module_path in valid_modules:
if compile_module(module_path):
success_count += 1
else:
failed_modules.append(module_path)
print("\n" + "=" * 60)
print(f"编译完成: {success_count}/{len(valid_modules)} 个模块成功")
if failed_modules:
print(f"失败模块: {failed_modules}")
if success_count == len(valid_modules):
print("✓ 所有模块编译成功")
return True
else:
print("✗ 部分模块编译失败")
return False
def main():
"""主函数"""
parser = argparse.ArgumentParser(description='跨平台 Python 模块编译脚本')
group = parser.add_mutually_exclusive_group()
group.add_argument('--compile', '-c', action='store_true', default=True,
help='编译指定的模块 (默认)')
group.add_argument('--list', '-l', action='store_true',
help='列出已编译的模块')
group.add_argument('--clean', '-k', action='store_true',
help='清理编译生成的文件')
args = parser.parse_args()
# 检查是否安装了 mypyc
try:
import mypyc
except ImportError:
print("错误: 未安装 mypyc，请先安装: pip install mypyc")
sys.exit(1)
if args.list:
list_compiled_modules()
elif args.clean:
clean_compiled_files()
else:
compile_all_modules()
print("\n使用 --list 选项查看已编译的模块")
print("使用 --clean 选项清理编译文件")
if __name__ == '__main__':
main()
main()

74
scripts/compile_modules.py
View File

@@ -1,74 +0,0 @@
#!/usr/bin/env python3
"""
编译模块脚本
这个脚本会单独编译每个Python模块，确保每个模块都在正确位置生成独立的.pyd文件。
"""
import os
import sys
import glob
from mypyc.build import mypycify
try:
from setuptools import setup
except ImportError:
from distutils.core import setup
def compile_module(module_path):
"""
编译单个模块
Args:
module_path: 要编译的Python模块路径
"""
print(f"\nCompiling {module_path}...")
try:
ext_modules = mypycify([module_path])
setup(name=f'compiled_{os.path.basename(module_path).replace(".py", "")}',
ext_modules=ext_modules)
return True
except Exception as e:
print(f"Error compiling {module_path}: {e}")
return False
def main():
"""
主函数
"""
# 检查 Python 版本
if not (sys.version_info.major == 3 and sys.version_info.minor == 14):
print("警告: 推荐使用 Python 3.14 以获得最佳性能")
print(f"当前版本: {sys.version_info.major}.{sys.version_info.minor}.{sys.version_info.micro}")
print("继续编译可能导致兼容性问题")
print()
# 要编译的模块列表
modules = [
'core/utils/json_utils.py', # JSON 处理
'core/utils/executor.py', # 代码执行引擎
'core/managers/command_manager.py', # 指令匹配和分发
'core/managers/permission_manager.py', # 权限管理（包含管理员管理功能）
'core/ws.py', # WebSocket 核心
'core/managers/plugin_manager.py', # 插件管理器
'core/bot.py', # Bot 核心抽象
'core/config_loader.py', # 配置加载
]
# 自动添加 events 模型
event_models = glob.glob('models/events/*.py')
event_models = [m for m in event_models if not m.endswith('__init__.py')]
modules.extend(event_models)
print(f"Found {len(modules)} modules to compile.")
success_count = 0
for module in modules:
if compile_module(module):
success_count += 1
print("\n--- Compilation Summary ---")
print(f"Total modules: {len(modules)}")
print(f"Successfully compiled: {success_count}")
print(f"Failed: {len(modules) - success_count}")
if __name__ == '__main__':
main()