我的flask钉钉企业内部开发机器人

人工智障火的一塌糊涂，智死方休。虽然博主也在使用，使用过程中多少还是有问题的（遇到死循环代码生成，重复回答等）。但是此篇文章主角是dingtalk的机器人，而不是那玩意，可以自行集成哈，查得太严，容易封站~

钉钉企业内部开发机器人

企业内部开发的机器人是钉钉为用户提供的组织内部使用的机器人，为组织数字化转型业务服务。开发者可通过本文所描述步骤进行机器人的自主开发和上架，组织内其它成员可通过方便快捷地使用机器人的能力。

基于企业机器人的outgoing（回调）机制，用户发消息给机器人之后，钉钉会将消息内容POST到开发者的消息接收地址。

开发者解析出消息内容、发送者身份，根据企业的业务逻辑，组装响应的消息内容返回，钉钉会将响应内容发送给用户。

官方样例

某企业开发了一个工具，用于检测某个网址是否安全。在上架为一个企业机器人之后，企业成员可以直接给这个机器人发消息，询问该机器人一个网址，机器人自动答复是否安全。

创建机器人步骤文档自行参考：​​https://open.dingtalk.com/document/robots/enterprise-created-chatbot​​

钉钉机器人开发

当用户@机器人时，钉钉会通过机器人开发者的HTTPS服务地址，把消息内容发送出去，报文协议如下

{
  "Content-Type": "application/json; charset=utf-8",
  "timestamp": "1577262236757",
  "sign":"xxxxxxxxxx"
}

参数	说明
timestamp	消息发送的时间戳，单位是毫秒。
sign	签名值。

我们先用flask模拟下并打印头部日志和body日志

from flask import Flask
from flask import jsonify
from flask import request
import logging
import json
LOG_FORMAT = '%(asctime)s -%(name)s- %(threadName)s-%(thread)d - %(levelname)s - %(message)s'
DATE_FORMAT = "%m/%d/%Y %H:%M:%S %p"
#日志配置
logging.basicConfig(level=logging.INFO,format=LOG_FORMAT,datefmt=DATE_FORMAT)

app = Flask(__name__)

@app.route('/')
def hello_world():
    return 'Hello dingding bot!'

@app.route('/aichat',methods=['POST'])
def aichat():
    data = request.get_json()
    header = request.headers
    logging.info("dingding request body: data:= "+ json.dumps(data) + "; headers:= "+ json.dumps(dict(header)) )
    rsp_j = {
        "status": "ok"
    }
    return jsonify(rsp_j)

if __name__ == '__main__':
    app.run(host='0.0.0.0',port=5000)

本地调试运行可以查看到日志中会输出body和headers

02/17/2023 17:25:12 PM -root- Thread-1-13720 - INFO - dingding request body: data:= {"name": "xadocker"}; headers:= {"Content-Type": "application/json", "User-Agent": "PostmanRuntime/7.29.0", "Accept": "*/*", "Postman-Token": "abf1ddde-d991-410d-abad-2030ee1a912f", "Host": "127.0.0.1:5000", "Accept-Encoding": "gzip, deflate, br", "Connection": "keep-alive", "Content-Length": "28"}
02/17/2023 17:25:12 PM -werkzeug- Thread-1-13720 - INFO - 127.0.0.1 - - [17/Feb/2023 17:25:12] "POST /aichat HTTP/1.1" 200 -

把该项目放到服务器上，并用钉钉去调试看下获取到输出

