feat(dingtalk): support group subscriptions and card interaction events

src/langbot/pkg/platform/adapters/dingtalk/adapter.py

@@ -7,6 +7,8 @@ import typing
 import dingtalk_stream
 import pydantic
 
+from .subscriptions import DingTalkSubscriptionHandler
+
 from langbot.libs.dingtalk_api.api import DingTalkClient, is_stream_card_configured
 from langbot.libs.dingtalk_api.dingtalkevent import DingTalkEvent
 import langbot_plugin.api.definition.abstract.platform.adapter as abstract_platform_adapter
@@ -40,6 +42,7 @@ class DingTalkCardCallbackHandler(dingtalk_stream.CallbackHandler):
             {
                 'conversation_type': 'CardCallback',
                 'Type': 'card_callback',
+                'timestamp': float(message.headers.time or 0) / 1000,
                 'CardCallback': {
                     'extension': callback.extension,
                     'corp_id': callback.corp_id,
@@ -107,6 +110,10 @@ class DingTalkAdapter(DingTalkAPIMixin, abstract_platform_adapter.AbstractPlatfo
         return [
             'message.received',
             'feedback.received',
+            'group.member_joined',
+            'group.member_left',
+            'group.info_updated',
+            'dingtalk.card_action',
             'platform.specific',
         ]
 
@@ -280,6 +287,7 @@ class DingTalkAdapter(DingTalkAPIMixin, abstract_platform_adapter.AbstractPlatfo
         async def on_message(event: DingTalkEvent):
             await self._handle_native_event(event)
 
+        self.bot.client.register_all_event_handler(DingTalkSubscriptionHandler(self))
         self.bot.on_message('FriendMessage')(on_message)
         self.bot.on_message('GroupMessage')(on_message)
         self.bot.client.register_callback_handler(
@@ -298,7 +306,9 @@ class DingTalkAdapter(DingTalkAPIMixin, abstract_platform_adapter.AbstractPlatfo
                 'DingTalk event received: '
                 f'conversation={event.conversation}, message_id={getattr(event.incoming_message, "message_id", None)}'
             )
-            if platform_events.FriendMessage in self.listeners or platform_events.GroupMessage in self.listeners:
+            if event.conversation in {'FriendMessage', 'GroupMessage'} and (
+                platform_events.FriendMessage in self.listeners or platform_events.GroupMessage in self.listeners
+            ):
                 legacy_event = await self.event_converter.target2legacy(event, self.bot_account_id)
                 if legacy_event:
                     callback = self.listeners.get(type(legacy_event))

src/langbot/pkg/platform/adapters/dingtalk/event_converter.py

@@ -23,7 +23,18 @@ class DingTalkEventConverter(abstract_platform_adapter.AbstractEventConverter):
             feedback = DingTalkEventConverter.card_callback_to_feedback(event)
             if feedback is not None:
                 return feedback
-            return DingTalkEventConverter.platform_specific(event, 'card.callback')
+            callback = event.get('CardCallback') or {}
+            content = callback.get('content') or {}
+            return platform_events.DingTalkCardActionEvent(
+                adapter_name=ADAPTER_NAME,
+                user=platform_entities.User(id=str(callback.get('user_id') or '')),
+                space_id=str(callback.get('space_id') or ''),
+                card_instance_id=str(callback.get('card_instance_id') or ''),
+                action=str(content.get('action') or content.get('actionKey') or ''),
+                data=callback,
+                timestamp=float(event.get('timestamp') or 0),
+                source_platform_object=event,
+            )
         return DingTalkEventConverter.platform_specific(event, f'message.{event.conversation or "unknown"}')
 
     @staticmethod

src/langbot/pkg/platform/adapters/dingtalk/manifest.yaml

@@ -151,6 +151,10 @@ spec:
   supported_events:
     - message.received
     - feedback.received
+    - group.member_joined
+    - group.member_left
+    - group.info_updated
+    - dingtalk.card_action
     - platform.specific
 
   supported_apis:

src/langbot/pkg/platform/adapters/dingtalk/subscriptions.py

@@ -0,0 +1,77 @@
+"""Translate DingTalk Stream EVENT envelopes, keeping unknown payloads intact."""
+
+import asyncio
+from collections import OrderedDict
+
+import dingtalk_stream
+from langbot_plugin.api.entities.builtin.platform import entities, events
+
+from .types import ADAPTER_NAME
+
+
+def subscription_events(message):
+    data = message.data
+    headers = message.headers
+    action = headers.event_type or data.get('EventType') or data.get('eventType') or ''
+    raw = {'headers': vars(headers).copy(), 'data': data}
+    stamp = headers.event_born_time or data.get('TimeStamp') or data.get('EventTime') or data.get('timestamp') or headers.time or 0
+    stamp = float(stamp)
+    common = dict(
+        adapter_name=ADAPTER_NAME,
+        timestamp=stamp / 1000 if stamp > 10_000_000_000 else stamp,
+        source_platform_object=raw,
+    )
+    group_id = data.get('OpenConversationId') or data.get('openConversationId') or data.get('chatId') or data.get('ChatId')
+    group = entities.UserGroup(id=str(group_id or ''), name=str(data.get('Title') or data.get('title') or ''))
+    owner_id = data.get('Owner') or data.get('owner')
+    if owner_id:
+        group.owner_id = str(owner_id)
+    operator_id = data.get('Operator') or data.get('operator')
+    operator = entities.User(id=str(operator_id)) if operator_id else None
+    members = data.get('UserId') or data.get('userId') or []
+    if isinstance(members, str):
+        members = [members]
+    if group_id and members and action in {'chat_add_member', 'chat_remove_member', 'chat_quit'}:
+        for member_id in dict.fromkeys(members):
+            member = entities.User(id=str(member_id))
+            if action == 'chat_add_member':
+                yield events.MemberJoinedEvent(group=group, member=member, inviter=operator, **common)
+            else:
+                yield events.MemberLeftEvent(
+                    group=group, member=member, operator=operator,
+                    is_kicked=action == 'chat_remove_member', **common,
+                )
+    elif group_id and action in {'chat_update_title', 'chat_update_owner'}:
+        yield events.GroupInfoUpdatedEvent(
+            group=group, operator=operator,
+            changed_fields=['name' if action == 'chat_update_title' else 'owner_id'], **common,
+        )
+    else:
+        yield events.PlatformSpecificEvent(action=action or 'unknown', data=raw, **common)
+
+
+class DingTalkSubscriptionHandler(dingtalk_stream.EventHandler):
+    def __init__(self, adapter):
+        super().__init__()
+        self.adapter = adapter
+        self.completed = OrderedDict()
+        self.lock = asyncio.Lock()
+
+    async def process(self, message):
+        # Serialize retries; mark each fan-out event only after successful dispatch.
+        async with self.lock:
+            try:
+                identity = message.headers.event_id or message.headers.message_id
+                for index, event in enumerate(subscription_events(message)):
+                    key = (message.headers.event_corp_id, identity, index)
+                    if identity and key in self.completed:
+                        continue
+                    await self.adapter._dispatch_eba_event(event)
+                    if identity:
+                        self.completed[key] = None
+                        while len(self.completed) > 4096:
+                            self.completed.popitem(last=False)
+                return dingtalk_stream.AckMessage.STATUS_OK, 'OK'
+            except Exception:
+                self.logger.exception('DingTalk subscription dispatch failed')
+                return 500, 'Event dispatch failed'