From 91b70ee2ce82f3e21f35fc5b86e4f266593cda28 Mon Sep 17 00:00:00 2001 From: jxxghp Date: Thu, 15 Oct 2026 09:02:59 +0800 Subject: [PATCH] =?UTF-8?q?fix(wechatclawbot):=20=E8=AF=86=E5=88=AB=20item?= =?UTF-8?q?=5Flist=20=E5=AA=92=E4=BD=93=E5=85=A5=E7=AB=99=E6=B6=88?= =?UTF-8?q?=E6=81=AF=E5=B9=B6=E4=BF=AE=E6=AD=A3=20CDN=20=E6=8E=A2=E6=B5=8B?= =?UTF-8?q?=E5=88=A4=E5=AE=9A=20(#7000)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(wechatclawbot): 识别 item_list 媒体入站消息并修正 CDN 探测判定 (#6992) iLink 下行媒体报文把附件放在 item_list 子项中，原解析存在三处缺口， 图片/语音/文件入站消息会被判为空消息丢弃： - item_list 仅按数字 type 处理 1 文本/2 图片/5、6 文件，缺少 3 语音/ 4 文件/5 视频分派；改为按实际出现的子项键名分派，同时兼容历史 type=5/6 文件报文。 - 附件引用只读取顶层 url/aeskey，识别不了 image_item.media.full_url / media.aes_key 等嵌套字段；统一按“顶层优先、media 子对象兜底”提取。 - 消息被解析为空时直接返回 None，日志只剩“未识别到有效消息”；补充 渠道级诊断日志，输出 message_id 与截断后的原始报文。 另外把“微信 ClawBot 媒体 CDN”网络探测的 400 纳入成功状态码：该 CDN 对无参数裸请求固定返回 400，此前探测项恒定误报失败。 tests/test_wechatclawbot.py 整文件迁移为 pytest 原生，并新增 item_list 媒体解析、仅图片消息可入链、空消息诊断日志回归；nettest 契约测试补充 CDN 400 探测用例。复杂度与 mypy 低水位随债务下降同步固化。 * fix(wechatclawbot): 同步架构清单中的 mypy 债务数字 (#6992) --- app/application/nettest/catalogue.py | 4 +- app/modules/wechatclawbot/__init__.py | 9 + app/modules/wechatclawbot/wechatclawbot.py | 128 +-- docs/refactor/optimization-checklist.md | 2 +- .../architecture/complexity-baseline.json | 2 +- .../fixtures/architecture/mypy-baseline.json | 3 +- tests/test_nettest_probe_contract.py | 19 + tests/test_wechatclawbot.py | 774 ++++++++++-------- 8 files changed, 560 insertions(+), 381 deletions(-) diff --git a/app/application/nettest/catalogue.py b/app/application/nettest/catalogue.py index 768c1b364..302af1b6a 100644 --- a/app/application/nettest/catalogue.py +++ b/app/application/nettest/catalogue.py @@ -308,7 +308,9 @@ def _build_notification_module_rules() -> list[NetworkTestRule]: proxy=True, allowed_redirect_prefixes=("https://novac2c.cdn.weixin.qq.com/",), module_ids=("WechatClawBotModule",), - success_status_codes=(200, 403, 404), + # CDN 对不带下载参数的裸请求固定返回 400，能返回 400 即证明连通正常， + # 不能据此判定探测失败。 + success_status_codes=(200, 400, 403, 404), ), ] diff --git a/app/modules/wechatclawbot/__init__.py b/app/modules/wechatclawbot/__init__.py index 774caaf74..413de54c8 100644 --- a/app/modules/wechatclawbot/__init__.py +++ b/app/modules/wechatclawbot/__init__.py @@ -292,6 +292,15 @@ class WechatClawBotModule(_MessageChannelModuleBase[WechatClawBot]): audio_refs = self._normalize_audio_refs(message.get("audio_refs")) files = self._normalize_files(message.get("files")) if not text and not images and not audio_refs and not files: + # 渠道报文被解析为空消息时必须留下诊断信息，否则只能在消息入口看到 + # “未识别到有效消息”，无法定位是哪种入站内容被丢弃。 + logger.info( + "微信 ClawBot 消息未识别到有效内容：source=%s, userid=%s, message_id=%s, body=%s", + client_config.name, + user_id, + message_id, + WechatClawBot._short_text(message, max_len=500), + ) return None if self._is_duplicate_message(client_config.name, message_id): logger.info( diff --git a/app/modules/wechatclawbot/wechatclawbot.py b/app/modules/wechatclawbot/wechatclawbot.py index 8b8f5e793..5020f945c 100644 --- a/app/modules/wechatclawbot/wechatclawbot.py +++ b/app/modules/wechatclawbot/wechatclawbot.py @@ -10,7 +10,7 @@ import time import uuid from dataclasses import dataclass, field from pathlib import Path -from typing import Any, Dict, List, Optional, Tuple +from typing import Any, Callable, Dict, List, Optional, Tuple from urllib.parse import quote from Crypto.Cipher import AES @@ -975,30 +975,36 @@ class ILinkClient: attachment: Dict[str, Any], default_name: Optional[str] = None, ) -> Optional[Tuple[str, Dict[str, Any]]]: - """从附件信息构建 wxclaw 协议引用。""" + """ + 从附件信息构建 wxclaw 协议引用。 + + 下行报文可能把下载地址与密钥放在附件的 `media` 子对象中 + （如 `image_item.media.full_url` / `media.aes_key`），也可能直接放在附件顶层， + 这里统一按“附件顶层优先、media 子对象兜底”的顺序提取字段。 + """ if not isinstance(attachment, dict): return None - download_url = ( - attachment.get("download_url") - or attachment.get("url") - or attachment.get("cdnurl") - or attachment.get("cdn_url") - or attachment.get("file_url") + media = attachment.get("media") + media = media if isinstance(media, dict) else {} + url_keys = ["full_url", "download_url", "url", "cdnurl", "cdn_url", "file_url"] + download_url = self._pick_value(attachment, url_keys) or self._pick_value( + media, url_keys ) if not download_url: return None + aeskey_keys = ["aeskey", "aes_key", "encoding_aes_key", "encrypt_key"] payload = { "url": download_url, - "aeskey": attachment.get("aeskey") - or attachment.get("encoding_aes_key") - or attachment.get("encrypt_key"), - "encrypt_type": attachment.get("encrypt_type"), - "mime_type": attachment.get("mime_type") - or attachment.get("content_type"), - "name": attachment.get("name") - or attachment.get("filename") + "aeskey": self._pick_value(attachment, aeskey_keys) + or self._pick_value(media, aeskey_keys), + "encrypt_type": self._pick_value(attachment, ["encrypt_type"]) + or self._pick_value(media, ["encrypt_type"]), + "mime_type": self._pick_value(attachment, ["mime_type", "content_type"]) + or self._pick_value(media, ["mime_type", "content_type"]), + "name": self._pick_value(attachment, ["name", "filename", "file_name"]) or default_name, - "size": attachment.get("size") or attachment.get("file_size"), + "size": self._pick_value(attachment, ["size", "file_size"]) + or self._pick_value(media, ["size", "file_size"]), } ref = self._encode_ref_payload(kind=kind, payload=payload) return ref, payload @@ -1012,6 +1018,58 @@ class ILinkClient: return None return value + def _append_item_list_entry( + self, + one: Dict[str, Any], + append_text: Callable[[Any], None], + images: List[Dict[str, Any]], + audio_refs: List[str], + files: List[Dict[str, Any]], + ) -> None: + """ + 把 item_list 的单个子项分派为文本或媒体附件。 + + iLink 下行报文的 item_list 存在两套类型约定：数字 `type` 与子项键名。 + 这里按实际出现的子项键名分派，兼容公开资料的 `2 图片/3 语音/4 文件/5 视频` + 以及历史实现里 `file_item` 使用 `type=5/6` 的报文，避免媒体消息被静默丢弃。 + """ + candidates = ( + ("text_item", "text", None), + ("image_item", "image", None), + ("voice_item", "voice", "voice.amr"), + ("file_item", "file", None), + ("video_item", "file", "video.mp4"), + ) + for key, kind, default_name in candidates: + child = one.get(key) + if not isinstance(child, dict): + continue + if kind == "text": + append_text(child.get("text")) + return + if kind == "voice": + append_text(child.get("content")) + attachment_ref = self._build_attachment_ref( + kind, child, default_name=default_name + ) + if not attachment_ref: + return + ref, payload = attachment_ref + if kind == "voice": + audio_refs.append(ref) + return + entry = { + "ref": ref, + "name": payload.get("name"), + "mime_type": payload.get("mime_type"), + "size": payload.get("size"), + } + if kind == "image": + images.append(entry) + else: + files.append(entry) + return + def _build_poll_result( self, success: bool, @@ -1228,41 +1286,7 @@ class ILinkClient: for one in item_list: if not isinstance(one, dict): continue - item_type = one.get("type") - if item_type == 1 and isinstance(one.get("text_item"), dict): - append_text((one.get("text_item") or {}).get("text")) - elif item_type == 2 and isinstance(one.get("image_item"), dict): - image_ref = self._build_attachment_ref( - "image", (one.get("image_item") or {}).get("media") or {} - ) - if image_ref: - ref, payload = image_ref - images.append( - { - "ref": ref, - "name": payload.get("name"), - "mime_type": payload.get("mime_type"), - "size": payload.get("size"), - } - ) - elif item_type in {5, 6}: - file_ref = self._build_attachment_ref( - "file", (one.get("file_item") or {}).get("media") or {} - ) - if file_ref: - ref, payload = file_ref - files.append( - { - "ref": ref, - "name": (one.get("file_item") or {}).get("name") - or (one.get("file_item") or {}).get("file_name") - or payload.get("name"), - "mime_type": (one.get("file_item") or {}).get("mime_type") - or payload.get("mime_type"), - "size": (one.get("file_item") or {}).get("size") - or payload.get("size"), - } - ) + self._append_item_list_entry(one, append_text, images, audio_refs, files) if not text_parts: append_text( diff --git a/docs/refactor/optimization-checklist.md b/docs/refactor/optimization-checklist.md index 25c517794..fc773cfcd 100644 --- a/docs/refactor/optimization-checklist.md +++ b/docs/refactor/optimization-checklist.md @@ -108,7 +108,7 @@ ARCH-201 至 ARCH-204 均达到实现、验证、提交、推送和远端门禁 | Event Contract | 53 | 均已有 payload model，但当前全部是 diagnostic enforcement | | Python 源码量 | 305,884 行 | 排除 `app/plugins/**`；61 个文件超过 1,000 行，11 个超过 2,000 行 | | 长方法 | 290 个超过 80 行 | AST 统计排除 `app/plugins/**`；65 个超过 150 行，21 个超过 250 行 | -| 全量 mypy 历史债务 | 9,163 / 508 文件 | 通知排版补齐 Slack 消息块类型，历史债务减少 2 项；工作流资源过滤改用领域对象调用过滤链，恢复的节点输出在消费前归一化，并收紧可空资源列表的类型低水位；消息媒体交互类型债务已清零，canonical Facade 与 endpoint 类型边界已补齐，低水位只允许继续下降 | +| 全量 mypy 历史债务 | 9,159 / 508 文件 | 通知排版补齐 Slack 消息块类型，历史债务减少 2 项；工作流资源过滤改用领域对象调用过滤链，恢复的节点输出在消费前归一化，并收紧可空资源列表的类型低水位；消息媒体交互类型债务已清零，canonical Facade 与 endpoint 类型边界已补齐，低水位只允许继续下降；微信 ClawBot 入站 item_list 媒体解析改为按子项键名分派后，类型错误减少 4 项 | | Ruff 历史诊断 | 100 | 低水位门禁通过，但规则集只覆盖 `E4/E7/E9/F/I` | | 覆盖率固定基线 | Application 80.00%，Domain 80.00% | Chain、Runtime、Agent、Adapter、Startup 未进入包级覆盖率门禁 | diff --git a/tests/fixtures/architecture/complexity-baseline.json b/tests/fixtures/architecture/complexity-baseline.json index 0f8ac8211..859b2b64a 100644 --- a/tests/fixtures/architecture/complexity-baseline.json +++ b/tests/fixtures/architecture/complexity-baseline.json @@ -110,7 +110,7 @@ "app/modules/trimemedia/api.py:Api.request": 16, "app/modules/ugreen/ugreen.py:Ugreen.get_tv_episodes": 21, "app/modules/wechat/__init__.py:WechatModule.message_parser": 20, - "app/modules/wechatclawbot/wechatclawbot.py:ILinkClient._parse_incoming": 34, + "app/modules/wechatclawbot/wechatclawbot.py:ILinkClient._parse_incoming": 29, "app/modules/zspace/zspace.py:ZSpace.get_items": 16, "app/modules/zspace/zspace.py:ZSpace.get_tv_episodes": 18, "app/modules/zspace/zspace.py:ZSpace.get_webhook_message": 17, diff --git a/tests/fixtures/architecture/mypy-baseline.json b/tests/fixtures/architecture/mypy-baseline.json index 10e38009a..80c307d78 100644 --- a/tests/fixtures/architecture/mypy-baseline.json +++ b/tests/fixtures/architecture/mypy-baseline.json @@ -2352,8 +2352,7 @@ }, "app/modules/wechatclawbot/wechatclawbot.py": { "arg-type": 2, - "assignment": 2, - "comparison-overlap": 3, + "assignment": 1, "no-any-return": 3, "no-untyped-def": 4, "type-arg": 2, diff --git a/tests/test_nettest_probe_contract.py b/tests/test_nettest_probe_contract.py index c0761d3ad..17f636299 100644 --- a/tests/test_nettest_probe_contract.py +++ b/tests/test_nettest_probe_contract.py @@ -121,6 +121,25 @@ def test_network_proxy_probe_uses_configured_github_identity(github_headers) -> assert "github-token" not in public_target.address +def test_wechatclawbot_cdn_probe_accepts_parameterless_400() -> None: + """微信 ClawBot 媒体 CDN 对无参数裸请求返回 400，探针应视为连通正常。""" + captured = {} + + async def respond(method, url, **options): + """模拟 CDN 对裸地址请求返回 400 空响应。""" + captured.update(method=method, url=url, **options) + return SimpleNamespace(status_code=400, headers={}, text="", aclose=AsyncMock()) + + transport = SimpleNamespace(request=AsyncMock(side_effect=respond)) + result = asyncio.run( + _network_test_service(transport).execute(target_id="wechatclawbot_cdn") + ) + + assert result.success + assert captured["method"] == "GET" + assert captured["url"] == "https://novac2c.cdn.weixin.qq.com/c2c" + + def test_tmdb_web_probe_identifies_as_moviepilot() -> None: """TMDB 网站探针应使用程序自身 UA，浏览器 UA 会被站点人机校验页拦成 403。""" captured = {} diff --git a/tests/test_wechatclawbot.py b/tests/test_wechatclawbot.py index 076e635a2..c3b32d989 100644 --- a/tests/test_wechatclawbot.py +++ b/tests/test_wechatclawbot.py @@ -1,95 +1,155 @@ import json import tempfile -import unittest from types import SimpleNamespace from unittest.mock import MagicMock, patch from app.modules.wechatclawbot import WechatClawBotModule from app.modules.wechatclawbot.wechatclawbot import ILinkClient, WechatClawBot +CLIENT_BASE_URL = "https://ilinkai.weixin.qq.com" -class WechatClawBotTest(unittest.TestCase): - def test_ilink_parse_incoming_uses_seq_as_message_id_fallback(self): - client = ILinkClient(base_url="https://ilinkai.weixin.qq.com") - message = client._parse_incoming( - { - "seq": 123456, - "from_user_id": "wxid_user_1", - "item_list": [{"type": 1, "text_item": {"text": "你好"}}], - } - ) +def _create_client(**kwargs) -> ILinkClient: + """构造指向 iLink 服务端的客户端实例。""" + return ILinkClient(base_url=CLIENT_BASE_URL, **kwargs) - self.assertIsNotNone(message) - self.assertEqual(message.message_id, "123456") - self.assertEqual(message.text, "你好") - def test_wechatclawbot_message_parser_deduplicates_message_id(self): - module = WechatClawBotModule() - body = json.dumps( - { - "__channel__": "wechatclawbot", - "userid": "wxid_user_1", - "username": "tester", - "message_id": "msg-1001", - "text": "刷新订阅", - } - ) +def _empty_state() -> dict: + """构造未登录的微信 ClawBot 持久化状态。""" + return { + "bot_token": None, + "account_id": None, + "sync_buf": None, + "qrcode": {}, + "known_targets": {}, + "user_context_tokens": {}, + "base_url": CLIENT_BASE_URL, + } - with patch.object( - module, - "get_config", - return_value=SimpleNamespace(name="wechatclawbot-test", config={}), - ): - first = module.message_parser( - source="wechatclawbot-test", - body=body, - form={}, - args={}, - ) - second = module.message_parser( - source="wechatclawbot-test", - body=body, - form={}, - args={}, - ) - self.assertIsNotNone(first) - self.assertEqual(first.message_id, "msg-1001") - self.assertIsNone(second) +def _create_bot() -> WechatClawBot: + """构造不自动轮询的微信 ClawBot 实例。""" + with patch.object(WechatClawBot, "_load_state", return_value=_empty_state()): + return WechatClawBot(name="wechatclawbot-test", auto_start_polling=False) - def test_ilink_extract_updates_keeps_empty_sync_buf(self): - client = ILinkClient( - base_url="https://ilinkai.weixin.qq.com", - bot_token="token", - sync_buf="cursor-old", - ) - items, sync_buf = client._extract_updates( - { - "ret": 0, - "get_updates_buf": "", - "msgs": [ - { - "message_id": "msg-1001", - "from_user_id": "wxid_user_1", - "item_list": [{"type": 1, "text_item": {"text": "你好"}}], - } - ], - } - ) +def _module_config() -> SimpleNamespace: + """构造消息解析所需的渠道配置。""" + return SimpleNamespace(name="wechatclawbot-test", config={}) - self.assertEqual(sync_buf, "") - self.assertEqual(len(items), 1) - def test_ilink_poll_updates_resets_sync_buf_when_server_returns_empty_cursor(self): - client = ILinkClient( - base_url="https://ilinkai.weixin.qq.com", - bot_token="token", - sync_buf="cursor-old", - ) - response = MagicMock() - response.json.return_value = { +def test_ilink_parse_incoming_uses_seq_as_message_id_fallback(): + client = _create_client() + + message = client._parse_incoming( + { + "seq": 123456, + "from_user_id": "wxid_user_1", + "item_list": [{"type": 1, "text_item": {"text": "你好"}}], + } + ) + + assert message is not None + assert message.message_id == "123456" + assert message.text == "你好" + + +def test_ilink_parse_incoming_extracts_item_list_media(): + """iLink 下行媒体放在 item_list 子项的 media 中，应解析为可直接下载的附件引用。""" + client = _create_client() + image_url = "https://novac2c.cdn.weixin.qq.com/c2c/download?fileid=image-1" + voice_url = "https://novac2c.cdn.weixin.qq.com/c2c/download?fileid=voice-1" + file_url = "https://novac2c.cdn.weixin.qq.com/c2c/download?fileid=file-1" + video_url = "https://novac2c.cdn.weixin.qq.com/c2c/download?fileid=video-1" + media_aes_key = "MDEyMzQ1Njc4OWFiY2RlZjAxMjM0NTY3ODlhYmNkZWY=" + + message = client._parse_incoming( + { + "message_id": "msg-media-1", + "from_user_id": "wxid_user_1", + "message_type": 1, + "item_list": [ + {"type": 1, "text_item": {"text": "看这张图"}}, + { + "type": 2, + "image_item": { + "aeskey": "0123456789abcdef0123456789abcdef", + "media": {"full_url": image_url, "aes_key": media_aes_key}, + }, + }, + { + "type": 3, + "voice_item": {"media": {"full_url": voice_url, "aes_key": media_aes_key}}, + }, + { + "type": 4, + "file_item": { + "file_name": "报告.pdf", + "size": 1024, + "media": {"full_url": file_url, "aes_key": media_aes_key}, + }, + }, + { + "type": 5, + "video_item": {"media": {"full_url": video_url, "aes_key": media_aes_key}}, + }, + ], + } + ) + + assert message is not None + assert message.text == "看这张图" + assert len(message.images) == 1 + image_payload = WechatClawBot._decode_ref_payload(message.images[0]["ref"], "image") + assert image_payload["url"] == image_url + assert image_payload["aeskey"] == "0123456789abcdef0123456789abcdef" + assert len(message.audio_refs) == 1 + voice_payload = WechatClawBot._decode_ref_payload(message.audio_refs[0], "voice") + assert voice_payload["url"] == voice_url + assert voice_payload["aeskey"] == media_aes_key + assert len(message.files) == 2 + file_payload = WechatClawBot._decode_ref_payload(message.files[0]["ref"], "file") + assert file_payload["url"] == file_url + assert file_payload["name"] == "报告.pdf" + assert message.files[0]["size"] == 1024 + video_payload = WechatClawBot._decode_ref_payload(message.files[1]["ref"], "file") + assert video_payload["url"] == video_url + assert message.files[1]["name"] == "video.mp4" + + +def test_ilink_parse_incoming_keeps_legacy_file_item_types(): + """历史报文把文件放在 type=5/6 的 file_item 中，仍需保持兼容。""" + client = _create_client() + file_url = "https://novac2c.cdn.weixin.qq.com/c2c/download?fileid=legacy-1" + + message = client._parse_incoming( + { + "message_id": "msg-file-legacy", + "from_user_id": "wxid_user_1", + "item_list": [ + { + "type": 6, + "file_item": { + "name": "archive.zip", + "media": {"full_url": file_url}, + }, + } + ], + } + ) + + assert message is not None + assert len(message.files) == 1 + file_payload = WechatClawBot._decode_ref_payload(message.files[0]["ref"], "file") + assert file_payload["url"] == file_url + assert file_payload["name"] == "archive.zip" + + +def test_ilink_extract_updates_keeps_empty_sync_buf(): + client = _create_client(bot_token="token", sync_buf="cursor-old") + + items, sync_buf = client._extract_updates( + { "ret": 0, "get_updates_buf": "", "msgs": [ @@ -100,276 +160,342 @@ class WechatClawBotTest(unittest.TestCase): } ], } + ) - with patch("app.modules.wechatclawbot.wechatclawbot.RequestUtils.post", return_value=response): - messages, sync_buf, result = client.poll_updates() + assert sync_buf == "" + assert len(items) == 1 - self.assertTrue(result["success"]) - self.assertEqual(sync_buf, "") - self.assertEqual(client.sync_buf, "") - self.assertEqual(len(messages), 1) - def test_ilink_poll_updates_accepts_canonical_payload_without_explicit_success_flag(self): - client = ILinkClient( - base_url="https://ilinkai.weixin.qq.com", - bot_token="token", - sync_buf="cursor-old", - ) - response = MagicMock() - response.json.return_value = { - "sync_buf": "cursor-new", - "msgs": [ +def test_ilink_poll_updates_resets_sync_buf_when_server_returns_empty_cursor(): + client = _create_client(bot_token="token", sync_buf="cursor-old") + response = MagicMock() + response.json.return_value = { + "ret": 0, + "get_updates_buf": "", + "msgs": [ + { + "message_id": "msg-1001", + "from_user_id": "wxid_user_1", + "item_list": [{"type": 1, "text_item": {"text": "你好"}}], + } + ], + } + + with patch("app.modules.wechatclawbot.wechatclawbot.RequestUtils.post", return_value=response): + messages, sync_buf, result = client.poll_updates() + + assert result["success"] + assert sync_buf == "" + assert client.sync_buf == "" + assert len(messages) == 1 + + +def test_ilink_poll_updates_accepts_canonical_payload_without_explicit_success_flag(): + client = _create_client(bot_token="token", sync_buf="cursor-old") + response = MagicMock() + response.json.return_value = { + "sync_buf": "cursor-new", + "msgs": [ + { + "message_id": "msg-1002", + "from_user_id": "wxid_user_2", + "item_list": [{"type": 1, "text_item": {"text": "收到"}}], + } + ], + } + + with patch("app.modules.wechatclawbot.wechatclawbot.RequestUtils.post", return_value=response): + messages, sync_buf, result = client.poll_updates() + + assert result["success"] + assert sync_buf == "cursor-new" + assert client.sync_buf == "cursor-new" + assert len(messages) == 1 + assert messages[0].text == "收到" + + +def test_ilink_poll_updates_rejects_noncanonical_nested_success_payload(): + client = _create_client(bot_token="token", sync_buf="cursor-old") + response = MagicMock() + response.json.return_value = { + "ret": 0, + "data": { + "get_updates_buf": "cursor-new", + "messages": [ { - "message_id": "msg-1002", - "from_user_id": "wxid_user_2", - "item_list": [{"type": 1, "text_item": {"text": "收到"}}], + "message_id": "msg-1001", + "from_user_id": "wxid_user_1", + "item_list": [{"type": 1, "text_item": {"text": "你好"}}], + } + ], + }, + } + + with patch("app.modules.wechatclawbot.wechatclawbot.RequestUtils.post", return_value=response): + messages, sync_buf, result = client.poll_updates() + + assert not result["success"] + assert result["message"] == "轮询响应结构非官方，缺少顶层 msgs 字段" + assert sync_buf == "cursor-old" + assert client.sync_buf == "cursor-old" + assert messages == [] + + +def test_ilink_poll_updates_rejects_failed_payload_even_if_it_contains_messages(): + client = _create_client(bot_token="token", sync_buf="cursor-old") + failed_response = MagicMock() + failed_response.json.return_value = { + "ret": -2, + "errmsg": "cursor invalid", + "data": { + "sync_buf": "cursor-old", + "messages": [ + { + "message_id": "msg-dup-1", + "from_user_id": "wxid_user_1", + "item_list": [{"type": 1, "text_item": {"text": "旧消息"}}], + } + ], + }, + } + + with patch( + "app.modules.wechatclawbot.wechatclawbot.RequestUtils.post", + return_value=failed_response, + ) as mock_post: + messages, sync_buf, result = client.poll_updates() + + assert not result["success"] + assert result["message"] == "cursor invalid" + assert sync_buf == "cursor-old" + assert client.sync_buf == "cursor-old" + assert messages == [] + mock_post.assert_called_once() + request_body = mock_post.call_args.kwargs["json"] + assert "get_updates_buf" in request_body + assert "sync_buf" not in request_body + assert "syncBuf" not in request_body + + +def test_ilink_test_connection_accepts_getconfig_ilink_user_id_limitation(): + client = _create_client(bot_token="token") + response = MagicMock() + response.json.return_value = { + "ret": -1, + "errmsg": "ilink_user_id required", + } + + with patch("app.modules.wechatclawbot.wechatclawbot.RequestUtils.post", return_value=response): + ok, message = client.test_connection() + + # `ilink_user_id required` 仅表示自检接口缺少额外参数，不代表连接失败：视为连接正常 + assert ok + assert message == "连接正常" + + +def test_wechatclawbot_message_parser_deduplicates_message_id(): + module = WechatClawBotModule() + body = json.dumps( + { + "__channel__": "wechatclawbot", + "userid": "wxid_user_1", + "username": "tester", + "message_id": "msg-1001", + "text": "刷新订阅", + } + ) + + with patch.object(module, "get_config", return_value=_module_config()): + first = module.message_parser( + source="wechatclawbot-test", + body=body, + form={}, + args={}, + ) + second = module.message_parser( + source="wechatclawbot-test", + body=body, + form={}, + args={}, + ) + + assert first is not None + assert first.message_id == "msg-1001" + assert second is None + + +def test_wechatclawbot_message_parser_accepts_image_only_message(): + """图片入站消息（无文本）应进入 Agent 链路，而不是被判为无效消息丢弃。""" + module = WechatClawBotModule() + client = _create_client() + parsed_message = client._parse_incoming( + { + "message_id": "msg-image-only", + "from_user_id": "wxid_user_1", + "item_list": [ + { + "type": 2, + "image_item": { + "aeskey": "0123456789abcdef0123456789abcdef", + "media": { + "full_url": "https://novac2c.cdn.weixin.qq.com/c2c/download?fileid=image-only", + "aes_key": "MDEyMzQ1Njc4OWFiY2RlZjAxMjM0NTY3ODlhYmNkZWY=", + }, + }, } ], } + ) + assert parsed_message is not None - with patch("app.modules.wechatclawbot.wechatclawbot.RequestUtils.post", return_value=response): - messages, sync_buf, result = client.poll_updates() - - self.assertTrue(result["success"]) - self.assertEqual(sync_buf, "cursor-new") - self.assertEqual(client.sync_buf, "cursor-new") - self.assertEqual(len(messages), 1) - self.assertEqual(messages[0].text, "收到") - - def test_ilink_poll_updates_rejects_noncanonical_nested_success_payload(self): - client = ILinkClient( - base_url="https://ilinkai.weixin.qq.com", - bot_token="token", - sync_buf="cursor-old", - ) - response = MagicMock() - response.json.return_value = { - "ret": 0, - "data": { - "get_updates_buf": "cursor-new", - "messages": [ - { - "message_id": "msg-1001", - "from_user_id": "wxid_user_1", - "item_list": [{"type": 1, "text_item": {"text": "你好"}}], - } - ], - }, - } - - with patch("app.modules.wechatclawbot.wechatclawbot.RequestUtils.post", return_value=response): - messages, sync_buf, result = client.poll_updates() - - self.assertFalse(result["success"]) - self.assertEqual(result["message"], "轮询响应结构非官方，缺少顶层 msgs 字段") - self.assertEqual(sync_buf, "cursor-old") - self.assertEqual(client.sync_buf, "cursor-old") - self.assertEqual(messages, []) - - def test_ilink_poll_updates_rejects_failed_payload_even_if_it_contains_messages(self): - client = ILinkClient( - base_url="https://ilinkai.weixin.qq.com", - bot_token="token", - sync_buf="cursor-old", - ) - failed_response = MagicMock() - failed_response.json.return_value = { - "ret": -2, - "errmsg": "cursor invalid", - "data": { - "sync_buf": "cursor-old", - "messages": [ - { - "message_id": "msg-dup-1", - "from_user_id": "wxid_user_1", - "item_list": [{"type": 1, "text_item": {"text": "旧消息"}}], - } - ], - }, - } - - with patch( - "app.modules.wechatclawbot.wechatclawbot.RequestUtils.post", - return_value=failed_response, - ) as mock_post: - messages, sync_buf, result = client.poll_updates() - - self.assertFalse(result["success"]) - self.assertEqual(result["message"], "cursor invalid") - self.assertEqual(sync_buf, "cursor-old") - self.assertEqual(client.sync_buf, "cursor-old") - self.assertEqual(messages, []) - mock_post.assert_called_once() - request_body = mock_post.call_args.kwargs["json"] - self.assertIn("get_updates_buf", request_body) - self.assertNotIn("sync_buf", request_body) - self.assertNotIn("syncBuf", request_body) - - def test_ilink_test_connection_accepts_getconfig_ilink_user_id_limitation(self): - client = ILinkClient( - base_url="https://ilinkai.weixin.qq.com", - bot_token="token", - ) - response = MagicMock() - response.json.return_value = { - "ret": -1, - "errmsg": "ilink_user_id required", - } - - with patch("app.modules.wechatclawbot.wechatclawbot.RequestUtils.post", return_value=response): - ok, message = client.test_connection() - - # `ilink_user_id required` 仅表示自检接口缺少额外参数，不代表连接失败：视为连接正常 - self.assertTrue(ok) - self.assertEqual(message, "连接正常") - - def test_wechatclawbot_send_msg_uses_plain_text_payload(self): - state = { - "bot_token": None, - "account_id": None, - "sync_buf": None, - "qrcode": {}, - "known_targets": {}, - "user_context_tokens": {}, - "base_url": "https://ilinkai.weixin.qq.com", - } - with patch.object(WechatClawBot, "_load_state", return_value=state): - bot = WechatClawBot(name="wechatclawbot-test", auto_start_polling=False) - - mock_client = MagicMock() - mock_client.send_text.return_value = True - - with patch.object(bot, "_build_client", return_value=mock_client): - result = bot.send_msg( - title="测试标题", - text="测试正文", - userid="wxid_user_1", - link="https://example.com/detail", - ) - - self.assertTrue(result) - mock_client.send_text.assert_called_once_with( - to_user="wxid_user_1", - text="测试标题\n\n测试正文\n\n查看详情：https://example.com/detail", - context_token=None, + with patch.object(module, "get_config", return_value=_module_config()): + parsed = module.message_parser( + source="wechatclawbot-test", + body=json.dumps(parsed_message.to_message_payload()), + form={}, + args={}, ) - def test_wechatclawbot_send_msg_prefers_text_when_text_and_image_coexist(self): - state = { - "bot_token": None, - "account_id": None, - "sync_buf": None, - "qrcode": {}, - "known_targets": {}, - "user_context_tokens": {}, - "base_url": "https://ilinkai.weixin.qq.com", + assert parsed is not None + assert parsed.images is not None + assert parsed.images[0].ref.startswith("wxclaw://image/") + assert not parsed.text + + +def test_wechatclawbot_message_parser_logs_unrecognized_content(): + """无法识别的入站内容应留下诊断日志，而不是静默丢弃。""" + module = WechatClawBotModule() + body = json.dumps( + { + "__channel__": "wechatclawbot", + "userid": "wxid_user_1", + "message_id": "msg-unknown-1", + "item_list": [{"type": 99, "unknown_item": {"foo": "bar"}}], } - with patch.object(WechatClawBot, "_load_state", return_value=state): - bot = WechatClawBot(name="wechatclawbot-test", auto_start_polling=False) + ) - mock_client = MagicMock() - mock_client.send_text.return_value = True - mock_client.send_image_png.return_value = True - mock_client.send_image_text_png.return_value = True + with ( + patch.object(module, "get_config", return_value=_module_config()), + patch("app.modules.wechatclawbot.logger") as mock_logger, + ): + parsed = module.message_parser( + source="wechatclawbot-test", + body=body, + form={}, + args={}, + ) + assert parsed is None + mock_logger.info.assert_called_once() + assert "未识别到有效内容" in mock_logger.info.call_args.args[0] + + +def test_wechatclawbot_send_msg_uses_plain_text_payload(): + bot = _create_bot() + mock_client = MagicMock() + mock_client.send_text.return_value = True + + with patch.object(bot, "_build_client", return_value=mock_client): + result = bot.send_msg( + title="测试标题", + text="测试正文", + userid="wxid_user_1", + link="https://example.com/detail", + ) + + assert result + mock_client.send_text.assert_called_once_with( + to_user="wxid_user_1", + text="测试标题\n\n测试正文\n\n查看详情：https://example.com/detail", + context_token=None, + ) + + +def test_wechatclawbot_send_msg_prefers_text_when_text_and_image_coexist(): + bot = _create_bot() + mock_client = MagicMock() + mock_client.send_text.return_value = True + mock_client.send_image_png.return_value = True + mock_client.send_image_text_png.return_value = True + + with ( + patch.object(bot, "_build_client", return_value=mock_client), + patch.object(bot, "_load_remote_image", return_value=b"image-bytes") as mock_load_image, + ): + result = bot.send_msg( + title="测试标题", + text="测试正文", + image="https://example.com/test.png", + userid="wxid_user_1", + ) + + assert result + mock_load_image.assert_not_called() + mock_client.send_text.assert_called_once_with( + to_user="wxid_user_1", + text="测试标题\n\n测试正文", + context_token=None, + ) + mock_client.send_image_png.assert_not_called() + mock_client.send_image_text_png.assert_not_called() + + +def test_wechatclawbot_send_file_prefers_image_when_image_file_has_caption(): + bot = _create_bot() + mock_client = MagicMock() + mock_client.send_text.return_value = True + mock_client.send_image_png.return_value = True + + with tempfile.NamedTemporaryFile(suffix=".png") as image_file: + image_file.write(b"\x89PNG\r\n\x1a\nfake-png") + image_file.flush() with ( patch.object(bot, "_build_client", return_value=mock_client), - patch.object(bot, "_load_remote_image", return_value=b"image-bytes") as mock_load_image, + patch.object(bot, "_guess_mime_type", return_value="image/png"), ): - result = bot.send_msg( - title="测试标题", - text="测试正文", - image="https://example.com/test.png", + result = bot.send_file( + file_path=image_file.name, + title="图片标题", + text="图片说明", userid="wxid_user_1", ) - self.assertTrue(result) - mock_load_image.assert_not_called() - mock_client.send_text.assert_called_once_with( - to_user="wxid_user_1", - text="测试标题\n\n测试正文", - context_token=None, - ) - mock_client.send_image_png.assert_not_called() - mock_client.send_image_text_png.assert_not_called() + assert result + mock_client.send_text.assert_not_called() + mock_client.send_image_png.assert_called_once_with( + to_user="wxid_user_1", + image_bytes=b"\x89PNG\r\n\x1a\nfake-png", + context_token=None, + ) - def test_wechatclawbot_send_file_prefers_image_when_image_file_has_caption(self): - state = { - "bot_token": None, - "account_id": None, - "sync_buf": None, - "qrcode": {}, - "known_targets": {}, - "user_context_tokens": {}, - "base_url": "https://ilinkai.weixin.qq.com", - } - with patch.object(WechatClawBot, "_load_state", return_value=state): - bot = WechatClawBot(name="wechatclawbot-test", auto_start_polling=False) - mock_client = MagicMock() - mock_client.send_text.return_value = True - mock_client.send_image_png.return_value = True +def test_wechatclawbot_send_file_prefers_file_when_generic_file_has_caption(): + bot = _create_bot() + mock_client = MagicMock() + mock_client.send_text.return_value = True + mock_client.send_file_bytes.return_value = True - with tempfile.NamedTemporaryFile(suffix=".png") as image_file: - image_file.write(b"\x89PNG\r\n\x1a\nfake-png") - image_file.flush() - with ( - patch.object(bot, "_build_client", return_value=mock_client), - patch.object(bot, "_guess_mime_type", return_value="image/png"), - ): - result = bot.send_file( - file_path=image_file.name, - title="图片标题", - text="图片说明", - userid="wxid_user_1", - ) + with tempfile.NamedTemporaryFile(suffix=".txt") as text_file: + text_file.write(b"plain-text") + text_file.flush() + with ( + patch.object(bot, "_build_client", return_value=mock_client), + patch.object(bot, "_guess_mime_type", return_value="text/plain"), + ): + result = bot.send_file( + file_path=text_file.name, + file_name="report.txt", + title="文件标题", + text="文件说明", + userid="wxid_user_1", + ) - self.assertTrue(result) - mock_client.send_text.assert_not_called() - mock_client.send_image_png.assert_called_once_with( - to_user="wxid_user_1", - image_bytes=b"\x89PNG\r\n\x1a\nfake-png", - context_token=None, - ) - - def test_wechatclawbot_send_file_prefers_file_when_generic_file_has_caption(self): - state = { - "bot_token": None, - "account_id": None, - "sync_buf": None, - "qrcode": {}, - "known_targets": {}, - "user_context_tokens": {}, - "base_url": "https://ilinkai.weixin.qq.com", - } - with patch.object(WechatClawBot, "_load_state", return_value=state): - bot = WechatClawBot(name="wechatclawbot-test", auto_start_polling=False) - - mock_client = MagicMock() - mock_client.send_text.return_value = True - mock_client.send_file_bytes.return_value = True - - with tempfile.NamedTemporaryFile(suffix=".txt") as text_file: - text_file.write(b"plain-text") - text_file.flush() - with ( - patch.object(bot, "_build_client", return_value=mock_client), - patch.object(bot, "_guess_mime_type", return_value="text/plain"), - ): - result = bot.send_file( - file_path=text_file.name, - file_name="report.txt", - title="文件标题", - text="文件说明", - userid="wxid_user_1", - ) - - self.assertTrue(result) - mock_client.send_text.assert_not_called() - mock_client.send_file_bytes.assert_called_once_with( - to_user="wxid_user_1", - file_bytes=b"plain-text", - file_name="report.txt", - mime_type="text/plain", - context_token=None, - ) + assert result + mock_client.send_text.assert_not_called() + mock_client.send_file_bytes.assert_called_once_with( + to_user="wxid_user_1", + file_bytes=b"plain-text", + file_name="report.txt", + mime_type="text/plain", + context_token=None, + )