from __future__ import annotations

import asyncio
import base64
import json
import time
import traceback
import uuid
import xml.etree.ElementTree as ET
from dataclasses import dataclass, field
import re
from typing import TYPE_CHECKING, Any, Callable, Optional, Tuple
from urllib.parse import unquote

from Crypto.Cipher import AES
from quart import Quart, request, Response, jsonify

from langbot.libs.wecom_ai_bot_api import wecombotevent
from langbot.libs.wecom_ai_bot_api.WXBizMsgCrypt3 import WXBizMsgCrypt

if TYPE_CHECKING:
    from langbot.pkg.platform.logger import EventLogger
from langbot.pkg.utils import httpclient

_CLIENT_TRANSIENT_CACHE_MAX = 4096
_MAX_STREAM_CONTENT_CHARS = 200000


@dataclass
class StreamChunk:
    """描述单次推送给企业微信的流式片段。"""

    # 需要返回给企业微信的文本内容
    content: str

    # 标记是否为最终片段，对应企业微信协议里的 finish 字段
    is_final: bool = False

    # 预留额外元信息，未来支持多模态扩展时可使用
    meta: dict[str, Any] = field(default_factory=dict)


@dataclass
class StreamSession:
    """维护一次企业微信流式会话的上下文。"""

    # 企业微信要求的 stream_id，用于标识后续刷新请求
    stream_id: str

    # 原始消息的 msgid，便于与流水线消息对应
    msg_id: str

    # 群聊会话标识（单聊时为空）
    chat_id: Optional[str]

    # 触发消息的发送者
    user_id: Optional[str]

    # 会话创建时间
    created_at: float = field(default_factory=time.time)

    # 最近一次被访问的时间，cleanup 依据该值判断过期
    last_access: float = field(default_factory=time.time)

    # 将流水线增量结果缓存到队列，刷新请求逐条消费
    queue: asyncio.Queue = field(default_factory=lambda: asyncio.Queue(maxsize=1))

    # 是否已经完成（收到最终片段）
    finished: bool = False

    # 缓存最近一次片段，处理重试或超时兜底
    last_chunk: Optional[StreamChunk] = None

    # 反馈 ID，用于接收用户点赞/点踩反馈
    feedback_id: Optional[str] = None

    # Dify 人工输入暂停态：runner 把 _form_data 传过来时填充。
    # 一旦设置，下次企微 followup 请求时返回 button_interaction 模板卡
    # 替代 stream chunk。点击按钮会回调 template_card_event，EventKey
    # 就是 Dify 的 action_id。
    pending_form: Optional[dict] = None

    # template_card task_id（企微要求 button_interaction 必填且不可重复）。
    # 创建 pending_form 时生成；按钮点击回调里用来反查 session。
    pending_form_task_id: Optional[str] = None


class StreamSessionManager:
    """管理 stream 会话的生命周期，并负责队列的生产消费。"""

    # Sessions with registered feedback_ids use a longer TTL to survive the
    # full like → cancel → dislike feedback flow. Must align with the adapter's
    # _stream_to_monitoring_msg TTL (wecombot.py).
    _FEEDBACK_SESSION_TTL = 600  # 10 minutes
    _MAX_SESSIONS = 4096

    def __init__(self, logger: EventLogger, ttl: int = 60) -> None:
        self.logger = logger

        self.ttl = ttl  # 超时时间（秒），超过该时间未被访问的会话会被清理由 cleanup
        self._sessions: dict[str, StreamSession] = {}  # stream_id -> StreamSession 映射
        self._msg_index: dict[str, str] = {}  # msgid -> stream_id 映射，便于流水线根据消息 ID 找到会话
        self._feedback_index: dict[str, str] = {}  # feedback_id -> stream_id 映射
        # task_id (button_interaction template_card 的) -> stream_id 映射，
        # 用于按钮点击回调里反查 pending_form。
        self._task_index: dict[str, str] = {}

    def get_stream_id_by_msg(self, msg_id: str) -> Optional[str]:
        if not msg_id:
            return None
        return self._msg_index.get(msg_id)

    def get_session(self, stream_id: str) -> Optional[StreamSession]:
        return self._sessions.get(stream_id)

    def get_session_by_feedback_id(self, feedback_id: str) -> Optional[StreamSession]:
        """根据 feedback_id 查找会话。

        Args:
            feedback_id: 企业微信反馈事件中的反馈 ID。

        Returns:
            Optional[StreamSession]: 找到的会话实例，未找到返回 None。
        """
        if not feedback_id:
            return None
        stream_id = self._feedback_index.get(feedback_id)
        if stream_id:
            return self._sessions.get(stream_id)
        return None

    def register_feedback_id(self, stream_id: str, feedback_id: str) -> None:
        """注册 feedback_id 与 stream_id 的映射。

        Args:
            stream_id: 企业微信流式会话 ID。
            feedback_id: 反馈 ID。
        """
        if feedback_id and stream_id:
            self._feedback_index[feedback_id] = stream_id

    def set_pending_form(self, stream_id: str, form_data: dict, task_id: str) -> None:
        """把 Dify 人工输入暂停态绑定到 stream session。

        下一次企微 followup 请求时，adapter 检测到 pending_form，
        返回 button_interaction 模板卡而不是 stream chunk。
        """
        session = self._sessions.get(stream_id)
        if not session:
            return
        session.pending_form = form_data
        session.pending_form_task_id = task_id
        if task_id:
            self._task_index[task_id] = stream_id

    def get_session_by_task_id(self, task_id: str) -> Optional[StreamSession]:
        """按按钮点击回调里的 TaskId 反查 session。"""
        if not task_id:
            return None
        stream_id = self._task_index.get(task_id)
        if not stream_id:
            return None
        return self._sessions.get(stream_id)

    def clear_pending_form(self, stream_id: str) -> None:
        """按钮点击消费完后清掉 pending_form，避免重复弹卡。"""
        session = self._sessions.get(stream_id)
        if not session:
            return
        task_id = session.pending_form_task_id
        session.pending_form = None
        session.pending_form_task_id = None
        if task_id:
            self._task_index.pop(task_id, None)

    def clear(self) -> None:
        """Release every retained stream and reverse index."""

        self._sessions.clear()
        self._msg_index.clear()
        self._feedback_index.clear()
        self._task_index.clear()

    def _drop_session(self, stream_id: str) -> StreamSession | None:
        session = self._sessions.pop(stream_id, None)
        if session is None:
            return None
        if session.msg_id and self._msg_index.get(session.msg_id) == stream_id:
            self._msg_index.pop(session.msg_id, None)
        if session.feedback_id:
            self._feedback_index.pop(session.feedback_id, None)
        if session.pending_form_task_id:
            self._task_index.pop(session.pending_form_task_id, None)
        return session

    def create_or_get(self, msg_json: dict[str, Any]) -> tuple[StreamSession, bool]:
        """根据企业微信回调创建或获取会话。

        Args:
            msg_json: 企业微信解密后的回调 JSON。

        Returns:
            Tuple[StreamSession, bool]: `StreamSession` 为会话实例，`bool` 指示是否为新建会话。

        Example:
            在首次回调中调用，得到 `is_new=True` 后再触发流水线。
        """
        msg_id = msg_json.get('msgid', '')
        if msg_id and msg_id in self._msg_index:
            stream_id = self._msg_index[msg_id]
            session = self._sessions.get(stream_id)
            if session:
                session.last_access = time.time()
                return session, False

        self.cleanup()
        while len(self._sessions) >= self._MAX_SESSIONS:
            oldest_stream_id = min(
                self._sessions,
                key=lambda candidate: self._sessions[candidate].last_access,
            )
            self._drop_session(oldest_stream_id)

        stream_id = str(uuid.uuid4())
        session = StreamSession(
            stream_id=stream_id,
            msg_id=msg_id,
            chat_id=msg_json.get('chatid'),
            user_id=msg_json.get('from', {}).get('userid'),
        )

        if msg_id:
            self._msg_index[msg_id] = stream_id
        self._sessions[stream_id] = session
        return session, True

    async def publish(self, stream_id: str, chunk: StreamChunk) -> bool:
        """向 stream 队列写入新的增量片段。

        Args:
            stream_id: 企业微信分配的流式会话 ID。
            chunk: 待发送的增量片段。

        Returns:
            bool: 当流式队列存在并成功入队时返回 True。

        Example:
            在收到模型增量后调用 `await manager.publish('sid', StreamChunk('hello'))`。
        """
        session = self._sessions.get(stream_id)
        if not session:
            return False

        session.last_access = time.time()
        session.last_chunk = chunk

        try:
            session.queue.put_nowait(chunk)
        except asyncio.QueueFull:
            # Each chunk is a complete snapshot. Coalesce a slow consumer to
            # the newest value instead of retaining every intermediate body.
            try:
                session.queue.get_nowait()
            except asyncio.QueueEmpty:
                pass
            session.queue.put_nowait(chunk)

        if chunk.is_final:
            session.finished = True

        return True

    async def consume(self, stream_id: str, timeout: float = 0.5) -> Optional[StreamChunk]:
        """从队列中取出一个片段，若超时返回 None。

        Args:
            stream_id: 企业微信流式会话 ID。
            timeout: 取片段的最长等待时间（秒）。

        Returns:
            Optional[StreamChunk]: 成功时返回片段，超时或会话不存在时返回 None。

        Example:
            企业微信刷新到达时调用，若队列有数据则立即返回 `StreamChunk`。
        """
        session = self._sessions.get(stream_id)
        if not session:
            return None

        session.last_access = time.time()

        try:
            chunk = await asyncio.wait_for(session.queue.get(), timeout)
            session.last_access = time.time()
            if chunk.is_final:
                session.finished = True
            return chunk
        except asyncio.TimeoutError:
            if session.finished and session.last_chunk:
                return session.last_chunk
            return None

    def mark_finished(self, stream_id: str) -> None:
        session = self._sessions.get(stream_id)
        if session:
            session.finished = True
            session.last_access = time.time()

    def cleanup(self) -> list[str]:
        """定期清理过期会话，防止队列与映射无上限累积。

        已注册 feedback_id 的会话使用更长的 TTL，确保用户在点赞/取消/点踩流程中
        不会因为 session 被提前清除而丢失上下文信息。
        """
        now = time.time()
        expired: list[str] = []
        for stream_id, session in self._sessions.items():
            # Sessions with registered feedback_ids use a longer TTL
            effective_ttl = self._FEEDBACK_SESSION_TTL if session.feedback_id else self.ttl
            if now - session.last_access > effective_ttl:
                expired.append(stream_id)

        removed_msg_ids: list[str] = []
        for stream_id in expired:
            session = self._drop_session(stream_id)
            if not session:
                continue
            if session.msg_id:
                removed_msg_ids.append(session.msg_id)
        return removed_msg_ids


