位置：首页>> 网络编程>> Python编程>> 教你如何使用Python开发一个钉钉群应答机器人

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

作者：zmister2016　　发布时间：2023-08-03 17:44:16　

标签：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/ 就是钉钉机器人的消息接收地址。