kris/ForcePilot
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

refactor(wechat-ilink): 重构事件工具与适配器，修复媒体类型映射

Browse Source 
1. 新增QQ Bot事件解析通用工具类，抽离重复的事件提取逻辑
2. 修复微信iLink适配器的媒体类型映射错误
3. 完善微信iLink状态适配器的事件类型与元数据支持
4. 新增运行时缓存清理方法，优化登出逻辑
5. 更新manifest配置，新增媒体相关开关
6. 重构出站适配器的媒体上传逻辑，修复参数问题
This commit is contained in:
Kris 2026-07-09 04:19:35 +08:00
parent c888db4e1f
commit ebbd552970
7 changed files with 182 additions and 45 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

43
backend/package/yuxi/channels/plugins/qqbot/adapters/_event_utils.py Normal file
View File

@ -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"]

34
backend/package/yuxi/channels/plugins/wechat_ilink/adapters/inbound_adapter.py
View File

@ -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"),
)
)

21
backend/package/yuxi/channels/plugins/wechat_ilink/adapters/login_adapter.py
View File

@ -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)

51
backend/package/yuxi/channels/plugins/wechat_ilink/adapters/outbound_adapter.py
View File

@ -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",

13
backend/package/yuxi/channels/plugins/wechat_ilink/adapters/status_adapter.py
View File

@ -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,
)

59
backend/package/yuxi/channels/plugins/wechat_ilink/ilink_client.py
View File

@ -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"]

6
backend/package/yuxi/channels/plugins/wechat_ilink/manifest.json
View File

@ -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": {}}
]
}