def _decrypt_file(encrypted_data: bytes, aes_key_str: str) -> bytes:
    """Decrypt AES-256-CBC encrypted file data.

    Aligned with the official WeCom AI Bot Python SDK (crypto_utils.py).

    Args:
        encrypted_data: The raw encrypted bytes.
        aes_key_str: Base64-encoded AES key (may lack padding).

    Returns:
        Decrypted bytes with PKCS#7 padding removed.
    """
    if not encrypted_data:
        raise ValueError('encrypted_data is empty')
    if not aes_key_str:
        raise ValueError('aes_key is empty')

    # Python's base64.b64decode requires proper padding (length % 4 == 0).
    # Node.js Buffer.from tolerates missing '=', so we must pad manually.
    remainder = len(aes_key_str) % 4
    if remainder != 0:
        aes_key_str = aes_key_str + '=' * (4 - remainder)
    key = base64.b64decode(aes_key_str)

    iv = key[:16]

    cipher = AES.new(key, AES.MODE_CBC, iv)

    # Ensure encrypted data is aligned to AES block size (16 bytes).
    # Node.js setAutoPadding(false) silently handles unaligned data,
    # but PyCryptodome will raise an error.
    block_size = 16
    data_remainder = len(encrypted_data) % block_size
    if data_remainder != 0:
        encrypted_data = encrypted_data + b'\x00' * (block_size - data_remainder)

    decrypted = cipher.decrypt(encrypted_data)

    # Remove PKCS#7 padding with validation
    if len(decrypted) == 0:
        raise ValueError('Decrypted data is empty')

    pad_len = decrypted[-1]
    if pad_len < 1 or pad_len > 32 or pad_len > len(decrypted):
        raise ValueError(f'Invalid PKCS#7 padding value: {pad_len}')

    # Verify all padding bytes are consistent
    for i in range(len(decrypted) - pad_len, len(decrypted)):
        if decrypted[i] != pad_len:
            raise ValueError('Invalid PKCS#7 padding: padding bytes mismatch')

    return decrypted[: len(decrypted) - pad_len]


def _extract_filename(content_disposition: str) -> Optional[str]:
    """Extract filename from a Content-Disposition header value."""
    if not content_disposition:
        return None
    # RFC 5987: filename*=UTF-8''xxx
    utf8_match = re.search(r"filename\*=UTF-8''([^;\s]+)", content_disposition, re.IGNORECASE)
    if utf8_match:
        return unquote(utf8_match.group(1))
    # Standard: filename="xxx" or filename=xxx
    match = re.search(r'filename="?([^";\s]+)"?', content_disposition, re.IGNORECASE)
    if match:
        return unquote(match.group(1))
    return None


def _bytes_to_data_uri(data: bytes) -> str:
    """Convert raw bytes to a data URI with auto-detected MIME type."""
    if data.startswith(b'\xff\xd8'):
        mime_type = 'image/jpeg'
    elif data.startswith(b'\x89PNG'):
        mime_type = 'image/png'
    elif data.startswith((b'GIF87a', b'GIF89a')):
        mime_type = 'image/gif'
    elif data.startswith(b'BM'):
        mime_type = 'image/bmp'
    elif data.startswith(b'II*\x00') or data.startswith(b'MM\x00*'):
        mime_type = 'image/tiff'
    elif data[:4] == b'%PDF':
        mime_type = 'application/pdf'
    elif data[:4] == b'PK\x03\x04':
        mime_type = 'application/zip'
    else:
        mime_type = 'application/octet-stream'

    base64_str = base64.b64encode(data).decode('utf-8')
    return f'data:{mime_type};base64,{base64_str}'


async def download_encrypted_file(
    download_url: str, aes_key: str, logger: EventLogger
) -> Tuple[Optional[bytes], Optional[str]]:
    """Download an AES-encrypted file from WeChat Work and decrypt it.

    Args:
        download_url: The encrypted file download URL.
        aes_key: The AES key for decryption (base64-encoded, per-message aeskey
                 or platform EncodingAESKey).
        logger: Logger instance.

    Returns:
        A tuple of (decrypted_bytes, filename) or (None, None) on failure.
    """
    if not download_url:
        return None, None
    if not aes_key:
        await logger.error('download_encrypted_file: aes_key is empty, cannot decrypt')
        return None, None

    filename: Optional[str] = None
    try:
        client = httpclient.get_session()
        async with client.get(download_url, timeout=30.0) as response:
            if response.status != 200:
                await logger.error(f'Failed to download file (HTTP {response.status})')
                return None, None
            encrypted_bytes = await httpclient.read_limited(response)
            filename = _extract_filename(response.headers.get('content-disposition', ''))
    except Exception:
        await logger.error(f'Failed to download file: {traceback.format_exc()}')
        return None, None

    try:
        decrypted = await asyncio.to_thread(_decrypt_file, encrypted_bytes, aes_key)
        return decrypted, filename
    except Exception:
        await logger.error(f'Failed to decrypt file: {traceback.format_exc()}')
        return None, None


