feat(wecom): add functions for template card action extraction and update, enhance button interaction handling

src/langbot/libs/wecom_ai_bot_api/api.py

@@ -770,6 +770,15 @@ async def parse_wecom_bot_message(
     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 build_button_interaction_payload(
     form_data: dict,
     task_id: str,
@@ -823,17 +832,10 @@ def build_button_interaction_payload(
     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}')
-        style_raw = (action.get('button_style') or '').lower()
-        if style_raw == 'primary' or (style_raw == '' and idx == 0):
-            style = 1
-        elif style_raw == 'danger':
-            style = 2
-        else:
-            style = 0
         button_list.append(
             {
                 'text': title,
-                'style': style,
+                'style': _wecom_button_style(action),
                 'key': action_id,
             }
         )
@@ -855,8 +857,112 @@ def build_button_interaction_payload(
     }
 
 
+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 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 = str(form_data.get('form_content') or '').strip()
+    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
+
+
 class WecomBotClient:
-    def __init__(self, Token: str, EnCodingAESKey: str, Corpid: str, logger: EventLogger, unified_mode: bool = False):
+    def __init__(
+        self,
+        Token: str,
+        EnCodingAESKey: str,
+        Corpid: str,
+        logger: EventLogger,
+        unified_mode: bool = False,
+    ):
         """企业微信智能机器人客户端。
 
         Args:
@@ -1197,9 +1303,7 @@ class WecomBotClient:
         """
         try:
             tce = msg_json.get('event', {}).get('template_card_event', {})
-            task_id = tce.get('TaskId') or tce.get('task_id') or ''
-            event_key = tce.get('EventKey') or tce.get('event_key') or ''
-            card_type = tce.get('CardType') or tce.get('card_type') or ''
+            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}')
 

src/langbot/libs/wecom_ai_bot_api/ws_client.py

@@ -24,6 +24,8 @@ from langbot.libs.wecom_ai_bot_api.api import (
     parse_wecom_bot_message,
     StreamSession,
     build_button_interaction_payload,
+    build_button_interaction_update_card,
+    extract_template_card_action,
 )
 from langbot.pkg.platform.logger import EventLogger
 
@@ -269,6 +271,30 @@ class WecomBotWsClient:
         """
         return await self._send_reply(req_id, card_payload)
 
+    async def update_template_card(
+        self,
+        req_id: str,
+        template_card: dict[str, Any],
+    ) -> Optional[dict]:
+        """Update an existing template_card via WebSocket.
+
+        Uses the ``aibot_respond_update_msg`` command.  Must be called
+        within 5 seconds of receiving the ``template_card_event`` callback,
+        using the **same req_id** from that callback.
+
+        The ``template_card`` dict should contain ``card_type`` and the
+        new content fields (e.g. ``main_title``, ``button_list`` with
+        disabled buttons and ``replace_text``).
+
+        Returns:
+            ACK frame dict, or None on failure.
+        """
+        body: dict[str, Any] = {
+            'response_type': 'update_template_card',
+            'template_card': template_card,
+        }
+        return await self._send_reply(req_id, body, cmd=CMD_RESPOND_UPDATE)
+
     def set_card_action_callback(self, callback: Callable) -> None:
         """Register the button-click handler.
 
@@ -702,28 +728,57 @@ class WecomBotWsClient:
 
             if event_type == 'template_card_event':
                 tce = event_info.get('template_card_event', {})
-                task_id = tce.get('TaskId') or tce.get('task_id') or ''
-                event_key = tce.get('EventKey') or tce.get('event_key') or ''
-                card_type = tce.get('CardType') or tce.get('card_type') or ''
+                task_id, event_key, card_type = extract_template_card_action(tce)
                 await self.logger.info(
                     f'收到按钮点击 (ws): task_id={task_id} event_key={event_key!r} card_type={card_type}'
                 )
                 pending = self._pending_forms_by_task.get(task_id)
                 if pending is None:
                     await self.logger.warning(f'未找到 task_id={task_id} 对应的 pending_form (ws)，按钮点击被丢弃')
-                elif self._card_action_callback is not None:
+                else:
+                    # Update the card in-place to show which button was clicked.
+                    # Must happen within 5s of the event, using the same req_id.
+                    req_id_for_update = frame.get('headers', {}).get('req_id', '')
+                    form_data = pending.get('form_data', {}) or {}
+                    action_title = event_key
+                    node_title = form_data.get('node_title', '') or ''
+                    main_title_text = f'{node_title} - 已选择' if node_title else '已选择'
+                    main_title_desc = f'✅ {action_title}'
+                    update_card = {
+                        'card_type': 'button_interaction',
+                        'main_title': {
+                            'title': main_title_text,
+                            'desc': main_title_desc,
+                        },
+                        'button_list': [],
+                        'task_id': task_id,
+                    }
+                    if self.card_source:
+                        update_card['source'] = self.card_source
+                    update_card = build_button_interaction_update_card(
+                        form_data,
+                        task_id,
+                        event_key,
+                        source=self.card_source,
+                    )
                     try:
-                        session = StreamSession(
-                            stream_id=pending.get('stream_id', ''),
-                            msg_id=pending.get('msg_id', ''),
-                            chat_id=pending.get('chat_id') or None,
-                            user_id=pending.get('user_id') or None,
-                        )
-                        session.pending_form = pending.get('form_data')
-                        session.pending_form_task_id = task_id
-                        await self._card_action_callback(session, event_key, task_id, body)
+                        await self.update_template_card(req_id_for_update, update_card)
                     except Exception:
-                        await self.logger.error(f'card action callback raised (ws): {traceback.format_exc()}')
+                        await self.logger.warning(f'更新卡片失败 (ws): {traceback.format_exc()}')
+
+                    if self._card_action_callback is not None:
+                        try:
+                            session = StreamSession(
+                                stream_id=pending.get('stream_id', ''),
+                                msg_id=pending.get('msg_id', ''),
+                                chat_id=pending.get('chat_id') or None,
+                                user_id=pending.get('user_id') or None,
+                            )
+                            session.pending_form = pending.get('form_data')
+                            session.pending_form_task_id = task_id
+                            await self._card_action_callback(session, event_key, task_id, body)
+                        except Exception:
+                            await self.logger.error(f'card action callback raised (ws): {traceback.format_exc()}')
                     # Consume — drop bookkeeping so a stale click can't re-fire.
                     self._pending_forms_by_task.pop(task_id, None)
                     msg_id = pending.get('msg_id', '')

tests/unit_tests/platform/test_wecombot_template_card.py

@@ -0,0 +1,72 @@
+import sys
+import types
+
+
+logger_module = types.ModuleType('langbot.pkg.platform.logger')
+logger_module.EventLogger = object
+sys.modules.setdefault('langbot.pkg.platform.logger', logger_module)
+
+from langbot.libs.wecom_ai_bot_api.api import (  # noqa: E402
+    build_button_interaction_payload,
+    build_button_interaction_update_card,
+    extract_template_card_action,
+)
+
+
+def test_extract_template_card_action_supports_nested_button_key():
+    task_id, event_key, card_type = extract_template_card_action(
+        {
+            'taskId': 'task-1',
+            'cardType': 'button_interaction',
+            'button': {'key': 'approve'},
+        }
+    )
+
+    assert task_id == 'task-1'
+    assert event_key == 'approve'
+    assert card_type == 'button_interaction'
+
+
+def test_build_button_interaction_update_card_marks_clicked_button():
+    card = build_button_interaction_update_card(
+        {
+            'node_title': 'Manual Review',
+            'form_content': 'Please choose one action.',
+            'actions': [
+                {'id': 'approve', 'title': 'Approve', 'button_style': 'primary'},
+                {'id': 'reject', 'title': 'Reject', 'button_style': 'danger'},
+            ],
+        },
+        task_id='task-1',
+        action_id='reject',
+        source={'desc': 'LangBot'},
+    )
+
+    assert card['main_title'] == {'title': 'Manual Review'}
+    assert card['sub_title_text'] == 'Please choose one action.'
+    assert card['button_list'][0] == {'text': 'Approve', 'style': 2, 'key': 'approve'}
+    assert card['button_list'][1] == {
+        'text': '已选择：Reject',
+        'style': 1,
+        'key': 'reject',
+        'replace_text': '已选择：Reject',
+    }
+    assert card['source'] == {'desc': 'LangBot'}
+
+
+def test_build_button_interaction_payload_uses_preselected_button_styles_before_click():
+    payload = build_button_interaction_payload(
+        {
+            'node_title': 'Manual Review',
+            'actions': [
+                {'id': 'approve', 'title': 'Approve', 'button_style': 'primary'},
+                {'id': 'reject', 'title': 'Reject', 'button_style': 'danger'},
+            ],
+        },
+        task_id='task-1',
+    )
+
+    assert payload['template_card']['button_list'] == [
+        {'text': 'Approve', 'style': 2, 'key': 'approve'},
+        {'text': 'Reject', 'style': 2, 'key': 'reject'},
+    ]