346 lines
10 KiB
Python
346 lines
10 KiB
Python
|
from flask import request, jsonify, current_app
|
||
|
|
from app import db
|
||
|
|
from app.models import Admin
|
||
|
|
from . import api_bp
|
||
|
|
from flask_login import current_user, login_required
|
||
|
|
from werkzeug.security import generate_password_hash
|
||
|
|
import functools
|
||
|
|
|
||
|
|
def require_admin(f):
|
||
|
|
"""管理员权限验证装饰器"""
|
||
|
|
@functools.wraps(f)
|
||
|
|
def decorated_function(*args, **kwargs):
|
||
|
|
# 检查用户是否已认证
|
||
|
|
if not current_user.is_authenticated:
|
||
|
|
return jsonify({
|
||
|
|
'success': False,
|
||
|
|
'message': '需要登录'
|
||
|
|
}), 401
|
||
|
|
|
||
|
|
# 检查是否为超级管理员
|
||
|
|
if not current_user.is_super_admin():
|
||
|
|
return jsonify({
|
||
|
|
'success': False,
|
||
|
|
'message': '需要超级管理员权限'
|
||
|
|
}), 403
|
||
|
|
|
||
|
|
return f(*args, **kwargs)
|
||
|
|
return decorated_function
|
||
|
|
|
||
|
|
def validate_admin_data(data, is_create=True):
|
||
|
|
"""验证管理员数据"""
|
||
|
|
if not data:
|
||
|
|
return False, '请求数据为空'
|
||
|
|
|
||
|
|
current_app.logger.info(f"Validating admin data: {data}, is_create: {is_create}")
|
||
|
|
|
||
|
|
if is_create:
|
||
|
|
username = data.get('username', '').strip()
|
||
|
|
if not username:
|
||
|
|
return False, '用户名不能为空'
|
||
|
|
|
||
|
|
# 检查用户名是否已存在
|
||
|
|
existing = Admin.query.filter_by(username=username).first()
|
||
|
|
if existing:
|
||
|
|
return False, '用户名已存在'
|
||
|
|
|
||
|
|
# 检查密码是否为空
|
||
|
|
password = data.get('password', '')
|
||
|
|
if not password or not password.strip():
|
||
|
|
return False, '密码不能为空'
|
||
|
|
|
||
|
|
# 检查密码长度
|
||
|
|
if len(password.strip()) < 6:
|
||
|
|
return False, '密码长度至少6位'
|
||
|
|
|
||
|
|
# 验证邮箱格式（如果提供）
|
||
|
|
email = data.get('email', '').strip()
|
||
|
|
if email and '@' not in email:
|
||
|
|
return False, '邮箱格式不正确'
|
||
|
|
|
||
|
|
# 验证角色
|
||
|
|
role = data.get('role')
|
||
|
|
current_app.logger.info(f"Role value: {role}, type: {type(role)}")
|
||
|
|
if role is not None and role not in [0, 1]:
|
||
|
|
return False, '角色值无效'
|
||
|
|
|
||
|
|
# 验证状态
|
||
|
|
status = data.get('status')
|
||
|
|
current_app.logger.info(f"Status value: {status}, type: {type(status)}")
|
||
|
|
if status is not None and status not in [0, 1]:
|
||
|
|
return False, '状态值无效'
|
||
|
|
|
||
|
|
return True, ''
|
||
|
|
|
||
|
|
@api_bp.route('/admins', methods=['GET'])
|
||
|
|
@require_admin
|
||
|
|
def get_admins():
|
||
|
|
"""获取管理员列表"""
|
||
|
|
try:
|
||
|
|
page = request.args.get('page', 1, type=int)
|
||
|
|
per_page = min(request.args.get('per_page', 10, type=int), 100)
|
||
|
|
keyword = request.args.get('keyword', '').strip()
|
||
|
|
role = request.args.get('role', type=int) if request.args.get('role') != '' else None
|
||
|
|
status = request.args.get('status', type=int) if request.args.get('status') != '' else None
|
||
|
|
|
||
|
|
query = Admin.query
|
||
|
|
|
||
|
|
# 关键词搜索
|
||
|
|
if keyword:
|
||
|
|
query = query.filter(Admin.username.contains(keyword))
|
||
|
|
|
||
|
|
# 角色筛选
|
||
|
|
if role is not None:
|
||
|
|
query = query.filter(Admin.role == role)
|
||
|
|
|
||
|
|
# 状态筛选
|
||
|
|
if status is not None:
|
||
|
|
query = query.filter(Admin.status == status)
|
||
|
|
|
||
|
|
# 分页
|
||
|
|
pagination = query.paginate(
|
||
|
|
page=page, per_page=per_page, error_out=False
|
||
|
|
)
|
||
|
|
|
||
|
|
admins = [admin.to_dict() for admin in pagination.items]
|
||
|
|
|
||
|
|
return jsonify({
|
||
|
|
'success': True,
|
||
|
|
'data': {
|
||
|
|
'admins': admins,
|
||
|
|
'pagination': {
|
||
|
|
'page': page,
|
||
|
|
'per_page': per_page,
|
||
|
|
'total': pagination.total,
|
||
|
|
'pages': pagination.pages,
|
||
|
|
'has_prev': pagination.has_prev,
|
||
|
|
'has_next': pagination.has_next
|
||
|
|
}
|
||
|
|
}
|
||
|
|
})
|
||
|
|
|
||
|
|
except Exception as e:
|
||
|
|
current_app.logger.error(f"获取管理员列表失败: {str(e)}")
|
||
|
|
return jsonify({
|
||
|
|
'success': False,
|
||
|
|
'message': '服务器内部错误'
|
||
|
|
}), 500
|
||
|
|
|
||
|
|
@api_bp.route('/admins', methods=['POST'])
|
||
|
|
@require_admin
|
||
|
|
def create_admin():
|
||
|
|
"""创建管理员"""
|
||
|
|
try:
|
||
|
|
data = request.get_json()
|
||
|
|
current_app.logger.info(f"Received data for create_admin: {data}")
|
||
|
|
|
||
|
|
if not data:
|
||
|
|
current_app.logger.warning("No data received for create_admin")
|
||
|
|
return jsonify({
|
||
|
|
'success': False,
|
||
|
|
'message': '请求数据为空'
|
||
|
|
}), 400
|
||
|
|
|
||
|
|
# 验证数据
|
||
|
|
is_valid, message = validate_admin_data(data, is_create=True)
|
||
|
|
if not is_valid:
|
||
|
|
current_app.logger.warning(f"Validation failed for create_admin: {message}")
|
||
|
|
return jsonify({
|
||
|
|
'success': False,
|
||
|
|
'message': message
|
||
|
|
}), 400
|
||
|
|
|
||
|
|
username = data.get('username', '').strip()
|
||
|
|
email = data.get('email', '').strip()
|
||
|
|
role = data.get('role', 0)
|
||
|
|
status = data.get('status', 1)
|
||
|
|
password = data.get('password', '')
|
||
|
|
|
||
|
|
current_app.logger.info(f"Creating admin with username: {username}, role: {role}, status: {status}")
|
||
|
|
|
||
|
|
# 创建管理员
|
||
|
|
admin = Admin(
|
||
|
|
username=username,
|
||
|
|
email=email,
|
||
|
|
role=role,
|
||
|
|
status=status
|
||
|
|
)
|
||
|
|
admin.set_password(password)
|
||
|
|
|
||
|
|
db.session.add(admin)
|
||
|
|
db.session.commit()
|
||
|
|
|
||
|
|
return jsonify({
|
||
|
|
'success': True,
|
||
|
|
'message': '管理员创建成功',
|
||
|
|
'data': admin.to_dict()
|
||
|
|
})
|
||
|
|
|
||
|
|
except Exception as e:
|
||
|
|
db.session.rollback()
|
||
|
|
current_app.logger.error(f"创建管理员失败: {str(e)}")
|
||
|
|
return jsonify({
|
||
|
|
'success': False,
|
||
|
|
'message': '服务器内部错误'
|
||
|
|
}), 500
|
||
|
|
|
||
|
|
@api_bp.route('/admins/<int:admin_id>', methods=['GET'])
|
||
|
|
@require_admin
|
||
|
|
def get_admin(admin_id):
|
||
|
|
"""获取管理员详情"""
|
||
|
|
try:
|
||
|
|
admin = Admin.query.get(admin_id)
|
||
|
|
if not admin:
|
||
|
|
return jsonify({
|
||
|
|
'success': False,
|
||
|
|
'message': '管理员不存在'
|
||
|
|
}), 404
|
||
|
|
|
||
|
|
return jsonify({
|
||
|
|
'success': True,
|
||
|
|
'data': admin.to_dict()
|
||
|
|
})
|
||
|
|
|
||
|
|
except Exception as e:
|
||
|
|
current_app.logger.error(f"获取管理员详情失败: {str(e)}")
|
||
|
|
return jsonify({
|
||
|
|
'success': False,
|
||
|
|
'message': '服务器内部错误'
|
||
|
|
}), 500
|
||
|
|
|
||
|
|
@api_bp.route('/admins/<int:admin_id>', methods=['PUT'])
|
||
|
|
@require_admin
|
||
|
|
def update_admin(admin_id):
|
||
|
|
"""更新管理员"""
|
||
|
|
try:
|
||
|
|
admin = Admin.query.get(admin_id)
|
||
|
|
if not admin:
|
||
|
|
return jsonify({
|
||
|
|
'success': False,
|
||
|
|
'message': '管理员不存在'
|
||
|
|
}), 404
|
||
|
|
|
||
|
|
data = request.get_json()
|
||
|
|
if not data:
|
||
|
|
return jsonify({
|
||
|
|
'success': False,
|
||
|
|
'message': '请求数据为空'
|
||
|
|
}), 400
|
||
|
|
|
||
|
|
# 验证数据
|
||
|
|
is_valid, message = validate_admin_data(data, is_create=False)
|
||
|
|
if not is_valid:
|
||
|
|
return jsonify({
|
||
|
|
'success': False,
|
||
|
|
'message': message
|
||
|
|
}), 400
|
||
|
|
|
||
|
|
# 更新字段
|
||
|
|
if 'email' in data:
|
||
|
|
admin.email = data['email'].strip()
|
||
|
|
|
||
|
|
if 'role' in data:
|
||
|
|
admin.role = data['role']
|
||
|
|
|
||
|
|
if 'status' in data:
|
||
|
|
admin.status = data['status']
|
||
|
|
|
||
|
|
# 如果提供了密码，则更新密码
|
||
|
|
password = data.get('password', '')
|
||
|
|
if password and password.strip():
|
||
|
|
admin.set_password(password)
|
||
|
|
|
||
|
|
# 更新时间自动更新
|
||
|
|
db.session.commit()
|
||
|
|
|
||
|
|
return jsonify({
|
||
|
|
'success': True,
|
||
|
|
'message': '管理员更新成功',
|
||
|
|
'data': admin.to_dict()
|
||
|
|
})
|
||
|
|
|
||
|
|
except Exception as e:
|
||
|
|
db.session.rollback()
|
||
|
|
current_app.logger.error(f"更新管理员失败: {str(e)}")
|
||
|
|
return jsonify({
|
||
|
|
'success': False,
|
||
|
|
'message': '服务器内部错误'
|
||
|
|
}), 500
|
||
|
|
|
||
|
|
@api_bp.route('/admins/<int:admin_id>/toggle-status', methods=['POST'])
|
||
|
|
@require_admin
|
||
|
|
def toggle_admin_status(admin_id):
|
||
|
|
"""切换管理员状态"""
|
||
|
|
try:
|
||
|
|
admin = Admin.query.get(admin_id)
|
||
|
|
if not admin:
|
||
|
|
return jsonify({
|
||
|
|
'success': False,
|
||
|
|
'message': '管理员不存在'
|
||
|
|
}), 404
|
||
|
|
|
||
|
|
# 不允许禁用自己
|
||
|
|
if current_user.admin_id == admin_id and admin.status == 1:
|
||
|
|
return jsonify({
|
||
|
|
'success': False,
|
||
|
|
'message': '不能禁用当前登录的管理员'
|
||
|
|
}), 400
|
||
|
|
|
||
|
|
# 切换状态
|
||
|
|
admin.status = 0 if admin.status == 1 else 1
|
||
|
|
db.session.commit()
|
||
|
|
|
||
|
|
status_name = '正常' if admin.status == 1 else '禁用'
|
||
|
|
action = '启用' if admin.status == 1 else '禁用'
|
||
|
|
|
||
|
|
return jsonify({
|
||
|
|
'success': True,
|
||
|
|
'message': f'管理员已{action}',
|
||
|
|
'data': {
|
||
|
|
'status': admin.status,
|
||
|
|
'status_name': status_name
|
||
|
|
}
|
||
|
|
})
|
||
|
|
|
||
|
|
except Exception as e:
|
||
|
|
db.session.rollback()
|
||
|
|
current_app.logger.error(f"切换管理员状态失败: {str(e)}")
|
||
|
|
return jsonify({
|
||
|
|
'success': False,
|
||
|
|
'message': '服务器内部错误'
|
||
|
|
}), 500
|
||
|
|
|
||
|
|
@api_bp.route('/admins/<int:admin_id>', methods=['DELETE'])
|
||
|
|
@require_admin
|
||
|
|
def delete_admin(admin_id):
|
||
|
|
"""删除管理员"""
|
||
|
|
try:
|
||
|
|
admin = Admin.query.get(admin_id)
|
||
|
|
if not admin:
|
||
|
|
return jsonify({
|
||
|
|
'success': False,
|
||
|
|
'message': '管理员不存在'
|
||
|
|
}), 404
|
||
|
|
|
||
|
|
# 不允许删除自己
|
||
|
|
if current_user.admin_id == admin_id:
|
||
|
|
return jsonify({
|
||
|
|
'success': False,
|
||
|
|
'message': '不能删除当前登录的管理员'
|
||
|
|
}), 400
|
||
|
|
|
||
|
|
db.session.delete(admin)
|
||
|
|
db.session.commit()
|
||
|
|
|
||
|
|
return jsonify({
|
||
|
|
'success': True,
|
||
|
|
'message': '管理员删除成功'
|
||
|
|
})
|
||
|
|
|
||
|
|
except Exception as e:
|
||
|
|
db.session.rollback()
|
||
|
|
current_app.logger.error(f"删除管理员失败: {str(e)}")
|
||
|
|
return jsonify({
|
||
|
|
'success': False,
|
||
|
|
'message': '服务器内部错误'
|
||
|
|
}), 500
|