mirror of
https://github.com/langbot-app/LangBot.git
synced 2026-08-26 12:17:14 +00:00
QuasarRyan
1217 lines
51 KiB
Python
1217 lines
51 KiB
Python
from __future__ import annotations
|
||
import typing
|
||
import re
|
||
import asyncio
|
||
import traceback
|
||
|
||
import datetime
|
||
import time
|
||
|
||
import langbot_plugin.api.definition.abstract.platform.adapter as abstract_platform_adapter
|
||
import langbot_plugin.api.entities.builtin.platform.message as platform_message
|
||
import langbot_plugin.api.entities.builtin.platform.events as platform_events
|
||
import langbot_plugin.api.entities.builtin.platform.entities as platform_entities
|
||
from langbot.libs.qq_official_api.api import (
|
||
QQ_SELECT_ACTION_PREFIX,
|
||
QQOfficialClient,
|
||
build_keyboard_from_form,
|
||
build_keyboard_from_select_field,
|
||
resolve_select_button_action,
|
||
)
|
||
from langbot.libs.qq_official_api.qqofficialevent import QQOfficialEvent
|
||
from ...utils import image
|
||
from ..logger import EventLogger
|
||
|
||
|
||
def _is_base64_data(value: str) -> bool:
|
||
"""Check if a string contains base64-encoded data rather than a URL."""
|
||
if not value:
|
||
return False
|
||
# data: URI scheme (e.g. data:image/png;base64,xxx)
|
||
if value.startswith('data:'):
|
||
return True
|
||
# Only treat as base64 if it doesn't look like a URL/path and has valid base64 chars
|
||
if value.startswith(('http://', 'https://', '/', './', '../')):
|
||
return False
|
||
# Check if it looks like base64 (only valid chars, reasonable length)
|
||
return bool(re.fullmatch(r'[A-Za-z0-9+/=\s]{20,}', value))
|
||
|
||
|
||
class QQOfficialMessageConverter(abstract_platform_adapter.AbstractMessageConverter):
|
||
@staticmethod
|
||
async def yiri2target(message_chain: platform_message.MessageChain):
|
||
"""将 LangBot 消息链转换为 QQ Official 消息格式列表。"""
|
||
content_list = []
|
||
for msg in message_chain:
|
||
if type(msg) is platform_message.Plain:
|
||
content_list.append(
|
||
{
|
||
'type': 'text',
|
||
'content': msg.text,
|
||
}
|
||
)
|
||
elif type(msg) is platform_message.Image:
|
||
url = msg.url if hasattr(msg, 'url') and msg.url else None
|
||
b64 = msg.base64 if hasattr(msg, 'base64') and msg.base64 else None
|
||
# Some plugins (e.g. MimoTTS) store base64 data in the url field
|
||
if url and not b64 and _is_base64_data(url):
|
||
b64 = url
|
||
url = None
|
||
content_list.append(
|
||
{
|
||
'type': 'image',
|
||
'url': url,
|
||
'base64': b64,
|
||
}
|
||
)
|
||
elif type(msg) is platform_message.Voice:
|
||
url = msg.url if hasattr(msg, 'url') and msg.url else None
|
||
b64 = msg.base64 if hasattr(msg, 'base64') and msg.base64 else None
|
||
# Some plugins (e.g. MimoTTS) store base64 data in the url field
|
||
if url and not b64 and _is_base64_data(url):
|
||
b64 = url
|
||
url = None
|
||
content_list.append(
|
||
{
|
||
'type': 'voice',
|
||
'url': url,
|
||
'base64': b64,
|
||
}
|
||
)
|
||
elif type(msg) is platform_message.File:
|
||
url = msg.url if hasattr(msg, 'url') and msg.url else None
|
||
b64 = msg.base64 if hasattr(msg, 'base64') and msg.base64 else None
|
||
# Some plugins store base64 data in the url field
|
||
if url and not b64 and _is_base64_data(url):
|
||
b64 = url
|
||
url = None
|
||
content_list.append(
|
||
{
|
||
'type': 'file',
|
||
'url': url,
|
||
'base64': b64,
|
||
'name': msg.name if hasattr(msg, 'name') else 'file',
|
||
}
|
||
)
|
||
|
||
return content_list
|
||
|
||
@staticmethod
|
||
async def target2yiri(message: str, message_id: str, pic_url: str, content_type):
|
||
yiri_msg_list = []
|
||
yiri_msg_list.append(platform_message.Source(id=message_id, time=datetime.datetime.now()))
|
||
if pic_url is not None:
|
||
base64_url = await image.get_qq_official_image_base64(pic_url=pic_url, content_type=content_type)
|
||
yiri_msg_list.append(platform_message.Image(url=pic_url, base64=base64_url))
|
||
|
||
yiri_msg_list.append(platform_message.Plain(text=message))
|
||
chain = platform_message.MessageChain(yiri_msg_list)
|
||
return chain
|
||
|
||
|
||
class QQOfficialEventConverter(abstract_platform_adapter.AbstractEventConverter):
|
||
@staticmethod
|
||
async def yiri2target(event: platform_events.MessageEvent) -> QQOfficialEvent:
|
||
return event.source_platform_object
|
||
|
||
@staticmethod
|
||
async def target2yiri(event: QQOfficialEvent):
|
||
"""
|
||
QQ官方消息转换为LB对象
|
||
"""
|
||
yiri_chain = await QQOfficialMessageConverter.target2yiri(
|
||
message=event.content,
|
||
message_id=event.d_id,
|
||
pic_url=event.attachments,
|
||
content_type=event.content_type,
|
||
)
|
||
|
||
if event.t == 'C2C_MESSAGE_CREATE':
|
||
friend = platform_entities.Friend(
|
||
id=event.user_openid,
|
||
nickname=event.t,
|
||
remark='',
|
||
)
|
||
return platform_events.FriendMessage(
|
||
sender=friend,
|
||
message_chain=yiri_chain,
|
||
time=int(datetime.datetime.strptime(event.timestamp, '%Y-%m-%dT%H:%M:%S%z').timestamp()),
|
||
source_platform_object=event,
|
||
)
|
||
|
||
if event.t == 'DIRECT_MESSAGE_CREATE':
|
||
friend = platform_entities.Friend(
|
||
id=event.guild_id,
|
||
nickname=event.t,
|
||
remark='',
|
||
)
|
||
return platform_events.FriendMessage(sender=friend, message_chain=yiri_chain, source_platform_object=event)
|
||
if event.t == 'GROUP_AT_MESSAGE_CREATE':
|
||
yiri_chain.insert(0, platform_message.At(target='justbot'))
|
||
|
||
sender = platform_entities.GroupMember(
|
||
id=event.group_openid,
|
||
member_name=event.t,
|
||
permission='MEMBER',
|
||
group=platform_entities.Group(
|
||
id=event.group_openid,
|
||
name='MEMBER',
|
||
permission=platform_entities.Permission.Member,
|
||
),
|
||
special_title='',
|
||
)
|
||
time = int(datetime.datetime.strptime(event.timestamp, '%Y-%m-%dT%H:%M:%S%z').timestamp())
|
||
return platform_events.GroupMessage(
|
||
sender=sender,
|
||
message_chain=yiri_chain,
|
||
time=time,
|
||
source_platform_object=event,
|
||
)
|
||
if event.t == 'AT_MESSAGE_CREATE':
|
||
yiri_chain.insert(0, platform_message.At(target='justbot'))
|
||
sender = platform_entities.GroupMember(
|
||
id=event.channel_id,
|
||
member_name=event.t,
|
||
permission='MEMBER',
|
||
group=platform_entities.Group(
|
||
id=event.channel_id,
|
||
name='MEMBER',
|
||
permission=platform_entities.Permission.Member,
|
||
),
|
||
special_title='',
|
||
)
|
||
time = int(datetime.datetime.strptime(event.timestamp, '%Y-%m-%dT%H:%M:%S%z').timestamp())
|
||
return platform_events.GroupMessage(
|
||
sender=sender,
|
||
message_chain=yiri_chain,
|
||
time=time,
|
||
source_platform_object=event,
|
||
)
|
||
|
||
|
||
class QQOfficialAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter):
|
||
bot: QQOfficialClient
|
||
config: dict
|
||
bot_account_id: str
|
||
bot_uuid: str = None
|
||
enable_webhook: bool = False
|
||
message_converter: QQOfficialMessageConverter = QQOfficialMessageConverter()
|
||
event_converter: QQOfficialEventConverter = QQOfficialEventConverter()
|
||
ap: typing.Any = None
|
||
|
||
def __init__(self, config: dict, logger: EventLogger):
|
||
enable_webhook = config.get('enable-webhook', False)
|
||
|
||
bot = QQOfficialClient(
|
||
app_id=config['appid'],
|
||
secret=config['secret'],
|
||
token=config['token'],
|
||
logger=logger,
|
||
unified_mode=enable_webhook,
|
||
)
|
||
|
||
super().__init__(
|
||
config=config,
|
||
logger=logger,
|
||
bot=bot,
|
||
bot_account_id=config['appid'],
|
||
)
|
||
|
||
self.enable_webhook = enable_webhook
|
||
self._ws_task: asyncio.Task = None
|
||
self._stream_ctx: dict = {}
|
||
self._stream_ctx_ts: dict[str, float] = {}
|
||
self._fallback_text: dict[str, str] = {}
|
||
self._fallback_text_ts: dict[str, float] = {}
|
||
# Dify form-action bookkeeping for the human-input button flow.
|
||
# session_key = "<scene>_<id>" where scene is c2c/group/channel and
|
||
# id is user_openid / group_openid / channel_id.
|
||
# session_key -> {form_data, msg_id, event_id, scene, target_id,
|
||
# sender_id, posted_at}
|
||
# Set when we send a markdown+keyboard card and consulted when:
|
||
# (a) INTERACTION_CREATE fires — we look up the form by
|
||
# session_key (button's `data` carries the action_id),
|
||
# (b) the resumed-workflow query needs to find a passive-reply
|
||
# event_id (INTERACTION_CREATE id, 30-min validity).
|
||
self._pending_forms: dict[str, dict] = {}
|
||
# session_key -> most recent ``INTERACTION_CREATE`` event_id, used
|
||
# as the passive event_id for the resumed query's LLM output.
|
||
self._session_event_ids: dict[str, dict] = {}
|
||
# Per-anchor msg_seq counter. QQ accepts up to 5 passive replies
|
||
# per (msg_id|event_id) within 60 min, but each reuse needs a
|
||
# fresh ``msg_seq`` — re-sending with msg_seq=1 is silently dedup'd.
|
||
self._anchor_msg_seq: dict[str, int] = {}
|
||
self._background_tasks: set[asyncio.Task] = set()
|
||
|
||
# Wire button-click handler so webhook mode catches INTERACTION_CREATE.
|
||
# (ws mode is wired separately via on_event in _run_websocket so the
|
||
# raw payload bypasses get_message's message-only flattening.)
|
||
@self.bot.on_interaction()
|
||
async def _on_interaction(event_data: dict, interaction_id: typing.Optional[str]):
|
||
await self._handle_interaction_create(event_data, interaction_id)
|
||
|
||
def _start_background_task(self, coro) -> bool:
|
||
"""Start one bounded adapter-side auxiliary task."""
|
||
|
||
background_tasks = getattr(self, '_background_tasks', None)
|
||
if background_tasks is None:
|
||
background_tasks = set()
|
||
object.__setattr__(self, '_background_tasks', background_tasks)
|
||
for task in tuple(background_tasks):
|
||
if task.done():
|
||
background_tasks.discard(task)
|
||
if len(background_tasks) >= 100:
|
||
coro.close()
|
||
return False
|
||
task = asyncio.create_task(coro)
|
||
background_tasks.add(task)
|
||
|
||
def done(done_task: asyncio.Task) -> None:
|
||
background_tasks.discard(done_task)
|
||
if not done_task.cancelled():
|
||
done_task.exception()
|
||
|
||
task.add_done_callback(done)
|
||
return True
|
||
|
||
async def reply_message(
|
||
self,
|
||
message_source: platform_events.MessageEvent,
|
||
message: platform_message.MessageChain,
|
||
quote_origin: bool = False,
|
||
):
|
||
qq_official_event = await QQOfficialEventConverter.yiri2target(
|
||
message_source,
|
||
)
|
||
|
||
# Synthetic event (button-click resume): no inbound platform
|
||
# object → no msg_id. Route via the cached INTERACTION_CREATE
|
||
# event_id (valid 30 min, no quota cost).
|
||
if qq_official_event is None:
|
||
await self._reply_synthetic(message_source, message)
|
||
return
|
||
|
||
content_list = await QQOfficialMessageConverter.yiri2target(message)
|
||
|
||
# 确定 target_type 和 target_id
|
||
target_type = None
|
||
target_id = None
|
||
|
||
if qq_official_event.t == 'C2C_MESSAGE_CREATE':
|
||
target_type = 'c2c'
|
||
target_id = qq_official_event.user_openid
|
||
elif qq_official_event.t == 'GROUP_AT_MESSAGE_CREATE':
|
||
target_type = 'group'
|
||
target_id = qq_official_event.group_openid
|
||
elif qq_official_event.t == 'AT_MESSAGE_CREATE':
|
||
# 频道群聊使用频道 API，暂不支持富媒体
|
||
for content in content_list:
|
||
if content['type'] == 'text':
|
||
await self.bot.send_channle_group_text_msg(
|
||
qq_official_event.channel_id,
|
||
content['content'],
|
||
qq_official_event.d_id,
|
||
)
|
||
return
|
||
elif qq_official_event.t == 'DIRECT_MESSAGE_CREATE':
|
||
# 频道私聊使用频道 API，暂不支持富媒体
|
||
for content in content_list:
|
||
if content['type'] == 'text':
|
||
await self.bot.send_channle_private_text_msg(
|
||
qq_official_event.guild_id,
|
||
content['content'],
|
||
qq_official_event.d_id,
|
||
)
|
||
return
|
||
|
||
# C2C 和群聊：支持文字 + 富媒体
|
||
for content in content_list:
|
||
content_type = content.get('type', 'text')
|
||
|
||
if content_type == 'text':
|
||
if target_type in {'c2c', 'group'}:
|
||
await self._send_c2c_or_group_text_reply(
|
||
target_type,
|
||
target_id,
|
||
content['content'],
|
||
msg_id=qq_official_event.d_id,
|
||
)
|
||
|
||
elif content_type == 'image':
|
||
file_url = content.get('url')
|
||
file_data = content.get('base64')
|
||
if file_url or file_data:
|
||
await self.bot.send_image_msg(
|
||
target_type,
|
||
target_id,
|
||
file_url=file_url,
|
||
file_data=file_data,
|
||
msg_id=qq_official_event.d_id,
|
||
)
|
||
|
||
elif content_type == 'voice':
|
||
file_url = content.get('url')
|
||
file_data = content.get('base64')
|
||
if file_url or file_data:
|
||
await self.bot.send_voice_msg(
|
||
target_type,
|
||
target_id,
|
||
file_url=file_url,
|
||
file_data=file_data,
|
||
msg_id=qq_official_event.d_id,
|
||
)
|
||
|
||
elif content_type == 'file':
|
||
file_url = content.get('url')
|
||
file_data = content.get('base64')
|
||
file_name = content.get('name', 'file')
|
||
if file_url or file_data:
|
||
await self.bot.send_file_msg(
|
||
target_type,
|
||
target_id,
|
||
file_url=file_url,
|
||
file_data=file_data,
|
||
file_name=file_name,
|
||
msg_id=qq_official_event.d_id,
|
||
)
|
||
|
||
async def send_message(self, target_type: str, target_id: str, message: platform_message.MessageChain):
|
||
pass
|
||
|
||
async def _send_c2c_or_group_text_reply(
|
||
self,
|
||
target_type: str,
|
||
target_id: str,
|
||
content: str,
|
||
*,
|
||
msg_id: typing.Optional[str] = None,
|
||
event_id: typing.Optional[str] = None,
|
||
msg_seq: int = 1,
|
||
) -> None:
|
||
"""Send a text reply using the configured C2C/group render mode."""
|
||
use_markdown = self.config.get('enable-markdown-rendering', False)
|
||
if target_type == 'c2c':
|
||
send = self.bot.send_private_markdown_msg if use_markdown else self.bot.send_private_text_msg
|
||
await send(
|
||
user_openid=target_id,
|
||
content=content,
|
||
msg_id=msg_id,
|
||
event_id=event_id,
|
||
msg_seq=msg_seq,
|
||
)
|
||
elif target_type == 'group':
|
||
send = self.bot.send_group_markdown_msg if use_markdown else self.bot.send_group_text_msg
|
||
await send(
|
||
group_openid=target_id,
|
||
content=content,
|
||
msg_id=msg_id,
|
||
event_id=event_id,
|
||
msg_seq=msg_seq,
|
||
)
|
||
else:
|
||
raise ValueError(f'Unsupported QQ Official text reply target: {target_type}')
|
||
|
||
def register_listener(
|
||
self,
|
||
event_type: typing.Type[platform_events.Event],
|
||
callback: typing.Callable[
|
||
[platform_events.Event, abstract_platform_adapter.AbstractMessagePlatformAdapter], None
|
||
],
|
||
):
|
||
async def on_message(event: QQOfficialEvent):
|
||
self.bot_account_id = 'justbot'
|
||
try:
|
||
return await callback(await self.event_converter.target2yiri(event), self)
|
||
except Exception:
|
||
await self.logger.error(f'Error in qqofficial callback: {traceback.format_exc()}')
|
||
|
||
if event_type == platform_events.FriendMessage:
|
||
self.bot.on_message('DIRECT_MESSAGE_CREATE')(on_message)
|
||
self.bot.on_message('C2C_MESSAGE_CREATE')(on_message)
|
||
elif event_type == platform_events.GroupMessage:
|
||
self.bot.on_message('GROUP_AT_MESSAGE_CREATE')(on_message)
|
||
self.bot.on_message('AT_MESSAGE_CREATE')(on_message)
|
||
|
||
def set_bot_uuid(self, bot_uuid: str):
|
||
"""设置 bot UUID（用于生成 webhook URL）"""
|
||
self.bot_uuid = bot_uuid
|
||
|
||
async def handle_unified_webhook(self, bot_uuid: str, path: str, request):
|
||
"""处理统一 webhook 请求。
|
||
|
||
Args:
|
||
bot_uuid: Bot 的 UUID
|
||
path: 子路径（如果有的话）
|
||
request: Quart Request 对象
|
||
|
||
Returns:
|
||
响应数据
|
||
"""
|
||
return await self.bot.handle_unified_webhook(request)
|
||
|
||
async def run_async(self):
|
||
if not self.enable_webhook:
|
||
await self._run_websocket()
|
||
else:
|
||
# 统一 webhook 模式下，不启动独立的 Quart 应用
|
||
async def keep_alive():
|
||
while True:
|
||
await asyncio.sleep(1)
|
||
|
||
await keep_alive()
|
||
|
||
async def _run_websocket(self):
|
||
"""以 WebSocket 模式运行网关连接"""
|
||
await self.logger.info('QQ Official adapter starting in WebSocket mode')
|
||
|
||
async def on_ready():
|
||
await self.logger.info('QQ Official WebSocket connected and ready')
|
||
|
||
async def on_event(event_type: str, event_data: dict):
|
||
# INTERACTION_CREATE is dispatched via bot.on_interaction()
|
||
# (registered in __init__) so we get the top-level ws_event_id
|
||
# — needed as the passive-reply event_id. It never reaches here.
|
||
# 只处理消息事件，忽略 READY/RESUMED 等系统事件
|
||
message_event_types = {
|
||
'C2C_MESSAGE_CREATE',
|
||
'DIRECT_MESSAGE_CREATE',
|
||
'GROUP_AT_MESSAGE_CREATE',
|
||
'AT_MESSAGE_CREATE',
|
||
}
|
||
if event_type not in message_event_types:
|
||
return
|
||
if not isinstance(event_data, dict):
|
||
await self.logger.warning(f'Event data is not dict, skipping: {event_type} -> {type(event_data)}')
|
||
return
|
||
await self.logger.info(f'Processing message event: {event_type}')
|
||
# 构造与 webhook 模式相同的 payload 结构
|
||
payload = {'t': event_type, 'd': event_data}
|
||
message_data = await self.bot.get_message(payload)
|
||
if message_data:
|
||
event = QQOfficialEvent.from_payload(message_data)
|
||
await self.bot._handle_message(event)
|
||
|
||
async def on_error(error: Exception):
|
||
await self.logger.error(f'WebSocket error: {error}')
|
||
await self.logger.error(f'QQ Official WebSocket error: {error}')
|
||
|
||
self._ws_task = asyncio.create_task(self.bot.connect_gateway_loop(on_event, on_ready, on_error))
|
||
try:
|
||
await self._ws_task
|
||
except asyncio.CancelledError:
|
||
pass
|
||
|
||
async def kill(self) -> bool:
|
||
task_set = getattr(self, '_background_tasks', set())
|
||
background_tasks = list(task_set)
|
||
for task in background_tasks:
|
||
if not task.done():
|
||
task.cancel()
|
||
if background_tasks:
|
||
await asyncio.gather(*background_tasks, return_exceptions=True)
|
||
task_set.clear()
|
||
if self._ws_task:
|
||
self._ws_task.cancel()
|
||
try:
|
||
await self._ws_task
|
||
except asyncio.CancelledError:
|
||
pass
|
||
self._ws_task = None
|
||
await self.bot.close()
|
||
self._pending_forms.clear()
|
||
self._session_event_ids.clear()
|
||
self._anchor_msg_seq.clear()
|
||
self._stream_ctx.clear()
|
||
self._stream_ctx_ts.clear()
|
||
self._fallback_text.clear()
|
||
self._fallback_text_ts.clear()
|
||
return True
|
||
|
||
# --------------- 流式输出 ---------------
|
||
|
||
_STREAM_CTX_TTL = 300 # seconds
|
||
|
||
async def _cleanup_stale_streams(self):
|
||
"""Remove stream contexts that have not been updated for more than _STREAM_CTX_TTL seconds."""
|
||
now = time.time()
|
||
stale_ids = [mid for mid, ts in self._stream_ctx_ts.items() if now - ts > self._STREAM_CTX_TTL]
|
||
for mid in stale_ids:
|
||
self._stream_ctx.pop(mid, None)
|
||
self._stream_ctx_ts.pop(mid, None)
|
||
stale_fb = [mid for mid, ts in self._fallback_text_ts.items() if now - ts > self._STREAM_CTX_TTL]
|
||
for mid in stale_fb:
|
||
self._fallback_text.pop(mid, None)
|
||
self._fallback_text_ts.pop(mid, None)
|
||
while len(self._stream_ctx) > 1000:
|
||
oldest = min(self._stream_ctx_ts, key=self._stream_ctx_ts.__getitem__)
|
||
self._stream_ctx.pop(oldest, None)
|
||
self._stream_ctx_ts.pop(oldest, None)
|
||
while len(self._fallback_text) > 1000:
|
||
oldest = min(self._fallback_text_ts, key=self._fallback_text_ts.__getitem__)
|
||
self._fallback_text.pop(oldest, None)
|
||
self._fallback_text_ts.pop(oldest, None)
|
||
if stale_ids or stale_fb:
|
||
await self.logger.debug(f'Cleaned up {len(stale_ids)} stream contexts, {len(stale_fb)} fallback texts')
|
||
|
||
async def is_stream_output_supported(self) -> bool:
|
||
return self.config.get('enable-stream-reply', False)
|
||
|
||
@staticmethod
|
||
def _is_form_placeholder_chunk(text: str) -> bool:
|
||
"""Return True for invisible placeholder chunks used to carry forms."""
|
||
|
||
if not text:
|
||
return False
|
||
|
||
cleaned = text.replace('\u200b', '').replace('\u200c', '').replace('\u200d', '').replace('\ufeff', '').strip()
|
||
# Some Windows consoles/logs display the zero-width placeholder as
|
||
# mojibake. Treat those variants as the same non-user-facing marker.
|
||
return cleaned in {'', '鈥?', '​'}
|
||
|
||
async def create_message_card(self, message_id: str, event: platform_events.MessageEvent) -> bool:
|
||
source = event.source_platform_object
|
||
# Synthetic events (button-click resume) have no source object —
|
||
# they ride a cached INTERACTION_CREATE event_id, not a streamable
|
||
# msg_id. Skip stream setup; reply_message handles the one-shot
|
||
# send at is_final.
|
||
if source is None:
|
||
return False
|
||
# Streaming API only supports C2C private chat
|
||
if source.t != 'C2C_MESSAGE_CREATE':
|
||
return False
|
||
|
||
# The stream endpoint still consumes msg_seq for this inbound msg_id.
|
||
# Keep the passive-reply counter in sync so a follow-up form card uses
|
||
# msg_seq=2 instead of being deduplicated by QQ as another seq=1 send.
|
||
if source.d_id:
|
||
self._anchor_msg_seq[source.d_id] = max(self._anchor_msg_seq.get(source.d_id, 0), 1)
|
||
while len(self._anchor_msg_seq) > 4096:
|
||
self._anchor_msg_seq.pop(next(iter(self._anchor_msg_seq)), None)
|
||
|
||
ctx = {
|
||
'user_openid': source.user_openid,
|
||
'msg_id': source.d_id,
|
||
'stream_msg_id': None,
|
||
'msg_seq': 1,
|
||
'index': 0,
|
||
'last_update_ts': 0,
|
||
'accumulated_text': '',
|
||
'sent_length': 0,
|
||
'session_started': False,
|
||
}
|
||
|
||
self._stream_ctx[message_id] = ctx
|
||
self._stream_ctx_ts[message_id] = time.time()
|
||
return True
|
||
|
||
async def reply_message_chunk(
|
||
self,
|
||
message_source: platform_events.MessageEvent,
|
||
bot_message: dict,
|
||
message: platform_message.MessageChain,
|
||
quote_origin: bool = False,
|
||
is_final: bool = False,
|
||
):
|
||
# Periodically clean up stale stream contexts
|
||
await self._cleanup_stale_streams()
|
||
|
||
# Dify human-input pause: when the runner attaches `_form_data` to
|
||
# the final chunk, finalize any in-flight stream session and send
|
||
# a markdown + keyboard message instead. Plain-text content from
|
||
# earlier chunks is already on the stream; we close it cleanly
|
||
# and the buttons land as a separate reply.
|
||
form_data = getattr(bot_message, '_form_data', None) if not isinstance(bot_message, dict) else None
|
||
if is_final:
|
||
_resume = getattr(bot_message, '_resume_from_form', None) if not isinstance(bot_message, dict) else None
|
||
_open_new = getattr(bot_message, '_open_new_card', None) if not isinstance(bot_message, dict) else None
|
||
if self.ap is not None:
|
||
self.ap.logger.info(
|
||
f'QQ Official reply_message_chunk final: '
|
||
f'type={type(bot_message).__name__} '
|
||
f'is_final={is_final} '
|
||
f'form_data_present={form_data is not None} '
|
||
f'resume_from_form={_resume} open_new_card={_open_new} '
|
||
f'content_len={len(getattr(bot_message, "content", "") or "")}'
|
||
)
|
||
if form_data and is_final:
|
||
await self._handle_form_chunk(message_source, message, form_data)
|
||
return
|
||
|
||
# 提取纯文本内容（当前 chunk 的文本）
|
||
text_parts = []
|
||
for msg in message:
|
||
if type(msg) is platform_message.Plain:
|
||
text_parts.append(msg.text)
|
||
chunk_text = '\n\n'.join(text_parts)
|
||
if self._is_form_placeholder_chunk(chunk_text):
|
||
await self.logger.debug('QQ Official: skipped invisible form placeholder chunk')
|
||
return
|
||
|
||
message_id = (
|
||
bot_message.get('resp_message_id')
|
||
if isinstance(bot_message, dict)
|
||
else getattr(bot_message, 'resp_message_id', None)
|
||
)
|
||
if not message_id or message_id not in self._stream_ctx:
|
||
# 非流式场景（如群聊不支持流式），累积文本后一次性回复
|
||
if chunk_text:
|
||
# Chunks carry the latest full snapshot, not a text delta.
|
||
self._fallback_text[message_id] = chunk_text[:200000]
|
||
self._fallback_text_ts[message_id] = time.time()
|
||
if is_final:
|
||
full_text = self._fallback_text.pop(message_id, '')
|
||
if full_text:
|
||
fallback_msg = platform_message.MessageChain([platform_message.Plain(text=full_text)])
|
||
await self.reply_message(message_source, fallback_msg, quote_origin)
|
||
return
|
||
|
||
ctx = self._stream_ctx[message_id]
|
||
|
||
# 累积文本
|
||
if chunk_text:
|
||
ctx['accumulated_text'] = chunk_text[:200000]
|
||
|
||
# 未启动会话时，等第一个有内容的 chunk 来建立会话
|
||
if not ctx['session_started']:
|
||
if not ctx['accumulated_text']:
|
||
return
|
||
# 用第一个 chunk 的文本建立会话（不发 "..." 避免污染前缀）
|
||
ctx['session_started'] = True
|
||
|
||
# `replace` mode requires every update to contain the previously
|
||
# delivered content as its prefix. `sent_length` only tells us whether
|
||
# a non-final snapshot has new content; it must not truncate the
|
||
# content sent to QQ.
|
||
if len(ctx['accumulated_text']) <= ctx['sent_length'] and not is_final:
|
||
return
|
||
content_to_send = ctx['accumulated_text']
|
||
|
||
input_state = 10 if is_final else 1
|
||
|
||
# Rate limiting: skip non-final updates if last update was <0.5s ago
|
||
now = time.time()
|
||
if not is_final and (now - ctx['last_update_ts']) < 0.5:
|
||
return
|
||
ctx['last_update_ts'] = now
|
||
|
||
try:
|
||
resp = await self.bot.send_stream_msg(
|
||
user_openid=ctx['user_openid'],
|
||
content=content_to_send,
|
||
event_id=ctx['msg_id'],
|
||
msg_id=ctx['msg_id'],
|
||
msg_seq=ctx['msg_seq'],
|
||
index=ctx['index'],
|
||
stream_msg_id=ctx['stream_msg_id'],
|
||
input_state=input_state,
|
||
)
|
||
if resp and isinstance(resp, dict):
|
||
new_stream_id = resp.get('id')
|
||
if new_stream_id:
|
||
ctx['stream_msg_id'] = new_stream_id
|
||
ctx['sent_length'] = len(ctx['accumulated_text'])
|
||
ctx['index'] += 1
|
||
await self.logger.debug(
|
||
f'[QQ Official] 流式 chunk 已发送, index={ctx["index"]}, '
|
||
f'sent_len={ctx["sent_length"]}, is_final={is_final}'
|
||
)
|
||
except Exception as e:
|
||
await self.logger.error(f'Failed to send stream message: {e}')
|
||
|
||
if is_final:
|
||
self._stream_ctx.pop(message_id, None)
|
||
|
||
def unregister_listener(
|
||
self,
|
||
event_type: type,
|
||
callback: typing.Callable[
|
||
[platform_events.Event, abstract_platform_adapter.AbstractMessagePlatformAdapter], None
|
||
],
|
||
):
|
||
return super().unregister_listener(event_type, callback)
|
||
|
||
# ------------------------------------------------------------------
|
||
# Dify human-input button-interaction support
|
||
# ------------------------------------------------------------------
|
||
|
||
_PENDING_FORM_TTL = 1800 # 30 min — matches QQ passive-reply window.
|
||
_MAX_REPLIES_PER_ANCHOR = 5 # QQ hard limit per msg_id / event_id.
|
||
|
||
def _next_msg_seq(self, anchor: str) -> typing.Optional[int]:
|
||
"""Return the next msg_seq for an anchor, or ``None`` if the
|
||
anchor has already been used 5 times (further sends would be
|
||
silently dropped by QQ)."""
|
||
if not anchor:
|
||
return 1
|
||
used = self._anchor_msg_seq.get(anchor, 0)
|
||
if used >= self._MAX_REPLIES_PER_ANCHOR:
|
||
return None
|
||
self._anchor_msg_seq[anchor] = used + 1
|
||
while len(self._anchor_msg_seq) > 4096:
|
||
self._anchor_msg_seq.pop(next(iter(self._anchor_msg_seq)), None)
|
||
return used + 1
|
||
|
||
async def _reply_synthetic(
|
||
self,
|
||
message_source: platform_events.MessageEvent,
|
||
message: platform_message.MessageChain,
|
||
) -> None:
|
||
"""Deliver a reply for a synthetic (button-click-resume) event.
|
||
|
||
Synthetic events have ``source_platform_object=None`` and no
|
||
fresh inbound msg_id. The previous INTERACTION_CREATE id we
|
||
cached in :attr:`_session_event_ids` is a valid passive-reply
|
||
anchor (``event_id``) for up to 30 minutes — use it.
|
||
"""
|
||
if isinstance(message_source, platform_events.GroupMessage):
|
||
target_type = 'group'
|
||
group = getattr(message_source, 'group', None) or (
|
||
message_source.sender.group if hasattr(message_source.sender, 'group') else None
|
||
)
|
||
target_id = str(group.id) if group else None
|
||
else:
|
||
target_type = 'c2c'
|
||
target_id = str(message_source.sender.id) if message_source.sender else None
|
||
|
||
if not target_id:
|
||
await self.logger.warning('QQ Official: synthetic reply has no target_id; dropping')
|
||
return
|
||
|
||
session_key = f'{target_type}_{target_id}'
|
||
cached = self._session_event_ids.get(session_key)
|
||
event_id = cached.get('event_id') if cached else None
|
||
if cached and (time.time() - cached.get('posted_at', 0)) > self._PENDING_FORM_TTL:
|
||
event_id = None
|
||
|
||
if not event_id:
|
||
await self.logger.warning(
|
||
f'QQ Official: no cached event_id for {session_key}; '
|
||
f'cannot deliver synthetic reply within passive-reply window'
|
||
)
|
||
return
|
||
|
||
content_list = await QQOfficialMessageConverter.yiri2target(message)
|
||
text_parts = [c['content'] for c in content_list if c.get('type') == 'text' and c.get('content')]
|
||
if not text_parts:
|
||
await self.logger.info('QQ Official: synthetic reply has no text content; skipping')
|
||
return
|
||
text = '\n\n'.join(text_parts)
|
||
|
||
msg_seq = self._next_msg_seq(event_id)
|
||
if msg_seq is None:
|
||
await self.logger.warning(
|
||
f'QQ Official: anchor {event_id!r} exhausted (>5 passive replies); '
|
||
f'cannot deliver synthetic reply for {session_key}'
|
||
)
|
||
return
|
||
|
||
try:
|
||
await self._send_c2c_or_group_text_reply(
|
||
target_type,
|
||
target_id,
|
||
text,
|
||
event_id=event_id,
|
||
msg_seq=msg_seq,
|
||
)
|
||
except Exception:
|
||
await self.logger.error(f'QQ Official: synthetic reply delivery failed: {traceback.format_exc()}')
|
||
|
||
def _resolve_target_from_source(self, source: QQOfficialEvent) -> typing.Optional[tuple[str, str]]:
|
||
"""Return ``(target_type, target_id)`` for sending a reply, or
|
||
``None`` if the scene cannot host a markdown+keyboard message."""
|
||
if source is None:
|
||
return None
|
||
if source.t == 'C2C_MESSAGE_CREATE':
|
||
return 'c2c', source.user_openid
|
||
if source.t == 'GROUP_AT_MESSAGE_CREATE':
|
||
return 'group', source.group_openid
|
||
if source.t == 'AT_MESSAGE_CREATE':
|
||
return 'channel', source.channel_id
|
||
# DIRECT_MESSAGE_CREATE uses the guild DM API which does not accept
|
||
# markdown+keyboard at the time of writing — caller falls back to text.
|
||
return None
|
||
|
||
def _resolve_target_from_event(
|
||
self, message_source: platform_events.MessageEvent
|
||
) -> typing.Optional[tuple[str, str]]:
|
||
"""Resolve ``(target_type, target_id)`` from the public event.
|
||
|
||
Prefers the platform-native source when present; falls back to
|
||
the synthesized event's sender/group fields so button-click
|
||
resume queries can still find a destination.
|
||
"""
|
||
source = message_source.source_platform_object
|
||
if source is not None:
|
||
return self._resolve_target_from_source(source)
|
||
if isinstance(message_source, platform_events.GroupMessage):
|
||
group = getattr(message_source, 'group', None) or (
|
||
message_source.sender.group
|
||
if message_source.sender and hasattr(message_source.sender, 'group')
|
||
else None
|
||
)
|
||
if group and getattr(group, 'id', None):
|
||
return 'group', str(group.id)
|
||
if isinstance(message_source, platform_events.FriendMessage):
|
||
if message_source.sender and getattr(message_source.sender, 'id', None):
|
||
return 'c2c', str(message_source.sender.id)
|
||
return None
|
||
|
||
def _prune_pending_forms(self) -> None:
|
||
now = time.time()
|
||
stale = [k for k, v in self._pending_forms.items() if now - v.get('posted_at', 0) > self._PENDING_FORM_TTL]
|
||
for k in stale:
|
||
self._pending_forms.pop(k, None)
|
||
stale_e = [
|
||
k for k, v in self._session_event_ids.items() if now - v.get('posted_at', 0) > self._PENDING_FORM_TTL
|
||
]
|
||
for k in stale_e:
|
||
stale_event = self._session_event_ids.pop(k, None)
|
||
if stale_event:
|
||
self._anchor_msg_seq.pop(stale_event.get('event_id'), None)
|
||
|
||
async def _handle_form_chunk(
|
||
self,
|
||
message_source: platform_events.MessageEvent,
|
||
message: platform_message.MessageChain,
|
||
form_data: dict,
|
||
) -> None:
|
||
"""Send the markdown + keyboard form prompt for a Dify pause.
|
||
|
||
Called from ``reply_message_chunk`` when the runner attaches
|
||
``_form_data`` to the final chunk. Replaces what would otherwise
|
||
be a plain-text numbered-list fallback.
|
||
"""
|
||
if self.ap is not None:
|
||
self.ap.logger.info(
|
||
f'QQ Official _handle_form_chunk entered; '
|
||
f'source_present={message_source.source_platform_object is not None} '
|
||
f'form_actions={len(form_data.get("actions") or [])}'
|
||
)
|
||
self._prune_pending_forms()
|
||
|
||
source = message_source.source_platform_object
|
||
scene_target = self._resolve_target_from_event(message_source)
|
||
if scene_target is None:
|
||
# No rich-UI fit — fall through to existing text path.
|
||
await self.logger.info('QQ Official: form chunk on unsupported scene; falling back to text')
|
||
text_parts = [m.text for m in message if type(m) is platform_message.Plain]
|
||
fallback_msg = platform_message.MessageChain([platform_message.Plain(text='\n\n'.join(text_parts))])
|
||
try:
|
||
await self.reply_message(message_source, fallback_msg)
|
||
except Exception:
|
||
await self.logger.error(f'QQ Official: form fallback text send failed: {traceback.format_exc()}')
|
||
return
|
||
|
||
target_type, target_id = scene_target
|
||
session_key = f'{target_type}_{target_id}'
|
||
|
||
# Cancel any in-flight stream / fallback ctx so plain-text prefix
|
||
# doesn't continue alongside the keyboard message.
|
||
msg_id = getattr(source, 'd_id', '') or '' if source is not None else ''
|
||
if msg_id:
|
||
self._stream_ctx.pop(msg_id, None)
|
||
self._stream_ctx_ts.pop(msg_id, None)
|
||
self._fallback_text.pop(msg_id, None)
|
||
self._fallback_text_ts.pop(msg_id, None)
|
||
|
||
node_title = form_data.get('node_title') or 'Confirmation needed'
|
||
form_content = form_data.get('form_content') or ''
|
||
is_field_step = bool(form_data.get('_current_input_field')) and not form_data.get('_action_select_only')
|
||
parts = [f'### {node_title}']
|
||
plain_parts = [node_title]
|
||
if form_content.strip():
|
||
parts.append(form_content.strip())
|
||
plain_parts.append(form_content.strip())
|
||
markdown_content = '\n\n'.join(parts)
|
||
plain_content = '\n\n'.join(plain_parts)
|
||
|
||
keyboard = build_keyboard_from_select_field(form_data) if is_field_step else None
|
||
is_text_field_step = is_field_step and not keyboard.get('content', {}).get('rows')
|
||
if is_text_field_step:
|
||
keyboard = None
|
||
if keyboard is None and not is_text_field_step:
|
||
keyboard = build_keyboard_from_form(form_data, buttons_per_row=2)
|
||
if keyboard is not None and not keyboard.get('content', {}).get('rows') and not is_text_field_step:
|
||
# No actions to render — fall back to plain text.
|
||
text_msg = platform_message.MessageChain([platform_message.Plain(text=plain_content)])
|
||
try:
|
||
await self.reply_message(message_source, text_msg)
|
||
except Exception:
|
||
await self.logger.error(f'QQ Official: empty-keyboard fallback send failed: {traceback.format_exc()}')
|
||
return
|
||
|
||
# Prefer the inbound msg_id (no quota cost). If the source is a
|
||
# synthetic event from a prior click, the cached interaction id
|
||
# serves as event_id for up to 30 min.
|
||
event_id = None
|
||
if not msg_id:
|
||
cached = self._session_event_ids.get(session_key)
|
||
if cached and (time.time() - cached.get('posted_at', 0)) < self._PENDING_FORM_TTL:
|
||
event_id = cached.get('event_id')
|
||
|
||
anchor = msg_id or event_id or ''
|
||
msg_seq = self._next_msg_seq(anchor)
|
||
if msg_seq is None:
|
||
await self.logger.warning(
|
||
f'QQ Official: anchor {anchor!r} exhausted (>5 passive replies); '
|
||
f'cannot deliver form card for session={session_key}'
|
||
)
|
||
return
|
||
|
||
try:
|
||
await self.bot.send_markdown_keyboard(
|
||
target_type=target_type,
|
||
target_id=target_id,
|
||
markdown_content=markdown_content,
|
||
keyboard=keyboard,
|
||
msg_id=msg_id if (msg_id and not event_id) else None,
|
||
event_id=event_id,
|
||
msg_seq=msg_seq,
|
||
)
|
||
if self.ap is not None:
|
||
self.ap.logger.info(
|
||
f'QQ Official: form card sent '
|
||
f'target={target_type}/{target_id} '
|
||
f'msg_id={msg_id!r} event_id={event_id!r} msg_seq={msg_seq}'
|
||
)
|
||
except Exception:
|
||
if self.ap is not None:
|
||
self.ap.logger.error(
|
||
f'QQ Official: send_markdown_keyboard failed, falling back to text: {traceback.format_exc()}'
|
||
)
|
||
await self.logger.error(
|
||
f'QQ Official: send_markdown_keyboard failed, falling back to text: {traceback.format_exc()}'
|
||
)
|
||
text_msg = platform_message.MessageChain([platform_message.Plain(text=plain_content)])
|
||
try:
|
||
await self.reply_message(message_source, text_msg)
|
||
except Exception:
|
||
pass
|
||
return
|
||
|
||
sender_id = ''
|
||
if source is not None:
|
||
sender_id = (
|
||
getattr(source, 'user_openid', None)
|
||
or getattr(source, 'member_openid', None)
|
||
or getattr(source, 'd_author_id', None)
|
||
or ''
|
||
)
|
||
if not sender_id and message_source.sender is not None:
|
||
sender_id = str(getattr(message_source.sender, 'id', '') or '')
|
||
self._pending_forms[session_key] = {
|
||
'form_data': form_data,
|
||
'msg_id': msg_id,
|
||
'sender_id': sender_id,
|
||
'target_type': target_type,
|
||
'target_id': target_id,
|
||
'source_event_t': source.t if source is not None else None,
|
||
'posted_at': time.time(),
|
||
}
|
||
await self.logger.info(
|
||
f'QQ Official: form posted session={session_key} actions={len(form_data.get("actions") or [])}'
|
||
)
|
||
|
||
async def _handle_interaction_create(
|
||
self,
|
||
event_data: dict,
|
||
ws_event_id: typing.Optional[str] = None,
|
||
) -> None:
|
||
"""Handle a button-click INTERACTION_CREATE event.
|
||
|
||
Two IDs at play (QQ keeps them separate):
|
||
ws_event_id top-level payload ``id`` (or webhook ``X-Bot-
|
||
Event-Id``). The ONLY value accepted as
|
||
``event_id`` for subsequent passive replies.
|
||
d['id'] the interaction id — used for PUT
|
||
/interactions/{id} ack. Cannot be reused as
|
||
event_id (QQ returns 40034025 if you try).
|
||
|
||
Layout (https://bot.q.qq.com/.../msg-btn.html):
|
||
chat_type 0 channel / 1 group / 2 c2c
|
||
data.resolved.button_data what we set as ``action.data``
|
||
data.resolved.button_id ``id`` field on the button row
|
||
"""
|
||
import langbot_plugin.api.entities.builtin.provider.session as provider_session
|
||
|
||
if self.ap is not None:
|
||
self.ap.logger.info(
|
||
f'QQ Official _handle_interaction_create entered; '
|
||
f'ws_event_id={ws_event_id!r} '
|
||
f'interaction_id={(event_data.get("id") if isinstance(event_data, dict) else None)!r} '
|
||
f'chat_type={event_data.get("chat_type") if isinstance(event_data, dict) else None}'
|
||
)
|
||
|
||
if not isinstance(event_data, dict):
|
||
await self.logger.warning(f'QQ Official: INTERACTION_CREATE event_data is not dict: {type(event_data)}')
|
||
return
|
||
|
||
# ACK uses the interaction id, NOT the ws event id.
|
||
interaction_id = event_data.get('id') or ''
|
||
if interaction_id:
|
||
self._start_background_task(self.bot.ack_interaction(interaction_id, code=0))
|
||
|
||
resolved = (event_data.get('data') or {}).get('resolved') or {}
|
||
action_id = str(resolved.get('button_data') or resolved.get('button_id') or '').strip()
|
||
if not action_id:
|
||
await self.logger.warning('QQ Official: INTERACTION_CREATE missing button_data/button_id; ignoring')
|
||
return
|
||
|
||
chat_type = event_data.get('chat_type')
|
||
scene_target: typing.Optional[tuple[str, str]] = None
|
||
if chat_type == 2 or event_data.get('user_openid'):
|
||
scene_target = ('c2c', event_data.get('user_openid') or '')
|
||
elif chat_type == 1 or event_data.get('group_openid'):
|
||
scene_target = ('group', event_data.get('group_openid') or '')
|
||
elif chat_type == 0 or event_data.get('channel_id'):
|
||
scene_target = ('channel', event_data.get('channel_id') or '')
|
||
|
||
if not scene_target or not scene_target[1]:
|
||
await self.logger.warning(f'QQ Official: INTERACTION_CREATE missing scene/target; raw={event_data}')
|
||
return
|
||
|
||
target_type, target_id = scene_target
|
||
session_key = f'{target_type}_{target_id}'
|
||
|
||
self._prune_pending_forms()
|
||
pending = self._pending_forms.get(session_key)
|
||
if not pending:
|
||
await self.logger.warning(
|
||
f'QQ Official: no pending form for session {session_key}; click ignored (action_id={action_id!r})'
|
||
)
|
||
return
|
||
|
||
# Cache ws_event_id so a follow-up pause / text reply can use it
|
||
# as event_id for passive delivery (30-min window). Falls back to
|
||
# the interaction_id only if no ws_event_id was provided (e.g.
|
||
# tests / older payload shape) — QQ will reject that value but
|
||
# we log so the mismatch is debuggable.
|
||
cached_event_id = ws_event_id or interaction_id
|
||
if cached_event_id:
|
||
self._session_event_ids[session_key] = {
|
||
'event_id': cached_event_id,
|
||
'posted_at': time.time(),
|
||
}
|
||
# New anchor → fresh 5-reply budget.
|
||
self._anchor_msg_seq[cached_event_id] = 0
|
||
while len(self._anchor_msg_seq) > 4096:
|
||
self._anchor_msg_seq.pop(next(iter(self._anchor_msg_seq)), None)
|
||
if self.ap is not None and not ws_event_id:
|
||
self.ap.logger.warning(
|
||
'QQ Official: INTERACTION_CREATE lacked ws_event_id; '
|
||
'falling back to interaction_id (passive reply may be rejected)'
|
||
)
|
||
|
||
form_data: dict = pending.get('form_data') or {}
|
||
actions = form_data.get('actions') or []
|
||
select_choice = resolve_select_button_action(form_data, action_id)
|
||
if action_id.startswith(QQ_SELECT_ACTION_PREFIX) and select_choice is None:
|
||
await self.logger.warning(f'QQ Official: invalid select action_id={action_id!r} for {session_key}')
|
||
return
|
||
|
||
matched = None
|
||
if select_choice is None:
|
||
matched = next(
|
||
(a for a in actions if str(a.get('id', '')) == action_id),
|
||
None,
|
||
)
|
||
if matched is None:
|
||
await self.logger.warning(
|
||
f'QQ Official: action_id={action_id!r} is not present on pending form for {session_key}'
|
||
)
|
||
return
|
||
self._pending_forms.pop(session_key, None)
|
||
action_title = select_choice[1] if select_choice else matched.get('title') or action_id
|
||
|
||
initiator_id = str(pending.get('sender_id') or '')
|
||
actor_id = str(event_data.get('member_openid') or event_data.get('user_openid') or initiator_id)
|
||
|
||
# Build resume payload matching the shape every other adapter uses
|
||
# (DingTalk / Lark / Telegram / WeCom). The runner's
|
||
# _merge_pending_form_action consumes this verbatim.
|
||
if target_type == 'group' or target_type == 'channel':
|
||
launcher_type = provider_session.LauncherTypes.GROUP
|
||
launcher_id = target_id
|
||
else:
|
||
launcher_type = provider_session.LauncherTypes.PERSON
|
||
launcher_id = target_id
|
||
|
||
form_action_data = {
|
||
'form_token': form_data.get('form_token', ''),
|
||
'workflow_run_id': form_data.get('workflow_run_id', ''),
|
||
'action_id': '' if select_choice else action_id,
|
||
'action_title': action_title,
|
||
'node_title': form_data.get('node_title', ''),
|
||
'user': f'{launcher_type.value}_{launcher_id}',
|
||
'inputs': {'select': select_choice[1]} if select_choice else {},
|
||
}
|
||
if select_choice:
|
||
form_action_data['_current_input_field'] = select_choice[0]
|
||
form_action_data['_input_progress'] = True
|
||
|
||
event_label = 'Form Select' if select_choice else 'Form Action'
|
||
message_chain = platform_message.MessageChain([platform_message.Plain(text=f'[{event_label}: {action_title}]')])
|
||
|
||
if launcher_type == provider_session.LauncherTypes.GROUP:
|
||
synthetic_event: platform_events.MessageEvent = platform_events.GroupMessage(
|
||
sender=platform_entities.GroupMember(
|
||
id=actor_id or launcher_id,
|
||
member_name='',
|
||
permission='MEMBER',
|
||
group=platform_entities.Group(
|
||
id=launcher_id,
|
||
name='',
|
||
permission=platform_entities.Permission.Member,
|
||
),
|
||
special_title='',
|
||
),
|
||
message_chain=message_chain,
|
||
time=int(time.time()),
|
||
source_platform_object=None,
|
||
)
|
||
else:
|
||
synthetic_event = platform_events.FriendMessage(
|
||
sender=platform_entities.Friend(
|
||
id=actor_id or launcher_id,
|
||
nickname='',
|
||
remark='',
|
||
),
|
||
message_chain=message_chain,
|
||
time=int(time.time()),
|
||
source_platform_object=None,
|
||
)
|
||
|
||
if self.ap is None:
|
||
await self.logger.error('QQ Official: ap not injected; cannot enqueue button-click query')
|
||
return
|
||
|
||
bot_uuid = ''
|
||
pipeline_uuid = form_data.get('pipeline_uuid') or None
|
||
for bot in self.ap.platform_mgr.bots:
|
||
if bot.adapter is self:
|
||
bot_uuid = bot.bot_entity.uuid
|
||
pipeline_uuid = pipeline_uuid or bot.bot_entity.use_pipeline_uuid
|
||
break
|
||
|
||
try:
|
||
await self.ap.query_pool.add_query(
|
||
bot_uuid=bot_uuid,
|
||
launcher_type=launcher_type,
|
||
launcher_id=launcher_id,
|
||
sender_id=actor_id or launcher_id,
|
||
message_event=synthetic_event,
|
||
message_chain=message_chain,
|
||
adapter=self,
|
||
pipeline_uuid=pipeline_uuid,
|
||
variables={
|
||
'_dify_form_action': form_action_data,
|
||
'_routed_by_rule': True,
|
||
},
|
||
)
|
||
await self.logger.info(
|
||
f'QQ Official: button-click query enqueued action_id={action_id!r} '
|
||
f'session={session_key} actor_id={actor_id}'
|
||
)
|
||
except Exception:
|
||
await self.logger.error(f'QQ Official: enqueue button-click query failed: {traceback.format_exc()}')
|