gardel/Mofox-Core
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(core): 提升类型安全性并添加配置空值检查

Browse Source 
此提交在核心模块中引入了多项改进，以增强类型安全性和健壮性，主要通过为类型提示添加 `cast` 并在访问 `global_config` 属性前进行空值检查实现。

主要改动包括：
- **类型安全**：在 `message_handler.py`、`unified_manager.py` 和 `napcat_adapter` 的消息处理器中使用 `typing.cast` 来解决类型不一致问题并提高静态分析的准确性。
- **配置空值检查**：在 `message_handler.py` 和 `message_processor.py` 中添加对 `global_config` 及其嵌套属性的检查，以防止在应用启动或配置加载过程中出现 `NoneType` 错误。
- **内存管理提示**：优化了 `unified_manager.py` 中内存判断器的提示，使其在获取长期记忆时更加保守，从而提升简单交互的性能。
- **Napcat 适配器**：新增了视频处理的配置选项以及回复行为。同时改进了消息解析逻辑的鲁棒性。- **消息处理器**：重构了 `_process_message_segments` 及相关函数，移除了未使用的 `message_info` 参数，从而简化了函数签名。
This commit is contained in:
tt-P607
2025-11-28 10:15:53 +08:00
parent 883e391010
commit abfcf56941
6 changed files with 119 additions and 94 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
src/plugins/built_in/core_actions/plugin.py
View File

@@ -63,7 +63,7 @@ class CoreActionsPlugin(BasePlugin):
"""返回插件包含的组件列表"""
# --- 根据配置注册组件 ---
components: ClassVar = []
components = []
# 注册 reply 动作
if self.get_config("components.enable_reply", True):

3
src/plugins/built_in/napcat_adapter/plugin.py
View File

@@ -317,6 +317,9 @@ class NapcatAdapterPlugin(BasePlugin):
"ignore_non_self_poke": ConfigField(type=bool, default=False, description="是否忽略不是针对自己的戳一戳消息"),
"poke_debounce_seconds": ConfigField(type=float, default=2.0, description="戳一戳防抖时间（秒）"),
"enable_emoji_like": ConfigField(type=bool, default=True, description="是否启用群聊表情回复处理"),
"enable_reply_at": ConfigField(type=bool, default=True, description="是否在回复时自动@原消息发送者"),
"reply_at_rate": ConfigField(type=float, default=0.5, description="回复时@的概率（0.0-1.0）"),
"enable_video_processing": ConfigField(type=bool, default=True, description="是否启用视频消息处理（下载和解析）"),
},
}

91
src/plugins/built_in/napcat_adapter/src/handlers/to_core/message_handler.py
View File

