Merge pull request #6 from open-dingtalk/features/20230609_linya_addthreadexecutor

hzjiangjian · web-flow · commit 3923e7132cb1 · 2023-06-09T13:07:13.000+08:00
feat: 1、机器人模式下增加异步任务处理模式；2、增加一些实用的卡片模板
diff --git a/dingtalk_stream/__init__.py b/dingtalk_stream/__init__.py
@@ -10,4 +10,4 @@
 from .chatbot import ChatbotMessage
 from .chatbot import TextContent
 from .chatbot import AtUser
-from .chatbot import ChatbotHandler
+from .chatbot import ChatbotHandler, AsyncChatbotHandler
diff --git a/dingtalk_stream/chatbot.py b/dingtalk_stream/chatbot.py
@@ -1,11 +1,13 @@
 # -*- coding:utf-8 -*-
 
 import json
-import uuid
 import requests
 import platform
 import hashlib
-from .stream import CallbackHandler
+from .stream import CallbackHandler, CallbackMessage
+from .frames import AckMessage, Headers
+from concurrent.futures import ThreadPoolExecutor
+import uuid
 
 
 class AtUser(object):
@@ -272,6 +274,7 @@ def reply_card(self,
                            ) % platform.python_version(),
         }
 
+        card_biz_id = self._gen_card_id(incoming_message)
         body = {"cardTemplateId": "StandardCard",
                 "robotCode": self.dingtalk_client.credential.client_id,
                 "cardData": json.dumps(card_data),
@@ -281,7 +284,7 @@ def reply_card(self,
                     # "receiverListJson": "String",
                     # "cardPropertyJson": "String"
                 },
-                "cardBizId": self._gen_card_id(incoming_message)
+                "cardBizId": card_biz_id
                 }
 
         if incoming_message.conversation_type == '2':
@@ -304,11 +307,12 @@ def reply_card(self,
                                      json=body)
             response.raise_for_status()
 
-            return response.json
+            return card_biz_id
         except Exception as e:
-            return {}
+            self.logger.error('reply card failed, error=%s', e)
+            return ""
 
-    def update_card(self, card_data: dict, incoming_message: ChatbotMessage):
+    def update_card(self, card_biz_id: str, card_data: dict):
         """
         回复互动卡片。由于sessionWebhook不支持发送互动卡片，所以需要使用OpenAPI，当前仅支持自建应用。
         https://open.dingtalk.com/document/orgapp/robots-send-interactive-cards
@@ -330,9 +334,8 @@ def update_card(self, card_data: dict, incoming_message: ChatbotMessage):
                            ) % platform.python_version(),
         }
 
-        card_id = self._gen_card_id(incoming_message)
         values = {
-            'cardBizId': card_id,
+            'cardBizId': card_biz_id,
             'cardData': json.dumps(card_data),
         }
         url = 'https://api.dingtalk.com/v1.0/im/robots/interactiveCards'
@@ -348,7 +351,45 @@ def update_card(self, card_data: dict, incoming_message: ChatbotMessage):
 
     @staticmethod
     def _gen_card_id(msg: ChatbotMessage):
-        factor = '%s_%s_%s_%s' % (msg.sender_id, msg.sender_corp_id, msg.conversation_id, msg.message_id)
+        factor = '%s_%s_%s_%s_%s' % (
+            msg.sender_id, msg.sender_corp_id, msg.conversation_id, msg.message_id, str(uuid.uuid1()))
         m = hashlib.sha256()
         m.update(factor.encode('utf-8'))
         return m.hexdigest()
+
+
+class AsyncChatbotHandler(ChatbotHandler):
+    """
+    多任务执行handler，注意：process函数重载的时候不要用 async
+    """
+
+    async_executor: ThreadPoolExecutor = ThreadPoolExecutor(max_workers=8)
+
+    def __init__(self, max_workers: int = 8):
+        super(AsyncChatbotHandler, self).__init__()
+        self.async_executor = ThreadPoolExecutor(max_workers=max_workers)
+
+    def process(self, message):
+        '''
+        不要用 async 修饰
+        :param message:
+        :return:
+        '''
+        return AckMessage.STATUS_NOT_IMPLEMENT, 'not implement'
+
+    async def raw_process(self, callback_message: CallbackMessage):
+        def func():
+            try:
+                self.process(callback_message)
+            except Exception as e:
+                self.logger.error('async process exception, error=%s', e)
+
+        self.async_executor.submit(func)
+
+        ack_message = AckMessage()
+        ack_message.code = AckMessage.STATUS_OK
+        ack_message.headers.message_id = callback_message.headers.message_id
+        ack_message.headers.content_type = Headers.CONTENT_TYPE_APPLICATION_JSON
+        ack_message.message = "ok"
+        ack_message.data = callback_message.data
+        return ack_message
diff --git a/dingtalk_stream/interactive_card.py b/dingtalk_stream/interactive_card.py
@@ -1,5 +1,7 @@
 # -*- coding:utf-8 -*-
 
+import copy, uuid
+
 """
 这里是卡片模板库，提供一些必要的卡片组件组合。
 INTERACTIVE_CARD_JSON_SAMPLE_1 极简卡片组合：title-text-image-button
