如何打造一个能自动回复的钉钉机器人
source link: https://xie.infoq.cn/article/3340770024c49b5b1a54597d5
Go to the source link to view the article. You can view the picture content, updated content and better typesetting reading experience. If the link is broken, please click the button below to view the snapshot at that time.
如何打造一个能自动回复的钉钉机器人
- 2022 年 2 月 23 日
本文字数：5170 字
阅读完需：约 17 分钟
跟老表一起学云服务器开发相关文章（如果是第一次阅读该系列文章， 强烈建议先学习下面文章）：
和不安全访问 Say goodbye，手把手教大家如何给域名申请免费 SSL 证书
一、前情回顾
在之前分享的文章如何用Python发送告警通知到钉钉？中我们实现了钉钉群机器人定时/报警发送自定义消息，应用于一般的监控场景是很适用的，比如：每日早上发送自己关注的股票/基金开盘、走势数据、股票跌幅报警等。
但是在其他场景，比如：知识库查询、内容验证等，这类交互式的查询就没法实现，这个时候，我们需要更进阶的机器人：钉钉企业机器人，本文将手把手教你如何创建企业机器人，实现交互式查询。
二、分享概要
系统：阿里云 ECS 共享型 n4 服务器 1 核 2g 存储 50g
环境：自带 python3.6.8 方便演示，直接使用它
今天依然是用的 Linux 服务器演示，使用 Windows 服务器的操作类似。
三、开始动手动脑
3.1 自己创建一个企业
没错，自己创建一个企业，目前钉钉对于创建企业没有什么限制，人人可以创建，然后可以使用钉钉企业管理相关所有（免费）功能了。
钉钉创建企业非常简单，登录钉钉后，点击工作台，如果之前没加入过的话应该是可以直接看到创建企业/组织/团队按钮，如果之前加入过，如图所示，县级下拉按钮，就可以看到创建企业/组织/团队按钮啦，点击即可创建一个属于自己的企业。
3.2 登录钉钉开发者后台，创建企业机器人
这部操作可以查看钉钉官方文档：https://open.dingtalk.com/document/robots/enterprise-created-chatbot
登录钉钉开发者后台地址：
https://open-dev.dingtalk.com/?spm=ding_open_doc.document.0.0.5d08722fvbdReF登录后，我们点击应用开发->企业内部开发，进入企业内部开发应用管理后台。
点击机器人，然后点击创建应用，输入机器人基本信息即可。
创建成功后，我们会看到机器人的基本信息，主要包括：应用信息、开发管理、权限管理等，首先我们需要配置下开发管理，这里需要我们现在服务器开启一个服务，用于接收和发送钉钉消息。
3.3 开启一个 web 服务，用于接收和发送数据
一般情况，我们可能没有已经配备好证书且可以访问的域名，所以我们可以使用 http:公网 ip:端口号 来作为数据接收的地址，这里我们使用flask来搭建这个服务（比较简单）。
和不安全访问 Say goodbye，手把手教大家如何给域名申请免费 SSL 证书
首先我们需要连接上服务器，我直接使用宝塔面板登录，如果还不知道如何安装使用宝塔的读者可以看Linux里的宝塔，真正的宝塔！详细教程，直接通过宝塔连接上服务器后，点击终端，即可进入服务器命令模式，进行操作。
在写代码之前，我们需要安装相关库，
pip3 install flask接下来，我们进入项目目录，然后创建一个新的项目文件夹EnterpriseBot，并进入项目目录，这里大家可以将项目存放到自己存放项目的位置，和我一样起一个合适的名字即可。
cd Project/Little_project/DingdingBot && mkdir EnterpriseBotcd EnterpriseBot
我们先写一个简单的 flask 服务，如下：新建一个 rt_data.py 文件，这里我们直接使用nano指令来创建和编辑文件，
nano rt_data.py进入编辑模式后，写入以下代码，7 行代码简单写了一个 flask 服务，指定 host 为0.0.0.0，这样所有外网 ip 就都可以访问到服务器的服务了，另外设置了服务运行端口 port 为8083，默认端口为 5000，大家可以随便改，（注意哈，选的端口不要和自己正在使用的端口重复，避免服务冲突，程序无法正常运行）。
'''rt_data.pyReceiver Transmitter Data 数据接收和发送器'''from flask import Flaskapp = Flask(__name__)@app.route('/')def index():return '<h1>Hello World</h1>'if __name__ == '__main__':# 指定host和portapp.run(host='0.0.0.0', port=8083)
然后按ctrl+o保存文件，再按ctrl+x退出编辑模式。
在正式运行前，还需将服务所在端口添加到防火墙/安全策略组中，服务开启后才能正常使用ip地址:8083进行访问。
3.4 进入服务器后台，允许外部通过 8083 端口访问服务器
阿里云服务器是安全策略组，腾讯云服务器是防火墙，到对应位置添加端口即可。
进入服务器后台（以阿里云为例子），点击实例->安全组->配置规则，即可进入。
入方向，点击手动添加，其他不管，主要输入端口范围：8083/8083，授权对象：0.0.0.0，然后保存即可。（注意哈，选的端口不要和自己正在使用的端口重复，避免服务冲突，程序无法正常运行）
指定端口添加成功后，我们就可以启动服务啦，直接在终端输入以下指令即可。
python3 rt_data.py 然后我们再回到钉钉机器人开发管理后台，服务器公网 IP 和服务器地址（http://服务器公网ip:端口号）即可。
3.5 机器人开发：获取数据
前面我们完成了一些基本设置，现在我们来正式开发钉钉自动回复机器人功能。
钉钉开放平台官网上给出了用户 @机器人后，系统发送 post 请求时的 header 和 body 内容，如下：
http header
http body
首先我们需要利用到 header 里的 timestamp 和 sign 做数字签名验证，官网已经提供了 Python 验证的代码，直接复制即可使用，我们写到check_sig函数中。
# 消息数字签名计算核对def check_sig(timestamp):app_secret = '你自己的app_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')return sign
其中涉及两个参数，一个是：app_secret，固定的，我们在钉钉机器人开发管理后台中的应用信息中可以看到。
另一个参数是：timestamp，就是 http header 中的 timestamp，我们通过代码获取到传入函数即可。
# 节约篇幅，只放了主程序部分代码# 如何打造一个能自动回复的钉钉机器人@app.route("/", methods=["POST"])def get_data():# 第一步验证：是否是post请求if request.method == "POST":# print(request.headers)# 签名验证 获取headers中的Timestamp和Signtimestamp = request.headers.get('Timestamp')sign = request.headers.get('Sign')# 第二步验证：签名是否有效if check_sig(timestamp) == sign:# 获取数据 打印出来看看text_info = request.dataprint(text_info)print('验证通过')return str(text_info)print('验证不通过')return str(timestamp)print('有get请求')return str(request.headers)
返回来的数据是字节码（bytes 类型），所以我们需要先改编码，然后转换为字典类型，修改一行代码即可：
# text_info = request.data 改为text_info = json.loads(str(request.data, 'utf-8'))
现在返回数据就是正常的字典啦，我们可以方便的取出：发送人信息和发送内容，然后进行匹配返回对应内容。
3.6 机器人开发：返回数据
返回数据，逻辑和之前写的个人钉钉群机器人一样，获取到通信 url 后，发送 post 请求即可，在系统返回的 http body 中的sessionWebhook就是。
为了方便开发和测试，机器人正式发布前，我们可以先在钉钉机器人开发管理后台点击版本管理与调试->调试，系统会自动帮我们创建一个调试群。
代码部分，我们可以先写一个消息发送模板（以 markdown 格式消息为例子），看过如何用Python发送告警通知到钉钉？的读者应该很熟悉其格式和内容啦，大家可以看下之前的文章或者钉钉官方文档。
官方文档地址：https://developers.dingtalk.com/document/robots/customize-robot-security-settings
# 发送markdown消息def send_md_msg(userid, title, message, webhook_url):'''userid: @用户 钉钉idtitle : 消息标题message: 消息主体内容webhook_url: 通讯url'''data = {"msgtype": "markdown","markdown": {"title":title,"text": message},'''"msgtype": "text","text": {"content": message},'''"at": {"atUserIds": [userid],}}# 利用requests发送post请求req = requests.post(webhook_url, json=data)
然后我们写一个处理用户消息的函数，专门处理用户消息，然后返回指定内容，如下：
# 处理自动回复消息def handle_info(req_data):# 解析用户发送消息 通讯webhook_urltext_info = req_data['text']['content'].strip()webhook_url = req_data['sessionWebhook']senderid = req_data['senderId']# print('***************text_info：', text_info)# if判断用户消息触发的关键词，然后返回对应内容# python3.10 以上还可以用 switch case...if text_info == '福利':title = "【简说Python】今日福利"text = """### 福利介绍你好，我是老表，和我一起学习Python、云服务器开发知识啦！\n>**[老表的个人博客，已上线](https://python-brief.com/)**\n"""# 调用函数，发送markdown消息send_md_msg(senderid, title, text, webhook_url)if text_info == 'xxx':print('还可以写更多匹配、响应模式')
上面只是作为例子，如果真实使用中，我们每遇到一个自动回复就写一个 if 判断匹配、响应，那就有点麻烦了，所以这种情况下，大家可以维护一个知识库，比如一个 excel 表格，里面存放问题和回答，每次匹配直接读取文件数据，然后去匹配结果即可。
最后，就是主服务模块了，调用上面相关函数，传递参数即可，如下：
@app.route("/", methods=["POST"])def get_data():# 第一步验证：是否是post请求if request.method == "POST":# print(request.headers)# 签名验证 获取headers中的Timestamp和Signtimestamp = request.headers.get('Timestamp')sign = request.headers.get('Sign')# 第二步验证：签名是否有效if check_sig(timestamp) == sign:# 获取、处理数据req_data = json.loads(str(request.data, 'utf-8'))# print(req_data)# 调用数据处理函数handle_info(req_data)print('验证通过')return 'hhh'print('验证不通过')return 'ppp'print('有get请求')return 'sss'
服务器启动后效果如下，可以群内 @机器人，也可以私聊。
3.7 给程序创建守护进程
经过上面我们完成了功能开发，但是会发现，一旦我们关闭程序，自动回复服务也会停止，所以我们需要创建一个守护进程来保护我们的进程。
以我自己为例，我们登录宝塔面板后，进入/etc/systemd/system文件夹下，新建一个ding_bot.service文件，并写入下面内容：
[Unit]Description=Dingding Bot service[Service]Type=forkingExecStart=/usr/bin/python3 /root/Project/Little_project/DingdingBot/EnterpriseBot/rt_data.pyKillMode=processRestart=on-failureRestartSec=3s[Install]WantedBy=multi-user.target
保存好文件后，我们直接终端内执行下面指令即可开启进程守护，运行后会进入守护进程状态，我们可以按 ctrl+c 退出，不会影响守护进程：
systemctl start ding_bot代码修改后，需要重启守护进程，修改代码才会生效，重启指令如下：
systemctl restart ding_bot如果不想设置这个守护进程了，执行 stop 指令可以停止该 service（程序也会停止），指令如下：
systemctl stop ding_bot肝了一下午，出来了，不过还是有一些缺陷的，比如：没有实现机器人回复 @用户（http body 里是有 senderId，不过是加密的，没看出是啥加密方法），只是简单的自动回复（可以创建知识库管理问题和回答，还可以和之前的数据监测功能结合起来）等。
欢迎大家有了解的读者进行学习交流，一起进步。
最近安排时间优化一波，自己也确实有相关需求，比如：做灵感记录、待办事情提醒、网站查询等。
点赞 在看 留言 转发 ，四连支持，原创不易。
好的，那么下期见，我是爱猫爱技术，更爱思思的老表⁽⁽ଘ( ˙꒳˙ )ଓ⁾⁾
老表
公众号｜简说Python 2018.09.22 加入
【公众号：简说Python】爱猫爱技术，Python终身学习者、数据分析爱好者、Go语言内卷机。
Recommend
About Joyk
Aggregate valuable and interesting links.
Joyk means Joy of geeK