async def parse_wecom_bot_message(
    msg_json: dict[str, Any], encoding_aes_key: str, logger: EventLogger
) -> dict[str, Any]:
    """Parse a decrypted WeChat Work AI Bot message JSON into a unified message dict.

    This is the shared message parsing logic used by both webhook and WebSocket modes.

    Args:
        msg_json: The decrypted message JSON from WeChat Work.
        encoding_aes_key: AES key for file decryption.
        logger: Logger instance.

    Returns:
        A dict suitable for constructing a WecomBotEvent.
    """
    message_data: dict[str, Any] = {}

    msg_type = msg_json.get('msgtype', '')
    if msg_type:
        message_data['msgtype'] = msg_type

    if msg_json.get('chattype', '') == 'single':
        message_data['type'] = 'single'
    elif msg_json.get('chattype', '') == 'group':
        message_data['type'] = 'group'

    max_inline_file_size = 5 * 1024 * 1024

    async def _safe_download(url: str, per_msg_aeskey: str = '') -> Tuple[Optional[bytes], Optional[str]]:
        """Download and decrypt a file, preferring per-message aeskey over platform key."""
        if not url:
            return None, None
        key = per_msg_aeskey or encoding_aes_key
        if not key:
            await logger.warning('No AES key available for file decryption, skipping download')
            return None, None
        return await download_encrypted_file(url, key, logger)

    async def _safe_download_as_data_uri(url: str, per_msg_aeskey: str = '') -> Optional[str]:
        """Download, decrypt, and convert to data URI for backward compatibility."""
        data, _filename = await _safe_download(url, per_msg_aeskey)
        if data:
            return await asyncio.to_thread(_bytes_to_data_uri, data)
        return None

    if msg_type == 'text':
        message_data['content'] = msg_json.get('text', {}).get('content')
    elif msg_type == 'markdown':
        message_data['content'] = msg_json.get('markdown', {}).get('content') or msg_json.get('text', {}).get(
            'content', ''
        )
    elif msg_type == 'image':
        image_info = msg_json.get('image', {})
        picurl = image_info.get('url', '')
        per_msg_aeskey = image_info.get('aeskey', '')
        base64_data = await _safe_download_as_data_uri(picurl, per_msg_aeskey)
        if base64_data:
            message_data['picurl'] = base64_data
            message_data['images'] = [base64_data]
    elif msg_type == 'voice':
        voice_info = msg_json.get('voice', {}) or {}
        download_url = voice_info.get('url')
        per_msg_aeskey = voice_info.get('aeskey', '')
        message_data['voice'] = {
            'url': download_url,
            'md5sum': voice_info.get('md5sum') or voice_info.get('md5'),
            'filesize': voice_info.get('filesize') or voice_info.get('size'),
            'sdkfileid': voice_info.get('sdkfileid') or voice_info.get('fileid'),
        }
        if voice_info.get('content'):
            message_data['content'] = voice_info.get('content')
        # if (message_data['voice'].get('filesize') or 0) <= max_inline_file_size:
        #     voice_base64 = await _safe_download_as_data_uri(download_url, per_msg_aeskey)
        #     if voice_base64:
        #         message_data['voice']['base64'] = voice_base64
    elif msg_type == 'video':
        video_info = msg_json.get('video', {}) or {}
        download_url = video_info.get('url')
        per_msg_aeskey = video_info.get('aeskey', '')
        video_data = {
            'url': download_url,
            'filesize': video_info.get('filesize') or video_info.get('size'),
            'sdkfileid': video_info.get('sdkfileid') or video_info.get('fileid'),
            'md5sum': video_info.get('md5sum') or video_info.get('md5'),
            'filename': video_info.get('filename') or video_info.get('name'),
        }
        # if (video_data.get('filesize') or 0) <= max_inline_file_size:
        #     video_base64 = await _safe_download_as_data_uri(download_url, per_msg_aeskey)
        #     if video_base64:
        #         video_data['base64'] = video_base64
        # 应为需要解密，但是目前暂时不能下载到内部进行解密，所以先将下载链接拼接aeskey返回给用户，由插件去处理该链接的下载和解密逻辑
        video_data['download_url'] = download_url + f'?aeskey={per_msg_aeskey}'
        message_data['video'] = video_data
    elif msg_type == 'file':
        file_info = msg_json.get('file', {}) or {}
        download_url = file_info.get('url') or file_info.get('fileurl')
        per_msg_aeskey = file_info.get('aeskey', '')
        file_data = {
            'filename': file_info.get('filename') or file_info.get('name'),
            'filesize': file_info.get('filesize') or file_info.get('size'),
            'md5sum': file_info.get('md5sum') or file_info.get('md5'),
            'sdkfileid': file_info.get('sdkfileid') or file_info.get('fileid'),
            'download_url': download_url,
            'extra': file_info,
        }
        # if (file_data.get('filesize') or 0) <= max_inline_file_size:
        #     file_bytes, dl_filename = await _safe_download(download_url, per_msg_aeskey)
        #     if file_bytes:
        #         file_data['base64'] = _bytes_to_data_uri(file_bytes)
        #         if dl_filename and not file_data.get('filename'):
        #             file_data['filename'] = dl_filename

        # 应为需要解密，但是目前暂时不能下载到内部进行解密，所以先将下载链接拼接aeskey返回给用户，由插件去处理该链接的下载和解密逻辑
        file_data['download_url'] = download_url + f'?aeskey={per_msg_aeskey}'
        message_data['file'] = file_data
    elif msg_type == 'link':
        message_data['link'] = msg_json.get('link', {})
        if not message_data.get('content'):
            title = message_data['link'].get('title', '')
            desc = message_data['link'].get('description') or message_data['link'].get('digest', '')
            message_data['content'] = '\n'.join(filter(None, [title, desc]))
    elif msg_type == 'mixed':
        items = msg_json.get('mixed', {}).get('msg_item', [])
        texts = []
        images = []
        files = []
        voices = []
        videos = []
        links = []
        for item in items:
            item_type = item.get('msgtype')
            if item_type == 'text':
                texts.append(item.get('text', {}).get('content', ''))
            elif item_type == 'image':
                img_info = item.get('image', {})
                img_url = img_info.get('url')
                img_aeskey = img_info.get('aeskey', '')
                base64_data = await _safe_download_as_data_uri(img_url, img_aeskey)
                if base64_data:
                    images.append(base64_data)
            elif item_type == 'file':
                file_info = item.get('file', {}) or {}
                download_url = file_info.get('url') or file_info.get('fileurl')
                item_aeskey = file_info.get('aeskey', '')
                file_data = {
                    'filename': file_info.get('filename') or file_info.get('name'),
                    'filesize': file_info.get('filesize') or file_info.get('size'),
                    'md5sum': file_info.get('md5sum') or file_info.get('md5'),
                    'sdkfileid': file_info.get('sdkfileid') or file_info.get('fileid'),
                    'download_url': download_url,
                    'extra': file_info,
                }
                if (file_data.get('filesize') or 0) <= max_inline_file_size:
                    file_bytes, dl_filename = await _safe_download(download_url, item_aeskey)
                    if file_bytes:
                        file_data['base64'] = await asyncio.to_thread(
                            _bytes_to_data_uri,
                            file_bytes,
                        )
                        if dl_filename and not file_data.get('filename'):
                            file_data['filename'] = dl_filename
                files.append(file_data)
            elif item_type == 'voice':
                voice_info = item.get('voice', {}) or {}
                download_url = voice_info.get('url')
                item_aeskey = voice_info.get('aeskey', '')
                voice_data = {
                    'url': download_url,
                    'md5sum': voice_info.get('md5sum') or voice_info.get('md5'),
                    'filesize': voice_info.get('filesize') or voice_info.get('size'),
                    'sdkfileid': voice_info.get('sdkfileid') or voice_info.get('fileid'),
                }
                if voice_info.get('content'):
                    texts.append(voice_info.get('content'))
                if (voice_data.get('filesize') or 0) <= max_inline_file_size:
                    voice_base64 = await _safe_download_as_data_uri(download_url, item_aeskey)
                    if voice_base64:
                        voice_data['base64'] = voice_base64
                voices.append(voice_data)
            elif item_type == 'video':
                video_info = item.get('video', {}) or {}
                download_url = video_info.get('url')
                item_aeskey = video_info.get('aeskey', '')
                video_data = {
                    'url': download_url,
                    'filesize': video_info.get('filesize') or video_info.get('size'),
                    'sdkfileid': video_info.get('sdkfileid') or video_info.get('fileid'),
                    'md5sum': video_info.get('md5sum') or video_info.get('md5'),
                    'filename': video_info.get('filename') or video_info.get('name'),
                }
                if (video_data.get('filesize') or 0) <= max_inline_file_size:
                    video_base64 = await _safe_download_as_data_uri(download_url, item_aeskey)
                    if video_base64:
                        video_data['base64'] = video_base64
                videos.append(video_data)
            elif item_type == 'link':
                links.append(item.get('link', {}))

        if texts:
            message_data['content'] = ' '.join(texts)
        if images:
            message_data['images'] = images
            message_data['picurl'] = images[0]
        if files:
            message_data['files'] = files
            message_data['file'] = files[0]
        if voices:
            message_data['voices'] = voices
            message_data['voice'] = voices[0]
        if videos:
            message_data['videos'] = videos
            message_data['video'] = videos[0]
        if links:
            message_data['link'] = links[0]
        if items:
            message_data['attachments'] = items
    else:
        message_data['raw_msg'] = msg_json

    from_info = msg_json.get('from', {})
    message_data['userid'] = from_info.get('userid', '')
    message_data['username'] = from_info.get('alias', '') or from_info.get('name', '') or from_info.get('userid', '')

    if msg_json.get('chattype', '') == 'group':
        message_data['chatid'] = msg_json.get('chatid', '')
        message_data['chatname'] = msg_json.get('chatname', '') or msg_json.get('chatid', '')

    message_data['msgid'] = msg_json.get('msgid', '')

    if msg_json.get('aibotid'):
        message_data['aibotid'] = msg_json.get('aibotid', '')

    # Handle quote (referenced message) - important for group chat file references
    quote_info = msg_json.get('quote')
    if quote_info:
        quote_data: dict[str, Any] = {}
        quote_type = quote_info.get('msgtype', '')
        quote_data['msgtype'] = quote_type

        if quote_type == 'text':
            quote_data['content'] = quote_info.get('text', {}).get('content', '')
        elif quote_type == 'image':
            img_info = quote_info.get('image', {})
            img_url = img_info.get('url', '')
            img_aeskey = img_info.get('aeskey', '')
            base64_data = await _safe_download_as_data_uri(img_url, img_aeskey)
            if base64_data:
                quote_data['picurl'] = base64_data
                quote_data['images'] = [base64_data]
        elif quote_type == 'file':
            file_info = quote_info.get('file', {}) or {}
            download_url = file_info.get('url') or file_info.get('fileurl')
            item_aeskey = file_info.get('aeskey', '')
            file_data = {
                'filename': file_info.get('filename') or file_info.get('name'),
                'filesize': file_info.get('filesize') or file_info.get('size'),
                'md5sum': file_info.get('md5sum') or file_info.get('md5'),
                'sdkfileid': file_info.get('sdkfileid') or file_info.get('fileid'),
                'download_url': download_url,
                'extra': file_info,
            }
            # Same as private chat: append aeskey to download_url for plugin processing
            if download_url and item_aeskey:
                file_data['download_url'] = download_url + f'?aeskey={item_aeskey}'
            quote_data['file'] = file_data
        elif quote_type == 'voice':
            voice_info = quote_info.get('voice', {}) or {}
            download_url = voice_info.get('url')
            item_aeskey = voice_info.get('aeskey', '')
            voice_data = {
                'url': download_url,
                'md5sum': voice_info.get('md5sum') or voice_info.get('md5'),
                'filesize': voice_info.get('filesize') or voice_info.get('size'),
                'sdkfileid': voice_info.get('sdkfileid') or voice_info.get('fileid'),
            }
            if voice_info.get('content'):
                quote_data['content'] = voice_info.get('content')
            # Same as private chat: append aeskey to url for plugin processing
            if download_url and item_aeskey:
                voice_data['url'] = download_url + f'?aeskey={item_aeskey}'
            quote_data['voice'] = voice_data
        elif quote_type == 'video':
            video_info = quote_info.get('video', {}) or {}
            download_url = video_info.get('url')
            item_aeskey = video_info.get('aeskey', '')
            video_data = {
                'url': download_url,
                'filesize': video_info.get('filesize') or video_info.get('size'),
                'sdkfileid': video_info.get('sdkfileid') or video_info.get('fileid'),
                'md5sum': video_info.get('md5sum') or video_info.get('md5'),
                'filename': video_info.get('filename') or video_info.get('name'),
            }
            # Same as private chat: append aeskey to download_url for plugin processing
            if download_url and item_aeskey:
                video_data['download_url'] = download_url + f'?aeskey={item_aeskey}'
            quote_data['video'] = video_data
        elif quote_type == 'link':
            quote_data['link'] = quote_info.get('link', {})
            link = quote_data['link']
            title = link.get('title', '')
            desc = link.get('description') or link.get('digest', '')
            quote_data['content'] = '\n'.join(filter(None, [title, desc]))
        elif quote_type == 'mixed':
            # Handle mixed type in quote (text + images + files etc.)
            items = quote_info.get('mixed', {}).get('msg_item', [])
            texts = []
            images = []
            files = []
            for item in items:
                item_type = item.get('msgtype')
                if item_type == 'text':
                    texts.append(item.get('text', {}).get('content', ''))
                elif item_type == 'image':
                    img_info = item.get('image', {})
                    img_url = img_info.get('url')
                    img_aeskey = img_info.get('aeskey', '')
                    base64_data = await _safe_download_as_data_uri(img_url, img_aeskey)
                    if base64_data:
                        images.append(base64_data)
                elif item_type == 'file':
                    file_info = item.get('file', {}) or {}
                    download_url = file_info.get('url') or file_info.get('fileurl')
                    item_aeskey = file_info.get('aeskey', '')
                    file_data = {
                        'filename': file_info.get('filename') or file_info.get('name'),
                        'filesize': file_info.get('filesize') or file_info.get('size'),
                        'md5sum': file_info.get('md5sum') or file_info.get('md5'),
                        'sdkfileid': file_info.get('sdkfileid') or file_info.get('fileid'),
                        'download_url': download_url,
                        'extra': file_info,
                    }
                    # Same as private chat: append aeskey to download_url for plugin processing
                    if download_url and item_aeskey:
                        file_data['download_url'] = download_url + f'?aeskey={item_aeskey}'
                    files.append(file_data)
            if texts:
                quote_data['content'] = ' '.join(texts)
            if images:
                quote_data['images'] = images
                quote_data['picurl'] = images[0]
            if files:
                quote_data['files'] = files
                quote_data['file'] = files[0]

        message_data['quote'] = quote_data

    return message_data


