钉钉单聊/群聊机器人实现思路(代码片段)

亦非我所愿丶 亦非我所愿丶     2022-12-09     233

关键词:

钉钉官网文档: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)

使用nio实现一个简易的群聊和单聊(代码片段)

服务端:接收客户端发送的消息,并进行转发。packagesocket.demo2;importjava.io.IOException;importjava.net.InetSocketAddress;importjava.net.ServerSocket;importjava.nio.ByteBuffer;importjava.nio.channels.*;importjava.nio.chars 查看详情

springwebsocket和socketjs实现单聊群聊,广播的消息推送详解(代码片段)

springwebsocket和socketjs实现单聊群聊,广播的消息推送详解WebSocket简单介绍  随着互联网的发展,传统的HTTP协议已经很难满足Web应用日益复杂的需求了。近年来,随着HTML5的诞生,WebSocket协议被提出,它实现了浏览器与服务器的全... 查看详情

zabbix--钉钉告警(代码片段)

zabbix钉钉告警机制群机器人是钉钉群的高级扩展功能,群机器人可以将第三方服务的信息聚合到群聊中,实现自动化的信息同步。例如:通过聚合GitHub,GitLab等源码管理服务,实现源码更新同步;通过聚合Trello,JIRA等项目协调... 查看详情

创建钉钉群聊机器人,使用python发送消息(代码片段)

文章目录获取自定义机器人Webhook使用Python发送消息使用curl发送消息获取自定义机器人Webhook1.1、创建群(然后将别人移出群聊)1.2、单击群设置>智能群助手1.3、在机器人管理页面选择自定义机器人1.4、输入机器人名字1... 查看详情

基于flask写的web_socket单聊和群聊(代码片段)

群聊 fromflaskimportFlask,request,render_templatefromgeventwebsocket.handlerimportWebSocketHandlerfromgevent.pywsgiimportWSGIServerfromgeventwebsocket.websocketimportWebSocketapp=Flask(__name__)use 查看详情

java_钉钉发送私聊消息(代码片段)

...;只不过群消息的前提是先建群,然后在群里添加钉钉机器人,并且工作群一旦多了,消息看的也就没那么及时,所以群消息在某一些场景下没那么便捷,或者可以说没有那么高效。一、文档地址Api文档地址... 查看详情

zabbix使用机器人报警到钉钉群聊

钉钉机器人接口脚本下载:https://www.appgao.com/files/192.html钉钉机器人接口地址(可以点击机器人查看):webhook地址,https://oapi.dingtalk.com/robot/send?access_token=xxxxxxxx配置报警媒介类型650)this.width=650;"src="https://s3.51cto.com/wyfs02/M 查看详情

python3钉钉加签名钉钉群机器人告警脚本(代码片段)

...加灵活全网普遍用自定义关键词多,不予置评添加群机器人智能群助手添加机器人自定义机器人设置加签复制一下签名SEC,后边会用到复制webhook点击完成机器人创建成功!python3钉钉加签名钉钉群机器人告警脚本dd-aler... 查看详情

通过钉钉群聊机器人推送zabbix告警

1、首先在zabbix服务端的alertscripts目录下新建一个dingding.sh脚本,内容如下:#!/bin/bashContent=$2curl‘https://oapi.dingtalk.com/robot/send?access_token=XXXX‘  -H‘Content-Type:application/json‘  -d‘&nbs 查看详情

javascript基于websocketnodejsuniapp的聊天室(代码片段)

...术栈开发了一个简易聊天室,可以进行简单的群聊和单聊的功能。登陆后在聊天大厅可以进行群聊查看当前在线用户选择指定在线用户进行单聊项目地址:https://github.com/Undertone0809/ZeeChatroom效果图首页选择头像群聊大厅单... 查看详情

javascript基于websocketnodejsuniapp的聊天室(代码片段)

...术栈开发了一个简易聊天室,可以进行简单的群聊和单聊的功能。登陆后在聊天大厅可以进行群聊查看当前在线用户选择指定在线用户进行单聊项目地址:https://github.com/Undertone0809/ZeeChatroom效果图首页选择头像群聊大厅单... 查看详情

python3钉钉加签名钉钉群机器人告警脚本(代码片段)

...加灵活全网普遍用自定义关键词多,不予置评添加群机器人智能群助手添加机器人自定义机器人设置加签复制一下签名SEC,后边会用到复制webhook点击完成机器人创建成功!python3钉钉加签名钉钉群机器人告警脚本dd-aler... 查看详情

python实现钉钉机器人消息自动化通知(代码片段)

目录前言自定义机器人安全设置自定义关键字加签IP地址钉钉消息通知类型安装使用发送text消息text消息之@指定用户Link消息Markdown消息@所有人Markdown消息@指定用户发送自动化报告前言之前自动化执行成功之后,会发... 查看详情

使用genvent.socket实施群聊/单聊模式

使用genvent.socket实施群聊fromflaskimportFlask,request,render_templatefromgevent.pywsgiimportWSGIServerfromgeventwebsocket.handlerimportWebSocketHandlerfromgeventwebsocket.websocketimportWebSocketimportjson 查看详情

aop实现系统告警(代码片段)

...、告警的方式1、钉钉告警通过在企业钉钉群,添加群机器人的方式,通过机器人向群内发送报警信息。至于钉钉机器人怎么创建,发送消息 查看详情

pytest+yaml框架-14.钉钉机器人通知测试结果(代码片段)

.../飞书/企业微信等。pip安装插件pipinstallpytest-yaml-yoyo钉钉机器人通知测试结果功能在v1.1.1版本实现钉钉机器人设置钉钉机器人的设置请参考官方API文档https://open.dingtalk.com/document/group/custom-robo 查看详情

zabbix配置钉钉的带图片报警(代码片段)

...kdown语法格式构造请求利用钉钉的webhook发送消息设置钉钉机器人钉钉机器人需要进行安全设置,这里我定义了关键词”通知“,也就是说发送的消息中必须包含这个词脚本实现由于需要将图片从本地拷到远端,免密需要提前配置... 查看详情

im即时通讯开发:高可用易伸缩高并发的im群聊单聊架构方案设计

要实现一整套能用于大用户量、高并发场景下的IM群聊,技术难度远超IM系统中的其它功能,原因在于:IM群聊消息的实时写扩散特性带来了一系列技术难题。举个例子:如一个2000人群里,一条普通消息的发出问题,将瞬间写扩... 查看详情