02/17/2023 09:27:23 AM -root- Dummy-1-139914028918432 - INFO - dingding request body: data:= {"conversationId": "cidJdfNhnyORlKCkSd8tjcq6Q==", "atUsers": [{"dingtalkId": "$:LWCP_v1:$HZvIOckG+jz8y7AZ+jXQJUNauE0S4VW2"}], "chatbotCorpId": "ding6ae2e50d88743ec8f2c783f7214b6d69", "chatbotUserId": "$:LWCP_v1:$HZvIOckG+jz8y7AZ+jXQJUNauE0S4VW2", "msgId": "msgtkshoVuY9Asx0Mrq+SpL9w==", "senderNick": "\u5b89\u4e1c", "isAdmin": true, "senderStaffId": "manager5345", "sessionWebhookExpiredTime": 1676631443578, "createAt": 1676626043272, "senderCorpId": "ding6ae2e50d88743ec8f2c783f7214b6d69", "conversationType": "2", "senderId": "$:LWCP_v1:$PhpOmv6+MlWvju2T20KTTDcyeCzOZ0EN", "conversationTitle": "\u76d1\u63a7\u5927\u5e08", "isInAtList": true, "sessionWebhook": "https://oapi.dingtalk.com/robot/sendBySession?session=8a46e032f2fb4c0e80fdf1defef8d08c", "text": {"content": " hello"}, "robotCode": "dingm6xywyp2xixpcy6d", "msgtype": "text"}; headers:= {"Host": "dingtalk.xadocker.cn", "X-Real-Ip": "59.82.61.12", "X-Forwarded-Proto": "http", "X-Forwarded-For": "59.82.61.12", "Connection": "close", "Content-Length": "783", "Sign": "A3ojN8BXI8fu44KoK7HKVBKfJXDLV4JDy0Bt3e3Svw0=", "Token": "a091f7d6-3a4e-41ae-a5d6-1ec6d6e7aadd", "Timestamp": "1676626043594", "Content-Type": "application/json; charset=utf-8", "Accept-Encoding": "gzip", "User-Agent": "okhttp/3.5.0"}

钉钉回调请求header

{
  "Host": "dingtalk.xadocker.cn",
  "X-Real-Ip": "59.82.61.12",
  "X-Forwarded-Proto": "http",
  "X-Forwarded-For": "59.82.61.12",
  "Connection": "close",
  "Content-Length": "783",
  "Sign": "A3ojN8BXI8fu44KoK7HKVBKfJXDLV4JDy0Bt3e3Svw0=",
  "Token": "a091f7d6-3a4e-41ae-a5d6-1ec6d6e7aadd",
  "Timestamp": "1676626043594",
  "Content-Type": "application/json; charset=utf-8",
  "Accept-Encoding": "gzip",
  "User-Agent": "okhttp/3.5.0"
}

官网提示：

开发者需对header中的timestamp和sign进行验证，以判断是否是来自钉钉的合法请求，避免其他仿冒钉钉调用开发者的HTTPS服务传送数据，具体验证逻辑如下：

• timestamp 与系统当前时间戳如果相差1小时以上，则认为是非法的请求。
• sign 与开发者自己计算的结果不一致，则认为是非法的请求。

必须当timestamp和sign同时验证通过，才能认为是来自钉钉的合法请求。

sign的计算方法

header中的timestamp + "\\n" + 机器人的appSecret当做签名字符串，使用HmacSHA256算法计算签名，然后进行Base64 encode，得到最终的签名值。

python样例

#python 3.8
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)

所以此时结合上面代码示例调整flask服务，​​讲sign校验提取到sign.py​​:

注意替换为自己机器人得secret

import hmac
import hashlib
import base64
from datetime import datetime
import pytz
import logging
LOG_FORMAT = '%(asctime)s -%(name)s- %(threadName)s-%(thread)d - %(levelname)s - %(message)s'
DATE_FORMAT = "%m/%d/%Y %H:%M:%S %p"
#日志配置
logging.basicConfig(level=logging.INFO,format=LOG_FORMAT,datefmt=DATE_FORMAT)

