helpers.py
3.8 KB
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
# _*_ coding: utf-8 _*_
# @Time : 2020/10/7 11:26
# @Author vanwhebin
from django.views.generic import View
from django.core.exceptions import PermissionDenied
from rest_framework.views import APIView
from rest_framework.permissions import AllowAny
from rest_framework_simplejwt.serializers import TokenObtainSerializer, TokenObtainPairSerializer
from libs.qywx.conf import Conf
from libs.qywx.CorpApi import CorpApi, CORP_API_TYPE
from usercenter.models import UserManager, User
from utils.util import response
from wxProject.qywx_settings import Conf
class WxRequiredMixin(View):
""" 验证是否企业微信登录成功；"""
def dispatch(self, request, *args, **kwargs):
# 状态和文章实例有user属性
if self.get_object().user.username != self.request.user.username:
raise PermissionDenied
return super(WxRequiredMixin, self).dispatch(request, *args, **kwargs)
class WxPushHelper:
api = None
conf = None
def __init__(self, conf):
if not conf:
raise ValueError("企业微信配置有误，配置项不能为空")
self.conf = conf
self.api = CorpApi(self.conf['CORP_ID'], self.conf['APP_SECRET'])
def get_corp_user_id_by_code(self, code):
return self.api.httpCall(CORP_API_TYPE['GET_USER_INFO_BY_CODE'], {"CODE": code})
def get_user_info_by_corp_user_id(self, corp_user_id):
return self.api.httpCall(CORP_API_TYPE['USER_GET'], {'userid': corp_user_id})
def push_text(self, user_wx_id, content):
data = {
"agentid": self.conf['APP_ID'], # 企业应用ID
"msgtype": 'text', # 消息类型为文本
"touser": user_wx_id, # 接受消息的对象
"text": {
"content": content # 消息文本
}
}
return self.api.httpCall(CORP_API_TYPE['MESSAGE_SEND'], data)
def push_card(self, user_wx_id, url, description):
data = {
"touser": str(user_wx_id),
"msgtype": "textcard",
"agentid": self.conf['APP_ID'], # 企业应用ID
"textcard": {
"title": "产品立项流程通知",
"description": description,
"url": url,
"btntxt": "点击查看"
},
"enable_id_trans": 0,
"enable_duplicate_check": 0,
"duplicate_check_interval": 1800
}
return self.api.httpCall(CORP_API_TYPE['MESSAGE_SEND'], data)
class WxUserlogin(APIView):
allowed_methods = ['post']
permission_classes = (AllowAny, )
@staticmethod
def _get_user_info(code, app_id):
client = WxPushHelper(Conf[app_id])
corp_user = client.get_corp_user_id_by_code(code)
if "errcode" in corp_user and corp_user['errcode'] == 0 and "UserId" in corp_user:
return client.get_user_info_by_corp_user_id(corp_user['UserId'])
else:
raise RuntimeError(u"获取企业微信用户信息错误")
def post(self, request, *args, **kwargs):
code = request.data.get('code', '').strip()
app_id = request.data.get('state', '').strip()
if not code:
raise PermissionDenied
else:
info = self._get_user_info(code, app_id)
user = User.objects.filter(wx_token=info['userid']).first()
if not user:
user = User.objects.create(
username=info['name'],
email=info['email'],
wx_token=info['userid'],
is_active=True
)
user.set_password("123456")
user.save()
# 获取token
token_obj = TokenObtainPairSerializer.get_token(user)
return response({"access_token": str(token_obj.access_token)})