def _wecom_button_style(action: dict, *, selected: bool = False) -> int:
    """Map Dify button style to WeCom button style."""

    if not selected:
        return 2

    return 1


def _wecom_field_display_name(field: dict, fallback: str = '') -> str:
    label = (
        field.get('label') or field.get('title') or field.get('name') or field.get('output_variable_name') or fallback
    )
    return str(label or fallback).strip()


def _wecom_input_hint_lines(form_data: dict) -> list[str]:
    lines: list[str] = []
    current_field = str(form_data.get('_current_input_field') or '').strip()
    for field in form_data.get('input_defs') or []:
        field_name = str(field.get('output_variable_name') or '').strip()
        field_type = str(field.get('type') or 'text').strip().lower()
        field_label = _wecom_field_display_name(field, field_name)
        if current_field and field_name != current_field:
            continue
        if not field_name:
            continue
        if field_type in {'file', 'file-list'}:
            limit = field.get('number_limits') if field_type == 'file-list' else 1
            allowed_types = ', '.join(field.get('allowed_file_types') or [])
            suffix = f', up to {limit}' if field_type == 'file-list' and limit else ''
            allowed = f' ({allowed_types})' if allowed_types else ''
            lines.append(f'- {field_label}: upload file(s){allowed}{suffix} or reply `{field_name}: <url>`')
    return lines


def _wecom_pending_input_defs(form_data: dict) -> list[dict]:
    if form_data.get('_action_select_only'):
        return []
    inputs = form_data.get('inputs') or {}
    current_field = str(form_data.get('_current_input_field') or '').strip()
    pending = []
    for field in form_data.get('input_defs') or []:
        field_name = str(field.get('output_variable_name') or '').strip()
        if not field_name:
            continue
        if current_field and field_name != current_field:
            continue
        if str(field.get('type') or '').strip().lower() in {'file', 'file-list'}:
            continue
        if inputs.get(field_name) in (None, '', []):
            pending.append(field)
    return pending


def _wecom_select_options(field: dict) -> list[str]:
    source = field.get('option_source') or {}
    options = source.get('value') if isinstance(source, dict) else []
    if not isinstance(options, list):
        return []
    return [str(option) for option in options]


def _wecom_select_option_id(index: int) -> str:
    return f'opt_{index + 1}'


def _wecom_pending_select_defs(form_data: dict) -> list[dict]:
    return [
        field
        for field in _wecom_pending_input_defs(form_data)
        if str(field.get('type') or '').strip().lower() == 'select' and _wecom_select_options(field)
    ]


def _wecom_field_title(field: dict, fallback: str) -> str:
    title = _wecom_field_display_name(field, fallback)
    return str(title or fallback).strip()[:13] or fallback


def _wecom_form_desc(form_data: dict) -> str:
    form_content = _wecom_clean_form_content(form_data)
    return form_content[:512] if form_content else ''


def build_human_input_text_prompt(form_data: dict) -> Optional[str]:
    """Build a plain-text prompt for a current non-select input field."""

    current_field = str(form_data.get('_current_input_field') or '').strip()
    if not current_field:
        return None
    for field in form_data.get('input_defs') or form_data.get('all_input_defs') or []:
        if str(field.get('output_variable_name') or '').strip() != current_field:
            continue
        field_type = str(field.get('type') or 'text').strip().lower()
        if field_type == 'select':
            return None
        form_content = _wecom_clean_form_content(form_data)
        if not form_content:
            form_content = _wecom_field_display_name(field, current_field)
        node_title = str(form_data.get('node_title') or '人工介入').strip()
        return f'{node_title}\n\n{form_content}' if form_content else node_title
    return None


def build_multiple_interaction_payload(
    form_data: dict,
    task_id: str,
    *,
    source: Optional[dict] = None,
) -> dict[str, Any]:
    """Build a WeCom multiple_interaction card for pending select fields."""

    select_fields = _wecom_pending_select_defs(form_data)
    node_title = (form_data.get('node_title') or '').strip() or 'Human Input'
    inputs = form_data.get('inputs') or {}

    select_list = []
    for field_index, field in enumerate(select_fields[:10]):
        field_name = str(field.get('output_variable_name') or '').strip()
        if not field_name:
            continue
        options = _wecom_select_options(field)[:10]
        option_list = [
            {
                'id': _wecom_select_option_id(idx),
                'text': option_text[:10] or _wecom_select_option_id(idx),
            }
            for idx, option_text in enumerate(options)
        ]
        selected_id = _wecom_select_option_id(0)
        current_value = inputs.get(field_name)
        if current_value not in (None, '', []):
            for idx, option_text in enumerate(options):
                if str(current_value) == option_text:
                    selected_id = _wecom_select_option_id(idx)
                    break
        select_list.append(
            {
                'question_key': field_name,
                'title': _wecom_field_title(field, f'Select {field_index + 1}'),
                'selected_id': selected_id,
                'option_list': option_list,
            }
        )

    card: dict[str, Any] = {
        'card_type': 'multiple_interaction',
        'main_title': {
            'title': node_title,
            'desc': _wecom_form_desc(form_data),
        },
        'select_list': select_list,
        'submit_button': {
            'text': 'Submit',
            'key': 'submit_human_input',
        },
        'task_id': task_id,
    }
    if source:
        card['source'] = source
    return {
        'msgtype': 'template_card',
        'template_card': card,
    }


_SELECT_BUTTON_KEY_PREFIX = '__dify_select__'


def _encode_select_button_key(field_name: str, option_index: int) -> str:
    data = json.dumps({'f': field_name, 'i': option_index}, ensure_ascii=False, separators=(',', ':'))
    encoded = base64.urlsafe_b64encode(data.encode('utf-8')).decode('ascii').rstrip('=')
    return f'{_SELECT_BUTTON_KEY_PREFIX}:{encoded}'


def parse_select_button_action(action_id: str, form_data: dict) -> dict[str, str]:
    """Decode a select option represented as a button_interaction click."""

    action_id = str(action_id or '').strip()
    prefix = f'{_SELECT_BUTTON_KEY_PREFIX}:'
    if not action_id.startswith(prefix):
        return {}
    encoded = action_id[len(prefix) :]
    try:
        padded = encoded + '=' * (-len(encoded) % 4)
        data = json.loads(base64.urlsafe_b64decode(padded.encode('ascii')).decode('utf-8'))
    except Exception:
        return {}
    field_name = str(data.get('f') or '').strip()
    option_index = data.get('i')
    if not field_name or not isinstance(option_index, int):
        return {}
    for field in form_data.get('input_defs') or form_data.get('all_input_defs') or []:
        if str(field.get('output_variable_name') or '').strip() != field_name:
            continue
        options = _wecom_select_options(field)
        if 0 <= option_index < len(options):
            return {field_name: options[option_index]}
    return {}


def build_select_button_interaction_payload(
    form_data: dict,
    task_id: str,
    *,
    source: Optional[dict] = None,
) -> dict[str, Any]:
    """Build a button_interaction card that emulates a select field.

    WeCom AI Bot long-connection callbacks are reliable for button clicks, so
    this is used as a fallback when multiple_interaction submit callbacks are
    not delivered by the platform.
    """

    select_fields = _wecom_pending_select_defs(form_data)
    field = select_fields[0] if select_fields else {}
    field_name = str(field.get('output_variable_name') or '').strip()
    options = _wecom_select_options(field)[:10] if field else []
    visible_options = options[:6]
    overflow_options = options[6:]

    node_title = (form_data.get('node_title') or '').strip() or 'Human Input'
    form_content = _wecom_clean_form_content(form_data)

    sub_title_parts: list[str] = []
    if form_content:
        sub_title_parts.append(form_content)
    if overflow_options:
        extra_lines = [f'  - {idx + 7}. {option}' for idx, option in enumerate(overflow_options)]
        sub_title_parts.append(
            'More options can be entered by replying with the option text:\n' + '\n'.join(extra_lines)
        )

    button_list = [
        {
            'text': option_text[:10] or f'Option {idx + 1}',
            'style': 2 if idx == 0 else 0,
            'key': _encode_select_button_key(field_name, idx),
        }
        for idx, option_text in enumerate(visible_options)
    ]

    card: dict[str, Any] = {
        'card_type': 'button_interaction',
        'main_title': {
            'title': node_title,
        },
        'sub_title_text': '\n\n'.join(sub_title_parts),
        'button_list': button_list,
        'task_id': task_id,
    }
    if source:
        card['source'] = source
    return {
        'msgtype': 'template_card',
        'template_card': card,
    }


