390 lines
14 KiB
Python
390 lines
14 KiB
Python
#!/usr/bin/env python
|
||
# create time: 07/01/2018 11:35
|
||
__author__ = 'Devin -- http://zhangchuzhao.site'
|
||
|
||
import json
|
||
import logging
|
||
import time
|
||
|
||
import requests
|
||
|
||
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:
|
||
"""
|
||
钉钉群自定义机器人（每个机器人每分钟最多发送20条），支持文本（text）、连接（link）、markdown三种消息类型！
|
||
"""
|
||
|
||
def __init__(self, webhook):
|
||
"""
|
||
机器人初始化
|
||
:param webhook: 钉钉群自定义机器人webhook地址
|
||
"""
|
||
super().__init__()
|
||
self.headers = {'Content-Type': 'application/json; charset=utf-8'}
|
||
self.webhook = webhook
|
||
self.times = 0
|
||
self.start_time = time.time()
|
||
|
||
def send_text(self, msg, is_at_all=False, at_mobiles=[], at_dingtalk_ids=[]):
|
||
"""
|
||
text类型
|
||
:param msg: 消息内容
|
||
:param is_at_all: @所有人时：true，否则为false（可选）
|
||
:param at_mobiles: 被@人的手机号（可选）
|
||
:param at_dingtalk_ids: 被@人的dingtalkId（可选）
|
||
: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 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 is_not_null_and_blank_str(title) and is_not_null_and_blank_str(text) and is_not_null_and_blank_str(message_url):
|
||
data = {
|
||
'msgtype': 'link',
|
||
'link': {
|
||
'text': text,
|
||
'title': title,
|
||
'picUrl': pic_url,
|
||
'messageUrl': 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=[]):
|
||
"""
|
||
markdown类型
|
||
:param title: 首屏会话透出的展示内容
|
||
:param text: markdown格式的消息内容
|
||
:param is_at_all: 被@人的手机号（在text内容里要有@手机号，可选）
|
||
:param at_mobiles: @所有人时：true，否则为：false（可选）
|
||
:param at_dingtalk_ids: 被@人的dingtalkId（可选）
|
||
:return: 返回消息发送结果
|
||
"""
|
||
if is_not_null_and_blank_str(title) and is_not_null_and_blank_str(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 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()
|
||
logging.debug('ActionCard类型：%s' % data)
|
||
return self.post(data)
|
||
else:
|
||
logging.error('ActionCard类型：传入的实例类型不正确！')
|
||
raise TypeError('ActionCard类型：传入的实例类型不正确！')
|
||
|
||
def send_feed_card(self, links):
|
||
"""
|
||
FeedCard类型
|
||
:param links: 信息集（FeedLink数组）
|
||
:return: 返回消息发送结果
|
||
"""
|
||
link_data_list = []
|
||
for link in links:
|
||
if isinstance(link, FeedLink) or isinstance(link, CardItem):
|
||
link_data_list.append(link.get_data())
|
||
if link_data_list:
|
||
# 兼容：1、传入FeedLink或CardItem实例列表；2、传入数据字典列表；
|
||
links = link_data_list
|
||
data = {'msgtype': 'feedCard', 'feedCard': {'links': links}}
|
||
logging.debug('FeedCard类型：%s' % data)
|
||
return self.post(data)
|
||
|
||
def post(self, data):
|
||
"""
|
||
发送消息（内容UTF-8编码）
|
||
:param data: 消息数据（字典）
|
||
:return: 返回发送结果
|
||
"""
|
||
self.times += 1
|
||
if self.times % 20 == 0:
|
||
if time.time() - self.start_time < 60:
|
||
logging.debug('钉钉官方限制每个机器人每分钟最多发送20条，当前消息发送频率已达到限制条件，休眠一分钟')
|
||
time.sleep(60)
|
||
self.start_time = time.time()
|
||
|
||
post_data = json.dumps(data)
|
||
try:
|
||
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)
|
||
if result['errcode']:
|
||
error_data = {'msgtype': 'text', 'text': {
|
||
'content': '钉钉机器人消息发送失败，原因：%s' % result['errmsg']}, 'at': {'isAtAll': True}}
|
||
logging.error('消息发送失败，自动通知：%s' % error_data)
|
||
requests.post(self.webhook, headers=self.headers,
|
||
data=json.dumps(error_data))
|
||
return result
|
||
|
||
|
||
class ActionCard:
|
||
"""
|
||
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().__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 is_not_null_and_blank_str(self.title) and is_not_null_and_blank_str(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:
|
||
"""
|
||
FeedCard类型单条消息格式
|
||
"""
|
||
|
||
def __init__(self, title, message_url, pic_url):
|
||
"""
|
||
初始化单条消息文本
|
||
:param title: 单条消息文本
|
||
:param message_url: 点击单条信息后触发的URL
|
||
:param pic_url: 点击单条消息后面图片触发的URL
|
||
"""
|
||
super().__init__()
|
||
self.title = title
|
||
self.message_url = message_url
|
||
self.pic_url = pic_url
|
||
|
||
def get_data(self):
|
||
"""
|
||
获取FeedLink消息数据（字典）
|
||
:return: 本FeedLink消息的数据
|
||
"""
|
||
if is_not_null_and_blank_str(self.title) and is_not_null_and_blank_str(self.message_url) and is_not_null_and_blank_str(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:
|
||
"""
|
||
ActionCard和FeedCard消息类型中的子控件
|
||
"""
|
||
|
||
def __init__(self, title, url, pic_url=None):
|
||
"""
|
||
CardItem初始化
|
||
@param title: 子控件名称
|
||
@param url: 点击子控件时触发的URL
|
||
@param pic_url: FeedCard的图片地址，ActionCard时不需要，故默认为None
|
||
"""
|
||
super().__init__()
|
||
self.title = title
|
||
self.url = url
|
||
self.pic_url = pic_url
|
||
|
||
def get_data(self):
|
||
"""
|
||
获取CardItem子控件数据（字典）
|
||
@return: 子控件的数据
|
||
"""
|
||
if is_not_null_and_blank_str(self.pic_url) and is_not_null_and_blank_str(self.title) and is_not_null_and_blank_str(self.url):
|
||
# FeedCard类型
|
||
data = {
|
||
'title': self.title,
|
||
'messageURL': self.url,
|
||
'picURL': self.pic_url
|
||
}
|
||
return data
|
||
elif is_not_null_and_blank_str(self.title) and is_not_null_and_blank_str(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()
|