教你如何使用Python开发一个钉钉群应答机器人-巨人网络通讯

主页 > 知识库 > 教你如何使用Python开发一个钉钉群应答机器人

教你如何使用Python开发一个钉钉群应答机器人

前提

搭建钉钉应答机器人，需要先准备或拥有以下权限：

钉钉企业的管理员或子管理员（如果不是企业管理员，可以自己创建一个企业，很方便的）
有公网通信地址（内网穿透也可以）；

钉钉群机器人开发文档：https://developers.dingtalk.com/document/app/overview-of-group-robots

创建「机器人」应用

输入好机器人的基本信息之后，就会生成创建一个「钉钉机器人」

我们的后端应用通过其提供的「AgentId」、「AppKey」、「AppSecret」就能够与钉钉机器人进行通信。

接收消息

在钉钉机器人的设定中，当用户@机器人时，钉钉会通过机器人开发者的服务器地址，用 POST 请求方法把消息内容发送出去，其 HTTP header 如下所示：

{
  "Content-Type": "application/json; charset=utf-8",
  "timestamp": "1577262236757",
  "sign":"xxxxxxxxxx"
}

其中，timestamp是消息发送时的时间戳，sign是签名值，我们需要对这两个值进行校验。

如果timestamp与系统当前时间相差1小时以上，则为非法请求。

如果sign签名值与后台计算的值不一样，也为非法请求。

其中sign签名值的计算方法为：header中的timestamp + “\n” + 机器人的appSecret当做签名字符串，使用HmacSHA256算法计算签名，然后进行Base64 encode，得到最终的签名值。

其 Python 实现代码如下所示：

import hmac
import hashlib
import base64

timestamp = '1577262236757'
app_secret = 'this is a secret'
app_secret_enc = app_secret.encode('utf-8')
string_to_sign = '{}\n{}'.format(timestamp, app_secret)
string_to_sign_enc = string_to_sign.encode('utf-8')
hmac_code = hmac.new(app_secret_enc, string_to_sign_enc, digestmod=hashlib.sha256).digest()
sign = base64.b64encode(hmac_code).decode('utf-8')
print(sign)

其发送的消息格如下所示：

{
    "conversationId": "xxx",
    "atUsers": [
        {
            "dingtalkId": "xxx",
            "staffId":"xxx"
        }
    ],
    "chatbotCorpId": "dinge8a565xxxx",
    "chatbotUserId": "$:LWCP_v1:$Cxxxxx",
    "msgId": "msg0xxxxx",
    "senderNick": "杨xx",
    "isAdmin": true,
    "senderStaffId": "user123",
    "sessionWebhookExpiredTime": 1613635652738,
    "createAt": 1613630252678,
    "senderCorpId": "dinge8a565xxxx",
    "conversationType": "2",
    "senderId": "$:LWCP_v1:$Ff09GIxxxxx",
    "conversationTitle": "机器人测试-TEST",
    "isInAtList": true,
    "sessionWebhook": "https://oapi.dingtalk.com/robot/sendBySession?session=xxxxx",
    "text": {
        "content": " 你好"
    },
    "msgtype": "text"
}

其中，一些参数的说明如下图所示：

我们接收到钉钉的消息后，可以根据实际的业务需求解析出相应字段的数据来进行处理。

响应消息

钉钉机器人支持我们通过「text」、「Markdown」、「整体跳转actionCard」、「独立跳转actionCard」和「feedCard」这5种消息类型发送消息到群里。

下面我们通过实际的代码来展示接收钉钉机器人的消息，以及发送 5 种消息类型到钉钉群里。

创建一个后端应用

接下来，我们通过创建一个 Django 应用来接收的处理用户发送给钉钉机器人的消息。

首先，创建一个 Django 项目和应用：

django-admin startproject DdRobot
python manage.py startapp app_robot

然后打开 “C:\DdRobot\DdRobot\settings.py” 文件，修改 ALLOWED_HOSTS 变量：

ALLOWED_HOSTS = ['*']

