fix(acceptance): repair official account replies and processor navigation

This commit is contained in:
Hyu
2026-10-09 20:13:51 +08:00
parent fea35e8b3f
commit 396b09ecce
9 changed files with 256 additions and 101 deletions
+121 -66
View File
@@ -5,6 +5,7 @@ from langbot.libs.wecom_api.WXBizMsgCrypt3 import WXBizMsgCrypt
import xml.etree.ElementTree as ET
from quart import Quart, request
import hashlib
import hmac
from typing import Callable
from langbot.libs.official_account_api.oaevent import OAEvent
@@ -24,6 +25,28 @@ xml_template = """
_MAX_CALLBACK_BODY_BYTES = 1024 * 1024
async def _read_callback_xml(client, req):
body = await req.data
if len(body) > _MAX_CALLBACK_BODY_BYTES:
raise ValueError('Official Account callback body exceeds the size limit')
root = ET.fromstring(body)
if root.find('Encrypt') is not None:
crypt = WXBizMsgCrypt(client.token, client.aes, client.appid)
code, xml = await asyncio.to_thread(
crypt.DecryptMsg, body, req.args.get('msg_signature', ''),
req.args.get('timestamp', ''), req.args.get('nonce', ''),
)
if code != 0:
raise ValueError('Official Account callback decryption failed')
return xml.decode('utf-8') if isinstance(xml, bytes) else xml
digest = hashlib.sha1(''.join(sorted([
client.token, req.args.get('timestamp', ''), req.args.get('nonce', ''),
])).encode('utf-8')).hexdigest()
if not hmac.compare_digest(digest, req.args.get('signature', '')):
raise ValueError('Official Account callback signature verification failed')
return body.decode('utf-8')
class OAClient:
_STATE_TTL_SECONDS = 600
_STATE_MAX = 4096
@@ -63,6 +86,7 @@ class OAClient:
}
self.access_token_expiry_time = None
self.msg_id_map = {}
self._processing_tasks: set[asyncio.Task] = set()
self.generated_content = {}
self._msg_seen_at = {}
self._generated_at = {}
@@ -91,6 +115,9 @@ class OAClient:
self._generated_at.pop(message_id, None)
def clear(self) -> None:
for task in self._processing_tasks:
task.cancel()
self._processing_tasks.clear()
self.msg_id_map.clear()
self.generated_content.clear()
self._msg_seen_at.clear()
@@ -155,28 +182,17 @@ class OAClient:
)
return 'signature verification failed', 403
elif req.method == 'POST':
encryt_msg = await req.data
if len(encryt_msg) > _MAX_CALLBACK_BODY_BYTES:
raise ValueError('Official Account callback body exceeds the size limit')
wxcpt = WXBizMsgCrypt(self.token, self.aes, self.appid)
ret, xml_msg = await asyncio.to_thread(
wxcpt.DecryptMsg,
encryt_msg,
msg_signature,
timestamp,
nonce,
)
xml_msg = xml_msg.decode('utf-8')
if ret != 0:
await self.logger.error('消息解密失败')
raise Exception('消息解密失败')
xml_msg = await _read_callback_xml(self, req)
message_data = await self.get_message(xml_msg)
if message_data:
event = OAEvent.from_payload(message_data)
if event:
await self._handle_message(event)
from langbot.pkg.core.task_boundary import create_detached_task
task = create_detached_task(self._process_message(event))
self._processing_tasks.add(task)
task.add_done_callback(self._processing_tasks.discard)
root = await asyncio.to_thread(ET.fromstring, xml_msg)
from_user = root.find('FromUserName').text # 发送者
@@ -192,7 +208,7 @@ class OAClient:
to_user=from_user,
from_user=to_user,
create_time=int(time.time()),
content=content,
content=content.replace(']]>', ']]]]><![CDATA[>'),
)
return response_xml
@@ -202,15 +218,7 @@ class OAClient:
await asyncio.sleep(interval)
if self.msg_id_map.get(message_data['MsgId'], 1) == 3:
# response_xml = xml_template.format(
# to_user=from_user,
# from_user=to_user,
# create_time=int(time.time()),
# content = "请求失效:暂不支持公众号超过15秒的请求,如有需求,请联系 LangBot 团队。"
# )
print('请求失效:暂不支持公众号超过15秒的请求,如有需求,请联系 LangBot 团队。')
return ''
return 'success'
except Exception:
await self.logger.error(f'handle_callback_request失败: {traceback.format_exc()}')
@@ -219,14 +227,14 @@ class OAClient:
async def get_message(self, xml_msg: str):
root = await asyncio.to_thread(ET.fromstring, xml_msg)
message_data = {
'ToUserName': root.find('ToUserName').text,
'FromUserName': root.find('FromUserName').text,
'CreateTime': int(root.find('CreateTime').text),
'MsgType': root.find('MsgType').text,
'Content': root.find('Content').text if root.find('Content') is not None else None,
'MsgId': int(root.find('MsgId').text) if root.find('MsgId') is not None else None,
}
message_data = {element.tag: element.text for element in root}
message_data['CreateTime'] = int(message_data.get('CreateTime') or 0)
if message_data.get('MsgId'):
message_data['MsgId'] = int(message_data['MsgId'])
else:
message_data['MsgId'] = ':'.join(str(message_data.get(key) or '') for key in (
'FromUserName', 'CreateTime', 'Event', 'EventKey',
))
return message_data
@@ -249,6 +257,12 @@ class OAClient:
return decorator
async def _process_message(self, event: OAEvent):
try:
await self._handle_message(event)
except Exception:
await self.logger.error(f'OfficialAccount processing failed: {traceback.format_exc()}')
async def _handle_message(self, event: OAEvent):
"""
处理消息事件。
@@ -288,6 +302,7 @@ class OAClientForLongerResponse:
logger: None,
unified_mode: bool = False,
api_base_url: str = 'https://api.weixin.qq.com',
customer_service: bool = False,
):
self.token = token
self.aes = EncodingAESKey
@@ -313,8 +328,13 @@ class OAClientForLongerResponse:
}
self.access_token_expiry_time = None
self.loading_message = LoadingMessage
from .customer_service import CustomerServiceReplies
self.customer_service = CustomerServiceReplies(AppID, Appsecret, api_base_url) if customer_service else None
self._callback_responses = {}
self.msg_queue = {}
self.user_msg_queue = {}
self._processing_tasks: set[asyncio.Task] = set()
self._last_queue_cleanup = 0.0
self.logger = logger
@@ -336,6 +356,10 @@ class OAClientForLongerResponse:
def clear(self) -> None:
self.msg_queue.clear()
self.user_msg_queue.clear()
self._callback_responses.clear()
for task in self._processing_tasks:
task.cancel()
self._processing_tasks.clear()
async def handle_callback_request(self):
"""处理回调请求(独立端口模式,使用全局 request)。"""
@@ -393,27 +417,24 @@ class OAClientForLongerResponse:
return 'signature verification failed', 403
elif req.method == 'POST':
encryt_msg = await req.data
if len(encryt_msg) > _MAX_CALLBACK_BODY_BYTES:
raise ValueError('Official Account callback body exceeds the size limit')
wxcpt = WXBizMsgCrypt(self.token, self.aes, self.appid)
ret, xml_msg = await asyncio.to_thread(
wxcpt.DecryptMsg,
encryt_msg,
msg_signature,
timestamp,
nonce,
)
xml_msg = xml_msg.decode('utf-8')
if ret != 0:
await self.logger.error('消息解密失败')
raise Exception('消息解密失败')
xml_msg = await _read_callback_xml(self, req)
# 解析 XML
root = await asyncio.to_thread(ET.fromstring, xml_msg)
from_user = root.find('FromUserName').text
to_user = root.find('ToUserName').text
message_key = (from_user, root.findtext('MsgId') or (
root.findtext('CreateTime'), root.findtext('Event'), root.findtext('EventKey')
))
cached = self._callback_responses.get(message_key)
if cached and time.monotonic() - cached[0] < 600:
return cached[1]
def remember(response):
self._callback_responses[message_key] = (time.monotonic(), response)
while len(self._callback_responses) > self._MAX_USERS:
self._callback_responses.pop(next(iter(self._callback_responses)))
return response
if self.msg_queue.get(from_user) and self.msg_queue[from_user][0]['content']:
queue_top = self.msg_queue[from_user].pop(0)
@@ -428,20 +449,22 @@ class OAClientForLongerResponse:
to_user=from_user,
from_user=to_user,
create_time=int(time.time()),
content=queue_content,
content=queue_content.replace(']]>', ']]]]><![CDATA[>'),
)
return response_xml
return remember(response_xml)
else:
response_xml = xml_template.format(
to_user=from_user,
from_user=to_user,
create_time=int(time.time()),
content=self.loading_message,
content=self.loading_message.replace(']]>', ']]]]><![CDATA[>'),
)
if not self.loading_message:
response_xml = 'success'
if self.user_msg_queue.get(from_user) and self.user_msg_queue[from_user][0]['content']:
return response_xml
if self.user_msg_queue.get(from_user):
return remember(response_xml)
else:
message_data = await self.get_message(xml_msg)
@@ -457,9 +480,15 @@ class OAClientForLongerResponse:
-self._MAX_MESSAGES_PER_USER :
]
self._prune_queues()
await self._handle_message(event)
# The callback must return the loading reply immediately.
# Native listeners establish their own Workspace scope.
from langbot.pkg.core.task_boundary import create_detached_task
return response_xml
task = create_detached_task(self._process_pending_message(event))
self._processing_tasks.add(task)
task.add_done_callback(self._processing_tasks.discard)
return remember(response_xml)
except Exception:
await self.logger.error(f'handle_callback_request失败: {traceback.format_exc()}')
@@ -468,14 +497,14 @@ class OAClientForLongerResponse:
async def get_message(self, xml_msg: str):
root = await asyncio.to_thread(ET.fromstring, xml_msg)
message_data = {
'ToUserName': root.find('ToUserName').text,
'FromUserName': root.find('FromUserName').text,
'CreateTime': int(root.find('CreateTime').text),
'MsgType': root.find('MsgType').text,
'Content': root.find('Content').text if root.find('Content') is not None else None,
'MsgId': int(root.find('MsgId').text) if root.find('MsgId') is not None else None,
}
message_data = {element.tag: element.text for element in root}
message_data['CreateTime'] = int(message_data.get('CreateTime') or 0)
if message_data.get('MsgId'):
message_data['MsgId'] = int(message_data['MsgId'])
else:
message_data['MsgId'] = ':'.join(str(message_data.get(key) or '') for key in (
'FromUserName', 'CreateTime', 'Event', 'EventKey',
))
return message_data
@@ -508,7 +537,32 @@ class OAClientForLongerResponse:
for handler in self._message_handlers[msg_type]:
await handler(event)
async def _process_pending_message(self, event: OAEvent):
try:
await self._handle_message(event)
except asyncio.CancelledError:
raise
except Exception:
self.user_msg_queue.pop(event.user_id, None)
await self.logger.error(f'OfficialAccount passive processing failed: {traceback.format_exc()}')
async def set_message(self, from_user: int, message_id: int, content: str):
if self.customer_service is not None:
try:
code = await self.customer_service.send(str(from_user), content)
except Exception as exc:
# Do not expose request URLs containing access tokens or AppSecret.
self.user_msg_queue.pop(from_user, None)
raise RuntimeError(f'OfficialAccount customer message failed ({type(exc).__name__})') from None
if code == 0:
self.user_msg_queue.pop(from_user, None)
return False
if code not in {48001, 45015, 45047, 43004}:
self.user_msg_queue.pop(from_user, None)
raise RuntimeError(f'OfficialAccount customer message rejected: errcode={code}')
await self.logger.info(
f'OfficialAccount customer message unavailable (errcode={code}); reply queued for next message'
)
if from_user not in self.msg_queue:
self.msg_queue[from_user] = []
@@ -520,3 +574,4 @@ class OAClientForLongerResponse:
)
self.msg_queue[from_user] = self.msg_queue[from_user][-self._MAX_MESSAGES_PER_USER :]
self._prune_queues()
return True
@@ -0,0 +1,46 @@
"""Optional asynchronous replies through WeChat's customer service API."""
import asyncio
import time
import httpx
class CustomerServiceReplies:
def __init__(self, appid: str, secret: str, base_url: str):
self.appid = appid
self.secret = secret
self.base_url = base_url.rstrip('/')
self._token = ''
self._expires = 0.0
self._lock = asyncio.Lock()
async def send(self, user_id: str, content: str) -> int:
# Retry only explicit token rejection, never an ambiguous network failure.
async with self._lock, httpx.AsyncClient(timeout=20) as client:
for attempt in range(2):
if time.monotonic() >= self._expires:
response = await client.get(
f'{self.base_url}/cgi-bin/token',
params={'grant_type': 'client_credential', 'appid': self.appid, 'secret': self.secret},
)
if response.status_code != 200:
raise RuntimeError(f'OfficialAccount token request HTTP {response.status_code}')
data = response.json()
if data.get('errcode'):
return int(data['errcode'])
self._token = data['access_token']
self._expires = time.monotonic() + max(0, int(data['expires_in']) - 120)
response = await client.post(
f'{self.base_url}/cgi-bin/message/custom/send',
params={'access_token': self._token},
json={'touser': user_id, 'msgtype': 'text', 'text': {'content': content}},
)
if response.status_code != 200:
raise RuntimeError(f'OfficialAccount customer message HTTP {response.status_code}')
code = int(response.json()['errcode'])
if code in {40001, 40014, 42001} and attempt == 0:
self._expires = 0.0
continue
return code
raise RuntimeError('OfficialAccount token refresh failed')
+21 -2
View File
@@ -465,7 +465,11 @@ def build_platform_tool_resources(
reason = None
if definition is None:
reason = 'unknown_tool'
elif definition.api not in supported_apis:
elif definition.api not in supported_apis and not (
definition.name == 'event_reply'
and {'reply_message', 'get_message'} <= supported_apis
and (event.delivery.reply_target or {}).get('message_id')
):
reason = 'adapter_api_unsupported'
elif definition.scope == 'event' and not _event_matches(event.event_type, definition.event_patterns):
reason = 'event_incompatible'
@@ -727,9 +731,24 @@ async def execute_platform_tool(
bot = await ap.platform_mgr.get_bot_by_uuid(execution_context, bot_id)
if bot is None:
raise ValueError(f'Bot {bot_id} is not running')
if definition.api not in set(bot.adapter.get_supported_apis() or []):
supported_apis = set(bot.adapter.get_supported_apis() or [])
reply_via_source = (
definition.name == 'event_reply'
and 'send_message' not in supported_apis
and {'reply_message', 'get_message'} <= supported_apis
and (delivery.get('reply_target') or {}).get('message_id')
)
if definition.api not in supported_apis and not reply_via_source:
raise ValueError(f'Platform API {definition.api} is no longer supported by bot {bot_id}')
api_func = getattr(bot.adapter, definition.api, None)
if reply_via_source:
async def api_func(target_type, target_id, message):
source = await bot.adapter.get_message(
chat_type=target_type,
chat_id=target_id,
message_id=delivery['reply_target']['message_id'],
)
return await bot.adapter.reply_message(source, message)
if not callable(api_func):
raise ValueError(f'Platform API {definition.api} is declared but not implemented')
if definition.api == 'send_message':
+4 -3
View File
@@ -704,7 +704,8 @@ class BotService:
bot = await self.get_bot(context, bot_data['uuid'], include_secret=True)
try:
await self.ap.platform_mgr.load_bot(context, bot)
if bot.get('enable'):
await self.ap.platform_mgr.load_bot(context, bot)
except Exception as exc:
raise BotApplyError(bot_error_message(exc, bot), bot['uuid']) from exc
@@ -740,8 +741,8 @@ class BotService:
bot = await self.get_bot(context, bot_uuid, include_secret=True)
try:
await self.ap.platform_mgr.remove_bot(context, bot_uuid)
runtime_bot = await self.ap.platform_mgr.load_bot(context, bot)
if runtime_bot.enable:
if bot.get('enable'):
runtime_bot = await self.ap.platform_mgr.load_bot(context, bot)
await runtime_bot.run()
except Exception as exc:
raise BotApplyError(bot_error_message(exc, bot), bot['uuid']) from exc
@@ -56,13 +56,14 @@ class OfficialAccountAdapter(OfficialAccountAPIMixin, abstract_platform_adapter.
}
if mode == 'drop':
bot = OAClient(**common_kwargs)
elif mode == 'passive':
elif mode in ('passive', 'auto'):
bot = OAClientForLongerResponse(
**common_kwargs,
LoadingMessage=config.get('LoadingMessage', ''),
customer_service=mode == 'auto',
)
else:
raise KeyError('OfficialAccount Mode must be "drop" or "passive"')
raise KeyError('OfficialAccount Mode must be "drop", "passive" or "auto"')
super().__init__(
config=config,
@@ -112,11 +113,12 @@ class OfficialAccountAdapter(OfficialAccountAPIMixin, abstract_platform_adapter.
if not isinstance(source, OAEvent):
raise ValueError('OfficialAccount reply_message requires an OAEvent source object')
content = await OfficialAccountMessageConverter.yiri2target(message)
if self.config.get('Mode') == 'passive':
await self.bot.set_message(source.user_id, source.message_id, content)
if self.config.get('Mode') in ('passive', 'auto'):
queued = await self.bot.set_message(source.user_id, source.message_id, content)
else:
await self.bot.set_message(source.message_id, content)
return platform_events.MessageResult(message_id=source.message_id, raw={'queued': True})
queued = True
return platform_events.MessageResult(message_id=source.message_id, raw={'queued': queued})
async def call_platform_api(self, action: str, params: dict = {}) -> dict:
handler = PLATFORM_API_MAP.get(action)
@@ -34,6 +34,22 @@ spec:
type: array[string]
required: false
default: []
- name: AppID
label:
en_US: App ID
zh_Hans: 应用 ID
zh_Hant: 應用 ID
type: string
required: true
default: ""
- name: AppSecret
label:
en_US: App Secret
zh_Hans: 应用密钥
zh_Hant: 應用密鑰
type: string
required: true
default: ""
- name: webhook_url
label:
en_US: Webhook Callback URL
@@ -62,32 +78,32 @@ spec:
type: string
required: true
default: ""
- name: AppID
label:
en_US: App ID
zh_Hans: 应用 ID
zh_Hant: 應用 ID
type: string
required: true
default: ""
- name: AppSecret
label:
en_US: App Secret
zh_Hans: 应用密钥
zh_Hant: 應用密鑰
type: string
required: true
default: ""
- name: Mode
label:
en_US: Mode
zh_Hans: 接入模式
zh_Hant: 接入模式
description:
en_US: "drop replies within the current callback; passive returns a loading message first and queues the real reply for the user's next message."
zh_Hans: "drop 会在当前回调内等待回复;passive 会先返回加载提示,并将真实回复排队到用户下一条消息。"
zh_Hant: "drop 會在目前回調內等待回覆;passive 會先回傳載入提示,並將真實回覆排隊到使用者下一則訊息。"
type: string
en_US: "Auto uses customer service messaging when available, falling back to next-message retrieval for permission, window or quota limits. Passive always waits for the next message. Drop requires a reply within 5 seconds."
zh_Hans: "自动模式在生成完成后通过客服接口回复;无权限、超出回复窗口或额度不足时,改为下一条消息领取。passive 始终由下一条消息领取;drop 需要在 5 秒内完成回复。客服接口通常需企业主体认证,以微信实际权限为准。"
zh_Hant: "自動模式在生成完成後透過客服介面回覆;無權限、超出回覆時限或額度不足時,改為下一則訊息領取。passive 始終由下一則訊息領取;drop 需要在 5 秒內完成回覆。客服介面通常需企業主體認證,以微信實際權限為準。"
type: select
options:
- name: auto
label:
en_US: Automatic reply with fallback (auto)
zh_Hans: 自动回复,受限时转为领取(auto)
zh_Hant: 自動回覆,受限時轉為領取(auto)
- name: drop
label:
en_US: Reply in current callback (drop)
zh_Hans: 当前回调内回复(drop)
zh_Hant: 目前回調內回覆(drop)
- name: passive
label:
en_US: Queue reply for next message (passive)
zh_Hans: 下一条消息时返回回复(passive)
zh_Hant: 下一則訊息時傳回回覆(passive)
required: true
default: "drop"
- name: LoadingMessage
+9 -2
View File
@@ -450,10 +450,13 @@ class RuntimeBot:
return {}
if hasattr(model, 'model_dump'):
try:
return model.model_dump(mode='json')
# Native adapter objects are for replying, not event logging.
# Exclude them before JSON serialization, since some (OAEvent,
# for example) cannot be serialized by Pydantic.
return model.model_dump(mode='json', exclude={'source_platform_object'})
except TypeError:
try:
return model.model_dump()
return model.model_dump(exclude={'source_platform_object'})
except Exception:
return {}
except Exception:
@@ -2231,6 +2234,10 @@ class PlatformManager:
)
binding = _binding
for bot in result.all():
# Disabled drafts may not have the credentials needed to construct
# an adapter. They become runtime bots when explicitly enabled.
if not bot.enable:
continue
try:
if binding is None:
binding = await self.ap.workspace_service.get_execution_binding(workspace_uuid)
@@ -605,7 +605,6 @@ export default function PluginProcessorDetailContent({
hasUnsavedChanges={dirty}
beforeRun={save}
onRunFinished={() => {
setActiveTab('logs');
void refreshLatestRun();
}}
supportedEventPatterns={component.supported_event_patterns}
@@ -57,6 +57,7 @@ export default function GuidedTour({
useState<PopoverPosition | null>(null);
const popoverRef = useRef<HTMLDivElement | null>(null);
const previousStorageKeyRef = useRef(storageKey);
const enteredStepRef = useRef<string | null>(null);
const activeIndex = steps.findIndex((step) => step.id === activeStepId);
const activeStep = activeIndex >= 0 ? steps[activeIndex] : undefined;
@@ -86,8 +87,17 @@ export default function GuidedTour({
}, [activeStepId, enabled, finished, steps, storageKey]);
useEffect(() => {
if (enabled && !finished) activeStep?.onEnter?.();
}, [activeStep, enabled, finished]);
if (!enabled || finished || !activeStep) {
enteredStepRef.current = null;
return;
}
// Callback identities can change during navigation. Enter each step once,
// so a URL update cannot replay its navigation and override the user's tab.
const stepKey = JSON.stringify([storageKey, activeStep.id]);
if (enteredStepRef.current === stepKey) return;
enteredStepRef.current = stepKey;
activeStep.onEnter?.();
}, [activeStep, enabled, finished, storageKey]);
const measure = useCallback(() => {
if (!activeStep) return;