1 import time 2 import hmac 3 import hashlib 4 import base64 5 import urllib 6 import json 7 import requests 8 import logging 9 10 try: 11 JSONDecodeError = json.decoder.JSONDecodeError 12 except AttributeError: 13 JSONDecodeError = ValueError 14 15 16 def is_not_null_and_blank_str(content): 17 if content and content.strip(): 18 return True 19 else: 20 return False 21 22 23 class DingtalkRobot(object): 24 def __init__(self, webhook, sign=None): 25 super(DingtalkRobot, self).__init__() 26 self.webhook = webhook 27 self.sign = sign 28 self.headers = {'Content-Type': 'application/json; charset=utf-8'} 29 self.times = 0 30 self.start_time = time.time() 31 32 # 加密簽名 33 def __spliceUrl(self): 34 timestamp = int(round(time.time() * 1000)) 35 secret = self.sign 36 secret_enc = secret.encode('utf-8') 37 string_to_sign = '{}\n{}'.format(timestamp, secret) 38 string_to_sign_enc = string_to_sign.encode('utf-8') 39 hmac_code = hmac.new(secret_enc, string_to_sign_enc, digestmod=hashlib.sha256).digest() 40 sign = urllib.parse.quote_plus(base64.b64encode(hmac_code)) 41 url = f"{self.webhook}×tamp={timestamp}&sign={sign}" 42 return url 43 44 def send_text(self, msg, is_at_all=False, at_mobiles=[]): 45 data = {"msgtype": "text", "at": {}} 46 if is_not_null_and_blank_str(msg): 47 data["text"] = {"content": msg} 48 else: 49 logging.error("text類型，消息內容不能為空！") 50 raise ValueError("text類型，消息內容不能為空！") 51 52 if is_at_all: 53 data["at"]["isAtAll"] = is_at_all 54 55 if at_mobiles: 56 at_mobiles = list(map(str, at_mobiles)) 57 data["at"]["atMobiles"] = at_mobiles 58 59 logging.debug('text類型：%s' % data) 60 return self.__post(data) 61 62 def __post(self, data): 63 """ 64 發送消息（內容UTF-8編碼） 65 :param data: 消息數據（字典） 66 :return: 返回發送結果 67 """ 68 self.times += 1 69 if self.times > 20: 70 if time.time() - self.start_time < 60: 71 logging.debug('釘釘官方限制每個機器人每分鍾最多發送20條，當前消息發送頻率已達到限制條件，休眠一分鍾') 72 time.sleep(60) 73 self.start_time = time.time() 74 75 post_data = json.dumps(data) 76 try: 77 response = requests.post(self.__spliceUrl(), headers=self.headers, data=post_data) 78 except requests.exceptions.HTTPError as exc: 79 logging.error("消息發送失敗， HTTP error: %d, reason: %s" % (exc.response.status_code, exc.response.reason)) 80 raise 81 except requests.exceptions.ConnectionError: 82 logging.error("消息發送失敗，HTTP connection error!") 83 raise 84 except requests.exceptions.Timeout: 85 logging.error("消息發送失敗，Timeout error!") 86 raise 87 except requests.exceptions.RequestException: 88 logging.error("消息發送失敗, Request Exception!") 89 raise 90 else: 91 try: 92 result = response.json() 93 except JSONDecodeError: 94 logging.error("服務器響應異常，狀態碼：%s，響應內容：%s" % (response.status_code, response.text)) 95 return {'errcode': 500, 'errmsg': '服務器響應異常'} 96 else: 97 logging.debug('發送結果：%s' % result) 98 if result['errcode']: 99 error_data = {"msgtype": "text", "text": {"content": "釘釘機器人消息發送失敗，原因：%s" % result['errmsg']}, 100 "at": {"isAtAll": True}} 101 logging.error("消息發送失敗，自動通知：%s" % error_data) 102 requests.post(self.webhook, headers=self.headers, data=json.dumps(error_data)) 103 return result 104 105 106 if __name__ == '__main__': 107 URL = "你的釘釘機器人地址" 108 SIGN = "簽名" 109 ding = DingtalkRobot(URL, SIGN) 110 print(ding.send_text("Hello World"))
做通知實例，如圖：