将 app_robot 添加到 INSTALLED_APPS 变量列表中：

INSTALLED_APPS = [
    'django.contrib.admin',
    'django.contrib.auth',
    'django.contrib.contenttypes',
    'django.contrib.sessions',
    'django.contrib.messages',
    'django.contrib.staticfiles',
    'app_robot',
]

创建校验时间戳和签名函数

因为钉钉机器人会在请求头里面传入timestamp时间戳和sign签名供我们对请求的合法性进行校验，所以为了机器人的安全，我们需要编写 2 个函数对它们进行校验（在DdRobot/app_robot/views.py文件中进行）。

首先，是时间戳的校验：

def check_timestamp(timestamp):
    now_timestamp = int(time.time()*1000)
    if now_timestamp - int(timestamp) > 3600000:
        return False
    else:
        return True

然后是签名值的校验，签名值的计算方法和示例代码钉钉已经提供，我们借用即可：

def check_sign(timestamp,sign):
    import hmac
    import hashlib
    import base64

    # now_timestamp = str(int(time.time()*1000))
    app_secret = 'teTLGS3xZVLp6Z99mXvgVpINOUyJqFsKJ3jLb7crFdjRsJ3_77E-kxhlIbBGbNjX'
    app_secret_enc = app_secret.encode('utf-8')
    string_to_sign = '{}\n{}'.format(timestamp, app_secret)
    string_to_sign_enc = string_to_sign.encode('utf-8')
    hmac_code = hmac.new(app_secret_enc, string_to_sign_enc, digestmod=hashlib.sha256).digest()
    new_sign = base64.b64encode(hmac_code).decode('utf-8')
    # print(sign)
    # print(new_sign)
    if sign == new_sign:
        return True
    else:
        return False

对于这 2 个值，校验成功我们都返回 True，校验失败我们都返回 False。

创建视图函数

接着，我们创建一个视图函数，用来接收钉钉传输过来的消息，以及响应给钉钉。

@csrf_exempt
def resp_dd(request):
	pass

在 resp_dd() 函数中，首先从请求头中读取钉钉传输过来的时间戳和签名值，然后进行校验：

@csrf_exempt
def resp_dd(request):
    timestamp = request.headers.get('timestamp','')
    sign = request.headers.get('sign','')
    # 校验时间戳
    if check_timestamp(timestamp) is False:
        return JsonResponse({'status':False,'data':'非法请求'})
    # 校验签名
    if check_sign(timestamp,sign) is False:
        return JsonResponse({'status':False,'data':'非法请求'})

若是时间戳和签名值校验无误，我们继续从请求 body 里面获取消息信息：

@csrf_exempt
def resp_dd(request):
    timestamp = request.headers.get('timestamp','')
    sign = request.headers.get('sign','')
    # 校验时间戳
    if check_timestamp(timestamp) is False:
        return JsonResponse({'status':False,'data':'非法请求'})
    # 校验签名
    if check_sign(timestamp,sign) is False:
        return JsonResponse({'status':False,'data':'非法请求'})
    body = json.loads(request.body)
    # 获取用户id
    # user_id = body['senderStaffId'] 机器人上线后才会返回
    user_id = body['senderId']
    # 获取发送的消息
    msg_type = body['msgtype']
    if msg_type == 'text':
        content = body['text']['content']

目前钉钉机器人只支持text文本内容的消息接收，所以在此处我们只对消息类型为text的消息进行处理。

获取到钉钉机器人发送过来的信息之后，我们就可以根据自己的业务逻辑进行处理，然后返回特定的消息类型了。

在这里，我们只对消息进行简单的处理：

当发送来的消息文本为text时，机器人回复文本消息；
当发送来的消息文本为markdown时，机器人回复一个 Markdown 的示例消息；
当发送来的消息文本为整体跳转时，机器人回复一个「整体跳转卡片」的示例消息；
当发送来的消息文本为独立跳转时，机器人回复一个「独立跳转卡片」的示例消息；
当发送来的消息文本为feed时，机器人回复一个「feedCard」的示例消息；

