fix(WeiXin): auto-refresh expired QR code during login to improve success rate

nanobot/channels/weixin.py

@@ -63,6 +63,7 @@ SESSION_PAUSE_DURATION_S = 60 * 60
 MAX_CONSECUTIVE_FAILURES = 3
 BACKOFF_DELAY_S = 30
 RETRY_DELAY_S = 2
+MAX_QR_REFRESH_COUNT = 3
 
 # Default long-poll timeout; overridden by server via longpolling_timeout_ms.
 DEFAULT_LONG_POLL_TIMEOUT_S = 35
@@ -241,24 +242,25 @@ class WeixinChannel(BaseChannel):
     # QR Code Login  (matches login-qr.ts)
     # ------------------------------------------------------------------
 
+    async def _fetch_qr_code(self) -> tuple[str, str]:
+        """Fetch a fresh QR code. Returns (qrcode_id, scan_url)."""
+        data = await self._api_get(
+            "ilink/bot/get_bot_qrcode",
+            params={"bot_type": "3"},
+            auth=False,
+        )
+        qrcode_img_content = data.get("qrcode_img_content", "")
+        qrcode_id = data.get("qrcode", "")
+        if not qrcode_id:
+            raise RuntimeError(f"Failed to get QR code from WeChat API: {data}")
+        return qrcode_id, (qrcode_img_content or qrcode_id)
+
     async def _qr_login(self) -> bool:
         """Perform QR code login flow. Returns True on success."""
         try:
             logger.info("Starting WeChat QR code login...")
-
+            refresh_count = 0
-            data = await self._api_get(
+            qrcode_id, scan_url = await self._fetch_qr_code()
-                "ilink/bot/get_bot_qrcode",
-                params={"bot_type": "3"},
-                auth=False,
-            )
-            qrcode_img_content = data.get("qrcode_img_content", "")
-            qrcode_id = data.get("qrcode", "")
-
-            if not qrcode_id:
-                logger.error("Failed to get QR code from WeChat API: {}", data)
-                return False
-
-            scan_url = qrcode_img_content or qrcode_id
             self._print_qr_code(scan_url)
 
             logger.info("Waiting for QR code scan...")
@@ -298,8 +300,23 @@ class WeixinChannel(BaseChannel):
                 elif status == "scaned":
                     logger.info("QR code scanned, waiting for confirmation...")
                 elif status == "expired":
-                    logger.warning("QR code expired")
+                    refresh_count += 1
-                    return False
+                    if refresh_count > MAX_QR_REFRESH_COUNT:
+                        logger.warning(
+                            "QR code expired too many times ({}/{}), giving up.",
+                            refresh_count - 1,
+                            MAX_QR_REFRESH_COUNT,
+                        )
+                        return False
+                    logger.warning(
+                        "QR code expired, refreshing... ({}/{})",
+                        refresh_count,
+                        MAX_QR_REFRESH_COUNT,
+                    )
+                    qrcode_id, scan_url = await self._fetch_qr_code()
+                    self._print_qr_code(scan_url)
+                    logger.info("New QR code generated, waiting for scan...")
+                    continue
                 # status == "wait" — keep polling
 
                 await asyncio.sleep(1)

tests/channels/test_weixin_channel.py

@@ -206,6 +206,57 @@ async def test_poll_once_pauses_session_on_expired_errcode() -> None:
     assert channel._session_pause_remaining_s() > 0
 
 
+@pytest.mark.asyncio
+async def test_qr_login_refreshes_expired_qr_and_then_succeeds() -> None:
+    channel, _bus = _make_channel()
+    channel._running = True
+    channel._save_state = lambda: None
+    channel._print_qr_code = lambda url: None
+    channel._api_get = AsyncMock(
+        side_effect=[
+            {"qrcode": "qr-1", "qrcode_img_content": "url-1"},
+            {"status": "expired"},
+            {"qrcode": "qr-2", "qrcode_img_content": "url-2"},
+            {
+                "status": "confirmed",
+                "bot_token": "token-2",
+                "ilink_bot_id": "bot-2",
+                "baseurl": "https://example.test",
+                "ilink_user_id": "wx-user",
+            },
+        ]
+    )
+
+    ok = await channel._qr_login()
+
+    assert ok is True
+    assert channel._token == "token-2"
+    assert channel.config.base_url == "https://example.test"
+
+
+@pytest.mark.asyncio
+async def test_qr_login_returns_false_after_too_many_expired_qr_codes() -> None:
+    channel, _bus = _make_channel()
+    channel._running = True
+    channel._print_qr_code = lambda url: None
+    channel._api_get = AsyncMock(
+        side_effect=[
+            {"qrcode": "qr-1", "qrcode_img_content": "url-1"},
+            {"status": "expired"},
+            {"qrcode": "qr-2", "qrcode_img_content": "url-2"},
+            {"status": "expired"},
+            {"qrcode": "qr-3", "qrcode_img_content": "url-3"},
+            {"status": "expired"},
+            {"qrcode": "qr-4", "qrcode_img_content": "url-4"},
+            {"status": "expired"},
+        ]
+    )
+
+    ok = await channel._qr_login()
+
+    assert ok is False
+
+
 @pytest.mark.asyncio
 async def test_process_message_skips_bot_messages() -> None:
     channel, bus = _make_channel()