123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402 |
- import base64
- import json
- from io import BytesIO
- from flask import request, jsonify, Response, current_app
- from flask_restx import Resource, Namespace, reqparse
- from sqlalchemy import insert, select, update, delete, func
- from sqlalchemy.orm import Session
- from werkzeug.datastructures import FileStorage
- from app.defines import StatesCode, Module, OperationType
- from app.modle.users import User
- from app.utils.jwt_util import login_required
- from app.utils.save_log import save_log
- from app.utils.util import to_dict, MyXlwt
- ns = Namespace('users', description='用户管理接口')
- get_users = reqparse.RequestParser(bundle_errors=True)
- get_users.add_argument(name='status', type=float, required=True, location='args', help='用户状态')
- get_users.add_argument(name='page_size', type=int, location='args', required=False, help='每页记录数量,默认:20')
- get_users.add_argument(name='page', type=int, location='args', required=False, help='第几页')
- @ns.route('/user_list')
- class GetUserListApi(Resource):
- method_decorators = [login_required]
- @ns.doc(id='get_users_list', description='获取用户列表')
- @ns.expect(get_users)
- def get(self):
- """获取用户列表"""
- status = request.args.get('status')
- page_size = int(request.args.get('page_size', 20))
- page = int(request.args.get('page', 1))
- if status is None:
- return jsonify(code=StatesCode.PARA_ERROR, message='用户状态不能为空')
- try:
- with Session(current_app.engine) as session:
- count = select(func.count(User.id)).where(User.account_status == status)
- count_results = session.execute(count).scalars().first()
- stmt = select(User).where(User.account_status == status).offset(page_size * (page - 1)).limit(page_size)
- results = session.execute(stmt).scalars().all()
- save_log(request, Module.USER, OperationType.INQUIRE, StatesCode.SUCCESS)
- return jsonify(code=StatesCode.SUCCESS, message="成功", total=count_results, data=to_dict(results))
- except Exception as e:
- save_log(request, Module.USER, OperationType.INQUIRE, StatesCode.UNKNOWN_ERROR)
- return jsonify(code=StatesCode.UNKNOWN_ERROR, message=str(e))
- add_user = reqparse.RequestParser(bundle_errors=True)
- add_user.add_argument(name='id', type=str, required=False, location='form', help='用户id')
- add_user.add_argument(name='username', type=str, required=True, location='form', help='用户名')
- add_user.add_argument(name='password', type=str, required=True, location='form', help='密码')
- add_user.add_argument(name='photograph', type=FileStorage, required=False, location='files', help='照片')
- add_user.add_argument(name='name', type=str, required=False, location='form', help='姓名')
- add_user.add_argument(name='phone', type=str, required=False, location='form', help='电话')
- add_user.add_argument(name='email', type=str, required=False, location='form', help='邮箱')
- add_user.add_argument(name='company', type=str, required=False, location='form', help='公司')
- add_user.add_argument(name='department', type=str, required=False, location='form', help='部门')
- add_user.add_argument(name='role', type=str, required=False, location='form', help='角色')
- add_user.add_argument(name='permission', type=str, required=False, location='form', help='权限')
- add_user.add_argument(name='on_job_status', type=str, required=False, location='form', help='在职状态')
- add_user.add_argument(name='duty', type=str, required=False, location='form', help='职务')
- add_user.add_argument(name='nationality', type=str, required=False, location='form', help='国籍')
- user_details = reqparse.RequestParser(bundle_errors=True)
- user_details.add_argument(name='id', type=int, required=True, location='args', help='用户id')
- @ns.route('/user')
- class UsersApi(Resource):
- method_decorators = [login_required]
- @ns.doc(id='get_users_details', description='获取用户详情')
- @ns.expect(user_details)
- def get(self):
- """获取用户详情"""
- user_id = request.args.get('id')
- if user_id is None:
- return jsonify(code=StatesCode.PARA_ERROR, message="用户id不能为空")
- try:
- with Session(current_app.engine) as session:
- stmt = select(User).where(User.id == user_id)
- results = session.execute(stmt).scalars().all()
- save_log(request, Module.USER, OperationType.INQUIRE, StatesCode.SUCCESS)
- return jsonify(code=StatesCode.SUCCESS, message='成功', data=to_dict(results))
- except Exception as e:
- save_log(request, Module.USER, OperationType.INQUIRE, StatesCode.UNKNOWN_ERROR)
- return jsonify(code=StatesCode.UNKNOWN_ERROR, message=str(e))
- @ns.doc(id='add_users', description='添加用户')
- @ns.expect(add_user)
- def post(self):
- """添加用户"""
- username = request.form.get('username')
- password = request.form.get('password')
- photograph = request.files.get('photograph')
- name = request.form.get('name')
- phone = request.form.get('phone')
- email = request.form.get('email')
- company = request.form.get('company')
- department = request.form.get('department')
- role = request.form.get('role')
- permission = request.form.get('permission')
- on_job_status = request.form.get('on_job_status')
- duty = request.form.get('duty')
- nationality = request.form.get('nationality')
- if username is None or password is None:
- return jsonify(code=StatesCode.PARA_ERROR, message="用户名或密码不能为空")
- if photograph:
- ext = photograph.filename.split('.')[-1]
- base64_data = b'data:image/%s;base64,' % ext.encode('utf-8') + base64.b64encode(photograph.read())
- base64_data = base64_data.decode('utf-8')
- else:
- base64_data = None
- try:
- with Session(current_app.engine) as session:
- # 判断用户是否存在
- stmt = select(User).where(User.user_name == username)
- result = session.execute(stmt).scalars().first()
- if result:
- return jsonify(code=StatesCode.UNKNOWN_ERROR, message="用户已存在")
- # 添加用户
- session.execute(
- insert(User).values(
- user_name=username,
- password=User().generate_password(password),
- photograph=base64_data,
- name=name,
- phone=phone,
- email=email,
- company=company,
- department=department,
- role=role,
- permission=permission,
- on_job_status=on_job_status,
- duty=duty,
- nationality=nationality,
- )
- )
- session.commit()
- save_log(request, Module.USER, OperationType.ADD, StatesCode.SUCCESS)
- return jsonify(code=StatesCode.SUCCESS, message="添加用户成功")
- except Exception as e:
- save_log(request, Module.USER, OperationType.ADD, StatesCode.UNKNOWN_ERROR)
- return jsonify(code=StatesCode.UNKNOWN_ERROR, message=str(e))
- @ns.doc(id='update_users_details', description='更新用户详情')
- @ns.expect(add_user)
- def put(self):
- """更新用户详情"""
- user_id = request.form.get('id')
- username = request.form.get('username')
- password = request.form.get('password')
- photograph = request.files.get('photograph')
- name = request.form.get('name')
- phone = request.form.get('phone')
- email = request.form.get('email')
- company = request.form.get('company')
- department = request.form.get('department')
- role = request.form.get('role')
- permission = request.form.get('permission')
- on_job_status = request.form.get('on_job_status')
- duty = request.form.get('duty')
- nationality = request.form.get('nationality')
- if user_id is None:
- return jsonify(code=StatesCode.PARA_ERROR, message='用户id不能为空')
- if photograph:
- ext = photograph.filename.split('.')[-1]
- base64_data = b'data:image/%s;base64,' % ext.encode('utf-8') + base64.b64encode(photograph.read())
- base64_data = base64_data.decode('utf-8')
- else:
- base64_data = None
- try:
- with Session(current_app.engine) as session:
- stmt = update(User).where(User.id == user_id).values(
- user_name=username,
- password=User().generate_password(password),
- photograph=base64_data,
- name=name,
- phone=phone,
- email=email,
- company=company,
- department=department,
- role=role,
- permission=permission,
- on_job_status=on_job_status,
- duty=duty,
- nationality=nationality,
- )
- session.execute(stmt)
- session.commit()
- save_log(request, Module.USER, OperationType.UPDATE, StatesCode.SUCCESS)
- return jsonify(code=StatesCode.SUCCESS, message='修改成功')
- except Exception as e:
- save_log(request, Module.USER, OperationType.UPDATE, StatesCode.UNKNOWN_ERROR)
- return jsonify(code=StatesCode.UNKNOWN_ERROR, message=str(e))
- @ns.doc(id='delete_users', description='删除用户')
- @ns.expect(add_user)
- def delete(self):
- """删除用户"""
- user_id = request.form.get('id')
- if user_id is None:
- return jsonify(code=StatesCode.PARA_ERROR, message="用户id不能为空")
- try:
- with Session(current_app.engine) as session:
- stmt = delete(User).where(User.id == user_id)
- session.execute(stmt)
- session.commit()
- save_log(request, Module.USER, OperationType.DELETE, StatesCode.SUCCESS)
- return jsonify(code=StatesCode.SUCCESS, message="删除成功")
- except Exception as e:
- save_log(request, Module.USER, OperationType.DELETE, StatesCode.UNKNOWN_ERROR)
- return jsonify(code=StatesCode.UNKNOWN_ERROR, message=str(e))
- users_status = reqparse.RequestParser(bundle_errors=True)
- users_status.add_argument(name='id', type=int, required=True, location='form', help='用户id')
- users_status.add_argument(name='status', type=int, required=True, location='form', help='用户状态')
- @ns.route('/user_status')
- class GetUsersApi(Resource):
- method_decorators = [login_required]
- @ns.doc(id='modify_users_status', description='修改用户状态,激活、禁用')
- @ns.expect(users_status)
- def put(self):
- """修改用户状态,激活、禁用"""
- user_id = int(request.form.get('id'))
- account_status = int(request.form.get('status'))
- if user_id is None or account_status is None:
- return jsonify(code=StatesCode.PARA_ERROR, message="用户id或用户状态不能为空")
- try:
- with Session(current_app.engine) as session:
- stmt = update(User).where(User.id == user_id).values(account_status=account_status)
- session.execute(stmt)
- session.commit()
- save_log(request, Module.USER, OperationType.UPDATE, StatesCode.SUCCESS)
- return jsonify(code=StatesCode.SUCCESS, message="修改用户状态成功")
- except Exception as e:
- save_log(request, Module.USER, OperationType.UPDATE, StatesCode.UNKNOWN_ERROR)
- return jsonify(code=StatesCode.UNKNOWN_ERROR, message=str(e))
- batch_delete_user = reqparse.RequestParser(bundle_errors=True)
- batch_delete_user.add_argument(name='users_id', type=str, required=True, location='form', help='用户id列表')
- @ns.route('/batch_delete_user')
- class BatchDeleteUserApi(Resource):
- method_decorators = [login_required]
- @ns.doc(id='batch_delete_users', description='批量删除用户')
- @ns.expect(batch_delete_user)
- def delete(self):
- """批量删除用户"""
- users_id = request.form.get('users_id')
- if users_id is not None:
- users_id = json.loads(users_id)
- else:
- return jsonify(code=StatesCode.PARA_ERROR, message='用户id不能为空')
- try:
- with Session(current_app.engine) as session:
- stmt = select(User).where(User.id.in_(users_id))
- results = session.execute(stmt).scalars().all()
- for result in results:
- session.delete(result)
- session.commit()
- save_log(request, Module.USER, OperationType.BATCH_DELETE, StatesCode.SUCCESS)
- return jsonify(code=StatesCode.SUCCESS, message='批量删除成功')
- except Exception as e:
- save_log(request, Module.USER, OperationType.BATCH_DELETE, StatesCode.UNKNOWN_ERROR)
- return jsonify(code=StatesCode.UNKNOWN_ERROR, message=str(e))
- batch_modify_user_status = reqparse.RequestParser(bundle_errors=True)
- batch_modify_user_status.add_argument(name='users_id', type=list, required=True, location='form', help='用户id列表')
- batch_modify_user_status.add_argument(name='status', type=int, required=True, location='form',
- help='用户状态 0为正常,1为禁用')
- @ns.route('/batch_modify_user_status')
- class BatchModifyUsersStatusApi(Resource):
- method_decorators = [login_required]
- @ns.doc(id='batch_modify_user_status', description='批量修改用户状态,激活、禁用')
- @ns.expect(batch_modify_user_status)
- def put(self):
- """批量修改用户状态,激活、禁用"""
- users_id = request.form.get('users_id')
- account_status = request.form.get('status')
- if users_id is not None and account_status is not None:
- users_id = json.loads(users_id)
- else:
- return jsonify(code=StatesCode.PARA_ERROR, message='用户id和状态不能为空')
- try:
- values = []
- for user_id in users_id:
- values.append({"id": user_id, "account_status": account_status})
- with Session(current_app.engine) as session:
- session.execute(
- update(User),
- values
- )
- session.commit()
- save_log(request, Module.USER, OperationType.BATCH_UPDATE, StatesCode.SUCCESS)
- return jsonify(code=StatesCode.SUCCESS, message='批量修改用户状态成功')
- except Exception as e:
- save_log(request, Module.USER, OperationType.BATCH_UPDATE, StatesCode.UNKNOWN_ERROR)
- return jsonify(code=StatesCode.UNKNOWN_ERROR, message=str(e))
- @ns.route('/export_data')
- class ExportDataApi(Resource):
- method_decorators = [login_required]
- @ns.doc(id='export_data', description='导出用户数据')
- def get(self):
- """导出用户数据"""
- try:
- with Session(current_app.engine) as session:
- stmt = select(User)
- results = session.execute(stmt).scalars().all()
- my_xlwt = MyXlwt()
- title = ['id', 'user_name', 'password', 'name', 'phone', 'email', 'company', 'department',
- 'role',
- 'permission', 'on_job_status', 'duty', 'account_status', 'nationality', 'register_time',
- 'common_menus']
- # 写入标题
- my_xlwt.write_row(0, 0, title)
- # 写入内容
- start_row = 1 # 内容起始行
- for result in results:
- data = [result.id, result.user_name, result.password, result.name, result.phone,
- result.email, result.company, result.department, result.role, result.permission,
- result.on_job_status,
- result.duty, result.account_status, result.nationality, result.register_time,
- result.common_menus]
- my_xlwt.write_row(start_row, 0, data)
- start_row += 1
- output = BytesIO()
- my_xlwt.save(output)
- response = Response(output.getvalue())
- response.headers['Content-Type'] = 'application/octet-stream'
- response.headers['Content-Disposition'] = 'attachment; filename=%s' % 'users.xls'
- save_log(request, Module.USER, OperationType.EXPORT, StatesCode.SUCCESS)
- return response
- except Exception as e:
- save_log(request, Module.USER, OperationType.EXPORT, StatesCode.UNKNOWN_ERROR)
- return jsonify(code=StatesCode.UNKNOWN_ERROR, message=str(e))
|