feat(weixin):

1.align protocol headers with package.json metadata
2.support upload_full_url with fallback to upload_param

nanobot/channels/weixin.py

@@ -53,7 +53,41 @@ MESSAGE_TYPE_BOT = 2
 MESSAGE_STATE_FINISH = 2
 
 WEIXIN_MAX_MESSAGE_LEN = 4000
-WEIXIN_CHANNEL_VERSION = "1.0.3"
+
+
+def _read_reference_package_meta() -> dict[str, str]:
+    """Best-effort read of reference `package/package.json` metadata."""
+    try:
+        pkg_path = Path(__file__).resolve().parents[2] / "package" / "package.json"
+        data = json.loads(pkg_path.read_text(encoding="utf-8"))
+        return {
+            "version": str(data.get("version", "") or ""),
+            "ilink_appid": str(data.get("ilink_appid", "") or ""),
+        }
+    except Exception:
+        return {"version": "", "ilink_appid": ""}
+
+
+def _build_client_version(version: str) -> int:
+    """Encode semantic version as 0x00MMNNPP (major/minor/patch in one uint32)."""
+    parts = version.split(".")
+
+    def _as_int(idx: int) -> int:
+        try:
+            return int(parts[idx])
+        except Exception:
+            return 0
+
+    major = _as_int(0)
+    minor = _as_int(1)
+    patch = _as_int(2)
+    return ((major & 0xFF) << 16) | ((minor & 0xFF) << 8) | (patch & 0xFF)
+
+
+_PKG_META = _read_reference_package_meta()
+WEIXIN_CHANNEL_VERSION = _PKG_META["version"] or "unknown"
+ILINK_APP_ID = _PKG_META["ilink_appid"]
+ILINK_APP_CLIENT_VERSION = _build_client_version(_PKG_META["version"] or "0.0.0")
 BASE_INFO: dict[str, str] = {"channel_version": WEIXIN_CHANNEL_VERSION}
 
 # Session-expired error code
@@ -199,6 +233,8 @@ class WeixinChannel(BaseChannel):
             "X-WECHAT-UIN": self._random_wechat_uin(),
             "Content-Type": "application/json",
             "AuthorizationType": "ilink_bot_token",
+            "iLink-App-Id": ILINK_APP_ID,
+            "iLink-App-ClientVersion": str(ILINK_APP_CLIENT_VERSION),
         }
         if auth and self._token:
             headers["Authorization"] = f"Bearer {self._token}"
@@ -267,13 +303,10 @@ class WeixinChannel(BaseChannel):
             logger.info("Waiting for QR code scan...")
             while self._running:
                 try:
-                    # Reference plugin sends iLink-App-ClientVersion header for
-                    # QR status polling (login-qr.ts:81).
                     status_data = await self._api_get(
                         "ilink/bot/get_qrcode_status",
                         params={"qrcode": qrcode_id},
                         auth=False,
-                        extra_headers={"iLink-App-ClientVersion": "1"},
                     )
                 except httpx.TimeoutException:
                     continue
