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
#! /usr/bin/env python
# -*- coding: utf-8 -*-
# __author__ = "chenwei"
# Date: 2018/11/15
import json
from django.conf import settings
from gm_types.ascle import ERROR
from gm_upload import IMG_TYPE, upload_with_short
from utils.base import APIView
from django.contrib.auth.hashers import make_password
from utils.download import download_img_func
from utils.logger import error_logger
from utils.pic_tools import tailor_image
class UserListView(APIView):
def get(self, request):
pictorial_id = request.GET.get('pictorial_id', '')
gender = request.GET.get('gender', None)
page = int(request.GET.get('page', 1))
limit = int(request.GET.get('limit', 10))
filters = self.handle_filter(request.GET.get('filter', '{}'))
if pictorial_id:
filters.update({'pictorial_id': pictorial_id})
if gender:
filters.update({'gender': gender})
try:
data = self.rpc['venus/sun/user/list'](offset=(page-1)*limit, limit=limit, filters=filters).unwrap()
except Exception as e:
error_logger.error(u'获取用户列表失败%s', e)
raise
return data
def post(self, request):
ids = json.loads(request.POST.get('ids', '[]'))
updates = json.loads(request.POST.get('updates', '{}'))
try:
self.rpc['venus/sun/user/batch/update'](updates=updates, ids=ids).unwrap()
except Exception as e:
error_logger.error(u'批量更新user列表失败%s', e)
raise
return {
"message": "更新成功"
}
class UserUpdateOrCreate(APIView):
def get(self, request):
id = request.GET.get('id')
try:
data = self.rpc['venus/sun/user/get'](id=id).unwrap()
except Exception as e:
error_logger.error(u'获取%s用户详情失败%s'%(id, e))
raise
return data
def post(self, request):
id = request.POST.get('id', '')
user_id = request.POST.get('user_id', '')
tag_ids = list(map(lambda x: x.split(":")[0], json.loads(request.POST.get('tags', '[]'))))
password = make_password(settings.PUPPET_PASSWORD, None, 'pbkdf2_sha256')
body_esthetics_tag_ids = list(
map(lambda x: x.split(':')[0], json.loads(request.POST.get('body_esthetics_tags', '[]'))))
expert_tags = json.loads(request.POST.get('expert_tags', '[]'))
try:
expert_tag_ids = [[int(item[0].split(":")[0]), int(item[1])] for item in expert_tags]
except:
error_logger.error(u'创建/编辑%s用户详情失败%s' % (id, e))
raise
data = {
'user_id': user_id,
'password': password,
'is_recommend': int(request.POST.get('is_recommend')) if request.POST.get('is_recommend') else 0,
'profile_pic': request.POST.get('avatar')[:-2],
'nick_name': request.POST.get('nick_name'),
'tag_ids': tag_ids,
'is_puppet': int(request.POST.get('is_puppet')),
'city_id': request.POST.get('city', '').split(":")[0],
'phone': request.POST.get('phone'),
'email': request.POST.get('email'),
'gender': request.POST.get('gender'),
'is_tcc': int(request.POST.get('is_tcc', 0)),
'body_esthetics_tag_ids': body_esthetics_tag_ids,
'expert_tag_ids': expert_tag_ids,
'is_kol': int(request.POST.get('is_kol', 0)),
'is_edit': int(request.POST.get('is_edit', 0)),
}
try:
self.rpc['venus/sun/user/edit'](id=id, data=data).unwrap()
except Exception as e:
error_logger.error(u'创建/编辑%s用户详情失败%s' % (id, e))
raise
return {
'message': '更新成功'
}
class PictorialUserList(APIView):
def get(self, request):
id_ = request.GET.get('pictorial_id')
offset = int(request.GET.get('page', 0))
count = int(request.GET.get('limit', 10))
filters = {'pictorial_id': id_, 'is_online': True}
try:
data = self.rpc['venus/sun/pictorial/user/list'](filters=filters, offset=(offset - 1) * count, count=count).unwrap()
except Exception as e:
error_logger.error(u'获取画报用户列表失败%s' , e)
raise
return data
class UserImage(APIView):
def post(self, request):
image_id = request.POST.get('image_id', '')
image_url = request.POST.get('image_url', '')
user_id = request.POST.get('user_id', '')
if not all([image_id, image_url, user_id]):
error_logger.error(u'参数不完整')
return self.write_fail(code=ERROR.ARG_MISS, message='参数不完整')
resp = self.rpc['venus/sun/user/check_image'](
user_id=user_id, image_id=image_id).unwrap()
if resp:
return self.write_success(data='', message='操作成功')
# 目前不需要 从图片中切出人脸
# try:
# data = self.rpc['saber/face/detect'](url=image_url).unwrap()
# except Exception as e:
# error_logger.error(u'分析人脸失败：%s', e)
# return self.write_fail(code=-1, message='分析人脸失败')
# faces = [{
# "point": [
# item["rect"]["left"],
# item["rect"]["top"],
# item["rect"]["right"]-item["rect"]["left"],
# item["rect"]["bottom"]-item["rect"]["top"]]} for item in data.get("faces", [])]
# if faces:
# point = faces[0].get('point', [])
# else:
# return self.write_fail(code=-1, message='获取人脸失败')
# file_path = download_img_func(image_url, user_id, suffix="png")
# images = tailor_image(file_path, [point])
# header_pic_path = images[0]
# with open(header_pic_path, 'rb') as f:
# image_url = upload_with_short(f, img_type=IMG_TYPE.USEREXTRA)[0]
try:
result = self.rpc['venus/sun/user/create_update_image'](
image_url=image_url, user_id=user_id, image_id=image_id).unwrap()
except Exception as e:
raise e
return self.write_fail(
code=-1 if result else 0,
message="成功" if result else "失败"
)
def get(self, request):
user_id = request.GET.get('user_id')
offset = int(request.GET.get('page', 1))
count = int(request.GET.get('limit', 10))
if not user_id:
error_logger.error(u'参数不完整')
return self.write_fail(code=ERROR.ARG_MISS, message='参数不完整')
try:
data = self.rpc['venus/sun/user/user_image_list'](
user_id=user_id, offset=(offset - 1) * count, limit=count).unwrap()
except Exception as e:
error_logger.error(u'获取%s用户头像失败%s' % (user_id, e))
raise
return data