python调用钉钉机器人Webhook进行消息发送:

前置条件

PC端添加机器人

群设置→机器人→添加机器人→添加机器人→自定义→加签

记录access_token地址:

https://oapi.dingtalk.com/robot/send?access_token=<YOUR_ACCESS_TOKEN>

在Webhook可复制如上地址,记录下<YOUR_ACCESS_TOKEN>位置的内容为access_token

在打钩的加签右侧复制秘钥,记录为secret

想要了解详细使用请阅读此链接

代码实现

import time
import hmac  # 用于生成 HMAC 签名
import hashlib  # 用于 SHA256 哈希计算
import base64  # 用于 Base64 编码
import urllib.parse  # 用于 URL 编码
import json  # 用于 JSON 序列化
import requests  # 用于发送 HTTP 请求

class DingTalkBot:
    def __init__(self, bots_info: list):
        """
        初始化 DingTalkBot 类。
        
        :param bots_info: 包含多个机器人的信息,每个元素是一个元组 (access_token, secret)。
        """
        self.bots_info = bots_info

    def _generate_sign(self, secret: str):
        """
        生成签名和时间戳。
        
        :param secret: 机器人的 secret 值。
        :return: 时间戳和签名的元组 (timestamp, sign)。
        """
        timestamp = str(round(time.time() * 1000))  # 获取当前时间戳,单位为毫秒
        secret_enc = secret.encode('utf-8')  # 对 secret 进行 UTF-8 编码
        string_to_sign = '{}\n{}'.format(timestamp, secret)  # 生成待签名字符串
        string_to_sign_enc = string_to_sign.encode('utf-8')  # 对待签名字符串进行 UTF-8 编码
        hmac_code = hmac.new(secret_enc, string_to_sign_enc, digestmod=hashlib.sha256).digest()  # 生成 HMAC-SHA256 签名
        sign = urllib.parse.quote_plus(base64.b64encode(hmac_code))  # 对签名进行 Base64 编码并 URL 编码
        return timestamp, sign

    def _send_message(self, message: dict):
        """
        向所有机器人发送消息。
        
        :param message: 要发送的消息内容,格式为字典。
        :return: 所有消息发送成功返回 True,否则返回 False。
        """
        success = True  # 标记是否所有消息都发送成功
        for access_token, secret in self.bots_info:
            if not access_token or not secret:
                continue  # 跳过无效的机器人信息
            timestamp, sign = self._generate_sign(secret)  # 生成时间戳和签名
            webhook_url = f"https://oapi.dingtalk.com/robot/send?access_token={access_token}"  # 拼接 webhook URL
            url_with_sign = f"{webhook_url}&timestamp={timestamp}&sign={sign}"  # 添加签名和时间戳到 URL
            headers = {
                "Content-Type": "application/json"
            }
            response = requests.post(url_with_sign, data=json.dumps(message), headers=headers)  # 发送 HTTP POST 请求
            if response.json().get('errcode', 1) != 0:
                success = False  # 如果返回结果中 errcode 不为 0,则标记为发送失败
        return success

    def send_text(self, content: str, at_mobiles: list = []):
        """
        发送文本消息。
        
        :param content: 文本消息的内容。
        :param at_mobiles: 要 @ 的手机号列表。
        :return: 所有消息发送成功返回 True,否则返回 False。
        """
        message = {
            "msgtype": "text",
            "text": {
                "content": content
            },
            "at": {
                "atMobiles": at_mobiles,
                "isAtAll": True  # 是否 @ 所有人
            }
        }
        return self._send_message(message)

    def send_markdown(self, title: str, text: str, at_mobiles: list = []):
        """
        发送 Markdown 消息。
        
        :param title: Markdown 消息的标题。
        :param text: Markdown 消息的内容。
        :param at_mobiles: 要 @ 的手机号列表。
        :return: 所有消息发送成功返回 True,否则返回 False。
        """
        message = {
            "msgtype": "markdown",
            "markdown": {
                "title": title,
                "text": text
            },
            "at": {
                "atMobiles": at_mobiles,
                "isAtAll": True  # 是否 @ 所有人
            }
        }
        return self._send_message(message)

# 使用示例:
bots_info = [
    ("access_token_1", "secret_1"),
    ("access_token_2", "secret_2"),
    # 可以添加更多的机器人信息
]

bot = DingTalkBot(bots_info)
bot.send_text("你好,这是一个测试消息。")
bot.send_markdown("Markdown 标题", "这是 **Markdown** 消息。")

闲来无事,记录琐事