Python配置钉钉机器人发送信息

webhook

1. 概述

Webhook 是一个 API 概念，并且变得越来越流行。我们能用事件描述的事物越多，Webhook 的作用范围也就越大。Webhook 作为一个轻量的事件处理应用，正变得越来越有用。

准确的说 Webhook 是一种 web 回调或者 http 的 push API，是向 APP 或者其他应用提供实时信息的一种方式。Webhook 在数据产生时立即发送数据，也就是你能实时收到数据。这一种不同于典型的 API，需要用了实时性需要足够快的轮询。这无论是对生产还是对消费者都是高效的，唯一的缺点是初始建立困难。Webhook 有时也被称为反向 API，因为他提供了 API 规则，你需要设计要使用的 API。Webhook 将向你的应用发起 http 请求，典型的是 post 请求，应用程序由请求驱动。

2. 使用 Webhook

消费一个 Webhook 是为 Webhook 准备一个 URL，用于 Webhook 发送请求。这些通常由后台页面和或者 API 完成。这就意味你的应用要设置一个通过公网可以访问的 URL。多数 Webhook 以两种数据格式发布数据：JSON 或者 XML。另一种数据格式是 application/x-www-form-urlencoded or multipart/form-data。这两种方式都很容易解析，并且多数的 Web 应用架构都可以做这部分工作。

3. Webhook 调试

调试 Webhook 有时很复杂，因为 Webhook 原则来说是异步的。你首先要触发它，然后等待，接着检查是否有响应，枯燥并且相当低效。幸运的是还有其他方法：

明白 webhook 能提供什么，使用如 RequestBin 之类的工具收集 webhook 的请求；
用 cURL 或者 Postman 来模拟请求；
用 ngrok 这样的工具测试你的代码；
用 Runscope 工具来查看整个流程。

4. Webhook 安全

因为 Webhook 发送数据到应用上公开的 URL，这就给其他人找到这个 URL 并且发送错误数据的机会。你可采用技术手段，防止这样的事情发生。最简单的方法是采用 https（TLS connection）。除了使用 https 外，还可以采用以下的方法进一步提高安全性：

首先增加 Token，这个大多数 Webhook 都支持；
增加认证；
数据签名。

5. 重要的问题

当作为 Webhook 的消费者时有两件事需要铭记于心：

Webhook 通过请求发送数据到你的应用后，就不再关注这些数据。也就是说如果你的应用存在问题，数据会丢失。许多 Webhook 会处理回应，如果程序出现错误会重传数据。如果你的应用处理这个请求并且依然返回一个错误，你的应用就会收到重复数据。
Webhook 会发出大量的请求，这样会造成你的应用阻塞。确保你的应用能处理这些请求。

• 实例脚本：

import requests
import json


class drobot:


    def dingmessage(argv):
        """

        :rtype: object
        """
        # 请求的URL，WebHook地址
        webhook = "https://oapi.dingtalk.com/robot/send?access_token=******"
        # 构建请求头部
        header = {
            "Content-Type": "application/json",
            "Charset": "UTF-8"
        }
        # 构建请求数据
        # 利用sys.argv传递参数
        tex = argv
        #print(tex)
        message = {

            "msgtype": "text",
            "text": {
                "content": tex
            },
            "at": {

                "isAtAll": True
            }

        }
        # 对请求的数据进行json封装
        message_json = json.dumps(message)
        # 发送请求
        info = requests.post(url=webhook, data=message_json, headers=header)
        # 打印返回的结果
        print(info.text)
        pass

# if __name__ == "__main__":
#     dingmessage()

设置机器人安全认证方式：签名

• 实例代码：

import requests
import json
import time
import urllib
import hmac
import hashlib
import base64

webhook = "https://oapi.dingtalk.com/robot/send?access_token=***********"
#设置签名的方法
timestamp = int(round(time.time() * 1000))
#签名的安全码
secret = 'SEC8a36**************'
secret_enc = secret.encode('utf-8')
string_to_sign = '{}\n{}'.format(timestamp, secret)
string_to_sign_enc = string_to_sign.encode('utf-8')
hmac_code = hmac.new(secret_enc, string_to_sign_enc, digestmod=hashlib.sha256).digest()
sign = urllib.parse.quote_plus(base64.b64encode(hmac_code))
#字符串拼接URL
url = f"{webhook}&timestamp={timestamp}&sign={sign}"

#print(url)
# 请求的URL，WebHook地址
#secret = "SEC8a36416413f1b44b6aa68d6e7c43a55af405bf635592bca87f43f3424ba130b5"

# 构建请求头部
header = {
    "Content-Type": "application/json",
    "Charset": "UTF-8"
}
class drobot:


    def dingmessage(argv):
        # 构建请求数据
        # 利用sys.argv传递参数
        tex = argv
        print(tex)
        message = {

            "msgtype": "text",
            "text": {
                "content": tex
            },
            "at": {

                "isAtAll": True
            }

        }
        # 对请求的数据进行json封装
        message_json = json.dumps(message)
        # 发送请求
        info = requests.post(url=url, data=message_json, headers=header)
        # 打印返回的结果
        #print(info.text)
        pass

# if __name__ == "__main__":
#     dingmessage()

    def dd_send_mark(S1="", S2="", S3=""):
        # 增加S1\S2\S3进行参数传递
        # 利用参数传递
        message = {

            "msgtype": "markdown",
            "markdown": {
                "title": "%s" % S1,
                "text": "### %s\n" % S1 +
                        "> #### %s\n\n\n" % S2 +
                        "> #### %s\n" % S3
            },
            "at": {

                "isAtAll": True
            }

        }
        # 对请求的数据进行json封装
        message_json = json.dumps(message)
        # 发送请求
        info = requests.post(url=url, data=message_json, headers=header)
        # 打印返回的结果
        print(info.text)
        pass