fix: tighten wechatclawbot poll protocol handling

app/modules/wechatclawbot/wechatclawbot.py

@@ -170,6 +170,19 @@ class ILinkClient:
                 return obj.get(key)
         return None
 
+    @staticmethod
+    def _pick_present_value(obj: Dict[str, Any], keys: List[str]) -> Optional[Any]:
+        """
+        获取已存在的字段值，允许空字符串和 0 作为有效值。
+
+        sync_buf 这类游标字段可能会被服务端明确置为空串，表示游标已经回到初始状态。
+        这里不能复用默认的“忽略空字符串”逻辑，否则会错误沿用上一轮旧游标。
+        """
+        for key in keys:
+            if key in obj and obj.get(key) is not None:
+                return obj.get(key)
+        return None
+
     @classmethod
     def _find_first_value(
         cls, data: Any, keys: List[str], max_depth: int = 5
@@ -370,27 +383,6 @@ class ILinkClient:
             {"to_user_id": user_id, "msg_type": "text", "content": text},
         ]
 
-    @staticmethod
-    def _build_markdown_payloads(user_id: str, text: str) -> List[Dict[str, Any]]:
-        return [
-            {
-                "to_user": user_id,
-                "msg_type": "markdown",
-                "markdown": {"content": text},
-            },
-            {
-                "touser": user_id,
-                "msgtype": "markdown",
-                "markdown": {"content": text},
-            },
-            {"to": user_id, "type": "markdown", "content": text},
-            {
-                "to_user": user_id,
-                "message_type": "markdown",
-                "markdown": {"content": text},
-            },
-        ]
-
     @staticmethod
     def _aes_ecb_padded_size(plaintext_size: int) -> int:
         return ((int(plaintext_size) + 1 + 15) // 16) * 16
@@ -786,20 +778,6 @@ class ILinkClient:
         ]
         return self._send_payload_candidates(to_user=to_user, payload_candidates=payload_candidates)
 
