場(chǎng)景:測(cè)試完成后,通過(guò)釘釘提示測(cè)試團(tuán)隊(duì)測(cè)試已完成并輸送測(cè)試報(bào)告地址
安裝 pip install DingtalkChatbot
附上DingtalkChatbot源碼:DingtalkChatbot提供了幾種發(fā)送類型【欤可以自行選擇自己想要的類型
#!/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):
"""
釘釘群自定義機(jī)器人(每個(gè)機(jī)器人每分鐘最多發(fā)送20條)叹坦,支持文本(text)、連接(link)卑雁、markdown三種消息類型募书!
"""
def __init__(self, webhook, secret=None, pc_slide=False, fail_notice=False):
"""
機(jī)器人初始化
:param webhook: 釘釘群自定義機(jī)器人webhook地址
:param secret: 機(jī)器人安全設(shè)置頁(yè)面勾選“加簽”時(shí)需要傳入的密鑰
:param pc_slide: 消息鏈接打開方式,默認(rèn)False為瀏覽器打開测蹲,設(shè)置為True時(shí)為PC端側(cè)邊欄打開
:param fail_notice: 消息發(fā)送失敗提醒莹捡,默認(rèn)為False不提醒,開發(fā)者可以根據(jù)返回的消息發(fā)送結(jié)果自行判斷和處理
"""
super(DingtalkChatbot, self).__init__()
self.headers = {'Content-Type': 'application/json; charset=utf-8'}
self.queue = queue.Queue(20) # 釘釘官方限流每分鐘發(fā)送20條信息
self.webhook = webhook
self.secret = secret
self.pc_slide = pc_slide
self.fail_notice = fail_notice
self.start_time = time.time() # 加簽時(shí)扣甲,請(qǐng)求時(shí)間戳與請(qǐng)求時(shí)間不能超過(guò)1小時(shí)篮赢,用于定時(shí)更新簽名
if self.secret is not None and self.secret.startswith('SEC'):
self.update_webhook()
def update_webhook(self):
"""
釘釘群自定義機(jī)器人安全設(shè)置加簽時(shí),簽名中的時(shí)間戳與請(qǐng)求時(shí)不能超過(guò)一個(gè)小時(shí)琉挖,所以每個(gè)1小時(shí)需要更新簽名
"""
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启泣、默認(rèn)或不設(shè)置時(shí),為瀏覽器打開:pc_slide=False
2示辈、在PC端側(cè)邊欄打開: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: 消息內(nèi)容
:param is_at_all: @所有人時(shí):true寥茫,否則為false(可選)
:param at_mobiles: 被@人的手機(jī)號(hào)(注意:可以在msg內(nèi)容里自定義@手機(jī)號(hào)的位置,也支持同時(shí)@多個(gè)手機(jī)號(hào)矾麻,可選)
:param at_dingtalk_ids: 被@人的dingtalkId(可選)
:param is_auto_at: 是否自動(dòng)在msg內(nèi)容末尾添加@手機(jī)號(hào)纱耻,默認(rèn)自動(dòng)添加,可設(shè)置為False取消(可選)
:return: 返回消息發(fā)送結(jié)果
"""
data = {"msgtype": "text", "at": {}}
if is_not_null_and_blank_str(msg):
data["text"] = {"content": msg}
else:
logging.error("text類型射富,消息內(nèi)容不能為空膝迎!")
raise ValueError("text類型,消息內(nèi)容不能為空胰耗!")
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: 返回消息發(fā)送結(jié)果
"""
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: 消息標(biāo)題
:param text: 消息內(nèi)容(如果太長(zhǎng)自動(dòng)省略顯示)
:param message_url: 點(diǎn)擊消息觸發(fā)的URL
:param pic_url: 圖片URL(可選)
:return: 返回消息發(fā)送結(jié)果
"""
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類型中消息標(biāo)題或內(nèi)容或鏈接不能為空柴灯!")
raise ValueError("link類型中消息標(biāo)題或內(nèi)容或鏈接不能為空卖漫!")
def send_markdown(self, title, text, is_at_all=False, at_mobiles=[], at_dingtalk_ids=[], is_auto_at=True):
"""
markdown類型
:param title: 首屏?xí)捦赋龅恼故緝?nèi)容
:param text: markdown格式的消息內(nèi)容
:param is_at_all: @所有人時(shí):true,否則為:false(可選)
:param at_mobiles: 被@人的手機(jī)號(hào)(默認(rèn)自動(dòng)添加在text內(nèi)容末尾赠群,可取消自動(dòng)化添加改為自定義設(shè)置羊始,可選)
:param at_dingtalk_ids: 被@人的dingtalkId(可選)
:param is_auto_at: 是否自動(dòng)在text內(nèi)容末尾添加@手機(jī)號(hào),默認(rèn)自動(dòng)添加查描,可設(shè)置為False取消(可選)
:return: 返回消息發(fā)送結(jié)果
"""
if all(map(is_not_null_and_blank_str, [title, text])):
# 給Mardown文本消息中的跳轉(zhuǎn)鏈接添加上跳轉(zhuǎn)方式
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類型中消息標(biāo)題或內(nèi)容不能為空突委!")
raise ValueError("markdown類型中消息標(biāo)題或內(nèi)容不能為空柏卤!")
def send_action_card(self, action_card):
"""
ActionCard類型
:param action_card: 整體跳轉(zhuǎn)ActionCard類型實(shí)例或獨(dú)立跳轉(zhuǎn)ActionCard類型實(shí)例
:return: 返回消息發(fā)送結(jié)果
"""
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類型:傳入的實(shí)例類型不正確,內(nèi)容為:{}".format(str(action_card)))
raise TypeError("ActionCard類型:傳入的實(shí)例類型不正確匀油,內(nèi)容為:{}".format(str(action_card)))
def send_feed_card(self, links):
"""
FeedCard類型
:param links: FeedLink實(shí)例列表 or CardItem實(shí)例列表
:return: 返回消息發(fā)送結(jié)果
"""
if not isinstance(links, list):
logging.error("FeedLink類型:傳入的數(shù)據(jù)格式不正確缘缚,內(nèi)容為:{}".format(str(links)))
raise ValueError("FeedLink類型:傳入的數(shù)據(jù)格式不正確,內(nèi)容為:{}".format(str(links)))
link_list = []
for link in links:
# 兼容:1敌蚜、傳入FeedLink實(shí)例列表桥滨;2、CardItem實(shí)例列表弛车;
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類型齐媒,傳入的數(shù)據(jù)格式不正確,內(nèi)容為:{}".format(str(link)))
raise ValueError("FeedLink類型纷跛,傳入的數(shù)據(jù)格式不正確喻括,內(nèi)容為:{}".format(str(link)))
data = {"msgtype": "feedCard", "feedCard": {"links": link_list}}
logging.debug("FeedCard類型:%s" % data)
return self.post(data)
def post(self, data):
"""
發(fā)送消息(內(nèi)容UTF-8編碼)
:param data: 消息數(shù)據(jù)(字典)
:return: 返回消息發(fā)送結(jié)果
"""
now = time.time()
# 釘釘自定義機(jī)器人安全設(shè)置加簽時(shí),簽名中的時(shí)間戳與請(qǐng)求時(shí)不能超過(guò)一個(gè)小時(shí)忽舟,所以每個(gè)1小時(shí)需要更新簽名
if now - self.start_time >= 3600 and self.secret is not None and self.secret.startswith('SEC'):
self.start_time = now
self.update_webhook()
# 釘釘自定義機(jī)器人現(xiàn)在每分鐘最多發(fā)送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('釘釘官方限制機(jī)器人每分鐘最多發(fā)送20條双妨,當(dāng)前發(fā)送頻率已達(dá)限制條件,休眠 {}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("消息發(fā)送失敗叮阅, HTTP error: %d, reason: %s" % (exc.response.status_code, exc.response.reason))
raise
except requests.exceptions.ConnectionError:
logging.error("消息發(fā)送失敗刁品,HTTP connection error!")
raise
except requests.exceptions.Timeout:
logging.error("消息發(fā)送失敗,Timeout error!")
raise
except requests.exceptions.RequestException:
logging.error("消息發(fā)送失敗, Request Exception!")
raise
else:
try:
result = response.json()
except JSONDecodeError:
logging.error("服務(wù)器響應(yīng)異常浩姥,狀態(tài)碼:%s挑随,響應(yīng)內(nèi)容:%s" % (response.status_code, response.text))
return {'errcode': 500, 'errmsg': '服務(wù)器響應(yīng)異常'}
else:
logging.debug('發(fā)送結(jié)果:%s' % result)
# 消息發(fā)送失敗提醒(errcode 不為 0,表示消息發(fā)送異常)勒叠,默認(rèn)不提醒兜挨,開發(fā)者可以根據(jù)返回的消息發(fā)送結(jié)果自行判斷和處理
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": "[注意-自動(dòng)通知]釘釘機(jī)器人消息發(fā)送失敗,時(shí)間:%s眯分,原因:%s拌汇,請(qǐng)及時(shí)跟進(jìn),謝謝!" % (
time_now, result['errmsg'] if result.get('errmsg', False) else '未知異常')
},
"at": {
"isAtAll": False
}
}
logging.error("消息發(fā)送失敗弊决,自動(dòng)通知:%s" % error_data)
requests.post(self.webhook, headers=self.headers, data=json.dumps(error_data))
return result
class ActionCard(object):
"""
ActionCard類型消息格式(整體跳轉(zhuǎn)噪舀、獨(dú)立跳轉(zhuǎn))
"""
def __init__(self, title, text, btns, btn_orientation=0, hide_avatar=0):
"""
ActionCard初始化
:param title: 首屏?xí)捦赋龅恼故緝?nèi)容
:param text: markdown格式的消息
:param btns: 按鈕列表:(1)按鈕數(shù)量為1時(shí),整體跳轉(zhuǎn)ActionCard類型飘诗;(2)按鈕數(shù)量大于1時(shí)与倡,獨(dú)立跳轉(zhuǎn)ActionCard類型;
:param btn_orientation: 0:按鈕豎直排列昆稿,1:按鈕橫向排列(可選)
:param hide_avatar: 0:正常發(fā)消息者頭像纺座,1:隱藏發(fā)消息者頭像(可選)
"""
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净响、傳入數(shù)據(jù)字典列表
self.btns = btns
def get_data(self):
"""
獲取ActionCard類型消息數(shù)據(jù)(字典)
:return: 返回ActionCard數(shù)據(jù)
"""
if all(map(is_not_null_and_blank_str, [self.title, self.text])) and len(self.btns):
if len(self.btns) == 1:
# 整體跳轉(zhuǎn)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:
# 獨(dú)立跳轉(zhuǎn)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類型少欺,消息標(biāo)題或內(nèi)容或按鈕數(shù)量不能為空!")
raise ValueError("ActionCard類型馋贤,消息標(biāo)題或內(nèi)容或按鈕數(shù)量不能為空狈茉!")
class FeedLink(object):
"""
FeedCard類型單條消息格式
"""
def __init__(self, title, message_url, pic_url):
"""
初始化單條消息文本
:param title: 單條消息文本
:param message_url: 點(diǎn)擊單條信息后觸發(fā)的URL
:param pic_url: 點(diǎn)擊單條消息后面圖片觸發(fā)的URL
"""
super(FeedLink, self).__init__()
self.title = title
self.message_url = message_url
self.pic_url = pic_url
def get_data(self):
"""
獲取FeedLink消息數(shù)據(jù)(字典)
:return: 本FeedLink消息的數(shù)據(jù)
"""
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仁讨、發(fā)送FeedCard消息時(shí)羽莺,參數(shù)pic_url必須傳入?yún)?shù)值;
2洞豁、發(fā)送ActionCard消息時(shí)盐固,參數(shù)pic_url不需要傳入?yún)?shù)值;
"""
def __init__(self, title, url, pic_url=None):
"""
CardItem初始化
@param title: 子控件名稱
@param url: 點(diǎn)擊子控件時(shí)觸發(fā)的URL
@param pic_url: FeedCard的圖片地址丈挟,ActionCard時(shí)不需要刁卜,故默認(rèn)為None
"""
super(CardItem, self).__init__()
self.title = title
self.url = url
self.pic_url = pic_url
def get_data(self):
"""
獲取CardItem子控件數(shù)據(jù)(字典)
@return: 子控件的數(shù)據(jù)
"""
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的子控件時(shí),title曙咽、url不能為空蛔趴;是FeedCard的子控件時(shí),title例朱、url孝情、pic_url不能為空!")
raise ValueError("CardItem是ActionCard的子控件時(shí)洒嗤,title箫荡、url不能為空;是FeedCard的子控件時(shí)渔隶,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自動(dòng)化測(cè)試腳本執(zhí)行完成婉弹,測(cè)試結(jié)果請(qǐng)查看測(cè)試報(bào)告:報(bào)告地址為{0}'.format(report_url)
return msg
def sends_text(self):
# WebHook地址
webhook = 'https://oapi.dingtalk.com/robot/send?access_token=xxxxxxxxx'
# 初始化機(jī)器人小丁
xiaoding = DingtalkChatbot(webhook)
at_mobiles=['xxx','xxx'] #艾特釘釘群?jiǎn)T賬號(hào)
msg = self.messge()
xiaoding.send_text(msg=msg, is_at_all=False,at_mobiles=at_mobiles)
if __name__ == '__main__':
Message().sends_text()
webhook:由釘釘pc端機(jī)器人管理添加機(jī)器人生成:
image.png