如何申请钉钉机器人接口
Posted
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了如何申请钉钉机器人接口相关的知识，希望对你有一定的参考价值。
参考技术A 钉钉群机器人钉钉群机器人是一个高级扩展功能，只要有一个钉钉账号，就可以使用它。它可以将第三方信息聚合到钉钉群中，实现信息自动同步。支持Webhook协议的自定义接入，通过发明者量化机器人，将提醒、报警等信息聚合到钉钉群中。支持文本（text）、链接（link）、markdown三种消息格式，五种消息类型。同一条信息还可以同时发送至多个钉钉群。
参考官方链接： https://developers.dingtalk.com/document/app/overview-of-group-robots
三、创建机器人
第1步：创建钉钉群
钉钉群每创建一个自定义机器人都会产生唯一的Hook地址，我们称为WebHook地址，通过向该WebHook地址推送消息，钉钉群就会收到消息。我们以PC端钉钉为例，首先点击左上方“+”号发起群聊，如果只想自己接受消息，可以随便拉两个人再踢出去，填写群名称：“XYK推送”，群类型选择普通群即可。
第2步：添加钉钉群机器人
点击头像，选择机器人管理，然后选择自定义，点击添加。自定义机器人名字：“XYK推送”，添加到刚刚创建的钉钉群。机器人支持三种安全设置：
自定义关键字：只有信息包含这个关键词，信息才会被同步。
咱们添加自定义关键字： 消息
加签：相当于设置密码。
IP地址：固定第三方信息的IP地址段。
如果只用于提醒或报警，选择自定义关键词就可以了。在这里我们定义的关键词是“：”，也就是说当发明者量化机器人推送的信息中包含“：”时，这条信息才会推送到钉钉群中。然后点击同意协议完成。最后复制Webhook地址备用。
钉钉单聊/群聊机器人实现思路
钉钉官网文档：https://open.dingtalk.com/document/group/robot-overview
一、钉钉开发平台
1.申请机器人
应用开发-企业内部开发-基础信息-机器人-创建机器人
2.开发管理
设置开发服务器出口IP列表
3.回调接口
设置钉钉扫码后的回调后端接口 （用于消息接收）
4.权限设置
企业内机器人发送消息权限
二、消息接收
获取post后的消息数据及各字段
1.通用字段
conversationType = req_data['conversationType'] # 消息类型，1单聊，2群聊
conversationId = req_data['conversationId'] # 加密的会话id
msgId = req_data['msgId'] # 加密的消息id
senderNick = req_data['senderNick'] # 发送消息的人昵称
isAdmin = req_data['isAdmin'] # 是否为管理员
sessionWebhookExpiredTime = req_data['sessionWebhookExpiredTime'] # 当前会话的Webhook地址过期时间
createAt = req_data['createAt'] # 消息的时间戳
senderId = req_data['senderId'] # 加密的发送者id
chatbotUserId = req_data['chatbotUserId'] # 加密的机器人id
msgtype = req_data['msgtype'] # 消息类型，目前只支持text
text = req_data['text']['content'] # 消息的关键字
sessionWebhook = req_data['sessionWebhook'] # 当前会话的webhook地址
robotCode = req_data['robotCode'] # 消息机器人代码
post_timestamp = request.headers.get('Timestamp') # 消息timestamp
post_sign = request.headers.get('Sign') # 消息sign
2.单聊字段
chatbotCorpId = req_data['chatbotCorpId'] # 加密的机器人所在的企业corpId
senderStaffId = req_data['senderStaffId'] # 企业内部群中@该机器人的成员userid
senderCorpId = req_data['senderCorpId'] # 企业内部群有的发送者当前群的企业corpId。
3.群聊字段
atUsers = req_data['atUsers'] # 被@人的信息
conversationTitle = req_data['conversationTitle'] # 群聊时的群名称
isInAtList = req_data['isInAtList'] # 是否在@列表中
三、后端认证
四、发送消息
1.单聊
msg =
"text":
"content": "单聊: ".format(text)
,
"msgtype": "text"
return json.dumps(msg)
五、实现代码
import json
import hmac
import hashlib
import base64
import time
api_blueprint = Blueprint(__name__ + '_api', __name__)
logger = get_logger(logger_type='run')
def check_sign_result(post_timestamp, post_sign):
timestamp = round(time.time() * 1000)
app_secret = 'ASAO7-nwDBzT1ObEo1PC6htvjHwlIBeu-4sZ7oVDQjf86EOA0DASgylZg-LIKmcL'
app_secret_enc = app_secret.encode('utf-8')
string_to_sign = '\\n'.format(post_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')
# 校验时间戳/sign
if (timestamp - int(post_timestamp) < 36000000 or (int(post_timestamp) - timestamp) < 3600000) and sign == post_sign:
return True
else:
return False
@api_blueprint.route('/wanglei/robot', methods=["GET", "POST"])
def wanglei_robot():
if request.method == 'GET':
return 'get'
elif request.method == 'POST':
req_data = json.loads(request.data.decode())
post_timestamp = request.headers.get('Timestamp')
post_sign = request.headers.get('Sign')
result = check_sign_result(post_timestamp, post_sign)
if result:
conversationType = req_data['conversationType'] # 消息类型，1单聊，2群聊
conversationId = req_data['conversationId'] # 加密的会话id
msgId = req_data['msgId'] # 加密的消息id
senderNick = req_data['senderNick'] # 发送消息的人昵称
isAdmin = req_data['isAdmin'] # 是否为管理员
sessionWebhookExpiredTime = req_data['sessionWebhookExpiredTime'] # 当前会话的Webhook地址过期时间
createAt = req_data['createAt'] # 消息的时间戳
senderId = req_data['senderId'] # 加密的发送者id
chatbotUserId = req_data['chatbotUserId'] # 加密的机器人id
msgtype = req_data['msgtype'] # 消息类型，目前只支持text
text = req_data['text']['content'] # 消息的关键字
sessionWebhook = req_data['sessionWebhook'] # 当前会话的webhook地址
robotCode = req_data['robotCode'] # 消息机器人代码
if conversationType == '1':
# 单聊
chatbotCorpId = req_data['chatbotCorpId'] # 加密的机器人所在的企业corpId
senderStaffId = req_data['senderStaffId'] # 企业内部群中@该机器人的成员userid
senderCorpId = req_data['senderCorpId'] # 企业内部群有的发送者当前群的企业corpId。
msg =
"text":
"content": "单聊: ".format(text)
,
"msgtype": "text"
return json.dumps(msg)
else:
# 群聊
atUsers = req_data['atUsers'] # 被@人的信息
conversationTitle = req_data['conversationTitle'] # 群聊时的群名称
isInAtList = req_data['isInAtList'] # 是否在@列表中
# 如果回复的消息需要@用户，则通过senderNick获取userid，在atUserIds中列出userid即可
msg =
"at":
"atUserIds": [
'manager8575'
],
"isAtAll": False
,
"text":
"content": '群聊: '.format(text)
,
"msgtype": "text"
return json.dumps(msg)
以上是关于如何申请钉钉机器人接口的主要内容，如果未能解决你的问题，请参考以下文章