先来定义 5 个不同消息类型的响应格式。

文本消息类型

  # 响应文字
    resp_text = {
        "at": {
            "atUserIds": [
                user_id
            ],
            "isAtAll": False
        },
        "text": {
            "content": "你刚刚发的消息是：[{}]".format(content)
        },
        "msgtype": "text"
    }

Markdown消息类型：

 # 响应Markdown
    resp_markdown = {
        "msgtype": "markdown",
            "markdown": {
            "title":"州的先生机器人助理",
            "text": "## 这是什么？ \n 这是一个钉钉机器人 \n ![](https://zmister.com/wp-content/uploads/2019/06/login_logo.png)"
        },
        "at": {
            "atUserIds": [
              user_id
            ],
            "isAtAll": False
        }
    }

整体跳转卡片消息类型：

    # 响应整体跳转actionCard
    resp_actioncard = {
        "msgtype": "actionCard",
        "actionCard": {
            "title": "州的先生 Python 实战教程合集", 
            "text": "![](https://zmister.com/wp-content/uploads/2019/06/login_logo.png) \n #### 州的先生 Python 实战教程合集 \n\n 学习Python的一个好方法就是用实际的项目来熟练语言",
            "singleTitle" : "阅读全文",
            "singleURL" : "http://mrdoc.zmister.com"
        }
    }

独立跳转卡片消息类型：

resp_actioncard_2 = {
        "msgtype": "actionCard",
        "actionCard": {
            "title": "州的先生 Python 实战教程合集", 
            "text": "![](https://zmister.com/wp-content/uploads/2019/06/login_logo.png) \n #### 州的先生 Python 实战教程合集 \n\n 学习Python的一个好方法就是用实际的项目来熟练语言",
            "hideAvatar": "0", 
            "btnOrientation": "0", 
            "btns": [
                {
                    "title": "去看看", 
                    "actionURL": "http://mrdoc.zmister.com"
                }, 
                {
                    "title": "不感兴趣", 
                    "actionURL": "https://zmister.com/"
                }
            ]
        }
    }

Feed卡片消息类型：

# 响应feedCard
    resp_feedcard = {
        "msgtype": "feedCard",
        "feedCard": {
            "links": [
                {
                    "title": "时代的火车向前开1", 
                    "messageURL": "http://mrdoc.zmister.com", 
                    "picURL": "https://img.alicdn.com/tfs/TB1NwmBEL9TBuNjy1zbXXXpepXa-2400-1218.png"
                },
                {
                    "title": "时代的火车向前开2", 
                    "messageURL": "https://zmister.com/", 
                    "picURL": "https://img.alicdn.com/tfs/TB1NwmBEL9TBuNjy1zbXXXpepXa-2400-1218.png"
                }
            ]
        }
    }

其他的消息响应空：

# 响应空，不回复
    resp_empty = {
        "msgtype": "empty"
    }

定义好几个消息响应类型数据后，我们对获取到的 content 变量进行判断返回响应即可：

 if content[1:] == 'text':
        return JsonResponse(resp_text)
    elif content[1:] == 'markdown':
        return JsonResponse(resp_markdown)
    elif content[1:] == '整体跳转':
        return JsonResponse(resp_actioncard)
    elif content[1:] == '独立跳转':
        return JsonResponse(resp_actioncard_2)
    elif content[1:] == 'feed':
        return JsonResponse(resp_feedcard)
    else:
        return JsonResponse(resp_empty)

这样，我们这个钉钉机器人的后端处理函数就写好了。

配置路由

写好视图函数之后，我们配置一下这个函数的 URL 路由。

在 “C:\DdRobot\DdRobot\urls.py” 文件中把内容修改为如下代码所示：

from django.contrib import admin
from django.urls import path
from app_robot import views

urlpatterns = [
    path('admin/', admin.site.urls),
    path('dd_robot/',views.resp_dd, name="resp_dd"),
]

这样 http://ip地址/dd_robot/ 就是钉钉机器人的消息接收地址。