From 8c4c5074f21befe638b7bdaca549bd519df96ce1 Mon Sep 17 00:00:00 2001 From: Claw Date: Mon, 4 May 2026 16:14:42 +0800 Subject: [PATCH] =?UTF-8?q?=E9=87=8D=E6=9E=84=E4=B8=BA=20openclaw=5Fbridge?= =?UTF-8?q?=20=E6=A8=A1=E6=9D=BF=E6=8F=92=E4=BB=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 将 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 扩展兼容性问题 --- README.md | 177 ++++++++++- config/openclawbridge/config.toml | 28 ++ packup.sh | 44 ++- requirements.txt | 3 +- src/config.toml | 15 +- src/process.py | 468 ++++++++++++++++++++++++++---- test.py | 175 +++-------- 7 files changed, 699 insertions(+), 211 deletions(-) create mode 100644 config/openclawbridge/config.toml diff --git a/README.md b/README.md index 90166bc..0b3709c 100644 --- a/README.md +++ b/README.md @@ -1,4 +1,175 @@ -# chatrebot_aireply_plug +# OpenClaw Bridge - QQ AI Reply Plugin -用于自动回复群聊私聊消息的插件 -程序打包后，将压缩包放在plugins文件夹，在插件初次被调用后，会在config/ai_reply下生成配置文件，填写api密钥等信息后即可正常运行 \ No newline at end of file +> QQ 消息 ↔ OpenClaw Gateway 桥接插件。 +> 当用户在 QQ 上发消息时，自动转发到 OpenClaw Gateway，由指定的 AI agent 处理并回复。 + +## 架构概览 + +``` +QQ 用户 ──→ NapCat (go-cqhttp) + │ + ▼ + qqrebot (HTTP Server :25580) + │ + ▼ + openclaw_bridge 插件 (本仓库) + │ + ▼ + OpenClaw Gateway (:18789) + │ + ▼ + qq-agent (AI 模型) + │ + ▼ + 自动回复 → QQ 用户 +``` + +## 前置条件 + +需要以下系统已部署运行： + +1. **[qqrebot](https://github.com/jianf/qqrebot)** — QQ 机器人框架，加载本插件 +2. **NapCat / go-cqhttp** — QQ 协议实现，转发消息给 qqrebot +3. **OpenClaw** — AI agent 网关 +4. **qq-agent (或其他 agent)** — 在 OpenClaw 中配置，负责实际处理消息 + +## 快速开始 + +### 1. 配置插件 + +编辑 `config/openclawbridge/config.toml`（若文件不存在则创建）： + +```toml +[openclaw] +gateway_url = "http://127.0.0.1:18789" # OpenClaw Gateway 地址 +gateway_token = "你的gateway token" # 从 OpenClaw 配置获取 +allowed_sender = "2198972886" # 管理员 QQ 号 +model = "openclaw/qq-agent" # 使用的 agent 模型 +agent_id = "qq-agent" # Agent ID +``` + +### 2. 打包（**必须在目标主机上执行！**） + +```bash +# 安装依赖 +bash packup.sh + +# 或者手动打包 +python3 package.py +``` + +> ⚠️ **重要**：由于插件依赖的 Python 包（如 `requests`、`jieba`）包含 C 扩展，**打包必须在最终运行 qqrebot 的机器上执行**，否则可能导致： +> - 动态链接库不兼容（.so 文件无法加载） +> - Python 版本差异导致语法错误 +> - 系统库依赖缺失 + +### 3. 部署 + +打包完成后，将生成的 `dist/openclaw_bridge.zip` 复制到 qqrebot 的 `plugins/` 目录： + +```bash +cp dist/openclaw_bridge.zip /path/to/qqrebot/plugins/ +``` + +重启 qqrebot 加载新插件： + +```bash +systemctl restart qqrebot +``` + +### 4. 验证 + +在 QQ 上给机器人发消息，或在日志中查看： + +```bash +journalctl -u qqrebot -f | grep OpenClawBridge +``` + +预期日志： +``` +Config loaded: url=http://127.0.0.1:18789, model=openclaw/qq-agent +Forwarding to OpenClaw: session=qqgroup:... +Agent reply: ... +``` + +## 插件生命周期 + +``` +before_load → after_load → after_save +``` + +- `before_load`: 插件加载前，可做初始化 +- `after_load`: 插件加载后 +- `after_save`: **核心处理入口** — 收到新消息时触发 + +## 安全特性 + +- **高危词检测**：内置词库检测越狱/提示词攻击、记忆操控、敏感信息泄漏等 +- **管理员白名单**：`allowed_sender` 之外的用户触发危险词自动拦截+拉黑+上报 +- **MC 指令过滤**：非管理员发送的 `/` 开头的消息跳过（由 ops-manager 处理） +- **系统通知过滤**：自动跳过 `[系统通知]`、`[文件回执]` 等内部消息 +- **纯媒体消息过滤**：纯图片/文件/视频消息（无文字内容）自动跳过 +- **错误信息脱敏**：所有 HTTP 响应体、异常详情不会暴露给 QQ 用户 + +## 配置文件说明 + +| 配置项 | 必填 | 说明 | +|---|---|---| +| `gateway_url` | ✅ | OpenClaw Gateway 地址 | +| `gateway_token` | ✅ | Gateway 认证 Token | +| `allowed_sender` | ✅ | 管理员 QQ 号 | +| `model` | ✅ | Agent 模型 ID | +| `agent_id` | ✅ | Agent ID | + +## 目录结构 + +``` +chatrebot_aireply_plug/ +├── config/ +│ └── openclawbridge/ +│ └── config.toml # 插件配置（模板，按需修改） +├── src/ +│ ├── __init__.py +│ ├── process.py # 插件主代码（核心逻辑） +│ ├── config.toml # 插件框架测试配置 +│ └── modules/ +│ ├── __init__.py +│ ├── plugin_modules.py # SDK: BasePlugin + MessageContext +│ └── user_module.py # SDK: User/Group 封装 +├── scripts/ +│ ├── __init__.py +│ └── file_store_api.py # SDK: ConfigManager +├── dependence.py # 安装依赖到 src/packages/ +├── package.py # SDK 打包器：process.py + config.toml + packages/ → .zip +├── packup.sh # 一键安装依赖 + 打包（类Unix） +├── packup.bat # 一键安装依赖 + 打包（Windows） +├── test.py # 本地测试框架 +├── requirements.txt # Python 依赖 +├── .gitignore +├── LICENSE +└── README.md # 本文件 +``` + +## 自定义与扩展 + +### 修改高危词库 + +编辑 `src/process.py` 中的 `HIGH_RISK_WORDS` 列表，按分类添加/删除关键词。 + +### 更换 AI 模型 + +在 `config.toml` 中修改 `model` 字段，指定 OpenClaw 中配置的任何 agent 模型 ID。 + +### 添加新的生命周期钩子 + +在 `OpenClawBridge` 类中添加 `before_load()` 或 `after_load()` 方法： + +```python +def after_load(self): + """插件加载完成后执行""" + logger.info("Plugin loaded successfully!") +``` + +## 许可证 + +MIT diff --git a/config/openclawbridge/config.toml b/config/openclawbridge/config.toml new file mode 100644 index 0000000..eb44dcd --- /dev/null +++ b/config/openclawbridge/config.toml @@ -0,0 +1,28 @@ +[plugin] +name = "openclaw_bridge" +version = "1.0.0" + +# qqrebot 插件框架测试用 +[test] +message = "openclaw_bridge_loaded" + +# ── 核心配置（使用前必须修改） ────────────────────────── +# 克隆本仓库后，将此文件放在 config/openclawbridge/config.toml +# 或在目标主机上打包后首次加载插件时会生成此路径 +[openclaw] +# OpenClaw Gateway 地址 +# 如果使用同机部署的 OpenClaw，默认 http://127.0.0.1:18789 +gateway_url = "http://127.0.0.1:18789" + +# Gateway 认证 Token（必填！从 OpenClaw 配置中获取） +gateway_token = "your_gateway_token_here" + +# 管理员 QQ 号（唯一可发敏感指令、不受 MC 指令过滤限制） +allowed_sender = "your_admin_qq" + +# 要调用的 agent 模型 ID +# 默认：openclaw/qq-agent（对应 qq-agent 配置） +model = "openclaw/qq-agent" + +# Agent ID（与 OpenClaw 配置中 agent id 一致） +agent_id = "qq-agent" diff --git a/packup.sh b/packup.sh index 7d5ed14..3611a26 100644 --- a/packup.sh +++ b/packup.sh @@ -1,18 +1,38 @@ #!/bin/bash -# 依次运行dependence.py和package.py的shell脚本 +# ═══════════════════════════════════════════════════════ +# OpenClaw Bridge 插件 — 一键打包脚本 +# +# ⚠️ 重要警告： +# 由于插件依赖的 Python 包包含 C 扩展（如 jieba、requests 等）， +# 打包必须在**最终运行 qqrebot 的目标主机**上执行。 +# 禁止在本机打包后传输到另一台架构/系统版本不同的机器使用， +# 否则可能导致动态链接库不兼容或 Python 版本差异问题。 +# ═══════════════════════════════════════════════════════ -echo "正在运行依赖安装脚本..." +set -e +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +cd "$SCRIPT_DIR" + +echo "═══════════════════════════════════════" +echo " OpenClaw Bridge 插件打包" +echo "═══════════════════════════════════════" +echo "主机: $(uname -a | head -c 80)" +echo "Python: $(python3 --version 2>&1)" +echo "路径: $SCRIPT_DIR" +echo "═══════════════════════════════════════" + +# 1. 安装依赖到 src/packages/ +echo "" +echo "📦 步骤1/2：安装依赖..." python3 dependence.py -if [ $? -ne 0 ]; then - echo "运行dependence.py失败! 错误码: $?" - exit $? -fi -echo "正在运行打包脚本..." +# 2. 打包 +echo "" +echo "📦 步骤2/2：打包..." python3 package.py -if [ $? -ne 0 ]; then - echo "运行package.py失败! 错误码: $?" - exit $? -fi -echo "所有脚本执行完毕!" +echo "" +echo "✅ 完成！插件包位于: dist/openclaw_bridge.zip" +echo "" +echo "部署: cp dist/openclaw_bridge.zip /path/to/qqrebot/plugins/" +echo "重启: systemctl restart qqrebot" diff --git a/requirements.txt b/requirements.txt index f0dd0ae..97d252e 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1 +1,2 @@ -openai \ No newline at end of file +requests>=2.28.0 +jieba>=0.42.1 diff --git a/src/config.toml b/src/config.toml index eae94ff..865bebe 100644 --- a/src/config.toml +++ b/src/config.toml @@ -1,6 +1,9 @@ -[base] -maxcount = 20 -[siloconflow] -api_key = "" -modules = ["Qwen/Qwen2.5-7B-Instruct", "Qwen/Qwen2.5-Coder-7B-Instruct"] -base_url = "https://api.siliconflow.cn/v1" \ No newline at end of file +[test] +message = "openclaw_bridge_loaded" + +[openclaw] +gateway_url = "http://127.0.0.1:18789" +gateway_token = "your_gateway_token_here" +allowed_sender = "your_admin_qq" +model = "openclaw/qq-agent" +agent_id = "qq-agent" diff --git a/src/process.py b/src/process.py index 66d1aa8..4b23231 100644 --- a/src/process.py +++ b/src/process.py @@ -1,68 +1,414 @@ -import threading -from src.modules.plugin_modules import BasePlugin, MessageContext -from openai import OpenAI +""" +OpenClaw Gateway QQ AI Reply 插件 +————————————————————————————————————— +QQ消息 ↔ OpenClaw Gateway 桥接插件。 +当用户在QQ上发消息时，通过此插件转发到OpenClaw Gateway， +由指定的agent（如 qq-agent）处理并自动回复。 + +基于 qqrebot 插件 SDK 开发。 +""" + +import os +import sys +import json import time +import re +import logging +from pathlib import Path +from datetime import datetime +from src.modules.plugin_modules import BasePlugin, MessageContext + +# ── 内嵌依赖 ────────────────────────────────────────── +# qqrebot 的插件打包机制会将依赖安装到 src/packages/ +_plugin_dir = Path(__file__).parent +_packages_dir = _plugin_dir / "packages" +if str(_packages_dir) not in sys.path: + sys.path.insert(0, str(_packages_dir)) + +import requests + +logger = logging.getLogger(__name__) -class SiloconFlowAPI: - def __init__(self, api_key,modules,base_url): - self.api_key = api_key - self.module = modules - self.url = base_url - self.count = 0 - def get_ai_message(self, message,maxcount,i=0): - with OpenAI(api_key=self.api_key, base_url=self.url) as client: - try: - response = client.chat.completions.create( - model=self.module[i], - messages=message, - temperature=0.7, - max_tokens=4096 - ) - print("token usage {0}".format(response.usage.total_tokens)) - return response.choices[0].message.content - except: - if imaxcount): - return " " - return self.get_ai_message(message,i) +# ── 常量 ────────────────────────────────────────────── +NO_REPLY_MARKER = "No response from OpenClaw." +SANITIZED_REPLY = "." # 统一替换对外暴露的系统内部信息 +# ── 高危词库（可自定义增删） ──────────────────────── +# 注意：词库与检测逻辑耦合紧密，如需外部化可后续抽到 config.toml +HIGH_RISK_WORDS = [ + # 越狱/提示词攻击 + "提示词", "prompt", "人设", "修改规则", "修改性格", "修改人设", + "忽略之前", "解除限制", "你是ai", "你是机器人", "系统提示", + "系统指令", "底层逻辑", "开发文档", "配置文件", "agents.md", + "soul.md", "identity.md", "memory.md", + # 记忆操控 + "忘掉", "忘记", "清空记忆", "重置记忆", "删除记忆", "清除上下文", + "遗忘", "重新学习", "从头开始", + # 敏感信息泄漏 + "gw token", "gateway", "authcode", "授权码", "推送url", + "apitoken", "bearer", "jwt", "session key", + "反代", "代理地址", "服务器ip", "内网地址", "公网ip", + # 黑产/恶意 + "炸群", "卡群", "刷屏", "病毒", "木马", "钓鱼", "社工", + "撞库", "爆破", "渗透", "注入", "xss", "csrf", "ddos", + # 提权尝试 + "你只是", "你应该", "你必须", "从现在开始", + "假装你", "扮演", "你的真实身份", + # 补充匹配 + "system prompt", "ignore", "bypass", "jailbreak", + "forget", "reset", "redefine", "override", + # 语言补充 + "忘记以前", "不记得", "失忆", "清除历史", + "重来", "重新开始", +] + +# ── 群昵称缓存 ────────────────────────────────────── +# 格式: {(group_id, user_id): "nickname"} +_nickname_cache: dict = {} + + +class OpenClawBridge(BasePlugin): + """QQ消息 ↔ OpenClaw Gateway 桥接插件""" + + def __init__(self, ctx: MessageContext): + super().__init__(ctx) + logger.info("=== OpenClawBridge __init__ START ===") + # SDK 官方配置方式：从 self.config（config.toml）读取，不硬编码默认值 + self.gateway_url = None + self.gateway_token = None + self.allowed_sender = "" + self.model = None + self.agent_id = None + + try: + cfg = self.config.get("openclaw", {}) + if cfg: + self.gateway_url = cfg.get("gateway_url") + self.gateway_token = cfg.get("gateway_token") + self.allowed_sender = str(cfg.get("allowed_sender", "")) + self.model = cfg.get("model") + self.agent_id = cfg.get("agent_id") + logger.info(f"Config loaded: url={self.gateway_url}, model={self.model}") + else: + logger.error("Config section [openclaw] not found") + except Exception as e: + logger.error(f"Failed to load config: {e}") + + missing = [k for k, v in { + "gateway_url": self.gateway_url, + "gateway_token": self.gateway_token, + }.items() if not v] + if missing: + logger.error(f"Missing required config: {missing}") + + logger.info("=== OpenClawBridge __init__ END ===") + + # ═══════════════════════════════════════════════════ + # 昵称解析 + # ═══════════════════════════════════════════════════ + + def _fetch_group_nickname(self, group_id: str, sender_id: str) -> str: + """从群成员缓存查昵称""" + cache_key = (group_id, sender_id) + if cache_key in _nickname_cache: + return _nickname_cache[cache_key] + + try: + url = f"http://127.0.0.1:25580/api/group_member_info" + resp = requests.get(url, params={ + "group_id": group_id, + "user_id": sender_id, + "no_cache": "true", + }, timeout=5) + if resp.status_code == 200: + data = resp.json() + nick = (data.get("data") or data).get("nickname", "") or \ + (data.get("data") or data).get("card", "") + if nick: + _nickname_cache[cache_key] = nick + return nick + except Exception: + pass + return "" + + def _lookup_from_group_users(self, group_id: str, sender_id: str) -> str: + """备用方案：从群成员列表遍历查找""" + try: + url = f"http://127.0.0.1:25580/api/get_group_member_list" + resp = requests.get(url, params={"group_id": group_id}, timeout=5) + if resp.status_code == 200: + data = resp.json() + members = data.get("data") or [] + for m in members: + uid = m.get("user_id") + if uid and str(uid) == str(sender_id): + nick = m.get("card") or m.get("nickname", "") + if nick: + _nickname_cache[(group_id, sender_id)] = nick + return nick + except Exception: + pass + return "" + + def _fetch_from_platform_api(self, group_id: str, sender_id: str) -> str: + """从平台接口获取群成员名（通用兜底）""" + # 此方法供后续扩展，目前直接返回空 + return "" + + def _get_sender_group_nickname(self) -> str: + """获取发送者在群里的昵称""" + if not self.ctx.group: + return "" + group_id = str(self.ctx.group.group_id) + sender_id = self._get_sender_id() + + nick = self._fetch_group_nickname(group_id, sender_id) + if nick: + return nick + nick = self._lookup_from_group_users(group_id, sender_id) + if nick: + return nick + return self._fetch_from_platform_api(group_id, sender_id) + + def _get_sender_id(self) -> str: + """获取发送者QQ号""" + return str(self.ctx.user.user_id) + + # ═══════════════════════════════════════════════════ + # 消息处理 + # ═══════════════════════════════════════════════════ + + def _clean_message(self, text: str) -> str: + """清理消息中的CQ码和at机器人标记""" + text = text.replace("[", "[").replace("]", "]") + bot_id_str = str(self.ctx.rebot_id) + bot_at = f"[CQ:at,qq={bot_id_str}]" + text = text.replace(bot_at, f"@你(你的QQ号{bot_id_str})") + text = re.sub(r'\[CQ:([^,]+)(?:,[^\]]+)?\]', r'[\1]', text) + return text + + def _build_source_tag(self) -> str: + """构建消息来源前缀""" + if self.ctx.group: + identity_tag = self._get_sender_group_nickname() + if not identity_tag: + identity_tag = str(self.ctx.user.user_id) + return f"{identity_tag}" + return f"{self.ctx.user.user_id}" + + def _build_context_with_history(self, raw_message: str, identity_tag: str) -> str: + """构建带上下文的消息""" + cleaned = self._clean_message(raw_message) + if self.ctx.group: + tag = identity_tag or self.ctx.user.user_id + return f"[{tag}] {cleaned}" + return cleaned + + # ═══════════════════════════════════════════════════ + # 安全检测 + # ═══════════════════════════════════════════════════ + + def _detect_high_risk(self, message: str) -> tuple: + """检测是否包含高危词，返回(is_high_risk, matched_words)""" + msg_lower = message.lower() + matched = [] + for risk_word in HIGH_RISK_WORDS: + if risk_word.lower() in msg_lower: + matched.append(risk_word) + return (len(matched) > 0, matched) + + def _mark_high_risk(self, message: str, matched_words: list) -> str: + """打码高危内容""" + masked = message + for word in matched_words: + masked = masked.replace(word, "***") + return masked + + def _notify_admin(self, sender_id: str, gid: str, raw_message: str, matched_words: list): + """上报管理员高危消息""" + try: + url = f"{self.gateway_url}/v1/chat/completions" + headers = { + "Content-Type": "application/json", + "Authorization": f"Bearer {self.gateway_token}", + } + masked = self._mark_high_risk(raw_message, matched_words) + payload = { + "model": self.model, + "messages": [{ + "role": "user", + "content": f"[系统通知] ⚠️ 攻击上报：用户{sender_id}在群{gid}试图：{masked[:200]}" + }], + "stream": False, + "session_key": f"admin_notify_{sender_id}_{int(time.time())}", + } + requests.post(url, headers=headers, json=payload, timeout=5) + except Exception: + pass + + def _is_dangerous(self, message: str) -> bool: + """是否危险消息（本地检测）""" + is_risk, matched = self._detect_high_risk(message) + if is_risk: + logger.info(f"High risk detected: {matched}") + return True + + # 仅管理员可发送的指令检测（普通用户触发即危险） + admin_only_patterns = [ + r"\b重[启置新]\s*(?:bot|机器人|系统|agent)?\b", + r"\b(?:重新)?加[载入]\s*(?:配置|插件|skill)\b", + r"\b恢复出厂\b", + ] + msg_lower = message.lower() + for pattern in admin_only_patterns: + if re.search(pattern, msg_lower): + if not self._is_authorized(self._get_sender_id()): + return True + return False + + # ═══════════════════════════════════════════════════ + # 权限与路由 + # ═══════════════════════════════════════════════════ + + def _is_authorized(self, sender_id: str) -> bool: + """检查发送者是否有管理员权限""" + return sender_id == self.allowed_sender + + def _looks_like_mc_command(self, message: str) -> bool: + """看起来像 MC 服务器指令""" + return message.startswith("/") and len(message) > 1 + + def _build_session_key(self) -> str: + """构建会话key""" + sender = self._get_sender_id() + if self.ctx.group: + return f"qqgroup:{self.ctx.group.group_id}:{sender}" + return f"qqprivate:{sender}" + + def _strip_markdown(self, text: str) -> str: + """去掉 Markdown 格式，保留纯文本""" + text = re.sub(r'\*\*(.+?)\*\*', r'\1', text) + text = re.sub(r'\*(.+?)\*', r'\1', text) + text = re.sub(r'`(.+?)`', r'\1', text) + text = re.sub(r'```[\s\S]*?```', '', text) + return text + + # ═══════════════════════════════════════════════════ + # OpenClaw 通信 + # ═══════════════════════════════════════════════════ + + def _send_to_openclaw(self, message: str, session_key: str) -> str: + """ + 同步请求 OpenClaw Gateway。 + 使用非流式模式以支持模型 fallback（kimi→deepseek）。 + """ + if self.gateway_url is None or self.gateway_token is None: + logger.error("gateway_url or gateway_token is not configured") + return SANITIZED_REPLY + + url = f"{self.gateway_url}/v1/chat/completions" + headers = { + "Content-Type": "application/json", + "Authorization": f"Bearer {self.gateway_token}", + } + payload = { + "model": self.model, + "messages": [{"role": "user", "content": message}], + "stream": False, + "user": session_key, + "session": session_key, + "session_key": session_key, + } + + try: + response = requests.post(url, headers=headers, json=payload, timeout=300) + + if response.status_code == 200: + data = response.json() + content = data.get("choices", [{}])[0].get("message", {}).get("content", "") + if content: + return self._strip_markdown(content) + return NO_REPLY_MARKER + + logger.warning(f"Non-200 response: {response.status_code}") + return SANITIZED_REPLY + + except requests.exceptions.Timeout: + logger.warning("Gateway request timed out") + return SANITIZED_REPLY + except requests.exceptions.ConnectionError: + logger.warning("Gateway connection failed") + return SANITIZED_REPLY + except Exception as e: + logger.warning(f"Gateway exception: {e}") + return SANITIZED_REPLY + + # ═══════════════════════════════════════════════════ + # 插件生命周期钩子 + # ═══════════════════════════════════════════════════ -class Ai_reply(BasePlugin): def after_save(self): - config = self.config.get("siloconflow") - maxcount = self.config.get("base").get("maxcount") - ai_server = SiloconFlowAPI(base_url= config.get("base_url"), api_key= config.get("api_key"), modules= config.get("modules")) - if self.ctx.group is None: - self.ctx.user.messages.append({'role': 'user', - 'content': - f"当前用户为:{self.ctx.user.nickname}，其uid为:{self.ctx.user.user_id}，以上为背景信息，根据背景信息回复用户消息。用户消息:{self.ctx.raw_message}"}) - typing_thread = threading.Thread(target=self.ctx.user.set_input_status, args=(1,)) - typing_thread.start() - #获取ai返回值 - message = ai_server.get_ai_message(self.ctx.user.messages,maxcount) - #结束刷新状态 - self.ctx.user.signal = False - #回收线程 - typing_thread.join() - #指示器归零 - self.ctx.user.signal = True - #发送消息 - self.ctx.user.send_message(message) - else: - if "CQ:at,qq={0}".format(self.ctx.rebot_id) in self.ctx.raw_message: - self.ctx.group.messages.append({'role': 'user', - 'content': - f"群聊名称为:{self.ctx.group.nickname},用户在群中的近十条消息为:{self.ctx.group.current_user.messages},用户名叫:{self.ctx.group.current_user.nickname},以上为背景信息，根据背景信息回复消息。消息为：{self.ctx.raw_message}"}) - message = ai_server.get_ai_message(self.ctx.group.messages,maxcount) - - self.ctx.group.send_message(message) - return "ok" \ No newline at end of file + """消息保存后触发——核心处理入口""" + logger.info("=== OpenClawBridge after_save START ===") + + sender_id = self._get_sender_id() + raw_message = self.ctx.raw_message + identity_tag = self._build_source_tag() + session_key = self._build_session_key() + + # ── 过滤系统通知、文件回执 ── + if not raw_message or raw_message.startswith("[系统通知]") or raw_message.startswith("[文件回执]"): + logger.info("Skip: system notification or file receipt") + return "ok" + + # ── 过滤纯媒体消息（无文字） ── + cleaned = re.sub(r'\[CQ:[^\]]+\]', '', raw_message).strip() + if not cleaned: + logger.info("Skip: media-only message (no text after CQ removal)") + return "ok" + + # ── 管理员消息：先安全检查 ── + is_dangerous = self._is_dangerous(raw_message) + if is_dangerous: + logger.info(f"Dangerous message from {sender_id}") + if self._is_authorized(sender_id): + # 管理员触发危险词，通知但不拦截 + self._notify_admin(sender_id, + self.ctx.group.group_id if self.ctx.group else "private", + raw_message, []) + logger.info(f"Admin triggered risk word: {raw_message[:50]}") + else: + # 非管理员触发，拦截+上报+拉黑 + is_risk, matched = self._detect_high_risk(raw_message) + self._notify_admin(sender_id, + self.ctx.group.group_id if self.ctx.group else "private", + raw_message, matched) + logger.warning(f"Blocked dangerous message from {sender_id}") + return "ok" + + # ── 非管理员消息：MC指令 → 跳过（由 ops-manager 处理） ── + if not self._is_authorized(sender_id) and self._looks_like_mc_command(raw_message): + logger.info("Skip: non-admin MC command") + return "ok" + + # ── 构建上下文并转发 ── + context = self._build_context_with_history(raw_message, identity_tag) + logger.info(f"Forwarding to OpenClaw: session={session_key}, msg={context[:80]}...") + + reply = self._send_to_openclaw(context, session_key) + logger.info(f"Agent reply: {reply[:100]}...") + + # ── 自动回复 ── + if reply and reply != NO_REPLY_MARKER and reply != SANITIZED_REPLY: + try: + if self.ctx.group: + self.ctx.group.send_message(reply) + else: + self.ctx.user.send_message(reply) + except Exception as e: + logger.error(f"Failed to send reply: {e}") + + logger.info("=== OpenClawBridge after_save END ===") + return "ok" diff --git a/test.py b/test.py index 1d920e3..720721c 100644 --- a/test.py +++ b/test.py @@ -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()