场景：测试完成后，通过钉钉提示测试团队测试已完成并输送测试报告地址
安装 pip install DingtalkChatbot
附上DingtalkChatbot源码：DingtalkChatbot提供了几种发送类型。可以自行选择自己想要的类型
#!/usr/bin/env python
# _*_ coding:utf-8 _*_
# create time: 07/01/2018 11:35
__author__ = 'Devin -- http://zhangchuzhao.site'
import re
import sys
import json
import time
import logging
import requests
import urllib
import hmac
import base64
import hashlib
import queue
_ver = sys.version_info
is_py3 = (_ver[0] == 3)
try:
quote_plus = urllib.parse.quote_plus
except AttributeError:
quote_plus = urllib.quote_plus
try:
JSONDecodeError = json.decoder.JSONDecodeError
except AttributeError:
JSONDecodeError = ValueError
def is_not_null_and_blank_str(content):
"""
非空字符串
:param content: 字符串
:return: 非空 - True，空 - False
>>> is_not_null_and_blank_str('')
False
>>> is_not_null_and_blank_str(' ')
False
>>> is_not_null_and_blank_str(' ')
False
>>> is_not_null_and_blank_str('123')
True
"""
if content and content.strip():
return True
else:
return False
class DingtalkChatbot(object):
"""
钉钉群自定义机器人（每个机器人每分钟最多发送20条），支持文本（text）、连接（link）、markdown三种消息类型！
"""
def __init__(self, webhook, secret=None, pc_slide=False, fail_notice=False):
"""
机器人初始化
:param webhook: 钉钉群自定义机器人webhook地址
:param secret: 机器人安全设置页面勾选“加签”时需要传入的密钥
:param pc_slide: 消息链接打开方式，默认False为浏览器打开，设置为True时为PC端侧边栏打开
:param fail_notice: 消息发送失败提醒，默认为False不提醒，开发者可以根据返回的消息发送结果自行判断和处理
"""
super(DingtalkChatbot, self).__init__()
self.headers = {'Content-Type': 'application/json; charset=utf-8'}
self.queue = queue.Queue(20) # 钉钉官方限流每分钟发送20条信息
self.webhook = webhook
self.secret = secret
self.pc_slide = pc_slide
self.fail_notice = fail_notice
self.start_time = time.time() # 加签时，请求时间戳与请求时间不能超过1小时，用于定时更新签名
if self.secret is not None and self.secret.startswith('SEC'):
self.update_webhook()
def update_webhook(self):
"""
钉钉群自定义机器人安全设置加签时，签名中的时间戳与请求时不能超过一个小时，所以每个1小时需要更新签名
"""
if is_py3:
timestamp = round(self.start_time * 1000)
string_to_sign = '{}\n{}'.format(timestamp, self.secret)
hmac_code = hmac.new(self.secret.encode(), string_to_sign.encode(), digestmod=hashlib.sha256).digest()
else:
timestamp = long(round(self.start_time * 1000))
secret_enc = bytes(self.secret).encode('utf-8')
string_to_sign = '{}\n{}'.format(timestamp, self.secret)
string_to_sign_enc = bytes(string_to_sign).encode('utf-8')
hmac_code = hmac.new(secret_enc, string_to_sign_enc, digestmod=hashlib.sha256).digest()
sign = quote_plus(base64.b64encode(hmac_code))
self.webhook = '{}×tamp={}&sign={}'.format(self.webhook, str(timestamp), sign)
def msg_open_type(self, url):
"""
消息链接的打开方式
1、默认或不设置时，为浏览器打开：pc_slide=False
2、在PC端侧边栏打开：pc_slide=True
"""
encode_url = quote_plus(url)
if self.pc_slide:
final_link = 'dingtalk://dingtalkclient/page/link?url={}&pc_slide=true'.format(encode_url)
else:
final_link = 'dingtalk://dingtalkclient/page/link?url={}&pc_slide=false'.format(encode_url)
return final_link
def send_text(self, msg, is_at_all=False, at_mobiles=[], at_dingtalk_ids=[], is_auto_at=True):
"""
text类型
:param msg: 消息内容
:param is_at_all: @所有人时：true，否则为false（可选）
:param at_mobiles: 被@人的手机号（注意：可以在msg内容里自定义@手机号的位置，也支持同时@多个手机号，可选）
:param at_dingtalk_ids: 被@人的dingtalkId（可选）
:param is_auto_at: 是否自动在msg内容末尾添加@手机号，默认自动添加，可设置为False取消（可选）
:return: 返回消息发送结果
"""
data = {"msgtype": "text", "at": {}}
if is_not_null_and_blank_str(msg):
data["text"] = {"content": msg}
else:
logging.error("text类型，消息内容不能为空！")
raise ValueError("text类型，消息内容不能为空！")
if is_at_all:
data["at"]["isAtAll"] = is_at_all
if at_mobiles:
at_mobiles = list(map(str, at_mobiles))
data["at"]["atMobiles"] = at_mobiles
if is_auto_at:
mobiles_text = '\n@' + '@'.join(at_mobiles)
data["text"]["content"] = msg + mobiles_text
if at_dingtalk_ids:
at_dingtalk_ids = list(map(str, at_dingtalk_ids))
data["at"]["atDingtalkIds"] = at_dingtalk_ids
logging.debug('text类型：%s' % data)
return self.post(data)
def send_image(self, pic_url):
"""
image类型（表情）
:param pic_url: 图片链接
:return: 返回消息发送结果
"""
if is_not_null_and_blank_str(pic_url):
data = {
"msgtype": "image",
"image": {
"picURL": pic_url
}
}
logging.debug('image类型：%s' % data)
return self.post(data)
else:
logging.error("image类型中图片链接不能为空！")
raise ValueError("image类型中图片链接不能为空！")
def send_link(self, title, text, message_url, pic_url=''):
"""
link类型
:param title: 消息标题
:param text: 消息内容（如果太长自动省略显示）
:param message_url: 点击消息触发的URL
:param pic_url: 图片URL（可选）
:return: 返回消息发送结果
"""
if all(map(is_not_null_and_blank_str, [title, text, message_url])):
data = {
"msgtype": "link",
"link": {
"text": text,
"title": title,
"picUrl": pic_url,
"messageUrl": self.msg_open_type(message_url)
}
}
logging.debug('link类型：%s' % data)
return self.post(data)
else:
logging.error("link类型中消息标题或内容或链接不能为空！")
raise ValueError("link类型中消息标题或内容或链接不能为空！")
def send_markdown(self, title, text, is_at_all=False, at_mobiles=[], at_dingtalk_ids=[], is_auto_at=True):
"""
markdown类型
:param title: 首屏会话透出的展示内容
:param text: markdown格式的消息内容
:param is_at_all: @所有人时：true，否则为：false（可选）
:param at_mobiles: 被@人的手机号（默认自动添加在text内容末尾，可取消自动化添加改为自定义设置，可选）
:param at_dingtalk_ids: 被@人的dingtalkId（可选）
:param is_auto_at: 是否自动在text内容末尾添加@手机号，默认自动添加，可设置为False取消（可选）
:return: 返回消息发送结果
"""
if all(map(is_not_null_and_blank_str, [title, text])):
# 给Mardown文本消息中的跳转链接添加上跳转方式
text = re.sub(r'(?<!!)\[.*?\]\((.*?)\)', lambda m: m.group(0).replace(m.group(1), self.msg_open_type(m.group(1))), text)
data = {
"msgtype": "markdown",
"markdown": {
"title": title,
"text": text
},
"at": {}
}
if is_at_all:
data["at"]["isAtAll"] = is_at_all
if at_mobiles:
at_mobiles = list(map(str, at_mobiles))
data["at"]["atMobiles"] = at_mobiles
if is_auto_at:
mobiles_text = '\n@' + '@'.join(at_mobiles)
data["markdown"]["text"] = text + mobiles_text
if at_dingtalk_ids:
at_dingtalk_ids = list(map(str, at_dingtalk_ids))
data["at"]["atDingtalkIds"] = at_dingtalk_ids
logging.debug("markdown类型：%s" % data)
return self.post(data)
else:
logging.error("markdown类型中消息标题或内容不能为空！")
raise ValueError("markdown类型中消息标题或内容不能为空！")
def send_action_card(self, action_card):
"""
ActionCard类型
:param action_card: 整体跳转ActionCard类型实例或独立跳转ActionCard类型实例
:return: 返回消息发送结果
"""
if isinstance(action_card, ActionCard):
data = action_card.get_data()
if "singleURL" in data["actionCard"]:
data["actionCard"]["singleURL"] = self.msg_open_type(data["actionCard"]["singleURL"])
elif "btns" in data["actionCard"]:
for btn in data["actionCard"]["btns"]:
btn["actionURL"] = self.msg_open_type(btn["actionURL"])
logging.debug("ActionCard类型：%s" % data)
return self.post(data)
else:
logging.error("ActionCard类型：传入的实例类型不正确，内容为：{}".format(str(action_card)))
raise TypeError("ActionCard类型：传入的实例类型不正确，内容为：{}".format(str(action_card)))
def send_feed_card(self, links):
"""
FeedCard类型
:param links: FeedLink实例列表 or CardItem实例列表
:return: 返回消息发送结果
"""
if not isinstance(links, list):
logging.error("FeedLink类型：传入的数据格式不正确，内容为：{}".format(str(links)))
raise ValueError("FeedLink类型：传入的数据格式不正确，内容为：{}".format(str(links)))
link_list = []
for link in links:
# 兼容：1、传入FeedLink实例列表；2、CardItem实例列表；
if isinstance(link, FeedLink) or isinstance(link, CardItem):
link = link.get_data()
link['messageURL'] = self.msg_open_type(link['messageURL'])
link_list.append(link)
else:
logging.error("FeedLink类型，传入的数据格式不正确，内容为：{}".format(str(link)))
raise ValueError("FeedLink类型，传入的数据格式不正确，内容为：{}".format(str(link)))
data = {"msgtype": "feedCard", "feedCard": {"links": link_list}}
logging.debug("FeedCard类型：%s" % data)
return self.post(data)
def post(self, data):
"""
发送消息（内容UTF-8编码）
:param data: 消息数据（字典）
:return: 返回消息发送结果
"""
now = time.time()
# 钉钉自定义机器人安全设置加签时，签名中的时间戳与请求时不能超过一个小时，所以每个1小时需要更新签名
if now - self.start_time >= 3600 and self.secret is not None and self.secret.startswith('SEC'):
self.start_time = now
self.update_webhook()
# 钉钉自定义机器人现在每分钟最多发送20条消息
self.queue.put(now)
if self.queue.full():
elapse_time = now - self.queue.get()
if elapse_time < 60:
sleep_time = int(60 - elapse_time) + 1
logging.debug('钉钉官方限制机器人每分钟最多发送20条，当前发送频率已达限制条件，休眠 {}s'.format(str(sleep_time)))
time.sleep(sleep_time)
try:
post_data = json.dumps(data)
response = requests.post(self.webhook, headers=self.headers, data=post_data)
except requests.exceptions.HTTPError as exc:
logging.error("消息发送失败， HTTP error: %d, reason: %s" % (exc.response.status_code, exc.response.reason))
raise
except requests.exceptions.ConnectionError:
logging.error("消息发送失败，HTTP connection error!")
raise
except requests.exceptions.Timeout:
logging.error("消息发送失败，Timeout error!")
raise
except requests.exceptions.RequestException:
logging.error("消息发送失败, Request Exception!")
raise
else:
try:
result = response.json()
except JSONDecodeError:
logging.error("服务器响应异常，状态码：%s，响应内容：%s" % (response.status_code, response.text))
return {'errcode': 500, 'errmsg': '服务器响应异常'}
else:
logging.debug('发送结果：%s' % result)
# 消息发送失败提醒（errcode 不为 0，表示消息发送异常），默认不提醒，开发者可以根据返回的消息发送结果自行判断和处理
if self.fail_notice and result.get('errcode', True):
time_now = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(time.time()))
error_data = {
"msgtype": "text",
"text": {
"content": "[注意-自动通知]钉钉机器人消息发送失败，时间：%s，原因：%s，请及时跟进，谢谢!" % (
time_now, result['errmsg'] if result.get('errmsg', False) else '未知异常')
},
"at": {
"isAtAll": False
}
}
logging.error("消息发送失败，自动通知：%s" % error_data)
requests.post(self.webhook, headers=self.headers, data=json.dumps(error_data))
return result
class ActionCard(object):
"""
ActionCard类型消息格式（整体跳转、独立跳转）
"""
def __init__(self, title, text, btns, btn_orientation=0, hide_avatar=0):
"""
ActionCard初始化
:param title: 首屏会话透出的展示内容
:param text: markdown格式的消息
:param btns: 按钮列表：（1）按钮数量为1时，整体跳转ActionCard类型；（2）按钮数量大于1时，独立跳转ActionCard类型；
:param btn_orientation: 0：按钮竖直排列，1：按钮横向排列（可选）
:param hide_avatar: 0：正常发消息者头像，1：隐藏发消息者头像（可选）
"""
super(ActionCard, self).__init__()
self.title = title
self.text = text
self.btn_orientation = btn_orientation
self.hide_avatar = hide_avatar
btn_list = []
for btn in btns:
if isinstance(btn, CardItem):
btn_list.append(btn.get_data())
if btn_list:
btns = btn_list # 兼容：1、传入CardItem示例列表；2、传入数据字典列表
self.btns = btns
def get_data(self):
"""
获取ActionCard类型消息数据（字典）
:return: 返回ActionCard数据
"""
if all(map(is_not_null_and_blank_str, [self.title, self.text])) and len(self.btns):
if len(self.btns) == 1:
# 整体跳转ActionCard类型
data = {
"msgtype": "actionCard",
"actionCard": {
"title": self.title,
"text": self.text,
"hideAvatar": self.hide_avatar,
"btnOrientation": self.btn_orientation,
"singleTitle": self.btns[0]["title"],
"singleURL": self.btns[0]["actionURL"]
}
}
return data
else:
# 独立跳转ActionCard类型
data = {
"msgtype": "actionCard",
"actionCard": {
"title": self.title,
"text": self.text,
"hideAvatar": self.hide_avatar,
"btnOrientation": self.btn_orientation,
"btns": self.btns
}
}
return data
else:
logging.error("ActionCard类型，消息标题或内容或按钮数量不能为空！")
raise ValueError("ActionCard类型，消息标题或内容或按钮数量不能为空！")
class FeedLink(object):
"""
FeedCard类型单条消息格式
"""
def __init__(self, title, message_url, pic_url):
"""
初始化单条消息文本
:param title: 单条消息文本
:param message_url: 点击单条信息后触发的URL
:param pic_url: 点击单条消息后面图片触发的URL
"""
super(FeedLink, self).__init__()
self.title = title
self.message_url = message_url
self.pic_url = pic_url
def get_data(self):
"""
获取FeedLink消息数据（字典）
:return: 本FeedLink消息的数据
"""
if all(map(is_not_null_and_blank_str, [self.title, self.message_url, self.pic_url])):
data = {
"title": self.title,
"messageURL": self.message_url,
"picURL": self.pic_url
}
return data
else:
logging.error("FeedCard类型单条消息文本、消息链接、图片链接不能为空！")
raise ValueError("FeedCard类型单条消息文本、消息链接、图片链接不能为空！")
class CardItem(object):
"""
ActionCard和FeedCard消息类型中的子控件
注意：
1、发送FeedCard消息时，参数pic_url必须传入参数值；
2、发送ActionCard消息时，参数pic_url不需要传入参数值；
"""
def __init__(self, title, url, pic_url=None):
"""
CardItem初始化
@param title: 子控件名称
@param url: 点击子控件时触发的URL
@param pic_url: FeedCard的图片地址，ActionCard时不需要，故默认为None
"""
super(CardItem, self).__init__()
self.title = title
self.url = url
self.pic_url = pic_url
def get_data(self):
"""
获取CardItem子控件数据（字典）
@return: 子控件的数据
"""
if all(map(is_not_null_and_blank_str, [self.title, self.url, self.pic_url])):
# FeedCard类型
data = {
"title": self.title,
"messageURL": self.url,
"picURL": self.pic_url
}
return data
elif all(map(is_not_null_and_blank_str, [self.title, self.url])):
# ActionCard类型
data = {
"title": self.title,
"actionURL": self.url
}
return data
else:
logging.error("CardItem是ActionCard的子控件时，title、url不能为空；是FeedCard的子控件时，title、url、pic_url不能为空！")
raise ValueError("CardItem是ActionCard的子控件时，title、url不能为空；是FeedCard的子控件时，title、url、pic_url不能为空！")
if __name__ == '__main__':
import doctest
doctest.testmod()
新建sendDtb.py
from dingtalkchatbot.chatbot import DingtalkChatbot
class Message():
def messge(self):
report_url = 'http://localhost:63342/result_report/index.html#'
msg = 'app自动化测试脚本执行完成，测试结果请查看测试报告：报告地址为{0}'.format(report_url)
return msg
def sends_text(self):
# WebHook地址
webhook = 'https://oapi.dingtalk.com/robot/send?access_token=xxxxxxxxx'
# 初始化机器人小丁
xiaoding = DingtalkChatbot(webhook)
at_mobiles=['xxx','xxx'] #艾特钉钉群员账号
msg = self.messge()
xiaoding.send_text(msg=msg, is_at_all=False,at_mobiles=at_mobiles)
if __name__ == '__main__':
Message().sends_text()
webhook:由钉钉pc端机器人管理添加机器人生成：
image.png
网友评论