def build_human_input_template_card_payload(
    form_data: dict,
    task_id: str,
    *,
    source: Optional[dict] = None,
    select_as_buttons: bool = False,
) -> dict[str, Any]:
    """Build the best WeCom template card for a Dify human-input form."""

    if _wecom_pending_select_defs(form_data):
        if select_as_buttons:
            return build_select_button_interaction_payload(form_data, task_id, source=source)
        return build_multiple_interaction_payload(form_data, task_id, source=source)
    return build_button_interaction_payload(form_data, task_id, source=source)


def _wecom_clean_form_content(form_data: dict) -> str:
    is_field_step = bool(form_data.get('_current_input_field')) and not form_data.get('_action_select_only')
    raw_content = str(form_data.get('raw_form_content') or '')
    content = form_data.get('form_content') or raw_content
    input_defs = list(form_data.get('all_input_defs') or form_data.get('input_defs') or [])
    fields = {
        str(field.get('output_variable_name') or '').strip(): field
        for field in input_defs
        if str(field.get('output_variable_name') or '').strip()
    }

    if form_data.get('_action_select_only') and raw_content:
        placeholders = [
            match
            for match in re.finditer(r'\{\{#\$output\.([^#{}]+)#\}\}', raw_content)
            if match.group(1).strip() in fields
        ]
        if placeholders:
            content = raw_content[placeholders[-1].end() :]

    if is_field_step:
        inputs = form_data.get('inputs') or {}

        def replace_placeholder(match: re.Match[str]) -> str:
            field_name = match.group(1).strip()
            field = fields.get(field_name)
            value = inputs.get(field_name)
            if not field or value in (None, '', []):
                return ''
            return f'✅ {field_name}：{_wecom_display_input_value(field, value)}'

        content = re.sub(r'\{\{#\$output\.([^#{}]+)#\}\}', replace_placeholder, str(content))

    kept_lines: list[str] = []
    for line in str(content).splitlines():
        placeholder = re.fullmatch(r'\s*\{\{#\$output\.([^#{}]+)#\}\}\s*', line)
        if placeholder and placeholder.group(1).strip() in fields:
            continue
        kept_lines.append(line)
    return re.sub(r'\n{3,}', '\n\n', '\n'.join(kept_lines).strip())


def _wecom_display_input_value(field: dict, value: Any) -> str:
    field_type = str(field.get('type') or 'text').strip().lower()
    if field_type == 'file':
        if isinstance(value, dict):
            return str(value.get('url') or value.get('upload_file_id') or '1 file')
    elif field_type == 'file-list' and isinstance(value, list):
        return f'{len(value)} file(s)'
    return str(value)


def build_button_interaction_payload(
    form_data: dict,
    task_id: str,
    *,
    source: Optional[dict] = None,
) -> dict[str, Any]:
    """Build a `template_card` (button_interaction) WeCom payload.

    Shared by both the webhook-mode client (returns the payload as the
    response to a stream-followup callback) and the ws_client (sends it
    as a reply frame). Output shape is `{"msgtype": "template_card",
    "template_card": {...}}` per the WeCom spec.

    Args:
        form_data: Dify human-input form data with keys ``actions`` (list of
            ``{id, title, button_style}``), ``node_title``, ``form_content``.
        task_id: Unique per-card identifier. WeCom requires this for
            button_interaction. The click callback returns it as TaskId so we
            can find the originating session.
        source: Optional source header dict ``{icon_url, desc, desc_color}``
            shown at the top of the card. WeCom accepts arbitrary HTTPS
            URLs for ``icon_url`` (unlike DingTalk Avatar which requires
            a uploaded media id), so the LangBot logo URL can be passed
            straight through.

    Notes:
        * ``button.key`` is set directly to the Dify ``action_id``. The click
          callback's ``EventKey`` carries this back unchanged (1024-byte limit
          per the spec, far more than we ever need).
        * WeCom caps the button list at 6. Extra actions are appended to
          ``sub_title_text`` so users can still reply with the id as text.
        * Styles map ``primary``→1 (blue), ``danger``→2 (red), default→0
          (gray). First button is auto-promoted to primary when no style.
    """
    actions = list(form_data.get('actions') or [])
    node_title = (form_data.get('node_title') or '').strip() or '人工介入'
    form_content = _wecom_clean_form_content(form_data)
    should_show_actions = not _wecom_pending_input_defs(form_data)

    visible_actions = actions[:6] if should_show_actions else []
    overflow = actions[6:] if should_show_actions else []

    sub_title_parts: list[str] = []
    if form_content:
        sub_title_parts.append(form_content)
    input_hint_lines = _wecom_input_hint_lines(form_data)
    if input_hint_lines:
        sub_title_parts.append('Fill these fields in chat before choosing an action:\n' + '\n'.join(input_hint_lines))
    if overflow:
        extra_lines = [f'  - {a.get("title") or a.get("id") or ""} (回复 id: {a.get("id") or ""})' for a in overflow]
        sub_title_parts.append(f'另有 {len(overflow)} 个选项不在按钮列表中，可直接回复 id：\n' + '\n'.join(extra_lines))
    sub_title_text = '\n\n'.join(sub_title_parts)

    button_list = []
    for idx, action in enumerate(visible_actions):
        action_id = str(action.get('id') or '')
        title = str(action.get('title') or action_id or f'选项 {idx + 1}')
        button_list.append(
            {
                'text': title,
                'style': _wecom_button_style(action),
                'key': action_id,
            }
        )

    card: dict[str, Any] = {
        'card_type': 'button_interaction',
        'main_title': {
            'title': node_title,
        },
        'sub_title_text': sub_title_text,
        'button_list': button_list,
        'task_id': task_id,
    }
    if source:
        card['source'] = source
    return {
        'msgtype': 'template_card',
        'template_card': card,
    }


def extract_template_card_action(tce: dict[str, Any]) -> tuple[str, str, str]:
    """Extract task id, clicked button key, and card type from a WeCom callback."""

    task_id = tce.get('TaskId') or tce.get('task_id') or tce.get('taskid') or tce.get('taskId') or ''
    event_key = (
        tce.get('EventKey')
        or tce.get('event_key')
        or tce.get('eventkey')
        or tce.get('eventKey')
        or tce.get('key')
        or tce.get('Key')
        or ''
    )
    card_type = tce.get('CardType') or tce.get('card_type') or tce.get('cardtype') or tce.get('cardType') or ''

    for button_key in ('button', 'Button', 'selected_button', 'selectedButton'):
        button = tce.get(button_key)
        if isinstance(button, dict):
            if not event_key:
                event_key = (
                    button.get('key')
                    or button.get('Key')
                    or button.get('event_key')
                    or button.get('EventKey')
                    or button.get('id')
                    or button.get('Id')
                    or ''
                )
            break

    return str(task_id or ''), str(event_key or ''), str(card_type or '')


def extract_wecom_event_type(payload: dict[str, Any]) -> str:
    """Extract eventtype from common WeCom callback wrapper shapes."""

    event = payload.get('event') if isinstance(payload, dict) else {}
    if not isinstance(event, dict):
        event = {}
    event_type = (
        event.get('eventtype')
        or event.get('event_type')
        or event.get('eventType')
        or event.get('EventType')
        or payload.get('eventtype')
        or payload.get('event_type')
        or payload.get('eventType')
        or payload.get('EventType')
        or ''
    )
    if event_type:
        return str(event_type)

    tce = extract_template_card_event_payload(payload)
    task_id, event_key, card_type = extract_template_card_action(tce)
    if task_id or event_key or card_type or extract_template_card_selections(tce):
        return 'template_card_event'
    return ''


def extract_template_card_event_payload(payload: dict[str, Any]) -> dict[str, Any]:
    """Extract template_card_event from common WeCom callback wrapper shapes."""

    if not isinstance(payload, dict):
        return {}
    event = payload.get('event') if isinstance(payload.get('event'), dict) else {}
    candidates = (
        event.get('template_card_event'),
        event.get('templateCardEvent'),
        event.get('TemplateCardEvent'),
        event.get('template_card'),
        payload.get('template_card_event'),
        payload.get('templateCardEvent'),
        payload.get('TemplateCardEvent'),
        payload.get('template_card'),
    )
    for candidate in candidates:
        if isinstance(candidate, dict):
            return candidate
    if any(
        key in payload
        for key in (
            'TaskId',
            'task_id',
            'taskId',
            'EventKey',
            'event_key',
            'eventKey',
            'CardType',
            'card_type',
            'cardType',
            'ResponseData',
            'response_data',
            'select_list',
            'SelectList',
        )
    ):
        return payload
    if any(
        key in event
        for key in (
            'TaskId',
            'task_id',
            'taskId',
            'EventKey',
            'event_key',
            'eventKey',
            'CardType',
            'card_type',
            'cardType',
            'ResponseData',
            'response_data',
            'select_list',
            'SelectList',
        )
    ):
        return event
    return {}


