-
Notifications
You must be signed in to change notification settings - Fork 0
/
WeChatComponent.py
183 lines (169 loc) · 6.41 KB
/
WeChatComponent.py
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
import redis
import logging
import requests
from urllib.parse import urlencode
from . import conf
logger = logging.getLogger()
class ComponentWechat(object):
"""公众号授权"""
def get_component_access_token(self):
"""获取component_access_token"""
pool = redis.ConnectionPool(host='localhost', port=6379)
redis_conn = redis.Redis(connection_pool=pool)
component_access_token = redis_conn.get('component_access_token')
if component_access_token:
component_access_token = component_access_token.decode()
return component_access_token
else:
component_verify_ticket = redis_conn.get('component_verify_ticket')
if component_verify_ticket:
component_verify_ticket = component_verify_ticket.decode()
else:
return None
url = 'https://api.weixin.qq.com/cgi-bin/component/api_component_token'
data = {
"component_appid": conf.component_appid,
"component_appsecret": conf.component_APPSECRET,
"component_verify_ticket": component_verify_ticket
}
resp = requests.post(url, json=data).json()
# print(resp,'===========component_access_token=======req')
try:
component_access_token = resp['component_access_token']
except:
return None
# 保存到redis conf.EXPIRES redis_conn.setex('test_key',EXPIRES,'component_access_token')
redis_conn.setex("component_access_token", conf.EXPIRES, component_access_token)
return component_access_token
def get_pre_auth_code(self):
"""获取预授权码"""
component_access_token = self.get_component_access_token()
if not component_access_token:
return None
url = 'https://api.weixin.qq.com/cgi-bin/component/api_create_preauthcode?'
params = {
'component_access_token':component_access_token
}
postdata = {"component_appid": conf.component_appid}
params = urlencode(params)
url = url + params
resp = requests.post(url, json=postdata).json()
try:
pre_auth_code = resp['pre_auth_code']
except:
return None
return pre_auth_code
def get_authorization_qr_url(self, task_id):
"""获取授权二维码链接"""
pre_auth_code = self.get_pre_auth_code()
# print(pre_auth_code, '================pre_auth_code')
if not pre_auth_code:
return None
# 二维码授权
url = 'https://mp.weixin.qq.com/cgi-bin/componentloginpage?'
# 跳转链接直接授权
# url = 'https://mp.weixin.qq.com/safe/bindcomponent?'
params = {
# 'action':'bindcomponent',
# 'auth_type': 3,
# 'no_scan':1,
'component_appid':conf.component_appid,
'pre_auth_code':pre_auth_code,
'redirect_uri':conf.component_redirect_uri+'?task_id='+task_id,
}
params = urlencode(params)
url = url + params
return url
def get_authorization_url(self):
"""获取授权链接"""
pre_auth_code = self.get_pre_auth_code()
# print(pre_auth_code, '================pre_auth_code')
if not pre_auth_code:
return None
# 二维码授权
# url = 'https://mp.weixin.qq.com/cgi-bin/componentloginpage?'
# 跳转链接直接授权
url = 'https://mp.weixin.qq.com/safe/bindcomponent?'
params = {
'action':'bindcomponent',
'auth_type': 3,
'no_scan':1,
'component_appid':conf.component_appid,
'pre_auth_code':pre_auth_code,
'redirect_uri':conf.component_redirect_uri,
}
params = urlencode(params)
url = url + params
return url
def get_authorization_info(self,authorization_code):
"""获取授权信息"""
component_access_token = self.get_component_access_token()
if not component_access_token:
return None
url = 'https://api.weixin.qq.com/cgi-bin/component/api_query_auth?'
params = {
'component_access_token':component_access_token
}
postdata = {
'component_appid':conf.component_appid,
'authorization_code':authorization_code
}
params = urlencode(params)
url = url + params
resp = requests.post(url, data=json.dumps(postdata)).json()
try:
authorization_info = resp['authorization_info']
except:
return None
return authorization_info
def api_get_authorizer_info(self,authorizer_appid):
"""获取授权公众号信息"""
component_access_token = self.get_component_access_token()
if not component_access_token:
return None
url = 'https://api.weixin.qq.com/cgi-bin/component/api_get_authorizer_info?'
params = {
'component_access_token':component_access_token
}
postdata = {
'component_appid':conf.component_appid,
'authorizer_appid':authorizer_appid
}
params = urlencode(params)
url = url + params
resp = requests.post(url, json=postdata).json()
try:
authorizer_info = resp['authorizer_info']
except:
return None
return authorizer_info
@staticmethod
def save_component_verify_ticket(xml_tree):
"""
保存component_verify_ticket
:param xml_tree: 解析的xml对象
:return:
"""
tick = xml_tree.find("ComponentVerifyTicket")
try:
if tick.text:
# 内容保存到redis中
pool = redis.ConnectionPool(host='localhost', port=6379)
redis_conn = redis.Redis(connection_pool=pool)
redis_connect.set('component_verify_ticket', tick.text)
except Exception as e:
logger.error(e)
@staticmethod
def update_authorized_info(xml_tree,info_type):
"""
更新授权信息
:param xml_tree:
:return:
"""
AuthorizerAppid = xml_tree.find('AuthorizerAppid')
# 取消授权
if info_type == 'unauthorized':
try:
print('取消授权')
except Exception as e:
logger.error(e)