-    def send_markdown(
-        self, to_user: str, text: str, context_token: Optional[str] = None
-    ) -> bool:
-        if not self.bot_token:
-            logger.warning("发送 Markdown 失败：bot token 未配置")
-            return False
-        if not to_user or not text:
-            logger.warning("发送 Markdown 失败：to_user 或 text 为空")
-            return False
-        payload_candidates = self._build_markdown_payloads(str(to_user), text)
-        if self._send_payload_candidates(to_user=to_user, payload_candidates=payload_candidates):
-            return True
-        return self.send_text(to_user=to_user, text=text, context_token=context_token)
-
     def send_image_text_png(
         self,
         to_user: str,
@@ -1260,49 +1238,76 @@ class ILinkClient:
     def _extract_updates(
         self, payload: Dict[str, Any]
     ) -> Tuple[List[Dict[str, Any]], Optional[str]]:
-        """从轮询响应中提取消息列表和下一轮使用的 sync_buf 游标。"""
-        data = payload.get("data") or payload.get("result") or payload
-        sync_buf = (
-            data.get("get_updates_buf")
-            or payload.get("get_updates_buf")
-            or data.get("sync_buf")
-            or data.get("syncBuf")
-            or payload.get("sync_buf")
-            or payload.get("syncBuf")
-            or self._find_first_value(
-                data,
-                ["get_updates_buf", "sync_buf", "syncBuf", "cursor", "offset", "next_sync_buf"],
-            )
-        )
-        list_keys = [
-            "msgs",
-            "updates",
-            "messages",
-            "items",
-            "events",
-            "msg_list",
-            "msgList",
-            "add_msgs",
-            "addMsgs",
-            "records",
-            "list",
-        ]
-        candidates = [data.get(key) for key in list_keys] + [payload.get(key) for key in list_keys]
-        for candidate in candidates:
-            if isinstance(candidate, list):
-                return candidate, sync_buf
-        nested = self._find_first_list(data, prefer_keys=list_keys)
-        if isinstance(nested, list):
-            return nested, sync_buf
-        if isinstance(data, list):
-            return data, sync_buf
-        if isinstance(data, dict):
-            for key in ["message", "msg", "event", "item"]:
-                item = data.get(key)
-                if isinstance(item, dict):
-                    return [item], sync_buf
+        """按官方 getupdates 协议提取顶层 msgs 与 get_updates_buf。"""
+        sync_buf = self._pick_present_value(payload, ["get_updates_buf"])
+        items = payload.get("msgs")
+        if isinstance(items, list):
+            return items, sync_buf
         return [], sync_buf
 
+    def _has_canonical_poll_shape(self, payload: Dict[str, Any]) -> bool:
+        """官方响应至少应包含顶层 msgs 列表。"""
+        return isinstance(payload.get("msgs"), list)
+
+    def _is_poll_success(self, payload: Dict[str, Any]) -> bool:
+        """
+        判断 getupdates 是否明确成功。
+
+        轮询接口不能沿用“只要没有明显报错就算成功”的宽松策略，否则服务端返回旧消息列表、
+        但状态码其实失败时，会被误判为可消费响应，导致旧消息再次进入业务链路。
+        """
+        if not payload:
+            return False
+        code = self._find_first_value(
+            payload, ["errcode", "code", "ret", "result_code", "status_code"]
+        )
+        if code is not None:
+            try:
+                return int(str(code)) == 0
+            except Exception:
+                return str(code).strip().lower() in {"0", "ok", "success", "succeed"}
+        success_flag = self._find_first_value(
+            payload, ["success", "ok", "is_success"]
+        )
+        if isinstance(success_flag, bool):
+            return success_flag
+        if success_flag is not None:
+            return str(success_flag).strip().lower() in {
+                "1",
+                "true",
+                "ok",
+                "success",
+                "succeed",
+            }
+        state = self._find_first_value(payload, ["status", "state"])
+        if state is not None:
+            lowered = str(state).strip().lower()
+            if lowered in {"ok", "success", "succeed", "done"}:
+                return True
+            if lowered in {"failed", "fail", "error", "denied", "blocked"}:
+                return False
+        return False
+
+    def _build_poll_result(
+        self,
+        success: bool,
+        payload: Optional[Dict[str, Any]] = None,
+        message: Optional[str] = None,
+        item_count: int = 0,
+        parsed_count: int = 0,
+    ) -> Dict[str, Any]:
+        payload = payload or {}
+        resolved_message = message or self._find_first_value(
+            payload, ["errmsg", "message", "error", "error_msg", "detail"]
+        )
+        return {
+            "success": success,
+            "raw": payload,
+            "message": self._short_text(resolved_message) if resolved_message else None,
+            "item_count": item_count,
+            "parsed_count": parsed_count,
+        }
+
     def poll_updates(
         self, timeout_seconds: int = 25
     ) -> Tuple[List[ILinkIncomingMessage], Optional[str], Dict[str, Any]]:
@@ -1316,27 +1321,39 @@ class ILinkClient:
             return [], self.sync_buf, {"success": False, "message": "bot token 未配置"}
         url = f"{self.base_url}/ilink/bot/getupdates"
         payload = {}
-        body_candidates = [
-            {"get_updates_buf": self.sync_buf or ""},
-            {"sync_buf": self.sync_buf, "timeout": timeout_seconds},
-            {"syncBuf": self.sync_buf, "timeout": timeout_seconds},
-            {"sync_buf": self.sync_buf, "wait": timeout_seconds},
-        ]
-        for body in body_candidates:
-            request_body = self._with_base_info(body)
-            resp = RequestUtils(
-                headers=self._headers(auth_required=True),
-                timeout=timeout_seconds + 10,
-            ).post(url, json=request_body)
-            payload = self._json(resp)
-            if payload and self._ok(payload):
-                break
-            if payload and self._find_first_list(
-                payload, prefer_keys=["updates", "messages", "items", "events", "add_msgs", "msgs"]
-            ):
-                break
+        request_body = self._with_base_info({"get_updates_buf": self.sync_buf or ""})
+        resp = RequestUtils(
+            headers=self._headers(auth_required=True),
+            timeout=timeout_seconds + 10,
+        ).post(url, json=request_body)
+        payload = self._json(resp)
+        success = bool(payload and self._is_poll_success(payload))
+        last_message = None
+        if payload and not success:
+            last_message = self._find_first_value(
+                payload, ["errmsg", "message", "error", "error_msg", "detail"]
+            ) or self._short_text(payload)
         if not payload:
-            return [], self.sync_buf, {"success": False, "message": "轮询返回空响应"}
+            return [], self.sync_buf, self._build_poll_result(
+                success=False,
+                message="轮询返回空响应",
+            )
+        if not success:
+            return [], self.sync_buf, self._build_poll_result(
+                success=False,
+                payload=payload,
+                message=last_message or "轮询响应未明确成功",
+            )
+        if not self._has_canonical_poll_shape(payload):
+            logger.warning(
+                "getupdates 返回非官方结构，已拒绝消费: %s",
+                self._short_text(payload),
+            )
+            return [], self.sync_buf, self._build_poll_result(
+                success=False,
+                payload=payload,
+                message="轮询响应结构非官方，缺少顶层 msgs 字段",
+            )
         items, sync_buf = self._extract_updates(payload)
         parsed: List[ILinkIncomingMessage] = []
         for item in items:
@@ -1345,13 +1362,12 @@ class ILinkClient:
                 parsed.append(message)
         if sync_buf is not None:
             self.sync_buf = str(sync_buf)
-        return parsed, self.sync_buf, {
-            "success": self._ok(payload),
-            "raw": payload,
-            "message": payload.get("errmsg") or payload.get("message"),
-            "item_count": len(items),
-            "parsed_count": len(parsed),
-        }
+        return parsed, self.sync_buf, self._build_poll_result(
+            success=True,
+            payload=payload,
+            item_count=len(items),
+            parsed_count=len(parsed),
+        )
 
     def test_connection(self) -> Tuple[bool, str]:
         if not self.bot_token:
@@ -1549,14 +1565,14 @@ class WechatClawBot:
         return [chunk for chunk in chunks if chunk]
 
     @staticmethod
-    def _compose_markdown(
+    def _compose_text(
         title: Optional[str] = None,
         text: Optional[str] = None,
         link: Optional[str] = None,
     ) -> str:
         parts = []
         if title:
-            parts.append(f"## {title}")
+            parts.append(str(title).strip())
         if text:
             parts.append(str(text).replace("\n\n", "\n"))
         if link:
@@ -1994,7 +2010,7 @@ class WechatClawBot:
             logger.warning("未找到可发送的微信 ClawBot 目标")
             return False
         image_bytes = self._load_remote_image(image) if image else None
-        content = self._compose_markdown(title=title, text=text, link=link)
+        content = self._compose_text(title=title, text=text, link=link)
         ok = False
         for target in targets:
             context_token = self._get_context_token(target)
@@ -2015,7 +2031,7 @@ class WechatClawBot:
                 client = self._build_client()
                 sent = True
                 for chunk in self._split_content(content):
-                    if not client.send_markdown(
+                    if not client.send_text(
                         to_user=target,
                         text=chunk,
                         context_token=context_token,
@@ -2044,7 +2060,7 @@ class WechatClawBot:
         targets = self._get_targets(userid=userid)
         if not targets:
             return False
-        caption = self._compose_markdown(title=title, text=text)
+        caption = self._compose_text(title=title, text=text)
         ok = False
         for target in targets:
             context_token = self._get_context_token(target)
@@ -2052,7 +2068,7 @@ class WechatClawBot:
             client = self._build_client()
             if caption:
                 for chunk in self._split_content(caption):
-                    if not client.send_markdown(
+                    if not client.send_text(
                         to_user=target,
                         text=chunk,
                         context_token=context_token,

tests/test_wechatclawbot.py

@@ -1,10 +1,10 @@
 import json
 import unittest
 from types import SimpleNamespace
-from unittest.mock import patch
+from unittest.mock import MagicMock, patch
 
 from app.modules.wechatclawbot import WechatClawBotModule
-from app.modules.wechatclawbot.wechatclawbot import ILinkClient
+from app.modules.wechatclawbot.wechatclawbot import ILinkClient, WechatClawBot
 
 
 class WechatClawBotTest(unittest.TestCase):
@@ -57,6 +57,157 @@ class WechatClawBotTest(unittest.TestCase):
         self.assertEqual(first.message_id, "msg-1001")
         self.assertIsNone(second)
 
+    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": "你好"}}],
+                    }
+                ],
+            }
+        )
+
+        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 = {
+            "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()
+
+        self.assertTrue(result["success"])
+        self.assertEqual(sync_buf, "")
+        self.assertEqual(client.sync_buf, "")
+        self.assertEqual(len(messages), 1)
+
+    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_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,
+        )
+
 
 if __name__ == "__main__":
     unittest.main()