def extract_template_card_selections(tce: dict[str, Any], form_data: Optional[dict] = None) -> dict[str, str]:
    """Extract multiple_interaction select values from a WeCom callback.

    WeCom callback examples differ between webhook and websocket docs, so this
    parser accepts common snake_case/camelCase/PascalCase variants and maps the
    selected option id back to the Dify select option text when form_data is
    available.
    """

    fields_by_name: dict[str, dict] = {}
    if form_data:
        for field in form_data.get('input_defs') or form_data.get('all_input_defs') or []:
            field_name = str(field.get('output_variable_name') or '').strip()
            if field_name:
                fields_by_name[field_name] = field

    def _maybe_decode_json(value: Any) -> Any:
        if not isinstance(value, str):
            return value
        text = value.strip()
        if not text or text[0] not in '[{':
            return value
        try:
            return json.loads(text)
        except json.JSONDecodeError:
            return value

    def _walk(value: Any) -> list[dict]:
        value = _maybe_decode_json(value)
        found: list[dict] = []
        if isinstance(value, dict):
            found.append(value)
            for child in value.values():
                found.extend(_walk(child))
        elif isinstance(value, list):
            for child in value:
                found.extend(_walk(child))
        return found

    def _lookup(item: dict, *keys: str) -> Any:
        for key in keys:
            if key in item:
                return item.get(key)
        lower_map = {str(key).lower(): value for key, value in item.items()}
        for key in keys:
            lowered = key.lower()
            if lowered in lower_map:
                return lower_map[lowered]
        return ''

    def _normalise_selected_value(question_key: str, selected: Any) -> str:
        selected = _maybe_decode_json(selected)
        if isinstance(selected, dict):
            selected = _lookup(
                selected,
                'selected_id',
                'SelectedId',
                'selected_option_id',
                'SelectedOptionId',
                'option_id',
                'OptionId',
                'id',
                'Id',
                'value',
                'Value',
            )
        selected_id = str(selected or '').strip()
        if not selected_id:
            return ''
        selected_value = selected_id
        field = fields_by_name.get(question_key)
        if field:
            options = _wecom_select_options(field)
            for idx, option_text in enumerate(options):
                if selected_id in {_wecom_select_option_id(idx), option_text}:
                    selected_value = option_text
                    break
        return selected_value

    selections: dict[str, str] = {}
    for item in _walk(tce):
        question_key = _lookup(
            item,
            'question_key',
            'questionKey',
            'QuestionKey',
            'question',
            'Question',
            'key',
            'Key',
        )
        selected_id = _lookup(
            item,
            'selected_id',
            'selectedId',
            'SelectedId',
            'selected_option_id',
            'selectedOptionId',
            'SelectedOptionId',
            'option_id',
            'optionId',
            'OptionId',
            'value',
            'Value',
        )
        question_key = str(question_key or '').strip()
        selected_id = str(selected_id or '').strip()
        if question_key not in fields_by_name or not selected_id:
            continue

        selected_value = _normalise_selected_value(question_key, selected_id)
        if not selected_value:
            continue
        selections[question_key] = selected_value

    # Some WeCom callbacks encode ResponseData as a direct mapping:
    # {"xiala": "id_two"} rather than a select_list item array.
    for item in _walk(tce):
        if not isinstance(item, dict):
            continue
        for question_key, selected in item.items():
            question_key = str(question_key or '').strip()
            if question_key not in fields_by_name or question_key in selections:
                continue
            selected_value = _normalise_selected_value(question_key, selected)
            if selected_value:
                selections[question_key] = selected_value

    return selections


def resolve_form_action_title(form_data: dict, action_id: str) -> str:
    """Resolve a Dify form action title from its id."""

    clean_action_id = str(action_id or '').strip()
    for action in form_data.get('actions') or []:
        if str(action.get('id', '')) == clean_action_id:
            return str(action.get('title') or clean_action_id)
    return clean_action_id


def build_button_interaction_update_card(
    form_data: dict,
    task_id: str,
    action_id: str,
    source: Optional[dict[str, Any]] = None,
) -> dict[str, Any]:
    """Build the template_card body used to update a clicked form card."""

    node_title = str(form_data.get('node_title') or '').strip() or '人工介入'
    form_content = _wecom_clean_form_content(form_data)
    action_title = resolve_form_action_title(form_data, action_id)
    clean_action_id = str(action_id or '').strip()

    button_list = []
    matched = False
    for idx, action in enumerate(list(form_data.get('actions') or [])[:6]):
        action_key = str(action.get('id') or '')
        button_title = str(action.get('title') or action_key or f'Option {idx + 1}')
        button = {
            'text': button_title,
            'style': _wecom_button_style(action),
            'key': action_key,
        }
        if action_key == clean_action_id:
            button['style'] = _wecom_button_style(action, selected=True)
            button['text'] = f'✅ {button_title}'
            button['replace_text'] = f'✅ {button_title}'
            matched = True
        button_list.append(button)

    if clean_action_id and not matched:
        button_list.append(
            {
                'text': action_title or clean_action_id,
                'style': 1,
                'key': clean_action_id,
                'replace_text': f'✅ {action_title or clean_action_id}',
            }
        )

    card: dict[str, Any] = {
        'card_type': 'button_interaction',
        'main_title': {
            'title': node_title,
        },
        'sub_title_text': form_content,
        'button_list': button_list,
        'task_id': task_id,
    }
    if source:
        card['source'] = source
    return card


def build_multiple_interaction_update_card(
    form_data: dict,
    task_id: str,
    selections: Optional[dict[str, str]] = None,
    source: Optional[dict[str, Any]] = None,
) -> dict[str, Any]:
    """Build an update card that freezes submitted select values."""

    node_title = str(form_data.get('node_title') or '').strip() or 'Human Input'
    selected_values = dict(form_data.get('inputs') or {})
    selected_values.update(selections or {})

    select_list = []
    fields = _wecom_pending_select_defs(form_data)
    if not fields:
        fields = [
            field
            for field in (form_data.get('input_defs') or form_data.get('all_input_defs') or [])
            if str(field.get('type') or '').strip().lower() == 'select'
        ]
    for field_index, field in enumerate(fields[:10]):
        field_name = str(field.get('output_variable_name') or '').strip()
        if not field_name:
            continue
        options = _wecom_select_options(field)[:10]
        option_list = [
            {
                'id': _wecom_select_option_id(idx),
                'text': option_text[:10] or _wecom_select_option_id(idx),
            }
            for idx, option_text in enumerate(options)
        ]
        selected_id = _wecom_select_option_id(0)
        current_value = selected_values.get(field_name)
        if current_value not in (None, '', []):
            for idx, option_text in enumerate(options):
                if str(current_value) == option_text or str(current_value) == _wecom_select_option_id(idx):
                    selected_id = _wecom_select_option_id(idx)
                    break
        select_list.append(
            {
                'question_key': field_name,
                'title': _wecom_field_title(field, f'Select {field_index + 1}'),
                'disable': True,
                'selected_id': selected_id,
                'option_list': option_list,
            }
        )

    display_form_data = dict(form_data)
    display_form_data['inputs'] = selected_values
    form_content = _wecom_clean_form_content(display_form_data)

    card: dict[str, Any] = {
        'card_type': 'multiple_interaction',
        'main_title': {
            'title': node_title,
            'desc': form_content,
        },
        'select_list': select_list,
        'submit_button': {
            'text': '✅',
            'key': 'submit_human_input',
        },
        'task_id': task_id,
    }
    if source:
        card['source'] = source
    return card