@@ -8,6 +10,132 @@
 高阶需求请至卡片搭建平台：https://card.dingtalk.com/card-builder
 """
 
+'''
+实用卡片模板：多行文本
+'''
+INTERACTIVE_CARD_JSON_SAMPLE_MULTI_TEXT_LINE = {
+    "config": {
+        "autoLayout": True,
+        "enableForward": True
+    },
+    "header": {
+        "title": {
+            "type": "text",
+            "text": "钉钉卡片"
+        },
+        "logo": "@lALPDfJ6V_FPDmvNAfTNAfQ"
+    },
+    "contents": [
+        {
+            "type": "text",
+            "text": "钉钉正在为各行各业提供专业解决方案，沉淀钉钉1900万企业组织核心业务场景，提供专属钉钉、教育、医疗、新零售等多行业多维度的解决方案。",
+            "id": "text_1686281949314"
+        },
+        {
+            "type": "divider",
+            "id": "divider_1686281949314"
+        }
+    ]
+}
+
+
+def generate_multi_text_line_card_data(title: str, logo: str, texts: [str]) -> dict:
+    card_data = copy.deepcopy(INTERACTIVE_CARD_JSON_SAMPLE_MULTI_TEXT_LINE)
+
+    if title != "":
+        card_data["header"]["title"]["text"] = title
+
+    if logo != "":
+        card_data["header"]["logo"] = logo
+
+    card_data["contents"] = []
+    for text in texts:
+        text_line = {
+            "type": "text",
+            "text": text,
+            "id": "text_" + str(uuid.uuid1())
+        }
+        divider_line = {
+            "type": "divider",
+            "id": "divider_" + str(uuid.uuid1())
+        }
+        card_data["contents"].append(text_line)
+        card_data["contents"].append(divider_line)
+
+    return card_data
+
+
+'''
+实用卡片模板，多行文本+多图组合
+'''
+INTERACTIVE_CARD_JSON_SAMPLE_MULTI_TEXT_IMAGE = {
+    "config": {
+        "autoLayout": True,
+        "enableForward": True
+    },
+    "header": {
+        "title": {
+            "type": "text",
+            "text": "钉钉卡片"
+        },
+        "logo": "@lALPDfJ6V_FPDmvNAfTNAfQ"
+    },
+    "contents": [
+        {
+            "type": "text",
+            "text": "钉钉正在为各行各业提供专业解决方案，沉淀钉钉1900万企业组织核心业务场景，提供专属钉钉、教育、医疗、新零售等多行业多维度的解决方案。",
+            "id": "text_1686281949314"
+        },
+        {
+            "type": "divider",
+            "id": "divider_1686281949314"
+        },
+        {
+            "type": "imageList",
+            "images": [
+                "@lADPDe7s2ySi18PNA6XNBXg",
+                "@lADPDf0i1beuNF3NAxTNBXg",
+                "@lADPDe7s2ySRnIvNA6fNBXg"
+            ],
+            "id": "imageList_1686283179480"
+        }
+    ]
+}
+
+
+def generate_multi_text_image_card_data(title: str, logo: str, texts: [str], images: [str]) -> dict:
+    card_data = copy.deepcopy(INTERACTIVE_CARD_JSON_SAMPLE_MULTI_TEXT_IMAGE)
+
+    if title != "":
+        card_data["header"]["title"]["text"] = title
+
+    if logo != "":
+        card_data["header"]["logo"] = logo
+
+    card_data["contents"] = []
+    for text in texts:
+        text_line = {
+            "type": "text",
+            "text": text,
+            "id": "text_" + str(uuid.uuid1())
+        }
+        divider_line = {
+            "type": "divider",
+            "id": "divider_" + str(uuid.uuid1())
+        }
+        card_data["contents"].append(text_line)
+        card_data["contents"].append(divider_line)
+
+    image_list = {
+        "type": "imageList",
+        "images": images,
+        "id": "imageList_" + str(uuid.uuid1())
+    }
+    card_data["contents"].append(image_list)
+
+    return card_data
+
+
 '''
 极简卡片组合：title-text-image-button
 '''
diff --git a/examples/cardbot/cardbot.py b/examples/cardbot/cardbot.py
@@ -11,7 +11,7 @@
 from dingtalk_stream import AckMessage, interactive_card
 import dingtalk_stream
 import time
-import copy
+import copy, asyncio
 
 
 def setup_logger():
@@ -38,45 +38,66 @@ def define_options():
     return options
 
 
-class CardBotHandler(dingtalk_stream.ChatbotHandler):
+class CardBotHandler(dingtalk_stream.AsyncChatbotHandler):
     """
     接收回调消息。
     回复一个卡片，然后更新卡片的文本和图片。
     """
 
-    def __init__(self, logger: logging.Logger = None):
-        super(dingtalk_stream.ChatbotHandler, self).__init__()
+    def __init__(self, logger: logging.Logger = None, max_workers: int = 8):
+        super(CardBotHandler, self).__init__(max_workers=max_workers)
         if logger:
             self.logger = logger
 
-    async def process(self, callback: dingtalk_stream.CallbackMessage):
+    def process(self, callback: dingtalk_stream.CallbackMessage):
+        '''
+        多线程场景，process函数不要用 async 修饰
+        :param message:
+        :return:
+        '''
+
         incoming_message = dingtalk_stream.ChatbotMessage.from_dict(callback.data)
 
-        card_data = copy.deepcopy(interactive_card.INTERACTIVE_CARD_JSON_SAMPLE_1)
+        texts = [
+            "第一行文本"
+        ]
 
-        # 先回复一个卡片
-        self.reply_card(card_data,
-                        incoming_message, False)
+        # 先回复一个文本卡片
+        self.reply_card(
+            interactive_card.generate_multi_text_line_card_data(title="机器人名字", logo="@lALPDfJ6V_FPDmvNAfTNAfQ",
+                                                                texts=texts),
+            incoming_message, False)
 
-        time.sleep(1)
+        images = [
+            "@lADPDe7s2ySi18PNA6XNBXg",
+            "@lADPDf0i1beuNF3NAxTNBXg",
+            "@lADPDe7s2ySRnIvNA6fNBXg"
+        ]
 
-        card_data = copy.deepcopy(interactive_card.INTERACTIVE_CARD_JSON_SAMPLE_1)
+        # 再回复一个文本+图片卡片
+        card_biz_id = self.reply_card(
+            interactive_card.generate_multi_text_image_card_data(title="机器人名字", logo="@lALPDfJ6V_FPDmvNAfTNAfQ",
+                                                                 texts=texts, images=images),
+            incoming_message, False)
 
-        # 更新文本
-        card_data["contents"][0]["text"] = "钉钉，让进步发生！\n 更新时间：{tt}".format(
-            tt=time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()))
+        # 再试试更新卡片
+        time.sleep(3)
 
         # 上传图片
         media_id = self.dingtalk_client.upload_to_dingtalk(open('./img.png', 'rb'),
                                                            filetype='image',
                                                            filename='image.png',
                                                            mimetype='image/png')
-        # 更新图片
-        card_data["contents"][1]["image"] = media_id
 
-        # 更新卡片
-        self.update_card(card_data,
-                         incoming_message)
+        texts = [
+            "更新后的第一行文本",
+            "更新后的第二行文本"
+        ]
+
+        self.update_card(card_biz_id, interactive_card.generate_multi_text_image_card_data(title="机器人名字",
+                                                                                           logo="@lALPDfJ6V_FPDmvNAfTNAfQ",
+                                                                                           texts=texts,
+                                                                                           images=[media_id]))
 
         return AckMessage.STATUS_OK, 'OK'
 
