import json import json from api.views.base_view import BaseView, get_offset_count from libs.user import user_exists_by_names from alpha_types.venus.error import ERROR as CODES class UpdateGraspStatus(BaseView): """ 更新用户爬取状态 """ def post(self, request): user_id = int(request.POST.get("user_id", 0)) relation_account_id = request.POST.get("relation_account_id", "") error, _data = self.call_rpc( "venus/community/user/update_grasp_status", user_id=user_id, relation_account_id=relation_account_id ) if error: return self.error(error) return self.ok(data=_data) class ShadowUserList(BaseView): ''' 列表获取马甲用户 ''' def get(self, request): offset, count = get_offset_count(request) error, _data = self.call_rpc( "venus/community/user/shadow/list", offset=offset, count=count ) if error: return self.error(error) users = _data.get('users') result = { 'users': [{'user_id': item.get('user_id')} for item in users if item], 'total': _data.get('total') } return self.ok(data=result) class Validate3PartyAccount(BaseView): """ 验证三方账号是否已绑定过 """ def post(self, request): _ids = request.POST.get("account_ids") or '[]' account_ids = json.loads(_ids) err, _data = self.call_rpc( "venus/community/user/validate_has_bind_3party_account", bind_account_ids=account_ids ) if err: return self.error(err) data = {account_id: _data.get(account_id, False) for account_id in account_ids} return self.ok(data=data) class BatchCreateShadowUser(BaseView): """ 批量创建马甲用户 """ def post(self, request): request_data = json.loads(request.POST.get('ins_data', '[]')) print(request_data) err, _data = self.call_rpc( "venus/community/user/batch_create_shadow_user", data=request_data ) if err: return self.error(err) return self.ok(data=_data) class Validate3PartyOrAccount(BaseView): """判断用户是否在三方账号或者alpha账号体系中。""" def post(self, request): try: names = json.loads(request.POST.get("names", "[]")) except: names = [] if not names: return self.error(self.get_ErrorInfo(CODES.PARAMS_INCOMPLETE)) account_users = user_exists_by_names(names=names) users = {name: account_users.get(name, False) for name in names} err, _3party_account = self.call_rpc( "venus/community/user/validate_has_bind_3party_account", bind_account_ids=names ) if err: return self.error(err) data = { name: _3party_account.get(name, False) or users.get(name, False) for name in names } return self.ok(data={ "users": data, }) class BatchUpdateUserLevelFansNum(BaseView): """批量更新用户星级 粉丝数""" def post(self, request): try: user_infos = json.loads(request.POST.get("user_infos", "[]")) except: update_data = [] if not update_data: return self.error(self.get_ErrorInfo(CODES.PARAMS_INCOMPLETE)) err, ret = self.call_rpc( "venus/community/user/batch_update_user_leven_fans", user_infos=user_infos ) if err: return self.error(err) return self.ok()