@@ -5,7 +5,7 @@ from __future__ import annotations
import base64
import time
from pathlib import Path
from typing import TYPE_CHECKING, Any, Dict, List, Optional, Tuple
from typing import TYPE_CHECKING, Any, Dict, List, Optional, Tuple, cast
import uuid
from mofox_wire import MessageBuilder
@@ -214,6 +214,9 @@ class MessageHandler:
case RealMessageType.record:
return await self._handle_record_message(segment)
case RealMessageType.video:
if not config_api.get_plugin_config(self.plugin_config, "features.enable_video_processing", True):
logger.debug("视频消息处理已禁用，跳过")
return None
return await self._handle_video_message(segment)
case RealMessageType.rps:
return await self._handle_rps_message(segment)
@@ -331,10 +334,13 @@ class MessageHandler:
{"type": seg.get("type", "text"), "data": seg.get("data", "")} for seg in reply_segments
] or [{"type": "text", "data": "[无法获取被引用的消息]"}]
return {
"type": "seglist",
"data": [{"type": "text", "data": prefix_text}, *brief_segments, {"type": "text", "data": suffix_text}],
}
return cast(
SegPayload,
{
"type": "seglist",
"data": [{"type": "text", "data": prefix_text}, *brief_segments, {"type": "text", "data": suffix_text}],
},
)
async def _handle_record_message(self, segment: dict) -> SegPayload | None:
"""处理语音消息"""
@@ -380,14 +386,17 @@ class MessageHandler:
video_base64 = base64.b64encode(video_data).decode("utf-8")
logger.debug(f"视频文件大小: {len(video_data) / (1024 * 1024):.2f} MB")
return {
"type": "video",
"data": {
"base64": video_base64,
"filename": Path(file_path).name,
"size_mb": len(video_data) / (1024 * 1024),
return cast(
SegPayload,
{
"type": "video",
"data": {
"base64": video_base64,
"filename": Path(file_path).name,
"size_mb": len(video_data) / (1024 * 1024),
},
},
}
)
elif video_url:
# URL下载处理
from ..video_handler import get_video_downloader
@@ -401,15 +410,18 @@ class MessageHandler:
video_base64 = base64.b64encode(download_result["data"]).decode("utf-8")
logger.debug(f"视频下载成功，大小: {len(download_result['data']) / (1024 * 1024):.2f} MB")
return {
"type": "video",
"data": {
"base64": video_base64,
"filename": download_result.get("filename", "video.mp4"),
"size_mb": len(download_result["data"]) / (1024 * 1024),
"url": video_url,
return cast(
SegPayload,
{
"type": "video",
"data": {
"base64": video_base64,
"filename": download_result.get("filename", "video.mp4"),
"size_mb": len(download_result["data"]) / (1024 * 1024),
"url": video_url,
},
},
}
)
else:
logger.warning("既没有有效的本地文件路径，也没有有效的视频URL")
return None
@@ -454,34 +466,39 @@ class MessageHandler:
processed_message = handled_message
forward_hint = {"type": "text", "data": "这是一条转发消息：\n"}
return {"type": "seglist", "data": [forward_hint, processed_message]}
return cast(SegPayload, {"type": "seglist", "data": [forward_hint, processed_message]})
async def _recursive_parse_image_seg(self, seg_data: SegPayload, to_image: bool) -> SegPayload:
# sourcery skip: merge-else-if-into-elif
if seg_data.get("type") == "seglist":
new_seg_list = []
for i_seg in seg_data.get("data", []):
parsed_seg = await self._recursive_parse_image_seg(i_seg, to_image)
new_seg_list.append(parsed_seg)
if isinstance(i_seg, dict): # 确保是字典类型
parsed_seg = await self._recursive_parse_image_seg(i_seg, to_image)
new_seg_list.append(parsed_seg)
return {"type": "seglist", "data": new_seg_list}
if to_image:
if seg_data.get("type") == "image":
image_url = seg_data.get("data")
try:
encoded_image = await get_image_base64(image_url)
except Exception as e:
logger.error(f"图片处理失败: {str(e)}")
return {"type": "text", "data": "[图片]"}
return {"type": "image", "data": encoded_image}
if isinstance(image_url, str):
try:
encoded_image = await get_image_base64(image_url)
except Exception as e:
logger.error(f"图片处理失败: {str(e)}")
return {"type": "text", "data": "[图片]"}
return {"type": "image", "data": encoded_image}
return {"type": "text", "data": "[图片]"}
if seg_data.get("type") == "emoji":
image_url = seg_data.get("data")
try:
encoded_image = await get_image_base64(image_url)
except Exception as e:
logger.error(f"图片处理失败: {str(e)}")
return {"type": "text", "data": "[表情包]"}
return {"type": "emoji", "data": encoded_image}
if isinstance(image_url, str):
try:
encoded_image = await get_image_base64(image_url)
except Exception as e:
logger.error(f"图片处理失败: {str(e)}")
return {"type": "text", "data": "[表情包]"}
return {"type": "emoji", "data": encoded_image}
return {"type": "text", "data": "[表情包]"}
logger.debug(f"不处理类型: {seg_data.get('type')}")
return seg_data
@@ -595,7 +612,7 @@ class MessageHandler:
"id": file_id,
}
return {"type": "file", "data": file_data}
return cast(SegPayload, {"type": "file", "data": file_data})
async def _handle_json_message(self, segment: dict) -> SegPayload | None:
"""
@@ -623,7 +640,7 @@ class MessageHandler:
# 从回声消息中提取文件信息
file_info = self._extract_file_info_from_echo(nested_data)
if file_info:
return {"type": "file", "data": file_info}
return cast(SegPayload, {"type": "file", "data": file_info})
# 检查是否是QQ小程序分享消息
if "app" in nested_data and "com.tencent.miniapp" in str(nested_data.get("app", "")):