From 2a2385baccce6c664327f78363c0c9c87f933c41 Mon Sep 17 00:00:00 2001 From: Kris <2893855659@qq.com> Date: Wed, 8 Jul 2026 22:56:45 +0800 Subject: [PATCH] =?UTF-8?q?refactor(wechat-ilink):=20=E9=87=8D=E6=9E=84?= =?UTF-8?q?=E5=BE=AE=E4=BF=A1iLink=E6=8F=92=E4=BB=B6=E4=BB=A3=E7=A0=81?= =?UTF-8?q?=EF=BC=8C=E7=BB=9F=E4=B8=80=E5=B8=B8=E9=87=8F=E4=B8=8E=E7=BC=93?= =?UTF-8?q?=E5=AD=98=E9=94=AE=E6=A0=BC=E5=BC=8F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 本次重构对微信iLink渠道插件进行了全面整理： 1. 新增共享常量模块`_constants.py`，集中管理所有跨模块共享配置与标识，对齐manifest单真相源 2. 统一缓存键格式为`wechat_ilink:{account_id}:{key}`，修复白名单存储键顺序错误 3. 重构登录适配器，移除不必要的account_id参数，简化接口调用 4. 重构生命周期相关代码，新增缓存清理逻辑，支持按账户清理孤儿缓存 5. 移除冗余重复定义的常量与配置默认值，统一从常量模块导入 6. 更新manifest配置，补充传输模式与能力要求声明 7. 优化出站适配器，新增续传支持与分片发送逻辑 --- .../plugins/wechat_ilink/_constants.py | 81 ++++++++++++ .../wechat_ilink/adapters/inbound_adapter.py | 8 +- .../adapters/lifecycle_adapter.py | 125 +++++++++++++----- .../wechat_ilink/adapters/login_adapter.py | 7 +- .../wechat_ilink/adapters/outbound_adapter.py | 61 +++++---- .../wechat_ilink/adapters/puller_adapter.py | 8 +- .../adapters/whitelist_adapter.py | 4 +- .../plugins/wechat_ilink/dtos/__init__.py | 1 - .../channels/plugins/wechat_ilink/entry.py | 18 ++- .../plugins/wechat_ilink/ilink_client.py | 61 +++------ .../plugins/wechat_ilink/lifecycle.py | 52 ++++++-- .../plugins/wechat_ilink/manifest.json | 25 ++-- 12 files changed, 307 insertions(+), 144 deletions(-) create mode 100644 backend/package/yuxi/channels/plugins/wechat_ilink/_constants.py delete mode 100644 backend/package/yuxi/channels/plugins/wechat_ilink/dtos/__init__.py diff --git a/backend/package/yuxi/channels/plugins/wechat_ilink/_constants.py b/backend/package/yuxi/channels/plugins/wechat_ilink/_constants.py new file mode 100644 index 00000000..7287318e --- /dev/null +++ b/backend/package/yuxi/channels/plugins/wechat_ilink/_constants.py @@ -0,0 +1,81 @@ +"""微信 iLink 渠道插件共享常量。 + +集中定义在多个模块间共享的配置默认值、依赖标识与渠道标识，避免重复 +定义导致的不一致风险（README §3 推荐布局，对齐 wechat_woc 模式）。 + +仅汇集真正跨模块共享的常量；仅单一模块使用的常量仍保留在各自模块内。 +所有常量值需与 ``manifest.json`` 中 ``config_schema`` 的 ``default`` 字段 +保持一致（F-03 manifest 单真相源），本模块的常量作为 ConfigPort 未命中 +时的本地回退默认值。 + +依赖方向：仅标准库，不 import 任何框架层或同插件模块，避免循环依赖。 +""" + +from __future__ import annotations + +# --------------------------------------------------------------------------- +# 渠道与依赖标识 +# --------------------------------------------------------------------------- + +# 渠道标识，用于 ConfigPort CHANNEL 作用域 target（插件 provides 值， +# 与 manifest.json 对齐） +CHANNEL_TARGET = "wechat_ilink" + +# iLink API 依赖标识，统一用于 ``DependencyError.dep`` 字段 +ILINK_API_DEP = "ilink_api" + +# --------------------------------------------------------------------------- +# HTTP 客户端配置（lifecycle 与 ilink_client 共享） +# --------------------------------------------------------------------------- + +# HTTP 客户端默认超时（秒），对齐 manifest.json config_schema.long_poll_timeout_ms +HTTP_TIMEOUT_SECONDS = 30.0 + +# HTTP 连接池最大连接数默认值，manifest.json resource_quota.max_connections.default=10 +# （entry.py 与 lifecycle 共享，避免硬编码重复） +DEFAULT_MAX_CONNECTIONS = 10 + +# HTTP 连接池最大 keepalive 连接数 +DEFAULT_MAX_KEEPALIVE_CONNECTIONS = 5 + +# 在途请求 drain 超时（秒），onStop/onUnload 等待在途请求完成的最长时间 +DRAIN_TIMEOUT_SECONDS = 30.0 + +# --------------------------------------------------------------------------- +# 配置默认值（与 manifest.json config_schema default 一致，ConfigPort 未 +# 命中时本地回退使用） +# --------------------------------------------------------------------------- + +# iLink API 基础地址，manifest.json config_schema.baseurl.default +DEFAULT_BASEURL = "https://ilinkai.weixin.qq.com" + +# iLink 渠道协议版本，manifest.json config_schema.channel_version.default +DEFAULT_CHANNEL_VERSION = "2.0.0" + +# iLink CDN 地址，manifest.json config_schema.cdn_url.default +DEFAULT_CDN_URL = "https://novac2c.cdn.weixin.qq.com/c2c" + +# 长轮询超时（毫秒），manifest.json config_schema.long_poll_timeout_ms.default=35000 +DEFAULT_LONG_POLL_TIMEOUT_MS = 35000 + +# 长轮询客户端超时余量（ms）：长于服务端 35s hang，避免客户端先超时 +LONG_POLL_EXTRA_MS = 5000 + +# 单条消息最大长度，manifest.json config_schema.max_message_length.default=4000 +DEFAULT_MAX_MESSAGE_LENGTH = 4000 + + +__all__ = [ + "CHANNEL_TARGET", + "ILINK_API_DEP", + "HTTP_TIMEOUT_SECONDS", + "DEFAULT_MAX_CONNECTIONS", + "DEFAULT_MAX_KEEPALIVE_CONNECTIONS", + "DRAIN_TIMEOUT_SECONDS", + "DEFAULT_BASEURL", + "DEFAULT_CHANNEL_VERSION", + "DEFAULT_CDN_URL", + "DEFAULT_LONG_POLL_TIMEOUT_MS", + "LONG_POLL_EXTRA_MS", + "DEFAULT_MAX_MESSAGE_LENGTH", +] diff --git a/backend/package/yuxi/channels/plugins/wechat_ilink/adapters/inbound_adapter.py b/backend/package/yuxi/channels/plugins/wechat_ilink/adapters/inbound_adapter.py index 476af9c3..d9becef2 100644 --- a/backend/package/yuxi/channels/plugins/wechat_ilink/adapters/inbound_adapter.py +++ b/backend/package/yuxi/channels/plugins/wechat_ilink/adapters/inbound_adapter.py @@ -26,6 +26,7 @@ from yuxi.channels.contract.errors import DependencyError, Error from yuxi.channels.contract.ports.driven.logger_port import LoggerPort from .. import crypto +from .._constants import DEFAULT_CDN_URL from ..ilink_client import ILinkClient # 媒体 URL scheme @@ -39,9 +40,6 @@ _VIDEO_ITEM_TYPE = 3 _FILE_ITEM_TYPE = 4 _VOICE_ITEM_TYPE = 5 -# CDN 默认地址 -_DEFAULT_CDN_URL = "https://novac2c.cdn.weixin.qq.com/c2c" - class WeChatILinkInboundAdapter: """微信 iLink 入站消息规范化适配器。 @@ -104,7 +102,7 @@ class WeChatILinkInboundAdapter: text_parts: list[str] = [] attachments: list[Attachment] = [] - cdn_url = await self._client.get_channel_config("cdn_url", _DEFAULT_CDN_URL) + cdn_url = await self._client.get_channel_config("cdn_url", DEFAULT_CDN_URL) for item in item_list: item_type = item.get("type") @@ -205,7 +203,7 @@ class WeChatILinkInboundAdapter: encrypt_query_param = params.get("encrypt_query_param", "") aes_key = params.get("aes_key", "") - cdn_url = params.get("cdn_url", _DEFAULT_CDN_URL) + cdn_url = params.get("cdn_url", DEFAULT_CDN_URL) try: encrypted = await self._client.cdn_download(cdn_url, encrypt_query_param) diff --git a/backend/package/yuxi/channels/plugins/wechat_ilink/adapters/lifecycle_adapter.py b/backend/package/yuxi/channels/plugins/wechat_ilink/adapters/lifecycle_adapter.py index e367e3f1..ad64e858 100644 --- a/backend/package/yuxi/channels/plugins/wechat_ilink/adapters/lifecycle_adapter.py +++ b/backend/package/yuxi/channels/plugins/wechat_ilink/adapters/lifecycle_adapter.py @@ -5,19 +5,27 @@ 配置落库后回调、删除前缓存清理、启用/禁用回调。 设计要点： -- 适配器无 mutable 实例变量（INV-5）。 +- 适配器无 mutable 实例变量（INV-5），仅持有 DI 注入的 ``CachePort`` / + ``LoggerPort``。 - 长轮询由传输引擎通过 ``ChannelAccountOnline``/``ChannelAccountOffline`` - 领域事件触发 ``PullerWorker`` 启动/停止，本适配器不自管理轮询任务（决策 3）。 + 领域事件触发 ``PullerWorker`` 启停，本适配器不自管理轮询任务（决策 3）。 - 所有生命周期回调方法幂等：重复调用安全。 - ``resolveAccountId`` 纯本地解析，不调 API。 - ``applyAccountConfig`` 仅校验配置并规范化，不写库也不写 CachePort（凭证由 CredentialService 通过 ConfigPort→CachePort 统一管理）。 +- ``beforeAccountDelete`` / ``onAccountDisabled`` 清理 CachePort 缓存， + 按账户隔离清理运行时缓存，避免孤儿键残留。CachePort 异常不抛（仅日志 + 告警，避免阻塞生命周期流程），对齐 wechat_woc 模式。 + - ``beforeAccountDelete`` 使用 ``wechat_ilink:{account_id}:*`` 通配符 + 清空该账户全部缓存（context_token / cursor / 白名单等）+ 凭证缓存。 + - ``onAccountDisabled`` 仅清运行态（context_token / cursor）+ 凭证缓存， + 保留白名单以便快速恢复（对齐 wechat_woc DISABLE 保留白名单语义）。 - ``rotateCredentials`` 未实现（iLink bot_token 通过扫码获取，不支持运行时轮换）。 - ``validateCredentials`` 仅校验 bot_token 非空，完整验证由 ProbeableAdapter 完成。 - ``revokeOldCredentials`` 默认空实现：重新扫码即自动失效旧 token。 依赖方向：仅 import ``yuxi.channels.contract.*`` 与同插件内部模块 -（``ilink_client``），不污染框架层。 +（``_constants``），不污染框架层。 """ from __future__ import annotations @@ -32,17 +40,28 @@ from yuxi.channels.contract.errors import ( NotImplementedError, ValidationError, ) +from yuxi.channels.contract.ports.driven.cache_port import CachePort from yuxi.channels.contract.ports.driven.logger_port import LoggerPort -from ..ilink_client import ILinkClient +from .._constants import DEFAULT_BASEURL -_DEFAULT_BASEURL = "https://ilinkai.weixin.qq.com" +# CachePort 失效模式 +# beforeAccountDelete：清空该账户所有运行时缓存。使用 +# ``wechat_ilink:{account_id}:*`` 通配符，避免多实例场景下跨账户清理。 +# 同时失效凭证缓存与 cursor / context_token / 白名单缓存。 +_CACHE_PATTERN_DELETE = "wechat_ilink:{account_id}:*" +# onAccountDisabled：仅清运行态（context_token / cursor）+ 凭证缓存， +# 保留白名单以便快速恢复（对齐 wechat_woc DISABLE 保留白名单语义）。 +_CACHE_PATTERNS_DISABLE = ( + "wechat_ilink:{account_id}:context_token:*", + "wechat_ilink:{account_id}:cursor", +) class WeChatILinkLifecycleAdapter: """微信 iLink 账户生命周期适配器。 - 通过 DI 接收 ``ILinkClient`` / ``LoggerPort``，不访问全局 settings 或 + 通过 DI 接收 ``CachePort`` / ``LoggerPort``，不访问全局 settings 或 logger。长轮询由传输引擎通过 ``ChannelAccountOnline``/``ChannelAccountOffline`` 领域事件触发 ``PullerWorker`` 启动/停止，本适配器不自管理轮询任务 （决策 3，INV-8）。所有生命周期回调方法幂等：重复调用安全。 @@ -50,10 +69,10 @@ class WeChatILinkLifecycleAdapter: def __init__( self, - client: ILinkClient, + cache_port: CachePort, logger_port: LoggerPort, ) -> None: - self._client = client + self._cache = cache_port self._logger = logger_port # ------------------------------------------------------------------ @@ -99,7 +118,7 @@ class WeChatILinkLifecycleAdapter: field="bot_token", message="bot_token_missing", ) - baseurl = raw_config.get("baseurl") or _DEFAULT_BASEURL + baseurl = raw_config.get("baseurl") or DEFAULT_BASEURL normalized = dict(raw_config) normalized["baseurl"] = baseurl await self._logger.info( @@ -120,23 +139,46 @@ class WeChatILinkLifecycleAdapter: ) async def beforeAccountDelete(self, account: ChannelAccount) -> None: - """删除前回调。 + """删除前回调。清空该账户的所有 CachePort 缓存。幂等。 - 清除 context_token 运行时缓存（幂等）：清除已清空的缓存安全无副作用。 - 缓存清理失败时让异常自然传播，由框架决定是否继续删除。 + 使用 ``wechat_ilink:{account_id}:*`` 通配符按账户隔离清理运行时 + 缓存（context_token / cursor / 白名单等），避免多实例场景下跨账户 + 清理。同时删除凭证缓存。CachePort 异常 **不抛**（仅日志告警， + 避免阻塞删除流程）。 Note: 凭证（bot_token 等）的撤销由 ``CredentialService.revokeCredentials`` - 统一负责，本适配器不再调用 ``ILinkClient.clear_credentials``（该方法 - 已于 Task 8 移除）。本方法仅清理运行时会话上下文 token（context_token）， - 该 token 属于运行时缓存而非持久化凭证，由 ILinkClient 自身管理。 + 统一负责，本方法仅清理运行时缓存，确保无孤儿键残留。 """ - await self._client.clear_context_token( - account.account_id, - from_user_id=None, - ) + account_id = account.account_id + pattern = _CACHE_PATTERN_DELETE.format(account_id=account_id) + try: + await self._cache.invalidate(pattern=pattern) + except Exception as exc: + # CachePort 故障降级：不阻塞删除流程，但需记录完整堆栈 + await self._logger.exception( + "iLink beforeAccountDelete: cache invalidate failed, non-blocking", + account_id=account_id, + exc_info=exc, + ) + + # 精确删除凭证缓存（键名与 CredentialService 约定一致） + precise_keys = [ + f"credentials:{account_id}", + ] + for key in precise_keys: + try: + await self._cache.delete(key) + except Exception as exc: + await self._logger.exception( + "iLink beforeAccountDelete: cache delete failed, non-blocking", + account_id=account_id, + key=key, + exc_info=exc, + ) + await self._logger.info( - "iLink account delete: context_token cleared, credentials revocation delegated to CredentialService", - account_id=account.account_id, + "iLink account delete: cache invalidated, credentials revocation delegated to CredentialService", + account_id=account_id, ) async def onAccountEnabled(self, account: ChannelAccount) -> None: @@ -151,19 +193,40 @@ class WeChatILinkLifecycleAdapter: ) async def onAccountDisabled(self, account: ChannelAccount) -> None: - """禁用后回调。 + """禁用后回调。清空运行态缓存与凭证缓存（保留白名单）。 框架发布 ``ChannelAccountOffline`` 事件触发停止 ``PullerWorker``。 - 清除 context_token 缓存（保留 bot_token 缓存以便快速恢复）。 + 清空 ``CachePort`` 中运行态缓存（context_token / cursor）与 + ``credentials:{account_id}`` 凭证缓存，保留白名单以便快速恢复 + （对齐 wechat_woc DISABLE 保留白名单语义）。CachePort 异常不抛 + （仅日志告警，避免阻塞禁用流程）。 """ - await self._client.clear_context_token( - account.account_id, - from_user_id=None, - ) + account_id = account.account_id + for pattern in _CACHE_PATTERNS_DISABLE: + formatted = pattern.format(account_id=account_id) + try: + await self._cache.invalidate(pattern=formatted) + except Exception as exc: + # CachePort 故障降级：不阻塞禁用流程，但需记录完整堆栈 + await self._logger.exception( + "iLink onAccountDisabled: cache invalidate failed, non-blocking", + account_id=account_id, + pattern=formatted, + exc_info=exc, + ) + + try: + await self._cache.delete(f"credentials:{account_id}") + except Exception as exc: + await self._logger.exception( + "iLink onAccountDisabled: credentials delete failed, non-blocking", + account_id=account_id, + exc_info=exc, + ) + await self._logger.info( - "iLink account disabled: PullerWorker will be stopped by transport engine, " - "context_token cleared, bot_token cached for fast recovery", - account_id=account.account_id, + "iLink account disabled: runtime cache cleared, whitelist preserved", + account_id=account_id, ) def supportsRotateCredentials(self) -> bool: @@ -198,7 +261,7 @@ class WeChatILinkLifecycleAdapter: iLink bot_token 通过 QR 扫码登录获取，重新扫码即自动失效旧 token， 无需显式撤销。 """ - pass # 默认空实现 + return None __all__ = ["WeChatILinkLifecycleAdapter"] diff --git a/backend/package/yuxi/channels/plugins/wechat_ilink/adapters/login_adapter.py b/backend/package/yuxi/channels/plugins/wechat_ilink/adapters/login_adapter.py index 5690223c..eb4035e8 100644 --- a/backend/package/yuxi/channels/plugins/wechat_ilink/adapters/login_adapter.py +++ b/backend/package/yuxi/channels/plugins/wechat_ilink/adapters/login_adapter.py @@ -90,14 +90,15 @@ class WeChatILinkLoginAdapter: qrcode 令牌编码到 ``qr_data_url`` 的 fragment（``#qrcode=xxx``）中， 由 ``QrLoginService`` 会话管理承载，不落 ``CachePort``（INV-5）。 - @pre: ``account_id`` 为 ``None`` 时使用默认账户 ``default``。 + iLink ``get_bot_qrcode`` 为登录前接口，无需鉴权，``account_id`` + 仅用于 Protocol 签名兼容，不影响请求。 + @post: 返回 ``QrLoginStartResult``，``connected`` 始终为 False， ``qr_data_url`` 携带 qrcode 令牌 fragment 供 ``loginWithQrWait`` 提取。 @failure: 渠道 API 调用失败抛契约层异常（由 ``ILinkClient`` 转换）。 """ - acct = account_id or _DEFAULT_ACCOUNT - resp = await self._client.get_bot_qrcode(acct) + resp = await self._client.get_bot_qrcode() qr_content = resp.get("qrcode_img_content") qrcode = resp.get("qrcode", "") qr_data_url = self._normalize_qr_data_url(qr_content, qrcode) diff --git a/backend/package/yuxi/channels/plugins/wechat_ilink/adapters/outbound_adapter.py b/backend/package/yuxi/channels/plugins/wechat_ilink/adapters/outbound_adapter.py index 1c8e1ef9..76a5110b 100644 --- a/backend/package/yuxi/channels/plugins/wechat_ilink/adapters/outbound_adapter.py +++ b/backend/package/yuxi/channels/plugins/wechat_ilink/adapters/outbound_adapter.py @@ -37,7 +37,8 @@ from yuxi.channels.contract.ports.driven.config_port import ConfigPort from yuxi.channels.contract.ports.driven.logger_port import LoggerPort from .. import crypto -from ..ilink_client import CHANNEL_TARGET, ILinkClient +from .._constants import CHANNEL_TARGET, DEFAULT_MAX_MESSAGE_LENGTH +from ..ilink_client import ILinkClient # iLink 消息类型常量 _MESSAGE_TYPE_BOT = 2 @@ -47,9 +48,6 @@ _IMAGE_ITEM_TYPE = 2 _VIDEO_ITEM_TYPE = 3 _FILE_ITEM_TYPE = 4 -# 长文本分片默认阈值（Unicode 字符数），ConfigPort 未命中时回退 -_DEFAULT_MAX_MESSAGE_LENGTH = 4000 - class WeChatILinkOutboundAdapter: """微信 iLink 出站消息投递适配器。 @@ -135,6 +133,8 @@ class WeChatILinkOutboundAdapter: account_id: str, peer_id: str, payload: FormattedMessage, + *, + delivered_parts: list[int] | None = None, ) -> list[MultiPartReceipt]: """发送多部分消息。 @@ -147,6 +147,9 @@ class WeChatILinkOutboundAdapter: 发送前预校验 ``context_token``，避免文本已发送、媒体因缺 token 失败导致半截消息。每片独立调用 ``sendMessage`` 生成唯一 ``client_id``。 + H-15 续发：``delivered_parts`` 含已成功投递的分片序号，适配器跳过 + 这些序号对应的分片（含文本与媒体），仅发送未投递分片。 + @consistency: 最终一致（eventual），部分分片可能成功部分失败。 @idempotent: False — 每片生成独立 ``client_id``，重试会产生重复分片。 """ @@ -161,39 +164,21 @@ class WeChatILinkOutboundAdapter: # 分片阈值从 ConfigPort 读取（支持热更新） max_length = await self._read_channel_config( "max_message_length", - _DEFAULT_MAX_MESSAGE_LENGTH, + DEFAULT_MAX_MESSAGE_LENGTH, ) receipts: list[MultiPartReceipt] = [] seq = 0 + skip_seqs = set(delivered_parts) if delivered_parts else set() if payload.content: if len(payload.content) <= max_length: - msg_id = await self.sendMessage( - account_id, - peer_id, - FormattedMessage( - content=payload.content, - format=payload.format, - ), - ) - receipts.append( - MultiPartReceipt( - platform_msg_id=msg_id, - part_type="text", - sequence=seq, - success=True, - ) - ) - seq += 1 - else: - chunks = [payload.content[i : i + max_length] for i in range(0, len(payload.content), max_length)] - for chunk in chunks: + if seq not in skip_seqs: msg_id = await self.sendMessage( account_id, peer_id, FormattedMessage( - content=chunk, + content=payload.content, format=payload.format, ), ) @@ -205,11 +190,35 @@ class WeChatILinkOutboundAdapter: success=True, ) ) + seq += 1 + else: + chunks = [payload.content[i : i + max_length] for i in range(0, len(payload.content), max_length)] + for chunk in chunks: + if seq not in skip_seqs: + msg_id = await self.sendMessage( + account_id, + peer_id, + FormattedMessage( + content=chunk, + format=payload.format, + ), + ) + receipts.append( + MultiPartReceipt( + platform_msg_id=msg_id, + part_type="text", + sequence=seq, + success=True, + ) + ) seq += 1 for attachment in payload.attachments: if attachment.content is None: continue + if seq in skip_seqs: + seq += 1 + continue aes_key = crypto.generate_random_aes_key() ciphertext = crypto.encrypt(attachment.content, aes_key) file_size = crypto.calc_ciphertext_size(len(attachment.content)) diff --git a/backend/package/yuxi/channels/plugins/wechat_ilink/adapters/puller_adapter.py b/backend/package/yuxi/channels/plugins/wechat_ilink/adapters/puller_adapter.py index cc0a7eb8..29733d86 100644 --- a/backend/package/yuxi/channels/plugins/wechat_ilink/adapters/puller_adapter.py +++ b/backend/package/yuxi/channels/plugins/wechat_ilink/adapters/puller_adapter.py @@ -23,11 +23,9 @@ from yuxi.channels.contract.errors import DependencyError, NotFoundError from yuxi.channels.contract.ports.driven.config_port import ConfigPort from yuxi.channels.contract.ports.driven.logger_port import LoggerPort +from .._constants import CHANNEL_TARGET, DEFAULT_LONG_POLL_TIMEOUT_MS from ..error_translator import translate_poll_error -from ..ilink_client import CHANNEL_TARGET, ILinkClient - -# 配置默认值（与 manifest.json long_poll_timeout_ms 一致，ConfigPort 未命中时回退） -_DEFAULT_LONG_POLL_TIMEOUT_MS = 35000 +from ..ilink_client import ILinkClient class WeChatILinkPullerAdapter: @@ -104,7 +102,7 @@ class WeChatILinkPullerAdapter: """ timeout_ms = await self._read_channel_config( "long_poll_timeout_ms", - _DEFAULT_LONG_POLL_TIMEOUT_MS, + DEFAULT_LONG_POLL_TIMEOUT_MS, ) return PollingConfig( long_poll_timeout_ms=timeout_ms, diff --git a/backend/package/yuxi/channels/plugins/wechat_ilink/adapters/whitelist_adapter.py b/backend/package/yuxi/channels/plugins/wechat_ilink/adapters/whitelist_adapter.py index a8223e0e..a522976e 100644 --- a/backend/package/yuxi/channels/plugins/wechat_ilink/adapters/whitelist_adapter.py +++ b/backend/package/yuxi/channels/plugins/wechat_ilink/adapters/whitelist_adapter.py @@ -2,7 +2,7 @@ 实现 ``WhitelistAdapter`` Protocol，管理 C2C 私聊白名单。iLink（个人微信） 无服务端白名单 API，白名单条目本地存储于 ``CachePort``，键模式为 -``wechat_ilink:whitelist:{account_id}:{policy_type}``。每个 bot_token 对应 +``wechat_ilink:{account_id}:whitelist:{policy_type}``。每个 bot_token 对应 一个微信账号，``account_id`` 由调用方（``WhitelistHandler``）传入，多账户 场景下按账户隔离。 @@ -59,7 +59,7 @@ class WeChatILinkWhitelistAdapter: def _key(self, account_id: str, policy_type: WhitelistPolicyType) -> str: """构造白名单存储键。""" - return f"wechat_ilink:whitelist:{account_id}:{policy_type.value}" + return f"wechat_ilink:{account_id}:whitelist:{policy_type.value}" async def _load_entries( self, diff --git a/backend/package/yuxi/channels/plugins/wechat_ilink/dtos/__init__.py b/backend/package/yuxi/channels/plugins/wechat_ilink/dtos/__init__.py deleted file mode 100644 index 6bf25eb2..00000000 --- a/backend/package/yuxi/channels/plugins/wechat_ilink/dtos/__init__.py +++ /dev/null @@ -1 +0,0 @@ -"""微信 iLink 渠道内部 DTO。""" diff --git a/backend/package/yuxi/channels/plugins/wechat_ilink/entry.py b/backend/package/yuxi/channels/plugins/wechat_ilink/entry.py index 510832b0..0dac533d 100644 --- a/backend/package/yuxi/channels/plugins/wechat_ilink/entry.py +++ b/backend/package/yuxi/channels/plugins/wechat_ilink/entry.py @@ -21,12 +21,13 @@ client 实例化、LifecycleHandler 构造与 ``registerAdapter`` 调用。 适配器 DI 说明： - ``puller`` / ``outbound`` 适配器接收 ``(client, config_port, logger_port)``： 适配器直接持有 ConfigPort 读取渠道级配置（对齐 wechat_woc 模式）。 -- 5 个适配器（inbound / login / lifecycle / doctor / wizard）接收 +- 4 个适配器（inbound / login / doctor / wizard）接收 ``(client, logger_port)``。 +- ``lifecycle`` / ``whitelist`` 适配器接收 ``(cache_port, logger_port)``： + 仅做账户级缓存清理与白名单数据存储，不调用渠道侧 API，无需 ILinkClient + （对齐 wechat_woc 模式）。 - ``probeable`` 适配器接收 ``(client, persistence_port, logger_port)``： 需动态发现 account_id 调用 getconfig 探活。 -- ``whitelist`` 适配器接收 ``(cache_port, logger_port)``：白名单数据本地 - 存储于 CachePort，不调用渠道侧 API，无需 ILinkClient。 - ``session`` / ``status`` 适配器无状态（INV-5），无 ``__init__`` 参数， 仅做协议转换，所有数据从 raw_event.payload 提取。 @@ -110,7 +111,7 @@ def channel_entry(host: PluginHost, manifest: ChannelManifest) -> PluginManifest # 2. 实例化 ILinkClient 与 LifecycleHandler # 依赖关系为单向：handler 在 onInit 调 client.attach_http_client 注入连接池， # client 不引用 handler。先创建 client，再创建 handler 并通过构造器注入。 - # config_schema 取自 manifest（F-03 单真相源）。 + # config_schema / max_connections 取自 manifest（F-03 单真相源）。 client = ILinkClient(config_port, cache_port, logger_port) handler = WeChatILinkLifecycleHandler( config_port, @@ -118,21 +119,24 @@ def channel_entry(host: PluginHost, manifest: ChannelManifest) -> PluginManifest cache_port, manifest.config_schema, client, + max_connections=(manifest.resource_quota.max_connections if manifest.resource_quota is not None else 10), ) # 3. 实例化并注册 11 个适配器 # puller / outbound 接收 (client, config_port, logger_port)： # 适配器直接持有 ConfigPort 读取渠道级配置（对齐 wechat_woc 模式）。 - # inbound / login / lifecycle / doctor / wizard 接收 (client, logger_port)； + # inbound / login / doctor / wizard 接收 (client, logger_port)； + # lifecycle / whitelist 接收 (cache_port, logger_port)（账户级缓存清理， + # 不调用渠道侧 API，对齐 wechat_woc 模式）； # probeable 接收 (client, persistence_port, logger_port)； - # whitelist 接收 (cache_port, logger_port)；session / status 无状态。 + # session / status 无状态。 host.registerAdapter("puller", WeChatILinkPullerAdapter(client, config_port, logger_port)) host.registerAdapter("inbound", WeChatILinkInboundAdapter(client, logger_port)) host.registerAdapter("outbound", WeChatILinkOutboundAdapter(client, config_port, logger_port)) host.registerAdapter("session", WeChatILinkSessionAdapter()) host.registerAdapter("status", WeChatILinkStatusAdapter()) host.registerAdapter("login", WeChatILinkLoginAdapter(client, logger_port)) - host.registerAdapter("lifecycle", WeChatILinkLifecycleAdapter(client, logger_port)) + host.registerAdapter("lifecycle", WeChatILinkLifecycleAdapter(cache_port, logger_port)) host.registerAdapter("probeable", WeChatILinkProbeableAdapter(client, persistence_port, logger_port)) host.registerAdapter("doctor", WeChatILinkDoctorAdapter(client, logger_port)) host.registerAdapter("wizard", WeChatILinkWizardAdapter(client, logger_port)) diff --git a/backend/package/yuxi/channels/plugins/wechat_ilink/ilink_client.py b/backend/package/yuxi/channels/plugins/wechat_ilink/ilink_client.py index 1d3b2cba..558b4cdb 100644 --- a/backend/package/yuxi/channels/plugins/wechat_ilink/ilink_client.py +++ b/backend/package/yuxi/channels/plugins/wechat_ilink/ilink_client.py @@ -29,21 +29,17 @@ from yuxi.channels.contract.ports.driven.config_port import ConfigPort from yuxi.channels.contract.ports.driven.logger_port import LoggerPort from . import crypto, error_translator - -# 渠道级配置的 target（插件 provides 值） -# 公开供同插件适配器（puller / outbound）复用，避免各模块重复定义字符串字面量 -CHANNEL_TARGET = "wechat_ilink" - -# 渠道级配置默认值（与 manifest.json 保持一致） -_DEFAULT_BASEURL = "https://ilinkai.weixin.qq.com" -_DEFAULT_CHANNEL_VERSION = "2.0.0" -_DEFAULT_LONG_POLL_TIMEOUT_MS = 35000 - -# 长轮询客户端超时余量（ms）：长于服务端 35s hang，避免客户端先超时 -_LONG_POLL_EXTRA_MS = 5000 +from ._constants import ( + CHANNEL_TARGET, + DEFAULT_BASEURL, + DEFAULT_CHANNEL_VERSION, + DEFAULT_LONG_POLL_TIMEOUT_MS, + HTTP_TIMEOUT_SECONDS, + LONG_POLL_EXTRA_MS, +) # 普通请求默认超时（秒） -_DEFAULT_REQUEST_TIMEOUT_S = 30.0 +_DEFAULT_REQUEST_TIMEOUT_S = HTTP_TIMEOUT_SECONDS class ILinkClient: @@ -337,14 +333,14 @@ class ILinkClient: timeout_seconds: 请求超时秒数。 """ if base_url is None: - baseurl = await self.get_channel_config("baseurl", _DEFAULT_BASEURL) + baseurl = await self.get_channel_config("baseurl", DEFAULT_BASEURL) else: baseurl = base_url effective_body = body if body is not None else {} if inject_base_info: channel_version = await self.get_channel_config( "channel_version", - _DEFAULT_CHANNEL_VERSION, + DEFAULT_CHANNEL_VERSION, ) base_info = effective_body.get("base_info") if isinstance(base_info, dict): @@ -383,7 +379,7 @@ class ILinkClient: - 返回校验后的 JSON 响应体 """ if base_url is None: - base_url = await self.get_channel_config("baseurl", _DEFAULT_BASEURL) + base_url = await self.get_channel_config("baseurl", DEFAULT_BASEURL) elif not base_url.startswith(("http://", "https://")): base_url = f"https://{base_url}" url = f"{base_url}{path}" @@ -414,9 +410,9 @@ class ILinkClient: """ timeout_ms = await self.get_channel_config( "long_poll_timeout_ms", - _DEFAULT_LONG_POLL_TIMEOUT_MS, + DEFAULT_LONG_POLL_TIMEOUT_MS, ) - timeout_s = (int(timeout_ms) + _LONG_POLL_EXTRA_MS) / 1000.0 + timeout_s = (int(timeout_ms) + LONG_POLL_EXTRA_MS) / 1000.0 body = {"get_updates_buf": cursor} return await self._post( "/ilink/bot/getupdates", @@ -438,7 +434,7 @@ class ILinkClient: {"msg": msg}, ) - async def get_bot_qrcode(self, account_id: str = "") -> dict: + async def get_bot_qrcode(self) -> dict: """获取登录二维码（登录前，无需鉴权）。 GET /ilink/bot/get_bot_qrcode?bot_type=3 @@ -466,9 +462,9 @@ class ILinkClient: """ timeout_ms = await self.get_channel_config( "long_poll_timeout_ms", - _DEFAULT_LONG_POLL_TIMEOUT_MS, + DEFAULT_LONG_POLL_TIMEOUT_MS, ) - timeout_s = (int(timeout_ms) + _LONG_POLL_EXTRA_MS) / 1000.0 + timeout_s = (int(timeout_ms) + LONG_POLL_EXTRA_MS) / 1000.0 return await self._get( "/ilink/bot/get_qrcode_status", None, @@ -598,7 +594,7 @@ class ILinkClient: from_user_id: str, ) -> str | None: """读取 context_token，未命中返回 None。""" - key = f"wechat_ilink:context_token:{account_id}:{from_user_id}" + key = f"wechat_ilink:{account_id}:context_token:{from_user_id}" cached = await self._cache.get(key) if cached.is_some(): return cached.unwrap() @@ -611,26 +607,9 @@ class ILinkClient: context_token: str, ) -> None: """写入 context_token。""" - key = f"wechat_ilink:context_token:{account_id}:{from_user_id}" + key = f"wechat_ilink:{account_id}:context_token:{from_user_id}" await self._cache.set(key, context_token) - async def clear_context_token( - self, - account_id: str, - from_user_id: str | None = None, - ) -> None: - """清除 context_token。 - - from_user_id 为 None 时清除该 account_id 下所有 context_token - （使用 invalidate 模式匹配）。 - """ - if from_user_id is None: - pattern = f"wechat_ilink:context_token:{account_id}:*" - await self._cache.invalidate(pattern) - else: - key = f"wechat_ilink:context_token:{account_id}:{from_user_id}" - await self._cache.delete(key) - # ------------------------------------------------------------------ # cursor 缓存 # ------------------------------------------------------------------ @@ -641,7 +620,7 @@ class ILinkClient: 供 ``DoctorAdapter`` 健康检查使用：非空游标表示长轮询已建立。 游标写入由框架通过 ``PollResult.next_cursor`` 管理，适配器不持久化。 """ - key = f"wechat_ilink:cursor:{account_id}" + key = f"wechat_ilink:{account_id}:cursor" cached = await self._cache.get(key) if cached.is_some(): return cached.unwrap() diff --git a/backend/package/yuxi/channels/plugins/wechat_ilink/lifecycle.py b/backend/package/yuxi/channels/plugins/wechat_ilink/lifecycle.py index 5f58a082..e0338034 100644 --- a/backend/package/yuxi/channels/plugins/wechat_ilink/lifecycle.py +++ b/backend/package/yuxi/channels/plugins/wechat_ilink/lifecycle.py @@ -36,10 +36,15 @@ from yuxi.channels.contract.ports.driven.logger_port import LoggerPort if TYPE_CHECKING: from .ilink_client import ILinkClient -# httpx 连接池配置（对齐 manifest.json resource_quota: max_connections=10） -_HTTP_TIMEOUT_SECONDS = 30.0 -_HTTP_MAX_CONNECTIONS = 10 -_HTTP_MAX_KEEPALIVE_CONNECTIONS = 5 +from ._constants import ( + DEFAULT_MAX_CONNECTIONS, + DEFAULT_MAX_KEEPALIVE_CONNECTIONS, + DRAIN_TIMEOUT_SECONDS, + HTTP_TIMEOUT_SECONDS, +) + +# CachePort 失效模式（清空插件所有运行时缓存，避免孤儿键残留） +_CACHE_INVALIDATE_PATTERN = "wechat_ilink:*" class WeChatILinkLifecycleHandler: @@ -58,11 +63,13 @@ class WeChatILinkLifecycleHandler: cache_port: CachePort, config_schema: tuple[ConfigField, ...], client: ILinkClient | None = None, + max_connections: int = DEFAULT_MAX_CONNECTIONS, ) -> None: self._config = config_port self._logger = logger_port self._cache = cache_port self._client = client + self._max_connections = max_connections self._http_client: httpx.AsyncClient | None = None self._started: bool = False # 从 config_schema 派生不可热更新键（hot_reloadable=False 的字段 key） @@ -77,10 +84,10 @@ class WeChatILinkLifecycleHandler: async def onInit(self) -> None: """初始化资源：创建 httpx 连接池并注入 ILinkClient。""" self._http_client = httpx.AsyncClient( - timeout=_HTTP_TIMEOUT_SECONDS, + timeout=HTTP_TIMEOUT_SECONDS, limits=httpx.Limits( - max_connections=_HTTP_MAX_CONNECTIONS, - max_keepalive_connections=_HTTP_MAX_KEEPALIVE_CONNECTIONS, + max_connections=self._max_connections, + max_keepalive_connections=DEFAULT_MAX_KEEPALIVE_CONNECTIONS, ), ) if self._client is not None: @@ -106,7 +113,12 @@ class WeChatILinkLifecycleHandler: self._started = False if self._client is not None: self._client.set_active(False) - await self._client.drain(timeout=30.0) + drained = await self._client.drain(timeout=DRAIN_TIMEOUT_SECONDS) + if not drained: + await self._logger.warn( + "WeChat iLink plugin drain timed out, in-flight requests may be interrupted", + timeout=DRAIN_TIMEOUT_SECONDS, + ) async def onPause(self) -> None: """暂停插件：停止接收新请求。""" @@ -123,17 +135,24 @@ class WeChatILinkLifecycleHandler: await self._logger.info("WeChat iLink plugin resumed") async def onUnload(self) -> None: - """释放所有资源：drain 在途请求后关闭 httpx 连接池。 + """释放所有资源：drain 在途请求后关闭 httpx 连接池并清空缓存。 长轮询连接由 ``StreamWorker`` 在宿主关停时首先停止， 本方法先 drain 在途请求（默认 30s 超时），再解除 client 引用并 - 关闭连接池。 + 关闭连接池，最后清空 CachePort 中 ``wechat_ilink:*`` 缓存避免 + 孤儿键残留（FR-32 资源释放约束）。 """ - # 先 drain 在途请求，避免 aclose 打断正在进行的请求 + # 1. 先 drain 在途请求，避免 aclose 打断正在进行的请求 if self._client is not None: self._client.set_active(False) - await self._client.drain(timeout=30.0) + drained = await self._client.drain(timeout=DRAIN_TIMEOUT_SECONDS) + if not drained: + await self._logger.warn( + "WeChat iLink plugin unload drain timed out, in-flight requests may be interrupted", + timeout=DRAIN_TIMEOUT_SECONDS, + ) self._client.detach_http_client() + # 2. 关闭 httpx 连接池（异常告警不抛，避免阻断后续缓存清理） if self._http_client is not None: try: await self._http_client.aclose() @@ -144,6 +163,15 @@ class WeChatILinkLifecycleHandler: ) self._http_client = None + # 3. 清空 CachePort 中 wechat_ilink:* 缓存，避免孤儿键残留 + try: + await self._cache.invalidate(pattern=_CACHE_INVALIDATE_PATTERN) + except Exception as exc: + await self._logger.warn( + "WeChat iLink cache invalidate failed during unload", + error=str(exc), + ) + self._started = False await self._logger.info("WeChat iLink plugin unloaded") diff --git a/backend/package/yuxi/channels/plugins/wechat_ilink/manifest.json b/backend/package/yuxi/channels/plugins/wechat_ilink/manifest.json index 07600caa..402e909c 100644 --- a/backend/package/yuxi/channels/plugins/wechat_ilink/manifest.json +++ b/backend/package/yuxi/channels/plugins/wechat_ilink/manifest.json @@ -26,6 +26,7 @@ }, "max_message_length": 4000, "supports_credential_cloning": false, + "transport_mode": "pull", "capabilities": { "rich_message": false, "streaming": false, @@ -54,6 +55,18 @@ "supports_qr_login": true, "agent_collaboration": false }, + "capability_requirements": [ + ["supports_image_inbound", ["bot_token", "ilink_bot_id", "ilink_user_id", "baseurl"]], + ["supports_video_inbound", ["bot_token", "ilink_bot_id", "ilink_user_id", "baseurl"]], + ["supports_image_outbound", ["bot_token", "ilink_bot_id", "ilink_user_id", "baseurl"]], + ["supports_video_outbound", ["bot_token", "ilink_bot_id", "ilink_user_id", "baseurl"]], + ["doctor", ["bot_token", "ilink_bot_id", "ilink_user_id", "baseurl"]], + ["whitelist", ["bot_token", "ilink_bot_id", "ilink_user_id", "baseurl"]], + ["wizard", ["bot_token", "ilink_bot_id", "ilink_user_id", "baseurl"]], + ["probeable", []], + ["lifecycle", []], + ["supports_qr_login", ["bot_token", "ilink_bot_id", "ilink_user_id", "baseurl"]] + ], "resource_quota": { "max_cpu": "15.0", "max_memory": "128MB", @@ -77,16 +90,6 @@ {"key": "long_poll_timeout_ms", "type": "integer", "required": false, "default": 35000, "hot_reloadable": true, "scope": "channel", "constraints": {"min": 10000, "max": 60000}}, {"key": "qr_poll_interval_ms", "type": "integer", "required": false, "default": 2000, "hot_reloadable": true, "scope": "channel", "constraints": {"min": 1000, "max": 5000}}, {"key": "qr_timeout_ms", "type": "integer", "required": false, "default": 120000, "hot_reloadable": true, "scope": "channel", "constraints": {"min": 30000, "max": 300000}}, - {"key": "max_message_length", "type": "integer", "required": false, "default": 4000, "hot_reloadable": true, "scope": "channel", "constraints": {"min": 500, "max": 10000}}, - {"key": "enable_media", "type": "boolean", "required": false, "default": true, "hot_reloadable": true, "scope": "channel", "constraints": {}}, - {"key": "enable_voice", "type": "boolean", "required": false, "default": false, "hot_reloadable": true, "scope": "channel", "constraints": {}} - ], - "env_vars": [ - {"name": "ILINK_BASE_URL", "description": "iLink API 基础地址", "required": false, "default": "https://ilinkai.weixin.qq.com", "sensitive": false}, - {"name": "ILINK_CDN_URL", "description": "iLink CDN 地址", "required": false, "default": "https://novac2c.cdn.weixin.qq.com/c2c", "sensitive": false}, - {"name": "ILINK_CHANNEL_VERSION", "description": "iLink 渠道协议版本", "required": false, "default": "2.0.0", "sensitive": false}, - {"name": "ILINK_QR_TIMEOUT_MS", "description": "扫码登录超时时间（毫秒）", "required": false, "default": "120000", "sensitive": false}, - {"name": "ILINK_LONGPOLL_TIMEOUT_MS", "description": "长轮询超时时间（毫秒）", "required": false, "default": "35000", "sensitive": false}, - {"name": "ILINK_QR_POLL_INTERVAL_MS", "description": "扫码状态轮询间隔（毫秒）", "required": false, "default": "2000", "sensitive": false} + {"key": "max_message_length", "type": "integer", "required": false, "default": 4000, "hot_reloadable": true, "scope": "channel", "constraints": {"min": 500, "max": 10000}} ] }