@@ -838,7 +871,7 @@ class WeixinChannel(BaseChannel):
         # Matches aesEcbPaddedSize: Math.ceil((size + 1) / 16) * 16
         padded_size = ((raw_size + 1 + 15) // 16) * 16
 
-        # Step 1: Get upload URL (upload_param) from server
+        # Step 1: Get upload URL from server (prefer upload_full_url, fallback to upload_param)
         file_key = os.urandom(16).hex()
         upload_body: dict[str, Any] = {
             "filekey": file_key,
@@ -855,19 +888,26 @@ class WeixinChannel(BaseChannel):
         upload_resp = await self._api_post("ilink/bot/getuploadurl", upload_body)
         logger.debug("WeChat getuploadurl response: {}", upload_resp)
 
-        upload_param = upload_resp.get("upload_param", "")
+        upload_full_url = str(upload_resp.get("upload_full_url", "") or "").strip()
-        if not upload_param:
+        upload_param = str(upload_resp.get("upload_param", "") or "")
-            raise RuntimeError(f"getuploadurl returned no upload_param: {upload_resp}")
+        if not upload_full_url and not upload_param:
+            raise RuntimeError(
+                "getuploadurl returned no upload URL "
+                f"(need upload_full_url or upload_param): {upload_resp}"
+            )
 
         # Step 2: AES-128-ECB encrypt and POST to CDN
         aes_key_b64 = base64.b64encode(aes_key_raw).decode()
         encrypted_data = _encrypt_aes_ecb(raw_data, aes_key_b64)
 
-        cdn_upload_url = (
+        if upload_full_url:
-            f"{self.config.cdn_base_url}/upload"
+            cdn_upload_url = upload_full_url
-            f"?encrypted_query_param={quote(upload_param)}"
+        else:
-            f"&filekey={quote(file_key)}"
+            cdn_upload_url = (
-        )
+                f"{self.config.cdn_base_url}/upload"
+                f"?encrypted_query_param={quote(upload_param)}"
+                f"&filekey={quote(file_key)}"
+            )
         logger.debug("WeChat CDN POST url={} ciphertextSize={}", cdn_upload_url[:80], len(encrypted_data))
 
         cdn_resp = await self._client.post(

tests/channels/test_weixin_channel.py

@@ -1,6 +1,7 @@
 import asyncio
 import json
 import tempfile
+from pathlib import Path
 from types import SimpleNamespace
 from unittest.mock import AsyncMock
 
@@ -42,10 +43,13 @@ def test_make_headers_includes_route_tag_when_configured() -> None:
 
     assert headers["Authorization"] == "Bearer token"
     assert headers["SKRouteTag"] == "123"
+    assert headers["iLink-App-Id"] == "bot"
+    assert headers["iLink-App-ClientVersion"] == str((2 << 16) | (1 << 8) | 1)
 
 
 def test_channel_version_matches_reference_plugin_version() -> None:
-    assert WEIXIN_CHANNEL_VERSION == "1.0.3"
+    pkg = json.loads(Path("package/package.json").read_text())
+    assert WEIXIN_CHANNEL_VERSION == pkg["version"]
 
 
 def test_save_and_load_state_persists_context_tokens(tmp_path) -> None:
@@ -278,3 +282,61 @@ async def test_process_message_skips_bot_messages() -> None:
     )
 
     assert bus.inbound_size == 0
+
+
+class _DummyHttpResponse:
+    def __init__(self, *, headers: dict[str, str] | None = None, status_code: int = 200) -> None:
+        self.headers = headers or {}
+        self.status_code = status_code
+
+    def raise_for_status(self) -> None:
+        return None
+
+
+@pytest.mark.asyncio
+async def test_send_media_uses_upload_full_url_when_present(tmp_path) -> None:
+    channel, _bus = _make_channel()
+
+    media_file = tmp_path / "photo.jpg"
+    media_file.write_bytes(b"hello-weixin")
+
+    cdn_post = AsyncMock(return_value=_DummyHttpResponse(headers={"x-encrypted-param": "dl-param"}))
+    channel._client = SimpleNamespace(post=cdn_post)
+    channel._api_post = AsyncMock(
+        side_effect=[
+            {
+                "upload_full_url": "https://upload-full.example.test/path?foo=bar",
+                "upload_param": "should-not-be-used",
+            },
+            {"ret": 0},
+        ]
+    )
+
+    await channel._send_media_file("wx-user", str(media_file), "ctx-1")
+
+    # first POST call is CDN upload
+    cdn_url = cdn_post.await_args_list[0].args[0]
+    assert cdn_url == "https://upload-full.example.test/path?foo=bar"
+
+
+@pytest.mark.asyncio
+async def test_send_media_falls_back_to_upload_param_url(tmp_path) -> None:
+    channel, _bus = _make_channel()
+
+    media_file = tmp_path / "photo.jpg"
+    media_file.write_bytes(b"hello-weixin")
+
+    cdn_post = AsyncMock(return_value=_DummyHttpResponse(headers={"x-encrypted-param": "dl-param"}))
+    channel._client = SimpleNamespace(post=cdn_post)
+    channel._api_post = AsyncMock(
+        side_effect=[
+            {"upload_param": "enc-need-fallback"},
+            {"ret": 0},
+        ]
+    )
+
+    await channel._send_media_file("wx-user", str(media_file), "ctx-1")
+
+    cdn_url = cdn_post.await_args_list[0].args[0]
+    assert cdn_url.startswith(f"{channel.config.cdn_base_url}/upload?encrypted_query_param=enc-need-fallback")
+    assert "&filekey=" in cdn_url