mirror/nanobot
1
0
Fork 0
mirror of https://github.com/HKUDS/nanobot.git synced 2026-04-24 03:46:02 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(WeXin): add route_tag property to adapt to WeChat official ilinkai 1.0.3 requirements

Browse Source
This commit is contained in:
xcosmosbox 2026-03-24 14:38:18 +08:00 committed by Xubin Ren
parent 7b31af2204
commit 3a9d6ea536
2 changed files with 17 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
nanobot/channels/weixin.py
View File

@ -83,6 +83,7 @@ class WeixinConfig(Base):
allow_from: list[str] = Field(default_factory=list) allow_from: list[str] = Field(default_factory=list)
base_url: str = "https://ilinkai.weixin.qq.com" base_url: str = "https://ilinkai.weixin.qq.com"
cdn_base_url: str = "https://novac2c.cdn.weixin.qq.com/c2c" cdn_base_url: str = "https://novac2c.cdn.weixin.qq.com/c2c"
route_tag: str | int | None = None
token: str = "" # Manually set token, or obtained via QR login token: str = "" # Manually set token, or obtained via QR login
state_dir: str = "" # Default: ~/.nanobot/weixin/ state_dir: str = "" # Default: ~/.nanobot/weixin/
poll_timeout: int = DEFAULT_LONG_POLL_TIMEOUT_S # seconds for long-poll poll_timeout: int = DEFAULT_LONG_POLL_TIMEOUT_S # seconds for long-poll
@ -187,6 +188,8 @@ class WeixinChannel(BaseChannel):
} }
if auth and self._token: if auth and self._token:
headers["Authorization"] = f"Bearer {self._token}" headers["Authorization"] = f"Bearer {self._token}"
if self.config.route_tag is not None and str(self.config.route_tag).strip():
headers["SKRouteTag"] = str(self.config.route_tag).strip()
return headers return headers
async def _api_get( async def _api_get(

14
tests/channels/test_weixin_channel.py
View File

@ -22,6 +22,20 @@ def _make_channel() -> tuple[WeixinChannel, MessageBus]:
return channel, bus return channel, bus
def test_make_headers_includes_route_tag_when_configured() -> None:
bus = MessageBus()
channel = WeixinChannel(
WeixinConfig(enabled=True, allow_from=["*"], route_tag=123),
bus,
)
channel._token = "token"
headers = channel._make_headers()
assert headers["Authorization"] == "Bearer token"
assert headers["SKRouteTag"] == "123"
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_process_message_deduplicates_inbound_ids() -> None: async def test_process_message_deduplicates_inbound_ids() -> None:
channel, bus = _make_channel() channel, bus = _make_channel()