应用场景：
用于提醒项目构建完成，提醒消息接收
核心代码：
# coding=utf-8
import json
import logging
import time
import hmac
import hashlib
import base64
import urllib
import requests
from manage_system.start.test_settings import WEB_HOOK, DING_SECRET
quote_plus = urllib.quote_plus
# https://oapi.dingtalk.com/robot/send?access_token=ce67d932ba4d62e490c26700cd1a922c474fe72e39a7d88b231e1e52949a531d
def is_not_null_and_blank_str(content):
"""
非空字符串
:param content: 字符串
:return: 非空 - True，空 - False
"""
if content and content.strip():
return True
else:
return False
class DingClient(object):
"""
钉钉群自定义机器人（每个机器人每分钟最多发送20条），支持文本（text）、连接（link）、markdown三种消息类型！
"""
def __init__(self, webhook, secret=None, fail_notice=False):
"""
机器人初始化
:param webhook: 钉钉群自定义机器人webhook地址
:param secret: 机器人安全设置页面勾选“加签”时需要传入的密钥
:param pc_slide: 消息链接打开方式，默认False为浏览器打开，设置为True时为PC端侧边栏打开
:param fail_notice: 消息发送失败提醒，默认为False不提醒，开发者可以根据返回的消息发送结果自行判断和处理
"""
super(DingClient, self).__init__()
self.headers = {'Content-Type': 'application/json; charset=utf-8'}
self.ls = list() # 钉钉官方限流每分钟发送20条信息
self.webhook = webhook
self.secret = secret
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小时需要更新签名
"""
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 send_text(self, msg, is_at_all=False, at_mobiles=None):
"""
text类型
:param msg: 消息内容
:param is_at_all: @所有人时：true，否则为false（可选）
:param at_mobiles: 被@人的手机号（注意：可以在msg内容里自定义@手机号的位置，也支持同时@多个手机号，可选）
:return: 返回消息发送结果
"""
if at_mobiles is None:
at_mobiles = []
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
logging.debug('text类型：%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.ls.append(now)
if len(self.ls) > 20:
# 拿出第一个元素进行比对时间
elapse_time = now - self.ls.pop(0)
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 json.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
def get_ding_instance():
return DingClient(WEB_HOOK, DING_SECRET)
if __name__ == '__main__':
# 环境配置完成后 放于dev_settings中
# WEB_HOOK = "https://oapi.dingtalk.com/robot/send?access_token=************************"
# DING_SECRET = "************************"
# d = DingClient(WEB_HOOK, DING_SECRET)
# d.send_text('测试', at_mobiles=['*****'])
print '测试完成'
