from flask import Blueprint, request from Modules.AdminUser.UserImpl1 import UserImpl from Modules.AdminUser.UserUtils1 import UserUtils from Utils.ErrorUtil import ReturnConditionCheckFailed, AttrCheckError from Modules.AdminUser.UserImpl import UserLoginImpl, SendLoginVcodeEmailImpl from Modules.AdminUser.UserAuthUtils import verify_token, authority_scope user_route = Blueprint('user', __name__) @user_route.route('/online_check', methods=['GET']) @verify_token def online_check_route(**kwargs): """ 在线检查,检查token是否有效 """ return {"info": "正常"}, 200 @user_route.route('/create', methods=['POST']) # @verify_token # @authority_scope(['admin']) def create_user_route(**kwargs): """新建用户""" try: req = request.json impl = UserImpl() impl.create( req['email'], req['name'], UserUtils.decrypt_data(encrypt_msg=req['pwd']), req['role'] ) return {"info": "用户创建成功"}, 200 except AssertionError as e: return {"info": e.__str__()}, 202 except AttrCheckError as e: return {"info": e.__str__()}, 202 @user_route.route('/login', methods=['POST']) def login_route(): """登录""" try: req = request.json user_login = UserLoginImpl() user_login.email, user_login.pwd, user_login.vcode = req['email'], req['pwd'], req['vcode'] user_login.check_obj() user_login.login() result = user_login.dict_to_show() return {"info": "登录成功", "result": result}, 200 except ReturnConditionCheckFailed as e: e.log_error() return {"info": e.__str__()}, e.status_code @user_route.route('/send_vcode', methods=['POST']) def send_vcode_to_user_route(): """发送验证码""" try: req = request.json obj = SendLoginVcodeEmailImpl() obj.email = req['email'] obj.check_obj() obj.send_vcode_email() return {"info": "发送成功"}, 200 except ReturnConditionCheckFailed as e: e.log_error() return {"info": e.__str__()}, e.status_code # @user_route.route('/list', methods=['POST']) # @verify_token # @authority_scope(['admin']) # def list_user_route(**kwargs): # """ # 用户信息列表接口 # """ # try: # req = request.json # list_user = ListUserImpl( # req['search'], # req['sort'], # req['page_size'], # req['page_no'] # ) # list_user.check_obj() # result = list_user.list() # return {"info": '查询结果', "result": result}, 200 # except ReturnConditionCheckFailed as e: # e.log_error() # return {"info": e.__str__()}, e.status_code # @user_route.route('/disable', methods=['GET']) # @verify_token # @authority_scope(['admin']) # def disable_user_route(**kwargs): # """停用用户接口""" # try: # user = UserManageImpl() # user.uid = request.args['UID'] # user.status = "disable" # user.check_obj(columns=["uid", "status"]) # user.disable() # return {"info": "已停用该用户"}, 200 # except ReturnConditionCheckFailed as e: # e.log_error() # return {"info": e.__str__()}, e.status_code # # # @user_route.route('/active', methods=['GET']) # @verify_token # @authority_scope(['admin']) # def active_user_route(**kwargs): # """激活用户接口""" # try: # user = UserManageImpl() # user.uid = request.args['UID'] # user.status = "normal" # user.check_obj(columns=["uid", "status"]) # user.active() # return {"info": "已激活该用户"}, 200 # except ReturnConditionCheckFailed as e: # e.log_error() # return {"info": e.__str__()}, e.status_code # # # @user_route.route('/delete', methods=['GET']) # @verify_token # @authority_scope(['admin']) # def delete_user_route(**kwargs): # """删除用户接口""" # return {"info": "调整中"}, 200 # # # @user_route.route('/manage_role', methods=['POST']) # @verify_token # @authority_scope(['admin']) # def manage_role_of_user_route(**kwargs): # """管理用户角色接口""" # return {"info": "调整中"}, 200