From ccd00d948f3edea6c42aafbc5f2bf0db86740c3e Mon Sep 17 00:00:00 2001 From: hp0912 <809211365@qq.com> Date: Sat, 12 Sep 2026 17:26:47 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=8F=91=E9=80=81=E5=BC=95=E7=94=A8?= =?UTF-8?q?=E6=B6=88=E6=81=AF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- skills/send-complex-message/SKILL.md | 223 ++++++++++++ .../send-complex-message/scripts/bootstrap.py | 111 ++++++ .../scripts/requirements.txt | 2 + .../scripts/send_complex_message.py | 336 ++++++++++++++++++ tests/test_send_complex_message.py | 99 ++++++ 5 files changed, 771 insertions(+) create mode 100644 skills/send-complex-message/SKILL.md create mode 100644 skills/send-complex-message/scripts/bootstrap.py create mode 100644 skills/send-complex-message/scripts/requirements.txt create mode 100644 skills/send-complex-message/scripts/send_complex_message.py create mode 100644 tests/test_send_complex_message.py diff --git a/skills/send-complex-message/SKILL.md b/skills/send-complex-message/SKILL.md new file mode 100644 index 0000000..2b88c5d --- /dev/null +++ b/skills/send-complex-message/SKILL.md @@ -0,0 +1,223 @@ +--- +name: send-complex-message +description: "在当前微信会话中发送纯文本、引用回复或群聊艾特/@/提及消息。用户要求发送一段文字、仅@成员或@所有人、引用某条消息回复、引用时同时@成员时使用；文本和引用支持私聊及群聊，可在发送后结束当前 Agent 对话。" +--- + +# Send Complex Message Skill + +## 描述 + +本技能是当前微信会话中纯文本、艾特和引用回复的统一发送入口。仅艾特时不需要正文；引用消息必须有正文，也可以附带成员艾特参数。艾特支持指定一个或多个成员，也支持微信原生的 `@所有人`。 + +技能脚本位于 `scripts/send_complex_message.py`，统一调用客户端的 `/message/send/refermessage` 接口。`refer_message_id` 是可选参数：不传时由客户端调用普通文本消息方法，传入时发送引用消息。指定成员时，根据昵称或备注查询当前群内未退群成员；@所有人时使用客户端协议值 `notify@all`，不要把 `@昵称` 或 `@所有人` 当普通正文拼接。 + +本技能不带引用 ID 时通过普通文本消息实现原生艾特，支持只艾特而不附加正文。带引用 ID 时，客户端接收 `at` 并显示艾特名称，但尚未实现引用消息中的原生艾特提醒，不能把引用发送成功表述为已经提醒成员。 + +## 触发条件 + +- 用户要求将一段话分成多条发送。 +- 用户要求「引用这条消息回复」「引用我刚才的话说收到」「回复我引用的那条消息」。 +- 需要发送微信原生引用回复，而不是在普通文本中复述原文。 +- 需要艾特、@、提及某个群成员或多个群成员。 +- 用户要求「帮我艾特下 xxx」「@ 一下 xxx」「提一下 xxx 和 yyy」。 +- 用户要求「@所有人」「提醒全体成员」「通知群里所有人」。 +- 需要在群聊里点名提醒某人。 +- 其它时候不应该使用本技能 + +不带艾特的纯文本和引用回复可用于私聊和群聊；只要提供艾特参数，`ROBOT_FROM_WX_ID` 就必须是群聊 ID。 + +## 入参规范 + +`refer_message_id` 不全局必填，只在用户明确要求引用时传入，值为 `messages.id`。仅文本、仅艾特时省略该参数，不查询引用目标，也不从环境变量自动补齐引用 ID。 + +| 发送方式 | 艾特参数 | `refer_message_id` | 正文 `content` | +| ---------- | ---------------------- | ------------------ | ------------------ | +| 仅艾特 | 指定成员或 `all: true` | 不传 | 可省略或为空 | +| 仅文本 | 不传 | 不传 | 必填且不能是纯空白 | +| 仅引用 | 不传 | 传入 `messages.id` | 必填且不能是纯空白 | +| 引用并艾特 | 指定成员或 `all: true` | 传入 `messages.id` | 必填且不能是纯空白 | + +不引用时也支持正文加艾特。下方 schema 没有全局必填字段，组合校验由 schema 和脚本共同约束。 + +```json +{ + "$schema": "http://json-schema.org/draft-07/schema#", + "type": "object", + "required": [], + "properties": { + "mention": { + "type": "string", + "description": "要艾特的群成员昵称或备注。按用户原话提取，不要改写。" + }, + "mentions": { + "type": "array", + "items": { + "type": "string" + }, + "description": "要艾特的多个群成员昵称或备注。" + }, + "all": { + "type": "boolean", + "description": "是否 @所有人。用户明确要求 @所有人或通知全体成员时设为 true，不能与 mention/mentions 同时使用。" + }, + "content": { + "type": "string", + "description": "要发送的文本内容。纯文本和引用回复必须提供非空正文；只艾特不附加正文时可以省略或为空字符串。" + }, + "refer_message_id": { + "type": "integer", + "minimum": 1, + "maximum": 9223372036854775807, + "description": "可选，仅引用时传入被引用消息的数据库主键 messages.id。仅文本、仅艾特时省略；传入后 content 必须为非空白文本。" + }, + "ended": { + "type": "boolean", + "description": "是否结束当前对话。当 Agent 已经完成消息发送、要说的话已说完、要做的事已做完时，设置为 true。" + } + }, + "anyOf": [ + { + "required": ["content"], + "properties": { "content": { "pattern": "\\S" } } + }, + { + "required": ["mention"], + "properties": { "mention": { "pattern": "\\S" } } + }, + { + "required": ["mentions"], + "properties": { + "mentions": { "contains": { "type": "string", "pattern": "\\S" } } + } + }, + { + "required": ["all"], + "properties": { "all": { "const": true } } + } + ], + "dependencies": { + "refer_message_id": { + "required": ["content"], + "properties": { "content": { "minLength": 1, "pattern": "\\S" } } + } + }, + "additionalProperties": false +} +``` + +对应命令行参数： + +- `--refer-message-id ` 可选，传入后发送引用消息，必须同时提供非空 `--content` +- `--mention <昵称或备注>` 指定成员时使用，可重复传入 +- `--mentions ` 指定成员时可选，用于一次传入多个昵称或备注 +- `--all`（也支持 `--mention-all`）可选，用于真正 @所有人，不能与 mention 参数同用 +- `--content <文本内容>` 纯文本和引用消息必填，单独艾特时可省略或为空 +- `--ended` 可选标志。当 Agent 已完成消息发送、要说的话已说完时传入。 + +## 引用消息的选择 + +- 仅在用户要求引用时选择消息，传给脚本的引用 ID 只使用 `messages.id`。 +- 引用当前用户触发本次对话的消息时，使用环境变量 `ROBOT_MESSAGE_ID` 中的消息主键。 +- 用户要求回复他引用的原消息时，使用 `ROBOT_REF_MESSAGE_ID` 中的消息主键；该值为空或 `0` 表示没有引用目标，不能改为引用当前消息。 +- 引用其他历史消息时，从当前机器人数据库 `messages` 表查找，限定 `from_wxid = ROBOT_FROM_WX_ID`，按用户描述确认原消息后取 `id`。不要使用 `msg_id`、`client_msg_id` 或 XML 中的 `svrid`。 +- 引用目标不明确时先确认，不能猜测消息 ID。不要因为上下文存在引用消息就自动发送引用回复。 +- 脚本接收明确的 `--refer-message-id`，不会自动选择最近一条消息；仅引用且不指定成员时不需要查询成员表。 + +## 成员匹配规则 + +仅指定成员时执行以下匹配；`--all` 不查询成员表： + +1. 只在当前群聊 `ROBOT_FROM_WX_ID` 对应的 `chat_room_members` 记录中查找。 +2. 只匹配 `is_leaved` 为空或 `0` 的成员，已经退群的成员不能被艾特。 +3. 使用用户给出的昵称或备注做模糊查询，字段优先级为 `remark`，然后是 `nickname`。 +4. 查询到候选成员后，优先选择 `remark` 完全等于输入值的成员。 +5. 如果没有完全相等的 `remark`，选择 `nickname` 完全等于输入值的成员。 +6. 如果没有完全相等结果，选择第一个 `remark` 包含输入值的成员。 +7. 如果仍未命中，选择第一个 `nickname` 包含输入值的成员。 + +## 执行步骤 + +1. 判断用户需要纯文本、仅艾特、引用回复，还是引用时同时艾特。纯文本和引用回复必须准备非空正文 `content`；引用时按上面的规则确定 `refer_message_id`。 +2. 如需指定成员，把用户原话中的昵称或备注写入 `mention`/`mentions`；@所有人时设置 `all: true` 并使用 `--all`。 +3. 在该技能目录执行脚本，例如： + +仅艾特某人，不附加正文： + +```bash +python3 scripts/send_complex_message.py --mention '张三' --ended +``` + +仅发送文本，不艾特、不引用： + +```bash +python3 scripts/send_complex_message.py --content '收到，我来处理' --ended +``` + +引用当前用户消息回复： + +```bash +python3 scripts/send_complex_message.py --refer-message-id "$ROBOT_MESSAGE_ID" --content '收到，我来处理' --ended +``` + +回复用户引用的原消息（先确认 `ROBOT_REF_MESSAGE_ID` 大于 `0`）： + +```bash +python3 scripts/send_complex_message.py --refer-message-id "$ROBOT_REF_MESSAGE_ID" --content '同意这个安排' +``` + +引用已确认的历史消息并附带成员显示（当前客户端不产生原生艾特提醒）： + +```bash +python3 scripts/send_complex_message.py --refer-message-id 12345 --mention '张三' --content '请看一下这个' +``` + +艾特群成员并附加正文： + +```bash +python3 scripts/send_complex_message.py --mention '张三' --content '看一下这个' +``` + +用户要求 @所有人时传 `--all`，不要把“所有人”当成员昵称查询： + +```bash +python3 scripts/send_complex_message.py --all --content '请大家查看群公告' +``` + +当 Agent 认为任务已完成、对话可以结束时，加上 `--ended` 标志： + +```bash +python3 scripts/send_complex_message.py --mention '张三' --content '看一下这个' --ended +``` + +4. 指定成员时，脚本查询数据库表 `chat_room_members` 并解析微信 ID；`--all` 时跳过数据库查询，使用 `at: ["notify@all"]`。如果指定成员未命中，可以查询记忆里是否记录了对方的别称。 +5. 脚本通过 `X-Private-Token` 请求头传递环境变量 `ROBOT_CLIENT_PRIVATE_TOKEN`，统一调用 `POST http://127.0.0.1:{ROBOT_WECHAT_CLIENT_PORT}/api/v1/robot/message/send/refermessage`。请求体包含 `to_wxid`、`content`、`at`；只有引用时才添加 `refer_message_id`。不引用时，客户端直接转入普通文本发送方法；纯文本的 `at` 为空数组，仅艾特的 `content` 为空字符串。 + +`to_wxid` 固定取当前会话 `ROBOT_FROM_WX_ID`。技能在客户端内执行，无需携带管理后台的机器人实例 query `id`。 + +## 校验规则 + +- `ROBOT_FROM_WX_ID` 必须配置；带艾特时必须以 `@chatroom` 结尾。 +- 没有艾特且正文为空或纯空白时不能发送；空的成员名称、空成员数组和 `all: false` 都不算有效艾特。 +- 仅艾特时，`content` 可直接为空，由客户端根据 `at` 生成艾特正文，不添加额外话语。 +- 传入 `refer_message_id` 时，必须是 int64 范围内的正整数，引用正文不能是空字符串或纯空白；不引用时直接省略 ID，不能为满足校验而猜测或自动填入消息 ID。 +- `--all` 必须独占，不能再指定成员。 +- 每个要艾特的人都必须能在当前群内匹配到未退群成员。 +- 如果同一个微信 ID 被多个昵称命中，只会艾特一次。 + +## 依赖安装 + +- 指定成员、需要查询数据库时，脚本会自动创建虚拟环境并安装依赖；纯文本、仅引用或 `--all` 不需要安装数据库依赖。 +- 如需手动重新安装，可执行：`python3 scripts/bootstrap.py` + +## ended 行为 + +- 当 `--ended` 传入且客户端返回业务状态 `code: 200` 时，脚本在正常输出末尾追加打印独立一行 `ended`。 +- `ended` 字符串必须位于输出的最末尾，前面不能跟其他字符。 +- Agent 检测到输出以 `ended` 结尾时，会自动退出 Agent 循环。 + +## 回复要求 + +- 成功时，脚本输出「文本消息发送成功」「引用消息发送成功」「艾特消息发送成功」或「艾特所有人消息发送成功」，表示消息已通过客户端接口直接发送，不要再重复发送正文。 +- HTTP 200 不等于业务成功，必须检查响应中的 `code`。当前引用接口成功时可能返回 `data: null`，不能因没有消息对象重发。 +- 如果传入 `--ended`，输出末尾会追加 `ended`，Agent 会自动结束对话。 +- 失败时，返回脚本输出的具体错误信息，不输出 `ended`。引用消息不存在、原消息内容不完整或原发送人不存在时，按客户端错误说明原因；请求超时或发送结果不确定时，不自动重试，以免重复发送。 diff --git a/skills/send-complex-message/scripts/bootstrap.py b/skills/send-complex-message/scripts/bootstrap.py new file mode 100644 index 0000000..45005f7 --- /dev/null +++ b/skills/send-complex-message/scripts/bootstrap.py @@ -0,0 +1,111 @@ +#!/usr/bin/env python3 + +from __future__ import annotations + +import hashlib +import subprocess +import sys +import traceback +from pathlib import Path + +sys.stderr = sys.stdout + + +def _skill_root_from(script_dir: Path) -> Path: + return script_dir.parent + + +def _venv_dir(script_dir: Path) -> Path: + return _skill_root_from(script_dir) / ".venv" + + +def _venv_python(venv_dir: Path) -> Path: + if sys.platform == "win32": + return venv_dir / "Scripts" / "python.exe" + return venv_dir / "bin" / "python" + + +def _stamp_file(venv_dir: Path) -> Path: + return venv_dir / ".req_hash" + + +def _file_hash(path: Path) -> str: + return hashlib.sha256(path.read_bytes()).hexdigest() + + +def _deps_up_to_date(requirements_file: Path, venv_dir: Path) -> bool: + stamp = _stamp_file(venv_dir) + if not stamp.is_file(): + return False + return stamp.read_text().strip() == _file_hash(requirements_file) + + +def _write_stamp(requirements_file: Path, venv_dir: Path) -> None: + _stamp_file(venv_dir).write_text(_file_hash(requirements_file)) + + +def _ensure_venv(venv_dir: Path, venv_python: Path) -> int: + if venv_python.is_file(): + return 0 + + sys.stdout.write(f"未检测到技能虚拟环境，正在创建: {venv_dir}\n") + import shutil + py = sys.executable or next( + (shutil.which(candidate) for candidate in ("python3", "python") if shutil.which(candidate)), + None, + ) + if not py: + raise RuntimeError("无法找到 Python 解释器路径") + + try: + subprocess.run([py, "-m", "venv", str(venv_dir)], check=True, stdout=sys.stdout, stderr=sys.stdout) + except subprocess.CalledProcessError as exc: + sys.stdout.write(f"创建虚拟环境失败，退出码: {exc.returncode}\n") + return exc.returncode or 1 + + return 0 + + +def main() -> int: + script_dir = Path(__file__).resolve().parent + requirements_file = script_dir / "requirements.txt" + venv_dir = _venv_dir(script_dir) + venv_python = _venv_python(venv_dir) + + if not requirements_file.is_file(): + sys.stdout.write(f"未找到依赖文件: {requirements_file}\n") + return 1 + + ensure_result = _ensure_venv(venv_dir, venv_python) + if ensure_result != 0: + return ensure_result + + if _deps_up_to_date(requirements_file, venv_dir): + sys.stdout.write("依赖已是最新，跳过安装\n") + return 0 + + try: + subprocess.run([str(venv_python), "-m", "pip", "install", "--upgrade", "pip"], check=True, stdout=sys.stdout, stderr=sys.stdout) + except subprocess.CalledProcessError as exc: + sys.stdout.write(f"升级 pip 失败，退出码: {exc.returncode}\n") + return exc.returncode or 1 + + try: + subprocess.run([str(venv_python), "-m", "pip", "install", "-r", str(requirements_file)], check=True, stdout=sys.stdout, stderr=sys.stdout) + except subprocess.CalledProcessError as exc: + sys.stdout.write(f"安装依赖失败，退出码: {exc.returncode}\n") + return exc.returncode or 1 + + _write_stamp(requirements_file, venv_dir) + sys.stdout.write(f"依赖安装完成，当前技能虚拟环境: {venv_dir}\n") + return 0 + + +if __name__ == "__main__": + try: + raise SystemExit(main()) + except SystemExit: + raise + except Exception: + traceback.print_exc(file=sys.stdout) + raise SystemExit(1) \ No newline at end of file diff --git a/skills/send-complex-message/scripts/requirements.txt b/skills/send-complex-message/scripts/requirements.txt new file mode 100644 index 0000000..35f2cf7 --- /dev/null +++ b/skills/send-complex-message/scripts/requirements.txt @@ -0,0 +1,2 @@ +cryptography +pymysql>=1.1,<2 \ No newline at end of file diff --git a/skills/send-complex-message/scripts/send_complex_message.py b/skills/send-complex-message/scripts/send_complex_message.py new file mode 100644 index 0000000..7bb602a --- /dev/null +++ b/skills/send-complex-message/scripts/send_complex_message.py @@ -0,0 +1,336 @@ +#!/usr/bin/env python3 + +from __future__ import annotations + +import argparse +import json +import os +import subprocess +import sys +import traceback +import urllib.request +from pathlib import Path + +sys.stderr = sys.stdout + + +def _client_private_token() -> str: + return os.environ.get("ROBOT_CLIENT_PRIVATE_TOKEN", "").strip() + + +def _skill_root() -> Path: + return Path(__file__).resolve().parent.parent + + +def _skill_venv_python() -> Path: + venv_dir = _skill_root() / ".venv" + if sys.platform == "win32": + return venv_dir / "Scripts" / "python.exe" + return venv_dir / "bin" / "python" + + +def _get_python_executable() -> str: + if sys.executable: + return sys.executable + import shutil + for candidate in ("python3", "python"): + found = shutil.which(candidate) + if found: + return found + raise RuntimeError("无法找到 Python 解释器路径") + + +def _run_bootstrap() -> None: + bootstrap = Path(__file__).resolve().parent / "bootstrap.py" + result = subprocess.run([_get_python_executable(), str(bootstrap)]) + if result.returncode != 0: + raise SystemExit(result.returncode) + + +def _ensure_skill_venv_python() -> None: + venv_python = _skill_venv_python() + if not venv_python.is_file(): + _run_bootstrap() + venv_python = _skill_venv_python() + if not venv_python.is_file(): + sys.stdout.write("bootstrap 后仍未找到虚拟环境\n") + raise SystemExit(1) + + venv_dir = _skill_root() / ".venv" + if Path(sys.prefix) == venv_dir.resolve(): + return + + os.execv(str(venv_python), [str(venv_python), str(Path(__file__).resolve()), *sys.argv[1:]]) + + +def _mysql_connect(): + _ensure_skill_venv_python() + try: + import pymysql # type: ignore + except ModuleNotFoundError: + _run_bootstrap() + venv_python = _skill_venv_python() + os.execv(str(venv_python), [str(venv_python), str(Path(__file__).resolve()), *sys.argv[1:]]) + + host = os.environ.get("MYSQL_HOST", "127.0.0.1") + port = int(os.environ.get("MYSQL_PORT", "3306")) + user = os.environ.get("MYSQL_USER", "root") + password = os.environ.get("MYSQL_PASSWORD", "") + database = os.environ.get("ROBOT_CODE", "") + if not database: + raise RuntimeError("环境变量 ROBOT_CODE 未配置") + + return pymysql.connect( + host=host, + port=port, + user=user, + password=password, + database=database, + charset="utf8mb4", + connect_timeout=10, + read_timeout=30, + cursorclass=pymysql.cursors.DictCursor, + ) + + +def _http_post_json(url: str, body: dict, timeout: int = 300) -> dict: + data = json.dumps(body).encode("utf-8") + req = urllib.request.Request( + url, + data=data, + headers={ + "Content-Type": "application/json", + "X-Private-Token": _client_private_token(), + }, + method="POST", + ) + with urllib.request.urlopen(req, timeout=timeout) as resp: + text = resp.read().decode("utf-8") + if not text.strip(): + return {} + return json.loads(text) + + +def _expand_json_array_values(values: list[str], label: str) -> list[str]: + expanded: list[str] = [] + for value in values: + stripped = value.strip() + if not stripped: + continue + if stripped.startswith("["): + parsed = json.loads(stripped) + if not isinstance(parsed, list): + raise ValueError(f"{label} 必须是字符串数组") + for item in parsed: + if not isinstance(item, str): + raise ValueError(f"{label} 必须是字符串数组") + if item.strip(): + expanded.append(item.strip()) + continue + expanded.append(stripped) + return expanded + + +def _parse_cli_params(argv: list[str]) -> tuple[list[str], str, bool, bool, int | None]: + parser = argparse.ArgumentParser(add_help=False) + parser.add_argument("--mention", action="append", default=[]) + parser.add_argument("--mentions", action="append", default=[]) + parser.add_argument("--all", "--mention-all", dest="mention_all", action="store_true") + parser.add_argument("--refer-message-id", type=int) + parser.add_argument("--content", default="") + parser.add_argument("--ended", action="store_true", default=False) + + namespace, unknown = parser.parse_known_args(argv) + if unknown: + raise ValueError(f"存在不支持的参数: {' '.join(unknown)}") + + mentions = _expand_json_array_values(namespace.mention + namespace.mentions, "mentions") + deduped: list[str] = [] + seen = set() + for mention in mentions: + key = mention.casefold() + if key not in seen: + seen.add(key) + deduped.append(mention) + + if namespace.mention_all and deduped: + raise ValueError("all 不能和 mention 或 mentions 同时使用") + + if namespace.refer_message_id is not None: + if not 0 < namespace.refer_message_id <= 2**63 - 1: + raise ValueError("refer_message_id 必须是正整数，且不能超过 int64 范围") + if not namespace.content.strip(): + raise ValueError("发送引用消息必须提供非空 content") + + if not deduped and not namespace.mention_all and not namespace.content.strip(): + raise ValueError("请提供非空 content，或指定要艾特的成员/--all") + + return deduped, namespace.content, namespace.ended, namespace.mention_all, namespace.refer_message_id + + +def _escape_like(value: str) -> str: + return value.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_") + + +def _normalize(value) -> str: + if value is None: + return "" + return str(value).strip() + + +def _find_member(conn, chat_room_id: str, mention: str) -> dict | None: + keyword = mention.strip() + like_keyword = f"%{_escape_like(keyword)}%" + sql = """ + SELECT wechat_id, remark, nickname + FROM chat_room_members + WHERE chat_room_id = %s + AND (is_leaved IS NULL OR is_leaved = 0) + AND ( + (remark IS NOT NULL AND remark LIKE %s ESCAPE '\\\\') + OR (nickname IS NOT NULL AND nickname LIKE %s ESCAPE '\\\\') + ) + ORDER BY id ASC + LIMIT 50 + """ + + with conn.cursor() as cursor: + cursor.execute(sql, (chat_room_id, like_keyword, like_keyword)) + candidates = list(cursor.fetchall()) + + keyword_folded = keyword.casefold() + + for field in ("remark", "nickname"): + for candidate in candidates: + if _normalize(candidate.get(field)).casefold() == keyword_folded: + return candidate + + for field in ("remark", "nickname"): + for candidate in candidates: + value = _normalize(candidate.get(field)).casefold() + if keyword_folded in value: + return candidate + + return None + + +def _resolve_mentions(conn, chat_room_id: str, mentions: list[str]) -> tuple[list[str], list[str]]: + at_wechat_ids: list[str] = [] + seen = set() + missing: list[str] = [] + + for mention in mentions: + member = _find_member(conn, chat_room_id, mention) + if not member: + missing.append(mention) + continue + + wechat_id = _normalize(member.get("wechat_id")) + if wechat_id and wechat_id not in seen: + seen.add(wechat_id) + at_wechat_ids.append(wechat_id) + + return at_wechat_ids, missing + + +def _send_message( + client_port: str, + to_wxid: str, + content: str, + at_wechat_ids: list[str], + refer_message_id: int | None, +) -> None: + # 未传引用 ID 时，客户端会调用普通文本发送方法，并保留原生艾特功能。 + send_url = f"http://127.0.0.1:{client_port}/api/v1/robot/message/send/refermessage" + body = { + "to_wxid": to_wxid, + "content": content, + "at": at_wechat_ids, + } + if refer_message_id is not None: + body["refer_message_id"] = refer_message_id + result = _http_post_json(send_url, body) + if not isinstance(result, dict) or result.get("code") != 200: + message = result.get("message") if isinstance(result, dict) else None + raise RuntimeError(message or "客户端未返回成功状态") + + +def main() -> int: + try: + mentions, content, ended, mention_all, refer_message_id = _parse_cli_params(sys.argv[1:]) + except (ValueError, json.JSONDecodeError) as exc: + sys.stdout.write(f"参数格式错误: {exc}\n") + return 1 + + to_wxid = os.environ.get("ROBOT_FROM_WX_ID", "").strip() + if not to_wxid: + sys.stdout.write("环境变量 ROBOT_FROM_WX_ID 未配置\n") + return 1 + if (mention_all or mentions) and not to_wxid.endswith("@chatroom"): + sys.stdout.write("当前会话不是群聊，不能发送艾特消息\n") + return 1 + + client_port = os.environ.get("ROBOT_WECHAT_CLIENT_PORT", "").strip() + if not client_port: + sys.stdout.write("环境变量 ROBOT_WECHAT_CLIENT_PORT 未配置\n") + return 1 + + at_wechat_ids: list[str] = [] + if mention_all: + at_wechat_ids = ["notify@all"] + elif mentions: + try: + conn = _mysql_connect() + except Exception as exc: + sys.stdout.write(f"数据库连接失败: {exc}\n") + return 1 + + try: + at_wechat_ids, missing = _resolve_mentions(conn, to_wxid, mentions) + except Exception as exc: + sys.stdout.write(f"查询群成员失败: {exc}\n") + return 1 + finally: + try: + conn.close() + except Exception: + pass + + if missing: + sys.stdout.write(f"未找到当前群内未退群成员: {', '.join(missing)}\n") + return 1 + if not at_wechat_ids: + sys.stdout.write("未找到可艾特的群成员\n") + return 1 + + try: + _send_message(client_port, to_wxid, content, at_wechat_ids, refer_message_id) + if refer_message_id is not None: + sys.stdout.write("引用消息发送成功\n") + if at_wechat_ids: + sys.stdout.write("引用中的艾特仅作显示，当前客户端尚未实现原生艾特提醒\n") + elif at_wechat_ids: + sys.stdout.write("艾特所有人消息发送成功\n" if mention_all else "艾特消息发送成功\n") + else: + sys.stdout.write("文本消息发送成功\n") + if ended: + sys.stdout.write("ended") + return 0 + except Exception as exc: + sys.stdout.write(f"消息发送失败: {exc}\n") + return 1 + + +if __name__ == "__main__": + try: + exit_code = main() + if exit_code == 0: + # ended may have already been printed above in the success path. + # If main() returned non-zero, ended is not printed. + pass + raise SystemExit(exit_code) + except SystemExit: + raise + except Exception: + traceback.print_exc(file=sys.stdout) + raise SystemExit(1) diff --git a/tests/test_send_complex_message.py b/tests/test_send_complex_message.py new file mode 100644 index 0000000..cdd67cc --- /dev/null +++ b/tests/test_send_complex_message.py @@ -0,0 +1,99 @@ +from __future__ import annotations + +import contextlib +import importlib.util +import io +import os +import sys +import unittest +from pathlib import Path +from unittest import mock + + +SCRIPT_PATH = ( + Path(__file__).resolve().parents[1] + / "skills/send-complex-message/scripts/send_complex_message.py" +) + + +class SendComplexMessageTests(unittest.TestCase): + @classmethod + def setUpClass(cls) -> None: + spec = importlib.util.spec_from_file_location("send_complex_message", SCRIPT_PATH) + if spec is None or spec.loader is None: + raise RuntimeError(f"无法加载测试脚本：{SCRIPT_PATH}") + cls.module = importlib.util.module_from_spec(spec) + with mock.patch.object(sys, "stderr", sys.stderr): + spec.loader.exec_module(cls.module) + + def setUp(self) -> None: + env = mock.patch.dict(os.environ, { + "ROBOT_FROM_WX_ID": "room@chatroom", + "ROBOT_WECHAT_CLIENT_PORT": "9000", + # 已有上下文消息也不能让不引用的请求自动带上引用 ID。 + "ROBOT_MESSAGE_ID": "123", + "ROBOT_REF_MESSAGE_ID": "456", + }, clear=True) + env.start() + self.addCleanup(env.stop) + + def test_sending_modes_and_optional_reference(self) -> None: + cases = [ + ("mention only", ["--mention", "张三"], "", ["wxid_zhangsan"], None), + ("text only", ["--content", "收到"], "收到", [], None), + ("quote only", ["--refer-message-id", "12", "--content", "收到"], "收到", [], 12), + ("quote and mention", ["--refer-message-id", "12", "--mention", "张三", "--content", "收到"], "收到", ["wxid_zhangsan"], 12), + ("mention all only", ["--all"], "", ["notify@all"], None), + ("text and mention", ["--mention", "张三", "--content", "收到"], "收到", ["wxid_zhangsan"], None), + ("large primary ID", ["--refer-message-id", "9007199254740993", "--content", "收到"], "收到", [], 9007199254740993), + ] + for name, args, content, at, reference in cases: + with self.subTest(name=name), contextlib.ExitStack() as stack: + stack.enter_context(mock.patch.object(sys, "argv", [str(SCRIPT_PATH), *args, "--ended"])) + stack.enter_context(contextlib.redirect_stdout(io.StringIO())) + connect = stack.enter_context(mock.patch.object(self.module, "_mysql_connect")) + resolve = stack.enter_context(mock.patch.object(self.module, "_resolve_mentions", return_value=(["wxid_zhangsan"], []))) + post = stack.enter_context(mock.patch.object(self.module, "_http_post_json", return_value={"code": 200, "data": None})) + + self.assertEqual(self.module.main(), 0) + body = {"to_wxid": "room@chatroom", "content": content, "at": at} + if reference is not None: + body["refer_message_id"] = reference + post.assert_called_once_with( + "http://127.0.0.1:9000/api/v1/robot/message/send/refermessage", + body, + ) + if "--mention" in args: + resolve.assert_called_once_with(connect.return_value, "room@chatroom", ["张三"]) + connect.return_value.close.assert_called_once() + else: + connect.assert_not_called() + resolve.assert_not_called() + + def test_invalid_combinations_do_not_send(self) -> None: + cases = [ + [], + ["--content", " \t\n"], + ["--mentions", "[]"], + ["--refer-message-id", "12"], + ["--refer-message-id", "12", "--mention", "张三"], + ["--refer-message-id", "12", "--content", " \t\n", "--all"], + ["--refer-message-id", "0", "--content", "收到"], + ["--refer-message-id", "-1", "--content", "收到"], + ["--refer-message-id", "9223372036854775808", "--content", "收到"], + ] + for args in cases: + with self.subTest(args=args), contextlib.ExitStack() as stack: + output = stack.enter_context(contextlib.redirect_stdout(io.StringIO())) + stack.enter_context(mock.patch.object(sys, "argv", [str(SCRIPT_PATH), *args, "--ended"])) + connect = stack.enter_context(mock.patch.object(self.module, "_mysql_connect")) + post = stack.enter_context(mock.patch.object(self.module, "_http_post_json")) + + self.assertEqual(self.module.main(), 1) + connect.assert_not_called() + post.assert_not_called() + self.assertFalse(output.getvalue().endswith("ended")) + + +if __name__ == "__main__": + unittest.main()