Losita/mai-bot
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 优化表情包插件和 TTS 插件的异步调用，简化错误处理逻辑

Browse Source
This commit is contained in:
DrSmoothl
2026-03-13 16:31:59 +08:00
parent 26ba33ff74
commit 262efa2302
7 changed files with 192 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
plugins/emoji_manage_plugin/plugin.py
View File

@@ -198,10 +198,7 @@ class EmojiManagePlugin(MaiBotPlugin):
@Command("random_emojis", description="发送多张随机表情包", pattern=r"^/random_emojis$")
async def handle_random_emojis(self, stream_id: str = "", **kwargs):
"""发送多张随机表情包"""
result = await self.ctx.emoji.get_random(5)
if not result or not result.get("success"):
return False, "未找到表情包", False
emojis = result.get("emojis", [])
emojis = await self.ctx.emoji.get_random(5)
if not emojis:
return False, "未找到表情包", False
messages = [

5
plugins/hello_world_plugin/plugin.py
View File

@@ -84,10 +84,7 @@ class HelloWorldPlugin(MaiBotPlugin):
@Command("random_emojis", description="发送多张随机表情包", pattern=r"^/random_emojis$")
async def handle_random_emojis(self, stream_id: str = "", **kwargs):
"""发送多张随机表情包"""
result = await self.ctx.emoji.get_random(5)
if not result or not result.get("success"):
return False, "未找到表情包", False
emojis = result.get("emojis", [])
emojis = await self.ctx.emoji.get_random(5)
if not emojis:
return False, "未找到表情包", False
# 用转发消息发送多张图片

136
pytests/test_plugin_runtime.py
View File

@@ -350,6 +350,142 @@ class TestSDK:
assert runner._rpc_client.calls[0]["plugin_id"] == "owner_plugin"
assert runner._rpc_client.calls[0]["method"] == "cap.request"
@pytest.mark.asyncio
async def test_runner_applies_initial_plugin_config(self, tmp_path):
"""Runner 应在 on_load 前为支持的插件实例注入 config.toml。"""
from src.plugin_runtime.runner.runner_main import PluginRunner
class DummyPlugin:
def __init__(self):
self.configs = []
def set_plugin_config(self, config):
self.configs.append(config)
plugin_dir = tmp_path / "demo_plugin"
plugin_dir.mkdir()
(plugin_dir / "config.toml").write_text("[section]\nvalue = 1\n", encoding="utf-8")
runner = PluginRunner(host_address="dummy", session_token="token", plugin_dirs=[])
plugin = DummyPlugin()
meta = SimpleNamespace(plugin_id="demo_plugin", plugin_dir=str(plugin_dir), instance=plugin)
runner._apply_plugin_config(meta)
assert plugin.configs == [{"section": {"value": 1}}]
@pytest.mark.asyncio
async def test_runner_config_update_refreshes_plugin_config_before_callback(self):
"""配置更新时应先刷新插件配置，再调用 on_config_update。"""
from src.plugin_runtime.protocol.envelope import Envelope, MessageType
from src.plugin_runtime.runner.runner_main import PluginRunner
class DummyPlugin:
def __init__(self):
self.configs = []
self.updates = []
def set_plugin_config(self, config):
self.configs.append(config)
async def on_config_update(self, config, version):
self.updates.append((config, version, list(self.configs)))
runner = PluginRunner(host_address="dummy", session_token="token", plugin_dirs=[])
plugin = DummyPlugin()
runner._loader._loaded_plugins["demo_plugin"] = SimpleNamespace(instance=plugin)
envelope = Envelope(
request_id=1,
message_type=MessageType.REQUEST,
method="plugin.config_updated",
plugin_id="demo_plugin",
payload={"config_data": {"enabled": True}, "config_version": "v2"},
)
response = await runner._handle_config_updated(envelope)
assert response.payload["acknowledged"] is True
assert plugin.configs == [{"enabled": True}]
assert plugin.updates == [({"enabled": True}, "v2", [{"enabled": True}])]
class TestPluginSdkUsage:
"""验证仓库内插件按新 SDK 归一化返回值工作。"""
@pytest.mark.asyncio
async def test_builtin_emoji_plugin_handles_normalized_results(self):
from maibot_sdk.context import PluginContext
from src.plugins.built_in.emoji_plugin.plugin import EmojiPlugin
async def fake_rpc_call(method: str, plugin_id: str = "", payload: dict | None = None):
assert method == "cap.request"
assert payload is not None
capability = payload["capability"]
return {
"emoji.get_random": {
"success": True,
"emojis": [{"base64": "img-1", "emotion": "happy"}],
},
"message.get_recent": {"success": True, "messages": [{"id": 1}]},
"message.build_readable": {"success": True, "text": "最近消息"},
"llm.generate": {"success": True, "response": "happy", "reasoning": "", "model_name": "m"},
"send.emoji": {"success": True},
}[capability]
plugin = EmojiPlugin()
plugin._set_context(PluginContext(plugin_id="emoji", rpc_call=fake_rpc_call))
success, message = await plugin.handle_emoji(stream_id="stream-1", reasoning="测试", chat_id="chat-1")
assert success is True
assert "成功发送表情包" in message
@pytest.mark.asyncio
async def test_tts_plugin_uses_send_custom_bool_result(self):
from maibot_sdk.context import PluginContext
from src.plugins.built_in.tts_plugin.plugin import TTSPlugin
async def fake_rpc_call(method: str, plugin_id: str = "", payload: dict | None = None):
assert method == "cap.request"
assert payload is not None
assert payload["capability"] == "send.custom"
return {"success": True}
plugin = TTSPlugin()
plugin._set_context(PluginContext(plugin_id="tts", rpc_call=fake_rpc_call))
success, message = await plugin.handle_tts_action(
stream_id="stream-1",
action_data={"voice_text": "你好！！！"},
)
assert success is True
assert message == "TTS动作执行成功"
@pytest.mark.asyncio
async def test_hello_world_plugin_handles_random_emoji_list(self):
from maibot_sdk.context import PluginContext
from plugins.hello_world_plugin.plugin import HelloWorldPlugin
async def fake_rpc_call(method: str, plugin_id: str = "", payload: dict | None = None):
assert method == "cap.request"
assert payload is not None
capability = payload["capability"]
return {
"emoji.get_random": {"success": True, "emojis": [{"base64": "img-1"}, {"base64": "img-2"}]},
"send.forward": {"success": True},
}[capability]
plugin = HelloWorldPlugin()
plugin._set_context(PluginContext(plugin_id="hello", rpc_call=fake_rpc_call))
success, message, should_continue = await plugin.handle_random_emojis(stream_id="stream-1")
assert success is True
assert message == "已发送随机表情包"
assert should_continue is True
# ─── 端到端集成测试 ────────────────────────────────────────

44
src/plugin_runtime/runner/runner_main.py
View File

@@ -11,6 +11,8 @@
from typing import Any, List, Optional
from pathlib import Path
import asyncio
import contextlib
import inspect
@@ -19,6 +21,7 @@ import os
import signal
import sys
import time
import tomllib
from src.common.logger import get_console_handler, get_logger, initialize_logging
from src.plugin_runtime import ENV_IPC_ADDRESS, ENV_PLUGIN_DIRS, ENV_SESSION_TOKEN
@@ -95,6 +98,7 @@ class PluginRunner:
for meta in plugins:
instance = meta.instance
self._inject_context(meta.plugin_id, instance)
self._apply_plugin_config(meta)
if hasattr(instance, "on_load"):
try:
ret = instance.on_load()
@@ -212,6 +216,34 @@ class PluginRunner:
instance._set_context(ctx)
logger.debug(f"已为插件 {plugin_id} 注入 PluginContext")
def _apply_plugin_config(self, meta: PluginMeta, config_data: Optional[dict[str, Any]] = None) -> None:
"""在 Runner 侧为插件实例注入当前插件配置。"""
instance = meta.instance
if not hasattr(instance, "set_plugin_config"):
return
plugin_config = config_data if config_data is not None else self._load_plugin_config(meta.plugin_dir)
try:
instance.set_plugin_config(plugin_config)
except Exception as exc:
logger.warning(f"插件 {meta.plugin_id} 配置注入失败: {exc}")
@staticmethod
def _load_plugin_config(plugin_dir: str) -> dict[str, Any]:
"""从插件目录读取 config.toml。"""
config_path = Path(plugin_dir) / "config.toml"
if not config_path.exists():
return {}
try:
with config_path.open("rb") as handle:
loaded = tomllib.load(handle)
except Exception as exc:
logger.warning(f"读取插件配置失败 {config_path}: {exc}")
return {}
return loaded if isinstance(loaded, dict) else {}
def _register_handlers(self) -> None:
"""注册方法处理器"""
self._rpc_client.register_method("plugin.invoke_command", self._handle_invoke)
@@ -460,14 +492,16 @@ class PluginRunner:
"""处理配置更新事件"""
plugin_id = envelope.plugin_id
meta = self._loader.get_plugin(plugin_id)
if meta and hasattr(meta.instance, "on_config_update"):
if meta:
try:
config_data = envelope.payload.get("config_data", {})
config_version = envelope.payload.get("config_version", "")
ret = meta.instance.on_config_update(config_data, config_version)
# 兼容同步和异步的 on_config_update 实现
if asyncio.iscoroutine(ret):
await ret
self._apply_plugin_config(meta, config_data=config_data)
if hasattr(meta.instance, "on_config_update"):
ret = meta.instance.on_config_update(config_data, config_version)
# 兼容同步和异步的 on_config_update 实现
if asyncio.iscoroutine(ret):
await ret
except Exception as e:
logger.error(f"插件 {plugin_id} 配置更新失败: {e}")
return envelope.make_error_response(ErrorCode.E_UNKNOWN.value, str(e))

28
src/plugins/built_in/emoji_plugin/plugin.py
View File

@@ -30,11 +30,7 @@ class EmojiPlugin(MaiBotPlugin):
reason = reasoning or "表达当前情绪"
# 1. 随机获取30个表情包
result = await self.ctx.emoji.get_random(30)
if not result or not result.get("success"):
return False, "无法获取随机表情包"
sampled_emojis = result.get("emojis", [])
sampled_emojis = await self.ctx.emoji.get_random(30)
if not sampled_emojis:
return False, "无法获取随机表情包"
@@ -57,19 +53,13 @@ class EmojiPlugin(MaiBotPlugin):
# 3. 获取最近消息作为上下文
messages_text = ""
if chat_id:
recent_result = await self.ctx.message.get_recent(chat_id=chat_id, limit=5)
if recent_result and recent_result.get("success"):
readable_result = await self.ctx.call_capability(
"message.build_readable",
chat_id=chat_id,
start_time=0,
end_time=0,
limit=5,
recent_messages = await self.ctx.message.get_recent(chat_id=chat_id, limit=5)
if recent_messages:
messages_text = await self.ctx.message.build_readable(
recent_messages,
timestamp_mode="normal_no_YMD",
truncate=False,
)
if readable_result and readable_result.get("success"):
messages_text = readable_result.get("text", "")
# 4. 构建 prompt 让 LLM 选择情感
available_emotions_str = "\n".join(available_emotions)
@@ -100,16 +90,14 @@ class EmojiPlugin(MaiBotPlugin):
chosen = random.choice(sampled_emojis)
# 7. 发送
send_result = await self.ctx.send.emoji(chosen["base64"], stream_id)
if send_result and send_result.get("success"):
send_ok = await self.ctx.send.emoji(chosen["base64"], stream_id)
if send_ok:
return True, f"成功发送表情包:[表情包：{chosen_emotion}]"
return False, "发送表情包失败"
async def on_load(self):
# 从插件配置读取 emoji_chance 来覆盖默认概率
config_result = await self.ctx.config.get("emoji.emoji_chance")
if config_result and isinstance(config_result, dict) and config_result.get("success"):
pass # 配置已在宿主端管理
await self.ctx.config.get("emoji.emoji_chance")
def create_plugin():

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

@@ -36,7 +36,7 @@ class KnowledgePlugin(MaiBotPlugin):
except (TypeError, ValueError):
limit_value = 5
result = await self.ctx.call_capability("knowledge.search", query=query, limit=limit_value)
result = await self.ctx.knowledge.search(query=query, limit=limit_value)
if result and result.get("success"):
content = result.get("content", f"你不太了解有关{query}的知识")
return {"type": "lpmm_knowledge", "id": query, "content": content}

13
src/plugins/built_in/tts_plugin/plugin.py
View File

@@ -5,7 +5,7 @@
import re
from maibot_sdk import MaiBotPlugin, Action
from maibot_sdk import Action, MaiBotPlugin
from maibot_sdk.types import ActivationType
@@ -40,15 +40,14 @@ class TTSPlugin(MaiBotPlugin):
processed_text = f"{processed_text}。"
# 发送自定义 tts 消息
result = await self.ctx.call_capability(
"send.custom",
message_type="tts_text",
content=processed_text,
send_ok = await self.ctx.send.custom(
custom_type="tts_text",
data=processed_text,
stream_id=stream_id,
)
if result and result.get("success"):
if send_ok:
return True, "TTS动作执行成功"
return False, f"TTS动作执行失败: {result}"
return False, "TTS动作执行失败"
def create_plugin():