更新 APP/auth.py
This commit is contained in:
parent
be8628dceb
commit
342a6c4f99
763
APP/auth.py
763
APP/auth.py
@ -1,356 +1,407 @@
|
|||||||
from flask import Blueprint, render_template, redirect, url_for, flash, request, jsonify, current_app
|
from flask import Blueprint, render_template, redirect, url_for, flash, request, jsonify, current_app
|
||||||
from flask_login import login_user, logout_user, login_required, current_user
|
from flask_login import login_user, logout_user, login_required, current_user
|
||||||
from app import db
|
from app import db
|
||||||
from app.models import User, UserField, UserLoginInfo, UserPassword, UserDetail, UserPasswordHistory
|
from app.models import User, UserField, UserLoginInfo, UserPassword, UserDetail, UserPasswordHistory
|
||||||
from urllib.parse import urlparse # 替换 werkzeug.urls import
|
from urllib.parse import urlparse # 替换 werkzeug.urls import
|
||||||
import traceback
|
import traceback
|
||||||
import random
|
import random
|
||||||
import string
|
import string
|
||||||
from flask_jwt_extended import create_access_token, create_refresh_token, jwt_required, get_jwt_identity, \
|
from flask_jwt_extended import create_access_token, create_refresh_token, jwt_required, get_jwt_identity, \
|
||||||
verify_jwt_in_request
|
verify_jwt_in_request
|
||||||
from functools import wraps
|
from functools import wraps
|
||||||
from datetime import datetime, timedelta
|
from datetime import datetime, timedelta
|
||||||
|
from app.authorization import Authorization, department_and_role_required
|
||||||
bp = Blueprint('auth', __name__)
|
|
||||||
|
bp = Blueprint('auth', __name__)
|
||||||
@bp.route('/logout')
|
|
||||||
def logout():
|
@bp.route('/logout')
|
||||||
logout_user()
|
def logout():
|
||||||
return redirect(url_for('index'))
|
logout_user()
|
||||||
|
return redirect(url_for('index'))
|
||||||
@bp.route('/register', methods=['POST'])
|
|
||||||
def register():
|
@bp.route('/register', methods=['POST'])
|
||||||
current_app.logger.info('Received registration request')
|
def register():
|
||||||
data = request.get_json()
|
current_app.logger.info('Received registration request')
|
||||||
current_app.logger.info(f'Registration data: {data}')
|
data = request.get_json()
|
||||||
|
current_app.logger.info(f'Registration data: {data}')
|
||||||
email = data.get('email')
|
|
||||||
if email:
|
email = data.get('email')
|
||||||
existing_user = User.find_by_username_or_email(email)
|
if email:
|
||||||
if existing_user:
|
existing_user = User.find_by_username_or_email(email)
|
||||||
return jsonify({'message': '该邮箱已被注册'}), 400
|
if existing_user:
|
||||||
|
return jsonify({'message': '该邮箱已被注册'}), 400
|
||||||
required_fields = ['first_name', 'last_name', 'id_number', 'phone', 'gender']
|
|
||||||
missing_fields = [field for field in required_fields if field not in data or not data[field]]
|
required_fields = ['first_name', 'last_name', 'id_number', 'phone', 'gender']
|
||||||
|
missing_fields = [field for field in required_fields if field not in data or not data[field]]
|
||||||
if missing_fields:
|
|
||||||
current_app.logger.warning(f'Registration failed: Missing required fields: {", ".join(missing_fields)}')
|
if missing_fields:
|
||||||
return jsonify({'message': f'以下字段是必填的: {", ".join(missing_fields)}'}), 400
|
current_app.logger.warning(f'Registration failed: Missing required fields: {", ".join(missing_fields)}')
|
||||||
|
return jsonify({'message': f'以下字段是必填的: {", ".join(missing_fields)}'}), 400
|
||||||
try:
|
|
||||||
# 生成用户名和密码
|
try:
|
||||||
username = generate_username(data['first_name'], data['last_name'])
|
# 生成用户名和密码
|
||||||
password = generate_password()
|
username = generate_username(data['first_name'], data['last_name'])
|
||||||
current_app.logger.info(f'Generated username: {username}')
|
password = generate_password()
|
||||||
|
current_app.logger.info(f'Generated username: {username}')
|
||||||
user = User(username=username)
|
|
||||||
user.set_password(password)
|
user = User(username=username)
|
||||||
db.session.add(user)
|
user.set_password(password)
|
||||||
|
db.session.add(user)
|
||||||
# 设置用户详细信息
|
|
||||||
for field in UserField.query.all():
|
# 设置用户详细信息
|
||||||
if field.name in data:
|
for field in UserField.query.all():
|
||||||
user.set_detail(field.name, data[field.name])
|
if field.name in data:
|
||||||
current_app.logger.info(f'Set user detail: {field.name} = {data[field.name]}')
|
user.set_detail(field.name, data[field.name])
|
||||||
|
current_app.logger.info(f'Set user detail: {field.name} = {data[field.name]}')
|
||||||
# 设置新用户标记
|
|
||||||
user.login_info.is_new_user = True
|
# 设置新用户标记
|
||||||
user.login_info.has_changed_initial_password = False
|
user.login_info.is_new_user = True
|
||||||
|
user.login_info.has_changed_initial_password = False
|
||||||
db.session.commit()
|
|
||||||
current_app.logger.info(f'User {username} registered successfully')
|
db.session.commit()
|
||||||
return jsonify({
|
current_app.logger.info(f'User {username} registered successfully')
|
||||||
'message': '注册成功',
|
return jsonify({
|
||||||
'username': username,
|
'message': '注册成功',
|
||||||
'password': password
|
'username': username,
|
||||||
}), 201
|
'password': password
|
||||||
except Exception as e:
|
}), 201
|
||||||
current_app.logger.error(f'Registration error: {str(e)}')
|
except Exception as e:
|
||||||
current_app.logger.error(traceback.format_exc())
|
current_app.logger.error(f'Registration error: {str(e)}')
|
||||||
db.session.rollback()
|
current_app.logger.error(traceback.format_exc())
|
||||||
return jsonify({'message': f'注册失败，错误: {str(e)}'}), 500
|
db.session.rollback()
|
||||||
|
return jsonify({'message': f'注册失败，错误: {str(e)}'}), 500
|
||||||
def generate_username(first_name, last_name):
|
|
||||||
base_username = f"{first_name.lower()} {last_name.lower()}"
|
def generate_username(first_name, last_name):
|
||||||
username = base_username
|
base_username = f"{first_name.lower()} {last_name.lower()}"
|
||||||
counter = 1
|
username = base_username
|
||||||
while User.query.filter_by(username=username).first():
|
counter = 1
|
||||||
username = f"{base_username} {counter}"
|
while User.query.filter_by(username=username).first():
|
||||||
counter += 1
|
username = f"{base_username} {counter}"
|
||||||
return username
|
counter += 1
|
||||||
|
return username
|
||||||
def generate_password():
|
|
||||||
characters = string.ascii_letters + string.digits + string.punctuation
|
def generate_password():
|
||||||
return ''.join(random.choice(characters) for i in range(16))
|
characters = string.ascii_letters + string.digits + string.punctuation
|
||||||
|
return ''.join(random.choice(characters) for i in range(16))
|
||||||
|
|
||||||
@bp.route('/login', methods=['POST'])
|
|
||||||
def login_api():
|
@bp.route('/login', methods=['POST'])
|
||||||
current_app.logger.info('Received login request')
|
def login_api():
|
||||||
data = request.get_json()
|
current_app.logger.info('Received login request')
|
||||||
|
data = request.get_json()
|
||||||
if not data:
|
|
||||||
current_app.logger.warning('Invalid request data')
|
if not data:
|
||||||
return jsonify({'message': '无效的请求数据'}), 400
|
current_app.logger.warning('Invalid request data')
|
||||||
|
return jsonify({'message': '无效的请求数据'}), 400
|
||||||
username_or_email = data.get('username')
|
|
||||||
password = data.get('password')
|
username_or_email = data.get('username')
|
||||||
|
password = data.get('password')
|
||||||
current_app.logger.info(f'Login attempt for user: {username_or_email}')
|
|
||||||
|
current_app.logger.info(f'Login attempt for user: {username_or_email}')
|
||||||
if not username_or_email or not password:
|
|
||||||
current_app.logger.warning('Missing username/email or password')
|
if not username_or_email or not password:
|
||||||
return jsonify({'message': '缺少用户名/邮箱或密码'}), 400
|
current_app.logger.warning('Missing username/email or password')
|
||||||
|
return jsonify({'message': '缺少用户名/邮箱或密码'}), 400
|
||||||
try:
|
|
||||||
user = User.find_by_username_or_email(username_or_email)
|
try:
|
||||||
if user is None:
|
user = User.find_by_username_or_email(username_or_email)
|
||||||
current_app.logger.warning(f'User not found: {username_or_email}')
|
if user is None:
|
||||||
return jsonify({'message': '无效的用户名/邮箱或密码'}), 401
|
current_app.logger.warning(f'User not found: {username_or_email}')
|
||||||
|
return jsonify({'message': '无效的用户名/邮箱或密码'}), 401
|
||||||
if not user.check_password(password):
|
|
||||||
current_app.logger.warning(f'Invalid password for user: {username_or_email}')
|
if not user.check_password(password):
|
||||||
return jsonify({'message': '无效的用户名/邮箱或密码'}), 401
|
current_app.logger.warning(f'Invalid password for user: {username_or_email}')
|
||||||
|
return jsonify({'message': '无效的用户名/邮箱或密码'}), 401
|
||||||
current_app.logger.info(f'User authenticated: {username_or_email}')
|
|
||||||
current_app.logger.info(f'Is new user: {user.login_info.is_new_user}')
|
current_app.logger.info(f'User authenticated: {username_or_email}')
|
||||||
current_app.logger.info(f'Has changed initial password: {user.login_info.has_changed_initial_password}')
|
current_app.logger.info(f'Is new user: {user.login_info.is_new_user}')
|
||||||
|
current_app.logger.info(f'Has changed initial password: {user.login_info.has_changed_initial_password}')
|
||||||
# 检查是否是新用户或未更改初始密码
|
|
||||||
if user.login_info.is_new_user or not user.login_info.has_changed_initial_password:
|
# 检查是否是新用户或未更改初始密码
|
||||||
current_app.logger.info(f'New user or initial password not changed: {username_or_email}')
|
if user.login_info.is_new_user or not user.login_info.has_changed_initial_password:
|
||||||
return jsonify({
|
current_app.logger.info(f'New user or initial password not changed: {username_or_email}')
|
||||||
'message': '首次登录，请修改密码',
|
return jsonify({
|
||||||
'require_password_change': True
|
'message': '首次登录，请修改密码',
|
||||||
}), 403
|
'require_password_change': True
|
||||||
|
}), 403
|
||||||
# 检查密码是否过期
|
|
||||||
if user.needs_password_change():
|
# 检查密码是否过期
|
||||||
current_app.logger.info(f'Password expired for user: {username_or_email}')
|
if user.password_history:
|
||||||
return jsonify({
|
days_since_last_change = (datetime.utcnow() - user.password_history.last_password_change).days
|
||||||
'message': '密码已过期，请修改密码',
|
if days_since_last_change >= 90:
|
||||||
'require_password_change': True
|
current_app.logger.info(f'Password expired for user: {username_or_email}')
|
||||||
}), 403
|
return jsonify({
|
||||||
|
'message': '密码已过期，请修改密码',
|
||||||
login_user(user)
|
'require_password_change': True
|
||||||
user.login_info.update_login_info()
|
}), 403
|
||||||
|
elif days_since_last_change >= 75:
|
||||||
access_token = create_access_token(identity=user.id)
|
expiry_date = user.password_history.last_password_change + timedelta(days=90)
|
||||||
refresh_token = create_refresh_token(identity=user.id)
|
warning_message = f'您的密码将在 {expiry_date.strftime("%Y-%m-%d")} 过期，请尽快修改密码'
|
||||||
|
current_app.logger.info(f'Password expiry warning for user: {username_or_email}')
|
||||||
current_app.logger.info(f'User {user.username} logged in successfully')
|
else:
|
||||||
|
warning_message = None
|
||||||
# 获取用户详细信息
|
else:
|
||||||
user_fields = UserField.query.all()
|
warning_message = None
|
||||||
user_details = {}
|
|
||||||
for field in user_fields:
|
login_user(user)
|
||||||
detail = UserDetail.query.filter_by(user_id=user.id, field_id=field.id).first()
|
user.login_info.update_login_info()
|
||||||
user_details[field.name] = detail.value if detail else None
|
|
||||||
|
access_token = create_access_token(identity=user.id)
|
||||||
user_info = {
|
refresh_token = create_refresh_token(identity=user.id)
|
||||||
'id': user.id,
|
|
||||||
'username': user.username,
|
current_app.logger.info(f'User {user.username} logged in successfully')
|
||||||
'primary_department': user.primary_department.name if user.primary_department else None,
|
|
||||||
'secondary_departments': [dept.name for dept in user.secondary_departments],
|
# 获取用户详细信息
|
||||||
'roles': [role.name for role in user.roles],
|
user_fields = UserField.query.all()
|
||||||
'permissions': [perm.name for perm in user.get_all_permissions()], # 添加这行
|
user_details = {}
|
||||||
'details': user_details,
|
for field in user_fields:
|
||||||
'login_info': {
|
detail = UserDetail.query.filter_by(user_id=user.id, field_id=field.id).first()
|
||||||
'register_time': user.login_info.register_time.isoformat(),
|
user_details[field.name] = detail.value if detail else None
|
||||||
'login_count': user.login_info.login_count,
|
|
||||||
'last_login_time': user.login_info.last_login_time.isoformat() if user.login_info.last_login_time else None,
|
user_info = {
|
||||||
'is_new_user': user.login_info.is_new_user,
|
'id': user.id,
|
||||||
'has_changed_initial_password': user.login_info.has_changed_initial_password
|
'username': user.username,
|
||||||
}
|
'primary_department': user.primary_department.name if user.primary_department else None,
|
||||||
}
|
'secondary_departments': [dept.name for dept in user.secondary_departments],
|
||||||
|
'roles': [role.name for role in user.roles],
|
||||||
return jsonify({
|
'permissions': [perm.name for perm in user.get_all_permissions()],
|
||||||
'message': '登录成功',
|
'details': user_details,
|
||||||
'access_token': access_token,
|
'login_info': {
|
||||||
'refresh_token': refresh_token,
|
'register_time': user.login_info.register_time.isoformat(),
|
||||||
'user_info': user_info
|
'login_count': user.login_info.login_count,
|
||||||
}), 200
|
'last_login_time': user.login_info.last_login_time.isoformat() if user.login_info.last_login_time else None,
|
||||||
except Exception as e:
|
'is_new_user': user.login_info.is_new_user,
|
||||||
current_app.logger.error(f'Login error: {str(e)}')
|
'has_changed_initial_password': user.login_info.has_changed_initial_password
|
||||||
current_app.logger.error(traceback.format_exc())
|
}
|
||||||
return jsonify({'message': '登录失败，请稍后再试'}), 500
|
}
|
||||||
|
|
||||||
@bp.route('/refresh', methods=['POST'])
|
response_data = {
|
||||||
@jwt_required(refresh=True)
|
'message': '登录成功',
|
||||||
def refresh():
|
'access_token': access_token,
|
||||||
current_user_id = get_jwt_identity()
|
'refresh_token': refresh_token,
|
||||||
access_token = create_access_token(identity=current_user_id)
|
'user_info': user_info
|
||||||
return jsonify({'access_token': access_token}), 200
|
}
|
||||||
|
|
||||||
@bp.route('/logout', methods=['POST'])
|
if warning_message:
|
||||||
@login_required
|
response_data['warning'] = warning_message
|
||||||
def logout_api():
|
|
||||||
current_app.logger.info(f'User {current_user.username} logged out')
|
return jsonify(response_data), 200
|
||||||
logout_user()
|
except Exception as e:
|
||||||
return jsonify({'message': '登出成功'}), 200
|
current_app.logger.error(f'Login error: {str(e)}')
|
||||||
|
current_app.logger.error(traceback.format_exc())
|
||||||
@bp.route('/profile', methods=['GET'])
|
return jsonify({'message': '登录失败，请稍后再试'}), 500
|
||||||
@login_required
|
|
||||||
def profile():
|
@bp.route('/refresh', methods=['POST'])
|
||||||
current_app.logger.info(f'Profile requested for user {current_user.username}')
|
@jwt_required(refresh=True)
|
||||||
try:
|
def refresh():
|
||||||
profile_data = {
|
current_user_id = get_jwt_identity()
|
||||||
'username': current_user.username,
|
access_token = create_access_token(identity=current_user_id)
|
||||||
'login_info': {
|
return jsonify({'access_token': access_token}), 200
|
||||||
'register_time': current_user.login_info.register_time.isoformat(),
|
|
||||||
'login_count': current_user.login_info.login_count,
|
@bp.route('/logout', methods=['POST'])
|
||||||
'last_login_time': current_user.login_info.last_login_time.isoformat() if current_user.login_info.last_login_time else None,
|
@login_required
|
||||||
'is_new_user': current_user.login_info.is_new_user
|
def logout_api():
|
||||||
}
|
current_app.logger.info(f'User {current_user.username} logged out')
|
||||||
}
|
logout_user()
|
||||||
current_app.logger.info(f'Profile data retrieved for user {current_user.username}')
|
return jsonify({'message': '登出成功'}), 200
|
||||||
return jsonify(profile_data)
|
|
||||||
except Exception as e:
|
@bp.route('/profile', methods=['GET'])
|
||||||
current_app.logger.error(f'Profile retrieval error for user {current_user.username}: {str(e)}')
|
@login_required
|
||||||
current_app.logger.error(traceback.format_exc())
|
def profile():
|
||||||
return jsonify({'message': '获取用户资料失败，请稍后再试'}), 500
|
current_app.logger.info(f'Profile requested for user {current_user.username}')
|
||||||
|
try:
|
||||||
@bp.route('/change_password', methods=['POST'])
|
profile_data = {
|
||||||
@jwt_required()
|
'username': current_user.username,
|
||||||
def change_password():
|
'login_info': {
|
||||||
current_app.logger.info('Received change password request')
|
'register_time': current_user.login_info.register_time.isoformat(),
|
||||||
current_user_id = get_jwt_identity()
|
'login_count': current_user.login_info.login_count,
|
||||||
data = request.get_json()
|
'last_login_time': current_user.login_info.last_login_time.isoformat() if current_user.login_info.last_login_time else None,
|
||||||
old_password = data.get('old_password')
|
'is_new_user': current_user.login_info.is_new_user
|
||||||
new_password = data.get('new_password')
|
}
|
||||||
|
}
|
||||||
if not all([old_password, new_password]):
|
current_app.logger.info(f'Profile data retrieved for user {current_user.username}')
|
||||||
return jsonify({'message': '缺少必要的字段'}), 400
|
return jsonify(profile_data)
|
||||||
|
except Exception as e:
|
||||||
try:
|
current_app.logger.error(f'Profile retrieval error for user {current_user.username}: {str(e)}')
|
||||||
user = User.query.get(current_user_id)
|
current_app.logger.error(traceback.format_exc())
|
||||||
if not user:
|
return jsonify({'message': '获取用户资料失败，请稍后再试'}), 500
|
||||||
return jsonify({'message': '用户不存在'}), 404
|
|
||||||
|
@bp.route('/change_password', methods=['POST'])
|
||||||
if not user.check_password(old_password):
|
@jwt_required()
|
||||||
return jsonify({'message': '旧密码不正确'}), 401
|
def change_password():
|
||||||
|
current_app.logger.info('Received change password request')
|
||||||
user.set_password(new_password)
|
current_user_id = get_jwt_identity()
|
||||||
user.login_info.has_changed_initial_password = True
|
data = request.get_json()
|
||||||
|
old_password = data.get('old_password')
|
||||||
# 更新密码修改历史
|
new_password = data.get('new_password')
|
||||||
if user.password_history:
|
|
||||||
user.password_history.update_password_change()
|
if not all([old_password, new_password]):
|
||||||
else:
|
return jsonify({'message': '缺少必要的字段'}), 400
|
||||||
user.password_history = UserPasswordHistory(user=user)
|
|
||||||
|
try:
|
||||||
db.session.commit()
|
user = User.query.get(current_user_id)
|
||||||
|
if not user:
|
||||||
current_app.logger.info(f'Password changed successfully for user {user.username}')
|
return jsonify({'message': '用户不存在'}), 404
|
||||||
return jsonify({'message': '密码修改成功'}), 200
|
|
||||||
except Exception as e:
|
if not user.check_password(old_password):
|
||||||
current_app.logger.error(f'Password change error: {str(e)}')
|
return jsonify({'message': '旧密码不正确'}), 401
|
||||||
current_app.logger.error(traceback.format_exc())
|
|
||||||
db.session.rollback()
|
user.set_password(new_password)
|
||||||
return jsonify({'message': '密码修改失败，请稍后再试'}), 500
|
user.login_info.has_changed_initial_password = True
|
||||||
|
|
||||||
def token_required(f):
|
# 更新密码修改历史
|
||||||
@wraps(f)
|
if user.password_history:
|
||||||
def decorated(*args, **kwargs):
|
user.password_history.update_password_change()
|
||||||
try:
|
else:
|
||||||
# 验证 JWT token
|
user.password_history = UserPasswordHistory(user=user)
|
||||||
verify_jwt_in_request()
|
|
||||||
# 获取当前用户的身份（通常是用户ID）
|
db.session.commit()
|
||||||
current_user_id = get_jwt_identity()
|
|
||||||
# 从数据库中获取用户对象
|
current_app.logger.info(f'Password changed successfully for user {user.username}')
|
||||||
current_user = User.query.get(current_user_id)
|
return jsonify({'message': '密码修改成功'}), 200
|
||||||
if not current_user:
|
except Exception as e:
|
||||||
return jsonify({'message': '找不到用户'}), 404
|
current_app.logger.error(f'Password change error: {str(e)}')
|
||||||
except Exception as e:
|
current_app.logger.error(traceback.format_exc())
|
||||||
return jsonify({'message': '无效的令牌'}), 401
|
db.session.rollback()
|
||||||
# 将当前用户对象传递给被装饰的函数
|
return jsonify({'message': '密码修改失败，请稍后再试'}), 500
|
||||||
return f(current_user, *args, **kwargs)
|
|
||||||
return decorated
|
def token_required(f):
|
||||||
|
@wraps(f)
|
||||||
@bp.route('/change_initial_password', methods=['POST'])
|
def decorated(*args, **kwargs):
|
||||||
def change_initial_password():
|
try:
|
||||||
data = request.get_json()
|
# 验证 JWT token
|
||||||
|
verify_jwt_in_request()
|
||||||
if not data:
|
# 获取当前用户的身份（通常是用户ID）
|
||||||
return jsonify({'message': '缺少请求数据'}), 400
|
current_user_id = get_jwt_identity()
|
||||||
|
# 从数据库中获取用户对象
|
||||||
username_or_email = data.get('username_or_email') or data.get('username')
|
current_user = User.query.get(current_user_id)
|
||||||
old_password = data.get('old_password')
|
if not current_user:
|
||||||
new_password = data.get('new_password')
|
return jsonify({'message': '找不到用户'}), 404
|
||||||
|
except Exception as e:
|
||||||
if not all([username_or_email, old_password, new_password]):
|
return jsonify({'message': '无效的令牌'}), 401
|
||||||
return jsonify({'message': '缺少必要的字段'}), 400
|
# 将当前用户对象传递给被装饰的函数
|
||||||
|
return f(current_user, *args, **kwargs)
|
||||||
user = User.find_by_username_or_email(username_or_email)
|
return decorated
|
||||||
if not user:
|
|
||||||
return jsonify({'message': '用户不存在'}), 404
|
@bp.route('/change_initial_password', methods=['POST'])
|
||||||
|
def change_initial_password():
|
||||||
if not user.check_password(old_password):
|
data = request.get_json()
|
||||||
return jsonify({'message': '旧密码不正确'}), 401
|
|
||||||
|
if not data:
|
||||||
if user.login_info.has_changed_initial_password:
|
return jsonify({'message': '缺少请求数据'}), 400
|
||||||
return jsonify({'message': '初始密码已经被修改过'}), 400
|
|
||||||
|
username_or_email = data.get('username_or_email') or data.get('username')
|
||||||
user.set_password(new_password)
|
old_password = data.get('old_password')
|
||||||
user.login_info.has_changed_initial_password = True
|
new_password = data.get('new_password')
|
||||||
user.login_info.is_new_user = False
|
|
||||||
|
if not all([username_or_email, old_password, new_password]):
|
||||||
# 更新密码修改历史
|
return jsonify({'message': '缺少必要的字段'}), 400
|
||||||
if user.password_history:
|
|
||||||
user.password_history.update_password_change()
|
user = User.find_by_username_or_email(username_or_email)
|
||||||
else:
|
if not user:
|
||||||
user.password_history = UserPasswordHistory(user=user)
|
return jsonify({'message': '用户不存在'}), 404
|
||||||
|
|
||||||
db.session.commit()
|
if not user.check_password(old_password):
|
||||||
|
return jsonify({'message': '旧密码不正确'}), 401
|
||||||
access_token = create_access_token(identity=user.id)
|
|
||||||
refresh_token = create_refresh_token(identity=user.id)
|
if user.login_info.has_changed_initial_password:
|
||||||
|
return jsonify({'message': '初始密码已经被修改过'}), 400
|
||||||
current_app.logger.info(f'Initial password changed for user: {user.username}')
|
|
||||||
current_app.logger.info(f'New user status: {user.login_info.is_new_user}')
|
user.set_password(new_password)
|
||||||
current_app.logger.info(f'Has changed initial password status: {user.login_info.has_changed_initial_password}')
|
user.login_info.has_changed_initial_password = True
|
||||||
|
user.login_info.is_new_user = False
|
||||||
return jsonify({
|
|
||||||
'message': '密码修改成功，现在可以登录',
|
# 更新密码修改历史
|
||||||
'access_token': access_token,
|
if user.password_history:
|
||||||
'refresh_token': refresh_token
|
user.password_history.update_password_change()
|
||||||
}), 200
|
else:
|
||||||
|
user.password_history = UserPasswordHistory(user=user)
|
||||||
@bp.route('/current_user', methods=['GET'])
|
|
||||||
@jwt_required()
|
db.session.commit()
|
||||||
def get_current_user():
|
|
||||||
current_user_id = get_jwt_identity()
|
access_token = create_access_token(identity=user.id)
|
||||||
user = User.query.get(current_user_id)
|
refresh_token = create_refresh_token(identity=user.id)
|
||||||
|
|
||||||
if not user:
|
current_app.logger.info(f'Initial password changed for user: {user.username}')
|
||||||
return jsonify({'message': '用户不存在'}), 404
|
current_app.logger.info(f'New user status: {user.login_info.is_new_user}')
|
||||||
|
current_app.logger.info(f'Has changed initial password status: {user.login_info.has_changed_initial_password}')
|
||||||
try:
|
|
||||||
user_fields = UserField.query.all()
|
return jsonify({
|
||||||
user_details = {}
|
'message': '密码修改成功，现在可以登录',
|
||||||
for field in user_fields:
|
'access_token': access_token,
|
||||||
detail = UserDetail.query.filter_by(user_id=user.id, field_id=field.id).first()
|
'refresh_token': refresh_token
|
||||||
user_details[field.name] = detail.value if detail else None
|
}), 200
|
||||||
|
|
||||||
user_info = {
|
@bp.route('/current_user', methods=['GET'])
|
||||||
'id': user.id,
|
@jwt_required()
|
||||||
'username': user.username,
|
def get_current_user():
|
||||||
'details': user_details,
|
current_user_id = get_jwt_identity()
|
||||||
'primary_department': user.primary_department.name if user.primary_department else None,
|
user = User.query.get(current_user_id)
|
||||||
'secondary_departments': [dept.name for dept in user.secondary_departments],
|
|
||||||
'roles': [role.name for role in user.roles],
|
if not user:
|
||||||
'permissions': [perm.name for perm in user.get_all_permissions()],
|
return jsonify({'message': '用户不存在'}), 404
|
||||||
'login_info': {
|
|
||||||
'register_time': user.login_info.register_time.isoformat(),
|
try:
|
||||||
'login_count': user.login_info.login_count,
|
user_fields = UserField.query.all()
|
||||||
'last_login_time': user.login_info.last_login_time.isoformat() if user.login_info.last_login_time else None,
|
user_details = {}
|
||||||
'is_new_user': user.login_info.is_new_user,
|
for field in user_fields:
|
||||||
'has_changed_initial_password': user.login_info.has_changed_initial_password
|
detail = UserDetail.query.filter_by(user_id=user.id, field_id=field.id).first()
|
||||||
}
|
user_details[field.name] = detail.value if detail else None
|
||||||
}
|
|
||||||
|
user_info = {
|
||||||
return jsonify(user_info), 200
|
'id': user.id,
|
||||||
except Exception as e:
|
'username': user.username,
|
||||||
current_app.logger.error(f'Error retrieving current user info: {str(e)}')
|
'details': user_details,
|
||||||
current_app.logger.error(traceback.format_exc())
|
'primary_department': user.primary_department.name if user.primary_department else None,
|
||||||
return jsonify({'message': '获取用户信息失败，请稍后再试'}), 500
|
'secondary_departments': [dept.name for dept in user.secondary_departments],
|
||||||
|
'roles': [role.name for role in user.roles],
|
||||||
|
'permissions': [perm.name for perm in user.get_all_permissions()],
|
||||||
|
'login_info': {
|
||||||
|
'register_time': user.login_info.register_time.isoformat(),
|
||||||
|
'login_count': user.login_info.login_count,
|
||||||
|
'last_login_time': user.login_info.last_login_time.isoformat() if user.login_info.last_login_time else None,
|
||||||
|
'is_new_user': user.login_info.is_new_user,
|
||||||
|
'has_changed_initial_password': user.login_info.has_changed_initial_password
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
return jsonify(user_info), 200
|
||||||
|
except Exception as e:
|
||||||
|
current_app.logger.error(f'Error retrieving current user info: {str(e)}')
|
||||||
|
current_app.logger.error(traceback.format_exc())
|
||||||
|
return jsonify({'message': '获取用户信息失败，请稍后再试'}), 500
|
||||||
|
|
||||||
|
@bp.route('/force_password_change', methods=['POST'])
|
||||||
|
@jwt_required()
|
||||||
|
@department_and_role_required('信息技术', ['Global Administrator', 'frontline staff'])
|
||||||
|
def force_password_change():
|
||||||
|
data = request.get_json()
|
||||||
|
target_user_id = data.get('user_id')
|
||||||
|
new_password = data.get('new_password')
|
||||||
|
|
||||||
|
if not target_user_id or not new_password:
|
||||||
|
return jsonify({'message': '缺少必要的字段'}), 400
|
||||||
|
|
||||||
|
target_user = User.query.get(target_user_id)
|
||||||
|
if not target_user:
|
||||||
|
return jsonify({'message': '目标用户不存在'}), 404
|
||||||
|
|
||||||
|
try:
|
||||||
|
target_user.set_password(new_password)
|
||||||
|
target_user.login_info.has_changed_initial_password = False
|
||||||
|
target_user.login_info.is_new_user = False
|
||||||
|
|
||||||
|
# 更新密码修改历史
|
||||||
|
if target_user.password_history:
|
||||||
|
target_user.password_history.update_password_change()
|
||||||
|
else:
|
||||||
|
target_user.password_history = UserPasswordHistory(user=target_user)
|
||||||
|
|
||||||
|
db.session.commit()
|
||||||
|
|
||||||
|
current_app.logger.info(f'Password forcibly changed for user: {target_user.username}')
|
||||||
|
return jsonify({'message': '密码已成功强制修改'}), 200
|
||||||
|
except Exception as e:
|
||||||
|
current_app.logger.error(f'Force password change error: {str(e)}')
|
||||||
|
db.session.rollback()
|
||||||
|
return jsonify({'message': '强制修改密码失败，请稍后再试'}), 500
|
||||||
Loading…
Reference in New Issue
Block a user