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}×tamp={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** 消息。")