重构为 openclaw_bridge 模板插件

- 将 process.py 替换为 OpenClaw Gateway QQ AI Reply 插件
- 新增 config/openclawbridge/config.toml(SDK 标准路径)
- 遵循 SDK 官方配置方式,移除硬编码默认值
- 内置高危词检测、管理员白名单、MC指令过滤
- 完善的 README.md 文档(使用说明+架构图+安全特性)
- packup.sh 增加『必须在目标主机打包』警告
- 保留 SDK 框架文件(plugin_modules.py, file_store_api.py 等)
- 更新 test.py 适配 openclaw_bridge 测试

注意:打包必须在目标主机执行,避免 C 扩展兼容性问题
This commit is contained in:
Claw
2026-05-04 16:14:42 +08:00
parent b88e868160
commit 8c4c5074f2
7 changed files with 699 additions and 211 deletions

175
test.py
View File

@ -1,141 +1,60 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import sys
"""
测试框架 — 模拟 qqrebot 环境测试 openclaw_bridge 插件
"""
import os
import importlib
import sys
import logging
from pathlib import Path
from typing import Type, List
from unittest.mock import Mock
# 添加src目录到系统路径
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
)
# 导入基础类
from src.modules.plugin_modules import MessageContext, BasePlugin
# 模拟 qqrebot 插件加载路径
sys.path.insert(0, str(Path(__file__).parent))
class PluginTester:
"""
插件测试助手类
创建模拟上下文和测试配置环境
"""
def __init__(self, plugin_class: Type[BasePlugin], config_dir: str = "test_configs"):
self.plugin_class = plugin_class
self.config_dir = config_dir
os.makedirs(self.config_dir, exist_ok=True)
# 创建模拟上下文
self.mock_ctx = MessageContext(
uid="test_user",
gid="test_group",
raw_message="测试消息",
id="test_bot"
from src.modules.plugin_modules import MessageContext
from src.process import OpenClawBridge
class OpenclawBridgeTest:
def __init__(self):
self.test_simulate_chat()
def test_simulate_chat(self):
"""模拟接收到 QQ 消息"""
print("=" * 60)
print("OpenClawBridge 插件测试")
print("=" * 60)
# 模拟群聊消息上下文
mock_ctx = MessageContext(
uid="2198972886",
gid="12345678",
raw_message="你好",
id="123456789",
)
# 创建插件实例
self.plugin = self.plugin_class(self.mock_ctx)
def test_before_load(self):
"""测试before_load方法(如果存在)"""
if hasattr(self.plugin, 'before_load'):
print(f"\n正在测试 {self.plugin_class.__name__} 的 before_load")
try:
result = self.plugin.before_load()
print(f"before_load 执行成功,返回值: {result}")
return True
except Exception as e:
print(f"before_load 执行出错: {str(e)}")
return False
else:
print(f"\n{self.plugin_class.__name__} 中没有找到 before_load 方法")
return None
# 实例化插件(会加载配置并尝试连接 Gateway)
plugin = OpenClawBridge(mock_ctx)
def test_after_load(self):
"""测试after_load方法(如果存在)"""
if hasattr(self.plugin, 'after_load'):
print(f"\n正在测试 {self.plugin_class.__name__} 的 after_load")
try:
result = self.plugin.after_load()
print(f"after_load 执行成功,返回值: {result}")
return True
except Exception as e:
print(f"after_load 执行出错: {str(e)}")
return False
else:
print(f"\n{self.plugin_class.__name__} 中没有找到 after_load 方法")
return None
print(f"\n配置状态:")
print(f" gateway_url: {plugin.gateway_url}")
print(f" allowed_sender: {plugin.allowed_sender}")
print(f" model: {plugin.model}")
print(f" agent_id: {plugin.agent_id}")
def test_after_save(self):
"""测试after_save方法(如果存在)"""
if hasattr(self.plugin, 'after_save'):
print(f"\n正在测试 {self.plugin_class.__name__} 的 after_save")
try:
result = self.plugin.after_save()
print(f"after_save 执行成功,返回值: {result}")
return True
except Exception as e:
print(f"after_save 执行出错: {str(e)}")
return False
else:
print(f"\n{self.plugin_class.__name__} 中没有找到 after_save 方法")
return None
if not plugin.gateway_token or plugin.gateway_token == "your_gateway_token_here":
print("\n⚠️ 配置未修改,请先编辑 config/openclawbridge/config.toml")
print(" 填入正确的 gateway_token 后再运行测试\n")
def find_plugin_classes(module_path: str) -> List[Type[BasePlugin]]:
"""
在process.py中查找所有继承自BasePlugin的类
返回类类型列表
"""
plugin_classes = []
# 动态导入模块
module_name = os.path.splitext(os.path.basename(module_path))[0]
spec = importlib.util.spec_from_file_location(module_name, module_path)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
# 查找所有继承自BasePlugin的类
for name, obj in vars(module).items():
try:
if isinstance(obj, type) and issubclass(obj, BasePlugin) and obj != BasePlugin:
plugin_classes.append(obj)
except TypeError:
continue
return plugin_classes
print("\n执行 after_save...")
result = plugin.after_save()
print(f"返回: {result}")
print("=" * 60)
def main():
# process.py文件路径
process_path = Path("src/process.py")
if not process_path.exists():
print(f"错误: 在 {process_path} 没有找到 process.py")
return
# 查找process.py中的所有插件类
plugin_classes = find_plugin_classes(str(process_path))
if not plugin_classes:
print("在 process.py 中没有找到插件类")
return
print(f"发现了 {len(plugin_classes)} 个需要测试的插件类:")
for i, plugin_class in enumerate(plugin_classes, 1):
print(f"{i}. {plugin_class.__name__}")
# 测试每个插件类
for plugin_class in plugin_classes:
print(f"\n{'='*50}")
print(f"正在测试插件: {plugin_class.__name__}")
tester = PluginTester(plugin_class)
# 按照自然顺序测试生命周期方法
tester.test_before_load()
tester.test_after_load()
# 如果需要测试配置保存
if hasattr(plugin_class, 'after_save'):
tester.test_after_save()
print(f"{'='*50}\n")
if __name__ == "__main__":
main()
OpenclawBridgeTest()