class WecomBotClient:
    _MAX_DISPATCH_TASKS = 100

    def __init__(
        self,
        Token: str,
        EnCodingAESKey: str,
        Corpid: str,
        logger: EventLogger,
        unified_mode: bool = False,
    ):
        """企业微信智能机器人客户端。

        Args:
            Token: 企业微信回调验证使用的 token。
            EnCodingAESKey: 企业微信消息加解密密钥。
            Corpid: 企业 ID。
            logger: 日志记录器。
            unified_mode: 是否使用统一 webhook 模式（默认 False）。

        Example:
            >>> client = WecomBotClient(Token='token', EnCodingAESKey='aeskey', Corpid='corp', logger=logger)
        """

        self.Token = Token
        self.EnCodingAESKey = EnCodingAESKey
        self.Corpid = Corpid
        self.ReceiveId = ''
        self.unified_mode = unified_mode
        self.app = Quart(__name__)

        # 只有在非统一模式下才注册独立路由
        if not self.unified_mode:
            self.app.add_url_rule(
                '/callback/command', 'handle_callback', self.handle_callback_request, methods=['POST', 'GET']
            )

        self._message_handlers = {
            'example': [],
        }
        self.logger = logger
        self.generated_content: dict[str, str] = {}
        self.msg_id_map: dict[str, int] = {}
        self.stream_sessions = StreamSessionManager(logger=logger)
        self.stream_poll_timeout = 0.5

        self._feedback_callback: Optional[Callable] = None
        self._card_action_callback: Optional[Callable] = None
        self._stream_last_content: dict[str, str] = {}
        self._dispatch_tasks: set[asyncio.Task] = set()
        # Optional `source` block injected into every interactive template_card
        # the client builds. Set via `set_card_source` from the adapter after
        # reading config. Format: {icon_url, desc, desc_color}.
        self.card_source: Optional[dict] = None

    def set_card_source(self, source: Optional[dict]) -> None:
        """Set the `source` header dict injected into every
        button_interaction template_card. Pass None to clear."""
        self.card_source = source

    def set_feedback_callback(self, callback: Callable) -> None:
        """设置反馈回调函数。

        Args:
            callback: 反馈回调函数，签名: async def callback(feedback_id, feedback_type, feedback_content, inaccurate_reasons, session)
        """
        self._feedback_callback = callback

    def set_card_action_callback(self, callback: Callable) -> None:
        """设置按钮卡片点击回调函数。

        Signature: ``async def callback(session, action_id, task_id, raw_event) -> None``

        ``session`` is the StreamSession the card was attached to;
        ``action_id`` is the Dify action_id reflected back via the
        button's ``key`` field; ``task_id`` is the card's task_id
        (matches ``session.pending_form_task_id``); ``raw_event`` is the
        decoded callback JSON for any extra fields the adapter wants.
        """
        self._card_action_callback = callback

    @staticmethod
    def _build_stream_payload(
        stream_id: str, content: str, finish: bool, feedback_id: Optional[str] = None
    ) -> dict[str, Any]:
        """按照企业微信协议拼装返回报文。

        Args:
            stream_id: 企业微信会话 ID。
            content: 推送的文本内容。
            finish: 是否为最终片段。
            feedback_id: 反馈 ID，用于接收用户点赞/点踩反馈。

        Returns:
            dict[str, Any]: 可直接加密返回的 payload。

        Example:
            组装 `{'msgtype': 'stream', 'stream': {'id': 'sid', ...}}` 结构。
        """
        stream_payload = {
            'id': stream_id,
            'finish': finish,
            'content': content,
        }
        if feedback_id:
            stream_payload['feedback'] = {'id': feedback_id}
        return {
            'msgtype': 'stream',
            'stream': stream_payload,
        }

    def _build_button_interaction_payload(self, form_data: dict, task_id: str) -> dict[str, Any]:
        """Class-level shim — delegates to module-level builder and auto-
        injects the client's configured `source` block so every card emitted
        through this client carries the LangBot header."""
        return build_human_input_template_card_payload(form_data, task_id, source=self.card_source)

    async def _encrypt_and_reply(self, payload: dict[str, Any], nonce: str) -> tuple[Response, int]:
        """对响应进行加密封装并返回给企业微信。

        Args:
            payload: 待加密的响应内容。
            nonce: 企业微信回调参数中的 nonce。

        Returns:
            Tuple[Response, int]: Quart Response 对象及状态码。

        Example:
            在首包或刷新场景中调用以生成加密响应。
        """
        reply_plain_str = json.dumps(payload, ensure_ascii=False)
        reply_timestamp = str(int(time.time()))
        ret, encrypt_text = await asyncio.to_thread(
            self.wxcpt.EncryptMsg,
            reply_plain_str,
            nonce,
            reply_timestamp,
        )
        if ret != 0:
            await self.logger.error(f'加密失败: {ret}')
            return jsonify({'error': 'encrypt_failed'}), 500

        root = ET.fromstring(encrypt_text)
        encrypt = root.find('Encrypt').text
        resp = {
            'encrypt': encrypt,
        }
        return jsonify(resp), 200

    async def _dispatch_event(self, event: wecombotevent.WecomBotEvent) -> None:
        """异步触发流水线处理，避免阻塞首包响应。

        Args:
            event: 由企业微信消息转换的内部事件对象。
        """
        try:
            await self._handle_message(event)
        except Exception:
            await self.logger.error(traceback.format_exc())

    def _start_dispatch_task(self, event: wecombotevent.WecomBotEvent) -> bool:
        """Start one bounded pipeline dispatch task."""

        for task in tuple(self._dispatch_tasks):
            if task.done():
                self._dispatch_tasks.discard(task)
        if len(self._dispatch_tasks) >= self._MAX_DISPATCH_TASKS:
            return False

        task = asyncio.create_task(self._dispatch_event(event))
        self._dispatch_tasks.add(task)

        def done(done_task: asyncio.Task) -> None:
            self._dispatch_tasks.discard(done_task)
            if not done_task.cancelled():
                done_task.exception()

        task.add_done_callback(done)
        return True

    async def close(self) -> None:
        """Cancel callbacks and release retained webhook state."""

        dispatch_tasks = list(self._dispatch_tasks)
        for task in dispatch_tasks:
            if not task.done():
                task.cancel()
        if dispatch_tasks:
            await asyncio.gather(*dispatch_tasks, return_exceptions=True)
        self._dispatch_tasks.clear()
        self.generated_content.clear()
        self.msg_id_map.clear()
        self._stream_last_content.clear()
        self.stream_sessions.clear()

    async def _handle_post_initial_response(self, msg_json: dict[str, Any], nonce: str) -> tuple[Response, int]:
        """处理企业微信首次推送的消息，返回 stream_id 并开启流水线。

        Args:
            msg_json: 解密后的企业微信消息 JSON。
            nonce: 企业微信回调参数 nonce。

        Returns:
            Tuple[Response, int]: Quart Response 及状态码。

        Example:
            首次回调时调用，立即返回带 `stream_id` 的响应。
        """
        session, is_new = self.stream_sessions.create_or_get(msg_json)

        feedback_id = str(uuid.uuid4())
        session.feedback_id = feedback_id
        self.stream_sessions.register_feedback_id(session.stream_id, feedback_id)

        message_data = await self.get_message(msg_json)
        if message_data:
            message_data['stream_id'] = session.stream_id
            message_data['feedback_id'] = feedback_id
            try:
                event = wecombotevent.WecomBotEvent(message_data)
            except Exception:
                await self.logger.error(traceback.format_exc())
            else:
                if is_new:
                    if not self._start_dispatch_task(event):
                        await self.logger.warning('WeCom webhook dispatch capacity reached; dropping message')

        payload = self._build_stream_payload(session.stream_id, '', False, feedback_id)
        return await self._encrypt_and_reply(payload, nonce)

    async def _handle_post_followup_response(self, msg_json: dict[str, Any], nonce: str) -> tuple[Response, int]:
        """处理企业微信的流式刷新请求，按需返回增量片段。

        Args:
            msg_json: 解密后的企业微信刷新请求。
            nonce: 企业微信回调参数 nonce。

        Returns:
            Tuple[Response, int]: Quart Response 及状态码。

        Example:
            在刷新请求中调用，按需返回增量片段。
        """
        stream_info = msg_json.get('stream', {})
        stream_id = stream_info.get('id', '')
        if not stream_id:
            await self.logger.error('刷新请求缺少 stream.id')
            return await self._encrypt_and_reply(self._build_stream_payload('', '', True), nonce)

        session = self.stream_sessions.get_session(stream_id)

        # If a Dify human-input pause arrived during this stream, switch
        # the response from `msgtype: stream` to `msgtype: template_card`
        # (button_interaction). The session's stream is also marked
        # finished so future followups aren't expected (assuming the
        # WeCom client treats template_card as the terminal response —
        # we'll know from the next callback whether it kept polling).
        if session and session.pending_form and session.pending_form_task_id:
            await self.logger.info(
                f'WeComBot: returning button_interaction for stream_id={stream_id} '
                f'task_id={session.pending_form_task_id} actions={len(session.pending_form.get("actions") or [])}'
            )
            card_payload = self._build_button_interaction_payload(session.pending_form, session.pending_form_task_id)
            self.stream_sessions.mark_finished(stream_id)
            return await self._encrypt_and_reply(card_payload, nonce)

        chunk = await self.stream_sessions.consume(stream_id, timeout=self.stream_poll_timeout)

        if not chunk:
            cached_content = None
            if session and session.msg_id:
                cached_content = self.generated_content.pop(session.msg_id, None)
            if cached_content is not None:
                chunk = StreamChunk(content=cached_content, is_final=True)
            else:
                payload = self._build_stream_payload(stream_id, '', False)
                return await self._encrypt_and_reply(payload, nonce)

        payload = self._build_stream_payload(stream_id, chunk.content, chunk.is_final)
        if chunk.is_final:
            self.stream_sessions.mark_finished(stream_id)
        return await self._encrypt_and_reply(payload, nonce)

    async def handle_callback_request(self):
        """企业微信回调入口（独立端口模式，使用全局 request）。

        Returns:
            Quart Response: 根据请求类型返回验证、首包或刷新结果。

        Example:
            作为 Quart 路由处理函数直接注册并使用。
        """
        return await self._handle_callback_internal(request)

    async def handle_unified_webhook(self, req):
        """处理回调请求（统一 webhook 模式，显式传递 request）。

        Args:
            req: Quart Request 对象

        Returns:
            响应数据
        """
        return await self._handle_callback_internal(req)

    async def _handle_callback_internal(self, req):
        """处理回调请求的内部实现，包括 GET 验证和 POST 消息接收。

        Args:
            req: Quart Request 对象
        """
        try:
            self.wxcpt = WXBizMsgCrypt(self.Token, self.EnCodingAESKey, '')

            if req.method == 'GET':
                return await self._handle_get_callback(req)

            if req.method == 'POST':
                return await self._handle_post_callback(req)

            return Response('', status=405)

        except Exception:
            await self.logger.error(traceback.format_exc())
            return Response('Internal Server Error', status=500)

    async def _handle_get_callback(self, req) -> tuple[Response, int] | Response:
        """处理企业微信的 GET 验证请求。"""

        msg_signature = unquote(req.args.get('msg_signature', ''))
        timestamp = unquote(req.args.get('timestamp', ''))
        nonce = unquote(req.args.get('nonce', ''))
        echostr = unquote(req.args.get('echostr', ''))

        if not all([msg_signature, timestamp, nonce, echostr]):
            await self.logger.error('请求参数缺失')
            return Response('缺少参数', status=400)

        ret, decrypted_str = self.wxcpt.VerifyURL(msg_signature, timestamp, nonce, echostr)
        if ret != 0:
            await self.logger.error('验证URL失败')
            return Response('验证失败', status=403)

        return Response(decrypted_str, mimetype='text/plain')

    async def _handle_post_callback(self, req) -> tuple[Response, int] | Response:
        """处理企业微信的 POST 回调请求。"""

        for expired_msg_id in self.stream_sessions.cleanup():
            self.generated_content.pop(expired_msg_id, None)
            self._stream_last_content.pop(expired_msg_id, None)
            self.msg_id_map.pop(expired_msg_id, None)

        msg_signature = unquote(req.args.get('msg_signature', ''))
        timestamp = unquote(req.args.get('timestamp', ''))
        nonce = unquote(req.args.get('nonce', ''))

        encrypted_json = await req.get_json()
        encrypted_msg = (encrypted_json or {}).get('encrypt', '')
        if not encrypted_msg:
            await self.logger.error("请求体中缺少 'encrypt' 字段")
            return Response('Bad Request', status=400)

        xml_post_data = f'<xml><Encrypt><![CDATA[{encrypted_msg}]]></Encrypt></xml>'
        ret, decrypted_xml = await asyncio.to_thread(
            self.wxcpt.DecryptMsg,
            xml_post_data,
            msg_signature,
            timestamp,
            nonce,
        )
        if ret != 0:
            await self.logger.error('解密失败')
            return Response('解密失败', status=400)

        msg_json = await asyncio.to_thread(json.loads, decrypted_xml)

        event_type = extract_wecom_event_type(msg_json)

        if event_type == 'feedback_event':
            return await self._handle_feedback_event(msg_json, nonce)

        # Button click on a button_interaction template_card. The WeCom doc
        # calls this `template_card_event`; some routes wrap the button
        # event payload inside `event.template_card_event`.
        if event_type == 'template_card_event':
            return await self._handle_template_card_event(msg_json, nonce)

        if msg_json.get('msgtype') == 'stream':
            return await self._handle_post_followup_response(msg_json, nonce)

        return await self._handle_post_initial_response(msg_json, nonce)

    async def _handle_template_card_event(self, msg_json: dict[str, Any], nonce: str) -> tuple[Response, int]:
        """Handle a button click on a button_interaction template_card.

        WeCom carries the click info in ``event.template_card_event`` with
        ``TaskId`` matching the card we created and ``EventKey`` carrying
        the button's ``key`` (which we set to the Dify ``action_id``).
        """
        try:
            tce = extract_template_card_event_payload(msg_json)
            task_id, event_key, card_type = extract_template_card_action(tce)

            await self.logger.info(f'收到按钮点击: task_id={task_id} event_key={event_key!r} card_type={card_type}')

            session = self.stream_sessions.get_session_by_task_id(task_id)
            if session is None:
                await self.logger.warning(f'未找到 task_id={task_id} 对应的 session，按钮点击被丢弃')
            else:
                if self._card_action_callback is not None:
                    try:
                        await self._card_action_callback(session, event_key, task_id, msg_json)
                    except Exception:
                        await self.logger.error(f'card action callback raised: {traceback.format_exc()}')
                # Drop the form so a fresh chunk/followup doesn't re-render
                # the same card (and so the task_id can be GC'd).
                self.stream_sessions.clear_pending_form(session.stream_id)
        except Exception:
            await self.logger.error(f'_handle_template_card_event error: {traceback.format_exc()}')

        # WeCom expects an empty success ack for event callbacks.
        return await self._encrypt_and_reply({}, nonce)

    async def _handle_feedback_event(self, msg_json: dict[str, Any], nonce: str) -> tuple[Response, int]:
        """处理企业微信用户反馈事件（点赞/点踩）。

        Args:
            msg_json: 解密后的企业微信反馈事件 JSON。
            nonce: 企业微信回调参数 nonce。

        Returns:
            Tuple[Response, int]: Quart Response 及状态码。

        Note:
            企业微信协议要求：反馈事件目前仅支持回复空包。
        """
        try:
            feedback_event = msg_json.get('event', {}).get('feedback_event', {})
            feedback_id = feedback_event.get('id', '')
            feedback_type = feedback_event.get('type', 0)
            feedback_content = feedback_event.get('content', '')
            inaccurate_reasons = feedback_event.get('inaccurate_reason_list', [])

            await self.logger.info(
                f'收到用户反馈事件: feedback_id={feedback_id}, type={feedback_type}, '
                f'content={feedback_content}, reasons={inaccurate_reasons}'
            )

            session = self.stream_sessions.get_session_by_feedback_id(feedback_id)

            if session:
                await self.logger.info(
                    f'反馈关联到会话: stream_id={session.stream_id}, msg_id={session.msg_id}, user_id={session.user_id}'
                )
            else:
                await self.logger.warning(f'未找到 feedback_id={feedback_id} 对应的会话，仍将记录反馈')

            # Dispatch feedback event regardless of session availability
            for handler in self._message_handlers.get('feedback', []):
                try:
                    await handler(
                        feedback_id=feedback_id,
                        feedback_type=feedback_type,
                        feedback_content=feedback_content,
                        inaccurate_reasons=inaccurate_reasons,
                        session=session,
                    )
                except Exception:
                    await self.logger.error(traceback.format_exc())

            if self._feedback_callback:
                try:
                    await self._feedback_callback(
                        feedback_id=feedback_id,
                        feedback_type=feedback_type,
                        feedback_content=feedback_content,
                        inaccurate_reasons=inaccurate_reasons,
                        session=session,
                    )
                except Exception:
                    await self.logger.error(traceback.format_exc())

        except Exception:
            await self.logger.error(traceback.format_exc())

        return await self._encrypt_and_reply({}, nonce)

    async def get_message(self, msg_json):
        return await parse_wecom_bot_message(msg_json, self.EnCodingAESKey, self.logger)

    async def _handle_message(self, event: wecombotevent.WecomBotEvent):
        """
        处理消息事件。
        """
        try:
            message_id = event.message_id
            if message_id in self.msg_id_map.keys():
                self.msg_id_map[message_id] += 1
                return
            self.msg_id_map[message_id] = 1
            while len(self.msg_id_map) > _CLIENT_TRANSIENT_CACHE_MAX:
                self.msg_id_map.pop(next(iter(self.msg_id_map)), None)
            msg_type = event.type
            if msg_type in self._message_handlers:
                for handler in self._message_handlers[msg_type]:
                    await handler(event)
        except Exception:
            print(traceback.format_exc())

    async def push_stream_chunk(self, msg_id: str, content: str, is_final: bool = False) -> bool:
        """将流水线片段推送到 stream 会话。

        Args:
            msg_id: 原始企业微信消息 ID。
            content: 流水线传入的累计正文快照。
            is_final: 是否为最终片段。

        Returns:
            bool: 当成功写入流式队列时返回 True。

        Example:
            在流水线 `reply_message_chunk` 中调用，将增量推送至企业微信。
        """
        # 根据 msg_id 找到对应 stream 会话，如果不存在说明当前消息非流式
        stream_id = self.stream_sessions.get_stream_id_by_msg(msg_id)
        if not stream_id:
            return False

        previous_content = self._stream_last_content.get(msg_id, '')
        # Pipeline delivery provides a snapshot, not a provider delta.
        next_content = content or previous_content
        if len(next_content) > _MAX_STREAM_CONTENT_CHARS:
            next_content = next_content[-_MAX_STREAM_CONTENT_CHARS:]

        if not is_final and next_content == previous_content:
            return True

        # Follow-up responses replace the displayed stream body in WeCom.
        # Publish the complete snapshot so earlier chunks remain visible.
        chunk = StreamChunk(content=next_content, is_final=is_final)
        await self.stream_sessions.publish(stream_id, chunk)
        self._stream_last_content[msg_id] = next_content
        if is_final:
            self._stream_last_content.pop(msg_id, None)
            self.stream_sessions.mark_finished(stream_id)
        return True

    async def push_form_pause(
        self, msg_id: str, form_data: dict, task_id: Optional[str] = None
    ) -> tuple[bool, Optional[str], Optional[str]]:
        """Attach a Dify human-input pause to the active stream session.

        On the next WeCom followup poll, the response switches from
        ``msgtype: stream`` to ``msgtype: template_card`` (button_interaction)
        carrying the buttons. ``task_id`` is auto-generated if not provided
        and is what the button-click callback uses to look the session back up.

        Returns:
            ``(ok, stream_id, task_id)``. ``ok`` is False if the
            adapter's msg_id maps to no stream session (e.g. non-stream mode).
        """
        stream_id = self.stream_sessions.get_stream_id_by_msg(msg_id)
        if not stream_id:
            return False, None, None
        if not task_id:
            # WeCom requires task_id [A-Za-z0-9_-@], <= 128 bytes, unique per bot.
            task_id = f'dify-{uuid.uuid4().hex[:24]}'
        self.stream_sessions.set_pending_form(stream_id, form_data, task_id)
        return True, stream_id, task_id

    async def set_message(self, msg_id: str, content: str):
        """兼容旧逻辑：若无法流式返回则缓存最终结果。

        Args:
            msg_id: 企业微信消息 ID。
            content: 最终回复的文本内容。

        Example:
            在非流式场景下缓存最终结果以备刷新时返回。
        """
        handled = await self.push_stream_chunk(msg_id, content, is_final=True)
        if not handled:
            self.generated_content[msg_id] = content[-_MAX_STREAM_CONTENT_CHARS:]
            while len(self.generated_content) > _CLIENT_TRANSIENT_CACHE_MAX:
                self.generated_content.pop(next(iter(self.generated_content)), None)

    def on_message(self, msg_type: str):
        def decorator(func: Callable[[wecombotevent.WecomBotEvent], None]):
            if msg_type not in self._message_handlers:
                self._message_handlers[msg_type] = []
            self._message_handlers[msg_type].append(func)
            return func

        return decorator

    def on_feedback(self):
        def decorator(func: Callable):
            if 'feedback' not in self._message_handlers:
                self._message_handlers['feedback'] = []
            self._message_handlers['feedback'].append(func)
            return func

        return decorator

    async def download_url_to_base64(self, download_url, encoding_aes_key):
        data, _filename = await download_encrypted_file(download_url, encoding_aes_key, self.logger)
        if data:
            return await asyncio.to_thread(_bytes_to_data_uri, data)
        return None

    async def run_task(self, host: str, port: int, *args, **kwargs):
        """
        启动 Quart 应用。
        """
        await self.app.run_task(host=host, port=port, *args, **kwargs)