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

当前位置博文首页 > 教你如何使用Python开发一个钉钉群应答机器人

最大化缩小

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

作者：zmister2016 时间：2021-08-01 18:14

前提

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

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

钉钉群机器人开发文档：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/ 就是钉钉机器人的消息接收地址。

配置钉钉机器人

回到钉钉开发者平台的网页，在钉钉机器人的「开发管理」页面，我们需要把服务器的出口IP 和钉钉机器人的消息接收地址填写好：

在这里插入图片描述

调试钉钉机器人

在配置好机器人的「服务器出口IP」与「消息接收地址」之后，我们点击网页菜单的「版本管理与发布」，点击「调试按钮」，进入到钉钉机器人的调试群：

在这里插入图片描述

这回在「钉钉机器人名称-TEST」的群里面添加创建的钉钉机器人：

在这里插入图片描述

我们可以在这个群里面@创建的群机器人进行测试：

在这里插入图片描述

在测试没问题之后，我们就可以点击「上线」按钮。钉钉机器人上线之后，就可以在钉钉群内添加这个机器人。

在这里插入图片描述

这样，我们就实现了从 0 到 1 使用 Python 开发钉钉群机器人。

基本的框架和流程大抵如此，具体的业务逻辑则需要根据不同的需求进行额外处理。比如：

查询天气，就得解析消息中的城市，然后请求天气接口获取天气数据，进行消息的响应；

淘宝客，就得解析消息中的文本，进行分词或其他处理，再查询数据库中的商品优惠券数据或是直接请求淘客接口获取商品优惠券数据；

员工绩效，就得接入钉钉的应用开发，借助钉钉开发的用户接口进行数据查询和响应。

jsjbwy

上一篇：Python中的套接字编程是什么?

下一篇：没有了

立即下载 - IIS7 站长工具包