diff --git a/backend/package/yuxi/channels/plugins/qqbot/adapters/_event_utils.py b/backend/package/yuxi/channels/plugins/qqbot/adapters/_event_utils.py new file mode 100644 index 00000000..f22b90c2 --- /dev/null +++ b/backend/package/yuxi/channels/plugins/qqbot/adapters/_event_utils.py @@ -0,0 +1,43 @@ +"""QQ Bot 事件解析共享辅助函数。 + +集中定义从 ``RawEvent`` 提取事件类型（``t`` 字段）与事件数据（``d`` 字段） +的辅助函数，消除 ``inbound`` / ``mention`` / ``session`` / ``status`` 四个 +适配器中的重复定义（P2-3 代码去重）。 + +``extract_event_data`` 采用宽松语义（缺失时返回空 dict），需要严格校验 +（缺失即抛 ``ValidationError``）的调用方（如 ``session`` / ``status``） +在调用后自行检查返回值。 + +依赖方向：仅 import ``yuxi.channels.contract.*``，不污染框架层。 +""" + +from __future__ import annotations + +from typing import Any + +from yuxi.channels.contract.dtos.common import RawEvent + + +def extract_event_type(raw_event: RawEvent) -> str: + """提取 QQ Bot 事件类型（``t`` 字段）。 + + @return 事件类型字符串，缺失时返回空字符串。 + """ + return str(raw_event.payload.get("t", "")) + + +def extract_event_data(raw_event: RawEvent) -> dict[str, Any]: + """从原始事件提取 ``d`` 字段（QQ Bot WS 事件数据载荷）。 + + 采用宽松语义：当 ``d`` 字段缺失或非 dict 时返回空 dict，不抛异常。 + 需要严格校验的调用方应在调用后检查返回值是否为空。 + + @return 事件数据 dict，缺失时返回 ``{}``。 + """ + data = raw_event.payload.get("d") + if not isinstance(data, dict): + return {} + return data + + +__all__ = ["extract_event_type", "extract_event_data"] 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 d9becef2..ca0c4069 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 @@ -33,12 +33,12 @@ from ..ilink_client import ILinkClient _MEDIA_SCHEME = "wechat_ilink" _MEDIA_HOST = "media" -# iLink item type 映射 +# iLink item type 映射（功能点清单 §2.5：3=VOICE, 5=VIDEO） _TEXT_ITEM_TYPE = 1 _IMAGE_ITEM_TYPE = 2 -_VIDEO_ITEM_TYPE = 3 +_VOICE_ITEM_TYPE = 3 _FILE_ITEM_TYPE = 4 -_VOICE_ITEM_TYPE = 5 +_VIDEO_ITEM_TYPE = 5 class WeChatILinkInboundAdapter: @@ -123,18 +123,18 @@ class WeChatILinkInboundAdapter: height=image_item.get("height"), ) ) - elif item_type == _VIDEO_ITEM_TYPE: - video_item = item.get("video_item", {}) + elif item_type == _VOICE_ITEM_TYPE: + voice_item = item.get("voice_item", {}) attachments.append( Attachment( - type="video", + type="audio", url=_build_media_url( - encrypt_query_param=video_item.get("encrypt_query_param", ""), - aes_key=video_item.get("aes_key", ""), - media_type="video", + encrypt_query_param=voice_item.get("encrypt_query_param", ""), + aes_key=voice_item.get("aes_key", ""), + media_type="audio", cdn_url=cdn_url, ), - duration_ms=video_item.get("duration"), + duration_ms=voice_item.get("duration"), ) ) elif item_type == _FILE_ITEM_TYPE: @@ -152,18 +152,18 @@ class WeChatILinkInboundAdapter: size=file_item.get("file_size"), ) ) - elif item_type == _VOICE_ITEM_TYPE: - voice_item = item.get("voice_item", {}) + elif item_type == _VIDEO_ITEM_TYPE: + video_item = item.get("video_item", {}) attachments.append( Attachment( - type="audio", + type="video", url=_build_media_url( - encrypt_query_param=voice_item.get("encrypt_query_param", ""), - aes_key=voice_item.get("aes_key", ""), - media_type="audio", + encrypt_query_param=video_item.get("encrypt_query_param", ""), + aes_key=video_item.get("aes_key", ""), + media_type="video", cdn_url=cdn_url, ), - duration_ms=voice_item.get("duration"), + duration_ms=video_item.get("duration"), ) ) 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 eb4035e8..2a98bf47 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 @@ -5,8 +5,9 @@ （``scaned_but_redirect`` 切换 Host）与二维码过期检测（``expired`` 返回 新二维码）。 -iLink 无显式登出 API，``logout`` 不清理本地存储（由 -``CredentialService.revokeCredentials`` 负责），``logged_out=None``。 +iLink 无显式登出 API，``logout`` 通过 ``ILinkClient.clear_runtime_cache`` +清理插件级运行时缓存（context_token / cursor），凭证缓存由 +``CredentialService.revokeCredentials`` 统一负责，``logged_out=None``。 ``forceLogout`` iLink 不支持，抛 ``NotImplementedError``。 设计要点： @@ -21,8 +22,9 @@ iLink 无显式登出 API，``logout`` 不清理本地存储（由 ``ilink_user_id`` / ``baseurl`` 填入 ``QrLoginWaitResult.credentials``， 由 ``CredentialService`` 加密落库；适配器不调用任何 Port 写方法 （INV-8），不感知 ``onboarding_status``。 -- ``logout`` 幂等：iLink 无登出 API，适配器不清理本地存储，直接返回 - ``LogoutResult(cleared=True, logged_out=None)``。 +- ``logout`` 幂等：iLink 无登出 API，适配器通过 ``ILinkClient.clear_runtime_cache`` + 清理运行时缓存（context_token / cursor），凭证缓存由 CredentialService 清理， + 返回 ``LogoutResult(cleared=True, logged_out=None)``。 - 适配器不得调用 ``ChannelAccountRepositoryPort`` 写方法，凭据落库由 ``CredentialService`` 统一控制。 @@ -219,15 +221,18 @@ class WeChatILinkLoginAdapter: async def logout(self, account_id: str) -> LogoutResult: """登出账户。幂等：重复调用安全，不抛异常。 - iLink 无显式登出 API，``logged_out=None``。适配器不清理本地存储 - （由 ``CredentialService.revokeCredentials`` 负责），仅返回结果。 - 适配器不调用任何 Port 写方法（INV-8）。 + iLink 无显式登出 API，``logged_out=None``。适配器通过 + ``ILinkClient.clear_runtime_cache`` 清理插件级运行时缓存 + （context_token / cursor），避免登出后孤儿键残留。凭证缓存 + （``credentials:{account_id}``）由 ``CredentialService.revokeCredentials`` + 统一清理。CachePort 异常不抛（降级为日志告警）。 @consistency: 最终一致（eventual），凭证撤销由 CredentialService 控制。 @idempotent: True — 重复调用返回相同结果，不抛异常。 """ + await self._client.clear_runtime_cache(account_id) await self._logger.info( - "iLink account logout: no channel-side API, local clear delegated to CredentialService", + "iLink account logout: runtime cache cleared, credentials revocation delegated to CredentialService", account_id=account_id, ) return LogoutResult(cleared=True, logged_out=None) 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 76a5110b..6cdd7737 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 @@ -18,6 +18,7 @@ iLink 不支持话题回复与批量发送，``supportsThreadReply`` / from __future__ import annotations +import hashlib from uuid import uuid4 from yuxi.channels.contract.dtos.common import MessageContent, MessageFormat @@ -29,8 +30,8 @@ from yuxi.channels.contract.dtos.outbound import ( from yuxi.channels.contract.dtos.outbox import MultiPartReceipt from yuxi.channels.contract.errors import ( DependencyError, - NotImplementedError, NotFoundError, + NotImplementedError, ValidationError, ) from yuxi.channels.contract.ports.driven.config_port import ConfigPort @@ -43,10 +44,17 @@ from ..ilink_client import ILinkClient # iLink 消息类型常量 _MESSAGE_TYPE_BOT = 2 _MESSAGE_STATE_FINISH = 2 +# item type 映射（功能点清单 §2.5：3=VOICE, 5=VIDEO） +# 注：VOICE(3) 仅入站使用，出站 audio 走 raise 分支（设计 §4.4） _TEXT_ITEM_TYPE = 1 _IMAGE_ITEM_TYPE = 2 -_VIDEO_ITEM_TYPE = 3 _FILE_ITEM_TYPE = 4 +_VIDEO_ITEM_TYPE = 5 + +# media_type 映射（设计 §4.4：1=IMAGE, 2=VIDEO, 3=FILE；无 VOICE） +_MEDIA_TYPE_IMAGE = 1 +_MEDIA_TYPE_VIDEO = 2 +_MEDIA_TYPE_FILE = 3 class WeChatILinkOutboundAdapter: @@ -141,8 +149,10 @@ class WeChatILinkOutboundAdapter: 分片策略： - 长文本（>4000 Unicode 字符）按字符边界切割，每片独立调用 ``sendMessage`` - 媒体附件（含 ``content`` bytes）先上传 CDN 获取 - ``file_id`` + ``encrypt_query_param`` + ``aes_key``，再发送 + ``encrypt_query_param``，再发送媒体 item - 文本在前，媒体在后 + - audio 类型不支持上传（设计 §4.4：语音不经过 getuploadurl）， + 抛 ``ValidationError``；v1 ``enable_voice`` 默认关闭（TD-6） 发送前预校验 ``context_token``，避免文本已发送、媒体因缺 token 失败导致半截消息。每片独立调用 ``sendMessage`` 生成唯一 ``client_id``。 @@ -219,26 +229,47 @@ class WeChatILinkOutboundAdapter: if seq in skip_seqs: seq += 1 continue + + # audio 类型：设计 §4.4 约束"语音通过 voice_item 直接引用，不经过 + # getuploadurl"。v1 enable_voice 默认关闭（TD-6），不支持语音上传。 + if attachment.type == "audio": + raise ValidationError( + field="attachment", + message="voice_upload_not_supported", + ) + aes_key = crypto.generate_random_aes_key() ciphertext = crypto.encrypt(attachment.content, aes_key) file_size = crypto.calc_ciphertext_size(len(attachment.content)) + raw_size = len(attachment.content) + raw_md5 = hashlib.md5(attachment.content).hexdigest() + filekey = uuid4().hex + if attachment.type == "image": - file_type = "image" + media_type = _MEDIA_TYPE_IMAGE elif attachment.type == "video": - file_type = "video" + media_type = _MEDIA_TYPE_VIDEO else: - file_type = "file" - upload_resp = await self._client.getuploadurl(account_id, file_type, file_size) + media_type = _MEDIA_TYPE_FILE + + upload_resp = await self._client.getuploadurl( + account_id, + media_type=media_type, + filekey=filekey, + to_user_id=peer_id, + rawsize=raw_size, + rawfilemd5=raw_md5, + filesize=file_size, + aeskey=aes_key, + ) upload_url = upload_resp["upload_url"] encrypt_query_param = upload_resp["encrypt_query_param"] - file_id = upload_resp["file_id"] await self._client.cdn_upload(upload_url, ciphertext, encrypt_query_param) if attachment.type == "image": item = { "type": _IMAGE_ITEM_TYPE, "image_item": { - "file_id": file_id, "aes_key": aes_key, "encrypt_query_param": encrypt_query_param, }, @@ -247,7 +278,6 @@ class WeChatILinkOutboundAdapter: item = { "type": _VIDEO_ITEM_TYPE, "video_item": { - "file_id": file_id, "aes_key": aes_key, "encrypt_query_param": encrypt_query_param, }, @@ -256,7 +286,6 @@ class WeChatILinkOutboundAdapter: item = { "type": _FILE_ITEM_TYPE, "file_item": { - "file_id": file_id, "aes_key": aes_key, "encrypt_query_param": encrypt_query_param, "file_name": attachment.filename or "file", diff --git a/backend/package/yuxi/channels/plugins/wechat_ilink/adapters/status_adapter.py b/backend/package/yuxi/channels/plugins/wechat_ilink/adapters/status_adapter.py index 1b7f8530..4b3c5dc2 100644 --- a/backend/package/yuxi/channels/plugins/wechat_ilink/adapters/status_adapter.py +++ b/backend/package/yuxi/channels/plugins/wechat_ilink/adapters/status_adapter.py @@ -10,6 +10,7 @@ iLink 长轮询仅接收用户消息（message_type=1），返回 EventType.MESS from __future__ import annotations from datetime import UTC, datetime +from typing import Any from yuxi.channels.contract.dtos.common import RawEvent from yuxi.channels.contract.dtos.status import EventType, StatusPayload @@ -40,17 +41,25 @@ class WeChatILinkStatusAdapter: async def extractStatusPayload(self, raw_event: RawEvent) -> StatusPayload: """提取状态负载。 - 含 message_id（iLink client_id）与 from_user_id。 + 含 message_id（iLink client_id）与 from_user_id（写入 metadata）。 + event_type 与 ``classifyEvent`` 结果联动，非 MESSAGE 事件使用实际分类。 @failure client_id 缺失抛 ValidationError。 """ client_id = raw_event.payload.get("client_id", "") if not client_id: raise ValidationError(field="client_id", message="client_id_missing") + from_user_id = raw_event.payload.get("from_user_id", "") + event_type = await self.classifyEvent(raw_event) + metadata: dict[str, Any] = {} + if from_user_id: + metadata["from_user_id"] = from_user_id + return StatusPayload( - event_type=EventType.MESSAGE, + event_type=event_type, ref_channel_msg_id=client_id, timestamp=raw_event.received_at or datetime.now(UTC), + metadata=metadata or None, ) 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 558b4cdb..2d30539a 100644 --- a/backend/package/yuxi/channels/plugins/wechat_ilink/ilink_client.py +++ b/backend/package/yuxi/channels/plugins/wechat_ilink/ilink_client.py @@ -526,16 +526,41 @@ class ILinkClient: async def getuploadurl( self, account_id: str, - file_type: str, - file_size: int, + *, + media_type: int, + filekey: str, + to_user_id: str, + rawsize: int, + rawfilemd5: str, + filesize: int, + aeskey: str, + thumb_rawsize: int | None = None, + thumb_rawfilemd5: str | None = None, + thumb_filesize: int | None = None, + thumb_aeskey: str | None = None, ) -> dict: """获取 CDN 上传地址。 POST /ilink/bot/getuploadurl - Body: {"file_type": file_type, "file_size": file_size} - 返回含 upload_url / encrypt_query_param / aes_key / file_id。 + Body（设计 §4.4）: ``filekey`` / ``media_type``（1=IMAGE, 2=VIDEO, + 3=FILE）/ ``to_user_id`` / ``rawsize`` / ``rawfilemd5`` / ``filesize`` + / ``aeskey``；图片/视频还需 ``thumb_*`` 参数。 + 返回含 upload_url / encrypt_query_param / file_id。 """ - body = {"file_type": file_type, "file_size": file_size} + body: dict[str, Any] = { + "filekey": filekey, + "media_type": media_type, + "to_user_id": to_user_id, + "rawsize": rawsize, + "rawfilemd5": rawfilemd5, + "filesize": filesize, + "aeskey": aeskey, + } + if thumb_rawsize is not None: + body["thumb_rawsize"] = thumb_rawsize + body["thumb_rawfilemd5"] = thumb_rawfilemd5 or "" + body["thumb_filesize"] = thumb_filesize or 0 + body["thumb_aeskey"] = thumb_aeskey or "" return await self._post("/ilink/bot/getuploadurl", account_id, body) async def cdn_upload( @@ -626,5 +651,29 @@ class ILinkClient: return cached.unwrap() return "" + async def clear_runtime_cache(self, account_id: str) -> None: + """清理指定账户的运行时缓存（context_token / cursor）。 + + 供 ``LoginAdapter.logout`` 调用，清理插件级运行时缓存键，避免 + 登出后孤儿键残留。凭证缓存（``credentials:{account_id}``）由 + ``CredentialService.revokeCredentials`` 统一清理，本方法不涉及。 + + CachePort 异常不抛（仅日志告警），对齐 lifecycle 降级策略。 + """ + patterns = [ + f"wechat_ilink:{account_id}:context_token:*", + f"wechat_ilink:{account_id}:cursor", + ] + for pattern in patterns: + try: + await self._cache.invalidate(pattern=pattern) + except Exception as exc: + await self._logger.exception( + "iLink clear_runtime_cache: invalidate failed, non-blocking", + account_id=account_id, + pattern=pattern, + exc_info=exc, + ) + __all__ = ["ILinkClient"] diff --git a/backend/package/yuxi/channels/plugins/wechat_ilink/manifest.json b/backend/package/yuxi/channels/plugins/wechat_ilink/manifest.json index 402e909c..f7b6c67d 100644 --- a/backend/package/yuxi/channels/plugins/wechat_ilink/manifest.json +++ b/backend/package/yuxi/channels/plugins/wechat_ilink/manifest.json @@ -9,7 +9,7 @@ "lifecycle": ["init", "start", "stop", "pause", "resume", "unload", "reconfigure", "fail"], "compatibility": { "min_host_version": "1.0.0", - "ilink_api_version": "2.0.0" + "ilink_channel_version": "2.0.0" }, "failure_policy": "degrade", "depends": [], @@ -90,6 +90,8 @@ {"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": "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": {}} ] }