def validate_sign(sign, ts, **kwargs):

    if not sign or not ts:
        return False

    """时间有效期 10 秒"""
    # 获取服务端当前时间戳
    china_timezone = pytz.timezone('Asia/Shanghai')
    now = datetime.now(tz=china_timezone)
    server_timestamp = int(now.timestamp() * 1000)
    if server_timestamp - int(ts) > 10000:
        logging.info("dingding requests ts:= " + str(ts) + "server_ts:= " + str(server_timestamp))
        return False

    # 校验sign
    client_timestamp = ts
    app_secret = 'yoursecret'

    app_secret_enc = app_secret.encode('utf-8')
    string_to_sign = '{}\n{}'.format(client_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')

    if sign == new_sign:
        return True
    else:
        logging.info("dingding requests c_sign:= " + ts + "s_sign:= " + new_sign)
        return False

此时我们得app.py为

import logging
import json

from flask import Flask,abort
from flask import jsonify
from flask import request

from utils.sign import validate_sign
LOG_FORMAT = '%(asctime)s -%(name)s- %(threadName)s-%(thread)d - %(levelname)s - %(message)s'
DATE_FORMAT = "%m/%d/%Y %H:%M:%S %p"
#日志配置
logging.basicConfig(level=logging.INFO,format=LOG_FORMAT,datefmt=DATE_FORMAT)

app = Flask(__name__)

@app.route('/')
def hello_world():
    return 'Hello dingding bot!'

@app.route('/aichat',methods=['POST'])
def aichat():
    data = request.get_json()
    headers = request.headers

    """校验sign"""
    sign = headers.get("Sign")
    ts = headers.get("Timestamp")

    if not validate_sign(sign, ts):
        logging.error("dingding request sign is invalid")
        abort(403)

    logging.info("dingding request body: data:= " + json.dumps(data) + "; headers:= " + json.dumps(dict(headers)))

    rsp_j = {
        "status": "ok"
    }
    return jsonify(rsp_j)

if __name__ == '__main__':
    app.run(host='0.0.0.0',port=5000)

测试下

服务日志：

02/17/2023 11:30:41 AM -root- Dummy-1-140478451600640 - INFO - dingding request body: data:= {"conversationId": "cidJdfNhnyORlKCkSd8tjcq6Q==", "atUsers": [{"dingtalkId": "$:LWCP_v1:$HZvIOckG+jz8y7AZ+jXQJUNauE0S4VW2"}], "chatbotCorpId": "ding6ae2e50d88743ec8f2c783f7214b6d69", "chatbotUserId": "$:LWCP_v1:$HZvIOckG+jz8y7AZ+jXQJUNauE0S4VW2", "msgId": "msgKE/HhGcPDfUdTkq8DWSAHA==", "senderNick": "\u5b89\u4e1c", "isAdmin": true, "senderStaffId": "manager5345", "sessionWebhookExpiredTime": 1676638840881, "createAt": 1676633440616, "senderCorpId": "ding6ae2e50d88743ec8f2c783f7214b6d69", "conversationType": "2", "senderId": "$:LWCP_v1:$PhpOmv6+MlWvju2T20KTTDcyeCzOZ0EN", "conversationTitle": "\u76d1\u63a7\u5927\u5e08", "isInAtList": true, "sessionWebhook": "https://oapi.dingtalk.com/robot/sendBySession?session=8a46e032f2fb4c0e80fdf1defef8d08c", "text": {"content": " hello hello"}, "robotCode": "dingm6xywyp2xixpcy6d", "msgtype": "text"}; headers:= {"Host": "dingtalk.xadocker.cn", "X-Real-Ip": "59.82.61.58", "X-Forwarded-Proto": "http", "X-Forwarded-For": "59.82.61.58", "Connection": "close", "Content-Length": "789", "Sign": "73hJBLQQZui7AEGJJw4oJSwziAlpPFbZgqpAoPGC64w=", "Token": "a091f7d6-3a4e-41ae-a5d6-1ec6d6e7aadd", "Timestamp": "1676633440897", "Content-Type": "application/json; charset=utf-8", "Accept-Encoding": "gzip", "User-Agent": "okhttp/3.5.0"}

至此，我们简单得sign校验功能已完成

钉钉回调请求body

{
  "conversationId": "cidJdfNhnyORlKCkSd8tjcq6Q==",
  "atUsers": [
    {
      "dingtalkId": "$:LWCP_v1:$HZvIOckG+jz8y7AZ+jXQJUNauE0S4VW2"
    }
  ],
  "chatbotCorpId": "ding6ae2e50d88743ec8f2c783f7214b6d69",
  "chatbotUserId": "$:LWCP_v1:$HZvIOckG+jz8y7AZ+jXQJUNauE0S4VW2",
  "msgId": "msgtkshoVuY9Asx0Mrq+SpL9w==",
  "senderNick": "安东",
  "isAdmin": true,
  "senderStaffId": "manager5345",
  "sessionWebhookExpiredTime": 1676631443578,
  "createAt": 1676626043272,
  "senderCorpId": "ding6ae2e50d88743ec8f2c783f7214b6d69",
  "conversationType": "2",
  "senderId": "$:LWCP_v1:$PhpOmv6+MlWvju2T20KTTDcyeCzOZ0EN",
  "conversationTitle": "监控大师",
  "isInAtList": true,
  "sessionWebhook": "https://oapi.dingtalk.com/robot/sendBySession?session=8a46e032f2fb4c0e80fdf1defef8d08c",
  "text": {
    "content": " hello"
  },
  "robotCode": "dingm6xywyp2xixpcy6d",
  "msgtype": "text"
}

参数	是否必填	类型	说明
msgtype	是	String	目前只支持text。
content	是	String	消息文本。
msgId	是	String	加密的消息ID。
createAt	是	String	消息的时间戳，单位ms。
conversationType	是	String	1：单聊2：群聊
conversationId	是	String	加密的会话ID。
conversationTitle	否	String	群聊时才有的会话标题。
senderId	是	String	加密的发送者ID。说明使用senderStaffId，作为发送者userid值。
senderNick	是	String	发送者昵称。
senderCorpId	否	String	企业内部群有的发送者当前群的企业corpId。
sessionWebhook	是	String	当前会话的Webhook地址。
sessionWebhookExpiredTime	是	Long	当前会话的Webhook地址过期时间。
isAdmin	否	boolean	是否为管理员。说明机器人发布上线后生效。
chatbotCorpId	否	String	加密的机器人所在的企业corpId。
isInAtList	否	boolean	是否在@列表中。
senderStaffId	否	String	企业内部群中@该机器人的成员userid。说明该字段在机器人发布线上版本后，才会返回。
chatbotUserId	是	String	加密的机器人ID。
atUsers		Array	被@人的信息。dingtalkId：加密的发送者ID。staffId：当前企业内部群中员工userid值。

HTTP响应格式

开发者可以根据自己的业务需要，选择回复一段消息到群中，目前支持text、markdown、整体跳转actionCard类型、独立跳转actionCard类型、feedCard这5种消息类型。

text格式响应

前面博主只是响应：

rsp_j = {
        "status": "ok"
    }

钉钉不识别所以@机器人后无消息返回，此时我们将消息转为以下格式：

{
    "msgtype": "text",
    "text": {
        "content": "月会通知"
    }
}

则此时app.py为

import logging
import json

from flask import Flask, abort
from flask import jsonify
from flask import request

from utils.sign import validate_sign

LOG_FORMAT = '%(asctime)s -%(name)s- %(threadName)s-%(thread)d - %(levelname)s - %(message)s'
DATE_FORMAT = "%m/%d/%Y %H:%M:%S %p"
# 日志配置
logging.basicConfig(level=logging.INFO, format=LOG_FORMAT, datefmt=DATE_FORMAT)

app = Flask(__name__)

@app.route('/')
def hello_world():
    return 'Hello dingding bot!'

@app.route('/aichat', methods=['POST'])
def aichat():
    data = request.get_json()
    headers = request.headers

    """校验sign"""
    sign = headers.get("Sign")
    ts = headers.get("Timestamp")

    if not validate_sign(sign, ts):
        logging.error("dingding request sign is invalid")
        abort(403)

    logging.info("dingding request body: data:= " + json.dumps(data) + "; headers:= " + json.dumps(dict(headers)))

    #
    customer_context = data['text']['content']
    rsp_j = {
        "msgtype": "text",
        "text": {
            "content": "hello, [{}]".format(customer_context)
        }
    }
    return jsonify(rsp_j)

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000)

测试下

markdown格式响应

响应体格式为

{
    "msgtype": "markdown",
    "markdown": {
        "title": "钉钉机器人",
        "text": "# 这是支持markdown的文本   \n   ## 标题2    \n   * 列表1   \n  ![alt 啊](https://static.xadocker.cn/wp-content/uploads/7d949cac66ca42e1862241338b23bf08.jpg)"
    }

钉钉内可用得markdown格式为：

标题
# 一级标题
## 二级标题
### 三级标题
#### 四级标题
##### 五级标题
###### 六级标题

引用
> A man who stands for nothing will fall for anything.

文字加粗、斜体
**bold**
*italic*

链接
[this is a link](http://name.com)

图片
![](http://name.com/pic.jpg)

无序列表
- item1
- item2

有序列表
1. item1
2. item2

换行
  \n  (建议\n前后分别加2个空格)