1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
|
import time
import hmac
import hashlib
import base64
import urllib.parse
import requests
import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
class DDingWebHookPush:
def __init__(self, access_token, secret):
self.access_token = access_token
self.secret = secret
self.headers = {"Content-Type": "application/json ;charset=utf-8 "}
@staticmethod
def initialize_markdown_template(title):
"""
markdown格式，支持markdown语法
:param title:
:return:
"""
return {
"msgtype": "markdown",
"markdown": {
"title": title,
"text": ''
},
"at": {
"isAtAll": False
}
}
@staticmethod
def initialize_text_template():
"""
普通文本类消息
:return:
"""
return {
"msgtype": "text",
"text": {
"content": "",
},
"at": {
"isAtAll": False
}
}
@staticmethod
def initialize_link_template(title, messageUrl, picUrl=None):
"""
超链接类型
:param messageUrl: 跳转链接
:param title: 图片链接，可为空
:return:
"""
return {
"msgtype": "link",
"link": {
"text": "",
"title": title,
"picUrl": picUrl if picUrl else "",
"messageUrl": messageUrl
}
}
@staticmethod
def initialize_actionCard_template(title, singleTitle="阅读全文", singleURL=None, btns=None):
"""
卡片类型
:param btns:
:param singleURL: 跳转地址
:param singleTitle: 跳转链接绑定的文字
:param title:
:return:
"""
if btns:
return {
"msgtype": "actionCard",
"actionCard": {
"text": "",
"title": title,
"btnOrientation": "0",
"btns": btns
}
}
return {
"msgtype": "actionCard",
"actionCard": {
"text": "",
"title": title,
"btnOrientation": "0",
"singleURL": singleURL if singleURL else "",
"singleTitle": singleTitle
}
}
@staticmethod
def initialize_feedCard_template(links: list):
"""
卡片类型
:param links: [{"title": "", "messageURL": "", "picURL": ""}]
:return:
"""
return {
"msgtype": "actionCard",
"feedCard": {
"links": links
}
}
def makeSign(self):
"""
加签
:return:
"""
timestamp = str(round(time.time() * 1000))
secret_enc = self.secret.encode('utf-8')
string_to_sign = '{}\n{}'.format(timestamp, self.secret)
string_to_sign_enc = string_to_sign.encode('utf-8')
hmac_code = hmac.new(secret_enc, string_to_sign_enc, digestmod=hashlib.sha256).digest()
sign = urllib.parse.quote_plus(base64.b64encode(hmac_code))
return timestamp, sign
def setContent(self, template, __content):
"""
内容主体
:param template: 内容模版
:param __content: 内容
:return:
"""
msgtype = template.get("msgtype")
if msgtype == 'markdown':
template.get("markdown").update({'text': __content})
elif msgtype == 'text':
template.get("text").update({'content': __content})
elif msgtype == 'link':
template.get("link").update({'text': __content})
elif msgtype == 'actionCard':
template.get("actionCard").update({'text': __content})
else:
raise Exception("this type could not set content!")
def setAtmobiles(self, template, mobiles: list):
"""
通过手机号来找到群里人进行@，若群内无此人@的是手机号，有此人则正常@人昵称
:param template:
:param mobiles:['','']
:return:
"""
template.update({"at": {'atMobiles': mobiles}})
def setAtByUserIds(self, template, userIds: list):
"""
通过手机号来找到群里人进行@，若群内无此人@的是手机号，有此人则正常@人昵称
:param template:
:param userIds:['','']
:return:
"""
template.update({"at": {'atUserIds': userIds}})
def setAtAll(self, template):
"""
@ 所有人
:param template:
:return:
"""
template["at"] = {'isAtAll': True}
def sendMsg(self, template):
"""
发送消息
:param template:
:return:
"""
if not self.access_token or not self.secret:
return False
timestamp, sign = self.makeSign()
url = "https://oapi.dingtalk.com/robot/send?access_token=%s×tamp=%s&sign=%s" % (self.access_token, timestamp, sign)
res = requests.post(url, json=template, headers=self.headers, verify=False)
return res.json()
|
