4808a48594
一、统一为单一监听器实现
原先两套 SQLAlchemy 事件监听器并存：
· app/utils/audit_events.py —— 全局监听 db.Model、无白名单、无请求上下文守卫（实际在跑）
· app/core/audit_listener.py —— 白名单制、有守卫、有模型级开关（从未生效）
后者失效的根因：注册代码写在 extensions.py 的 init_extensions() 内，
而该函数全仓库只有定义、没有任何调用（create_app 直接内联调用 db.init_app 等）。
现统一由 app/core/audit_listener.py 承担，并在 create_app() 中显式注册。
extensions.py 的死函数 init_extensions 整体删除，避免后人误以为它是有效入口。
二、修复监听器三处致命缺陷（此前注册了也写不进数据）
1. 事件回调第二个参数是 Connection，原代码却调用 Connection.add()（不存在），
每次写日志都抛 AttributeError 并被 except 吞掉 → 改为 connection.execute()
2. register_audit_listeners 从 app.models 批量 import 多个未导出的模型，
ImportError 被上层 try/except 吞掉 → 改为按表名从 db.metadata 取模型
3. 本项目有 31 处函数体内延迟导入模型（如 scrap.py 内部才 import ScrapApproval），
一次性注册会静默漏表 → 增加 ensure_audit_listeners() 惰性补绑，
并在模型预加载段补全审批单/BOM/采购等模型
三、强约束
· WHITELIST_TABLES：仅 18 张核心业务表，系统表/草稿表/向量表不再自审
· has_request_context() 守卫：系统初始化与后台定时任务不再产生 username=system 噪声
· IGNORE_FIELDS 增加 password/password_hash/salt/token/secret/api_key（安全红线）
· created_at 显式写 beijing_time()，与全系统时间口径一致
四、清除僵尸装饰器
@audit_log 早已退化为直接透传的空壳（module/action 参数全被忽略，
数据库中零星的中文 action 即其历史遗留产物），却仍挂在 38 处路由上。
连同 13 个文件的 import 一并移除；audit_events.register_audit_events 改为空操作。
验证：应用上下文中的写操作不产生日志；HTTP 请求产生 5 条日志，
对象为业务单号（APR-SCRAP-... / SKU），模块中文，操作人真实，时间为北京时间。
513 lines
20 KiB
Python
513 lines
20 KiB
Python
# app/api/v1/auth.py
|
||
# ==============================================================================
|
||
# Flask & Extensions
|
||
# ==============================================================================
|
||
from flask import Blueprint, request, jsonify, current_app
|
||
from flask_jwt_extended import (
|
||
jwt_required,
|
||
get_jwt,
|
||
get_jwt_identity,
|
||
create_access_token,
|
||
)
|
||
from werkzeug.exceptions import BadRequest, Unauthorized, Forbidden, NotFound, InternalServerError
|
||
|
||
# ==============================================================================
|
||
# Database & Extensions
|
||
# ==============================================================================
|
||
from app.extensions import db
|
||
|
||
# ==============================================================================
|
||
# Service & Decorators
|
||
# ==============================================================================
|
||
from app.services.auth_service import AuthService
|
||
from app.utils.decorators import permission_required
|
||
|
||
auth_bp = Blueprint('auth', __name__)
|
||
|
||
|
||
# ==============================================================================
|
||
# 辅助函数：获取当前用户的完整权限列表（基于角色查询）
|
||
# ==============================================================================
|
||
def get_current_user_permissions():
|
||
"""
|
||
返回当前用户拥有的所有权限码列表（包括菜单和元素）
|
||
此函数根据角色+公司查询数据库得到权限。
|
||
"""
|
||
claims = get_jwt()
|
||
user_role = claims.get('role')
|
||
user_company = claims.get('company_name', '')
|
||
if not user_role:
|
||
return []
|
||
# 超级管理员返回所有字段权限
|
||
if user_role.upper() == 'SUPER_ADMIN':
|
||
return ['system_user:*']
|
||
perm_dict = AuthService.get_user_permissions(user_role, company_name=user_company)
|
||
# 合并菜单和元素权限
|
||
perms = perm_dict.get('menus', []) + perm_dict.get('elements', [])
|
||
return perms
|
||
|
||
|
||
def filter_item_by_permissions(item_dict, user_permissions):
|
||
"""
|
||
根据用户权限过滤 item 字典，无权限的字段值置为 None
|
||
"""
|
||
# 字段名到权限码的映射（与前端 permissionMap 保持一致）
|
||
field_to_perm = {
|
||
'id': 'system_user:id',
|
||
'username': 'system_user:username',
|
||
'account_id': 'system_user:account_id',
|
||
'email': 'system_user:email',
|
||
'department': 'system_user:department',
|
||
'role': 'system_user:role',
|
||
'status': 'system_user:status',
|
||
'created_at': 'system_user:created_at',
|
||
}
|
||
# 如果用户是超级管理员且有 'system_user:*'，则不过滤
|
||
if 'system_user:*' in user_permissions:
|
||
return item_dict
|
||
for field, perm_code in field_to_perm.items():
|
||
if field in item_dict and perm_code not in user_permissions:
|
||
item_dict[field] = None
|
||
return item_dict
|
||
|
||
|
||
# ==============================================================================
|
||
# 登录接口
|
||
# ==============================================================================
|
||
@auth_bp.route('/login', methods=['POST'])
|
||
def login():
|
||
try:
|
||
data = request.get_json()
|
||
if not data:
|
||
return jsonify({'msg': '无效的请求数据'}), 400
|
||
|
||
if not data.get('username') or not data.get('password'):
|
||
return jsonify({'msg': '请输入用户名和密码'}), 400
|
||
|
||
result = AuthService.login(data)
|
||
|
||
response_data = {
|
||
'msg': '登录成功',
|
||
'access_token': result.get('access_token'),
|
||
'refresh_token': result.get('refresh_token'),
|
||
'user': result.get('user')
|
||
}
|
||
return jsonify(response_data), 200
|
||
|
||
except ValueError as ve:
|
||
current_app.logger.warning(f"Login validation failed: {str(ve)}")
|
||
return jsonify({'msg': str(ve)}), 401
|
||
except Exception as e:
|
||
current_app.logger.error(f"Login Failed Error: {str(e)}")
|
||
return jsonify({'msg': f'服务器内部错误: {str(e)}'}), 500
|
||
|
||
|
||
# ==============================================================================
|
||
# Token 刷新接口
|
||
# ==============================================================================
|
||
@auth_bp.route('/refresh', methods=['POST'])
|
||
def refresh():
|
||
"""使用 refresh_token 换发新的 access_token"""
|
||
try:
|
||
data = request.get_json()
|
||
if not data or not data.get('refresh_token'):
|
||
return jsonify({'msg': '缺少 refresh_token'}), 400
|
||
|
||
refresh_token = data.get('refresh_token')
|
||
result = AuthService.refresh_access_token(refresh_token)
|
||
|
||
return jsonify({
|
||
'msg': 'Token 刷新成功',
|
||
'access_token': result.get('access_token')
|
||
}), 200
|
||
|
||
except ValueError as ve:
|
||
current_app.logger.warning(f"Token refresh validation failed: {str(ve)}")
|
||
return jsonify({'msg': str(ve)}), 401
|
||
except Exception as e:
|
||
current_app.logger.error(f"Token Refresh Error: {str(e)}")
|
||
return jsonify({'msg': f'Token 刷新失败: {str(e)}'}), 500
|
||
|
||
|
||
# ==============================================================================
|
||
# 创建用户（管理员）
|
||
# ==============================================================================
|
||
@auth_bp.route('/user/create', methods=['POST'])
|
||
@jwt_required()
|
||
@permission_required('system_user:operation')
|
||
def create_user():
|
||
try:
|
||
data = request.get_json()
|
||
# 数据清洗：移除用户没有权限的字段
|
||
user_permissions = get_current_user_permissions()
|
||
# 超级管理员不过滤
|
||
if 'system_user:*' not in user_permissions:
|
||
field_to_perm = {
|
||
'cn_name': 'system_user:username',
|
||
'username': 'system_user:username',
|
||
'password': 'system_user:password',
|
||
'department': 'system_user:department',
|
||
'role': 'system_user:role',
|
||
'email': 'system_user:email',
|
||
}
|
||
for field in list(data.keys()):
|
||
perm_code = field_to_perm.get(field)
|
||
if field == 'password':
|
||
if 'system_user:operation' not in user_permissions:
|
||
data.pop(field, None)
|
||
continue
|
||
if perm_code and perm_code not in user_permissions:
|
||
data.pop(field, None)
|
||
|
||
claims = get_jwt()
|
||
operator_role = claims.get('role')
|
||
|
||
result = AuthService.create_user(data, operator_role)
|
||
return jsonify({'msg': '用户创建成功', 'data': result}), 201
|
||
|
||
except Exception as e:
|
||
current_app.logger.error(f"User Create Failed: {str(e)}")
|
||
return jsonify({'msg': str(e)}), 400
|
||
|
||
|
||
# ==============================================================================
|
||
# 批量创建用户
|
||
# ==============================================================================
|
||
@auth_bp.route('/user/batch', methods=['POST'])
|
||
@jwt_required()
|
||
@permission_required('system_user:operation')
|
||
def batch_create_user():
|
||
try:
|
||
data_list = request.get_json()
|
||
if not data_list or not isinstance(data_list, list):
|
||
return jsonify({'msg': '请求数据必须是用户数组'}), 400
|
||
|
||
# 数据清洗：移除用户没有权限的字段
|
||
user_permissions = get_current_user_permissions()
|
||
for data in data_list:
|
||
if 'system_user:*' not in user_permissions:
|
||
field_to_perm = {
|
||
'cn_name': 'system_user:username',
|
||
'username': 'system_user:username',
|
||
'password': 'system_user:password',
|
||
'department': 'system_user:department',
|
||
'role': 'system_user:role',
|
||
'email': 'system_user:email',
|
||
}
|
||
for field in list(data.keys()):
|
||
perm_code = field_to_perm.get(field)
|
||
if field == 'password':
|
||
if 'system_user:operation' not in user_permissions:
|
||
data.pop(field, None)
|
||
continue
|
||
if perm_code and perm_code not in user_permissions:
|
||
data.pop(field, None)
|
||
|
||
claims = get_jwt()
|
||
operator_role = claims.get('role')
|
||
|
||
results = AuthService.batch_create_users(data_list, operator_role)
|
||
return jsonify({'msg': '批量处理完成', 'data': results}), 200
|
||
|
||
except Exception as e:
|
||
current_app.logger.error(f"Batch User Create Failed: {str(e)}")
|
||
return jsonify({'msg': str(e)}), 500
|
||
|
||
|
||
# ==============================================================================
|
||
# 更新用户（管理员）
|
||
# ==============================================================================
|
||
@auth_bp.route('/user/<int:user_id>', methods=['PUT'])
|
||
@jwt_required()
|
||
@permission_required('system_user:operation')
|
||
def update_user(user_id):
|
||
try:
|
||
data = request.get_json()
|
||
user_permissions = get_current_user_permissions()
|
||
if 'system_user:*' not in user_permissions:
|
||
field_to_perm = {
|
||
'cn_name': 'system_user:username',
|
||
'username': 'system_user:username',
|
||
'password': 'system_user:password',
|
||
'department': 'system_user:department',
|
||
'role': 'system_user:role',
|
||
'email': 'system_user:email',
|
||
}
|
||
for field in list(data.keys()):
|
||
perm_code = field_to_perm.get(field)
|
||
if field == 'password':
|
||
if 'system_user:operation' not in user_permissions:
|
||
data.pop(field, None)
|
||
continue
|
||
if perm_code and perm_code not in user_permissions:
|
||
data.pop(field, None)
|
||
|
||
claims = get_jwt()
|
||
operator_role = claims.get('role')
|
||
|
||
result = AuthService.update_user(user_id, data, operator_role)
|
||
return jsonify({'msg': '用户更新成功', 'data': result}), 200
|
||
|
||
except Exception as e:
|
||
current_app.logger.error(f"User Update Failed: {str(e)}")
|
||
return jsonify({'msg': str(e)}), 400
|
||
|
||
|
||
# ==============================================================================
|
||
# 获取所有用户列表（管理员）
|
||
# ==============================================================================
|
||
@auth_bp.route('/users', methods=['GET'])
|
||
@jwt_required()
|
||
@permission_required('system_user')
|
||
def get_users():
|
||
try:
|
||
from app.utils.decorators import get_current_company_filter
|
||
company_filter = get_current_company_filter()
|
||
users = AuthService.get_all_users(company_name=company_filter)
|
||
user_permissions = get_current_user_permissions()
|
||
filtered_users = [filter_item_by_permissions(user, user_permissions) for user in users]
|
||
return jsonify({'msg': '获取成功', 'data': filtered_users}), 200
|
||
except Exception as e:
|
||
current_app.logger.error(f"Get Users Failed: {str(e)}")
|
||
return jsonify({'msg': f'获取用户列表失败: {str(e)}'}), 500
|
||
|
||
|
||
# ==============================================================================
|
||
# 删除用户（管理员）
|
||
# ==============================================================================
|
||
@auth_bp.route('/user/<int:user_id>', methods=['DELETE'])
|
||
@jwt_required()
|
||
@permission_required('system_user:operation')
|
||
def delete_user(user_id):
|
||
try:
|
||
claims = get_jwt()
|
||
operator_role = claims.get('role')
|
||
operator_company = claims.get('company_name', '')
|
||
|
||
username = AuthService.delete_user(user_id, operator_role, operator_company)
|
||
return jsonify({'msg': '删除成功', 'username': username}), 200
|
||
except Exception as e:
|
||
current_app.logger.error(f"Delete User Failed: {str(e)}")
|
||
return jsonify({'msg': str(e)}), 400
|
||
|
||
|
||
# ==============================================================================
|
||
# 获取当前用户权限列表（登录后调用）
|
||
# ==============================================================================
|
||
@auth_bp.route('/my-permissions', methods=['GET'])
|
||
@jwt_required()
|
||
def get_my_permissions():
|
||
"""获取当前登录用户的权限列表"""
|
||
try:
|
||
claims = get_jwt()
|
||
role = claims.get('role')
|
||
company_name = claims.get('company_name', '')
|
||
permissions = AuthService.get_user_permissions(role, company_name=company_name)
|
||
return jsonify({'msg': '获取成功', 'data': permissions}), 200
|
||
except Exception as e:
|
||
current_app.logger.error(f"Get Permissions Failed: {str(e)}")
|
||
return jsonify({'msg': f'获取权限失败: {str(e)}'}), 500
|
||
|
||
|
||
# ==============================================================================
|
||
# 获取可指定审批人列表（SUPERVISOR / SUPER_ADMIN 且 status=active）
|
||
# ==============================================================================
|
||
@auth_bp.route('/users/approvers', methods=['GET'])
|
||
@jwt_required()
|
||
def get_approvers():
|
||
"""
|
||
查询角色为 SUPER_ADMIN 或 SUPERVISOR 且状态为活跃的用户列表
|
||
返回: [{id, username, email, role, is_self}]
|
||
其中 is_self=true 表示当前登录用户本人（用于前端标记）
|
||
"""
|
||
try:
|
||
from app.models.system import SysUser
|
||
|
||
current_user_id = get_jwt_identity()
|
||
me = SysUser.query.get(current_user_id) if current_user_id else None
|
||
|
||
from sqlalchemy import and_, or_
|
||
if me and (me.role or '').upper() == 'SUPER_ADMIN':
|
||
# 超管本人：不受部门限制，仍列出全部主管+超管
|
||
users = SysUser.query.filter(
|
||
SysUser.role.in_(['SUPER_ADMIN', 'SUPERVISOR']),
|
||
SysUser.status == 'active'
|
||
).all()
|
||
else:
|
||
# 普通/主管：只列“本公司(department)主管” + “所有超管”
|
||
my_company = me.department if me else None
|
||
users = SysUser.query.filter(
|
||
SysUser.status == 'active',
|
||
or_(
|
||
SysUser.role == 'SUPER_ADMIN',
|
||
and_(SysUser.role == 'SUPERVISOR', SysUser.department == my_company)
|
||
)
|
||
).all()
|
||
|
||
return jsonify({
|
||
'msg': '获取成功',
|
||
'data': [
|
||
{
|
||
'id': u.id,
|
||
'username': u.username,
|
||
'email': u.email or '',
|
||
'role': u.role,
|
||
'is_self': (u.id == current_user_id)
|
||
} for u in users
|
||
]
|
||
}), 200
|
||
|
||
except Exception as e:
|
||
current_app.logger.error(f"Get Approvers Failed: {str(e)}")
|
||
return jsonify({'msg': f'获取审批人列表失败: {str(e)}'}), 500
|
||
|
||
|
||
# ==============================================================================
|
||
# 获取当前用户个人资料（自我查看）
|
||
# ==============================================================================
|
||
@auth_bp.route('/me', methods=['GET'])
|
||
@jwt_required()
|
||
def get_my_profile():
|
||
"""
|
||
【重构】获取当前登录用户的个人资料（自我查看）
|
||
- 只返回姓名/账号和所属部门
|
||
- 严格脱敏：不暴露系统角色字段
|
||
"""
|
||
try:
|
||
from app.models.system import SysUser
|
||
|
||
user_id = get_jwt_identity()
|
||
|
||
# 超级管理员（user_id=0）
|
||
if user_id == 0:
|
||
return jsonify({
|
||
'msg': '获取成功',
|
||
'data': {
|
||
'id': 0,
|
||
'username': 'IRIS',
|
||
'display_name': '超级管理员(IRIS)',
|
||
'department': 'System',
|
||
}
|
||
}), 200
|
||
|
||
user = SysUser.query.get(user_id)
|
||
if not user:
|
||
return jsonify({'msg': '用户不存在'}), 404
|
||
|
||
return jsonify({
|
||
'msg': '获取成功',
|
||
'data': {
|
||
'id': user.id,
|
||
'username': user.username.split('/')[1] if '/' in user.username else user.username,
|
||
'display_name': user.username.split('/')[0] if '/' in user.username else user.username,
|
||
'department': user.department or '-',
|
||
}
|
||
}), 200
|
||
|
||
except Exception as e:
|
||
current_app.logger.error(f"Get Profile Failed: {str(e)}")
|
||
return jsonify({'msg': f'获取个人资料失败: {str(e)}'}), 500
|
||
|
||
|
||
# ==============================================================================
|
||
# 自我修改密码（无需旧密码）
|
||
# ==============================================================================
|
||
@auth_bp.route('/me/password', methods=['PUT'])
|
||
@jwt_required()
|
||
def change_my_password():
|
||
"""
|
||
【重构】自我修改密码接口
|
||
- 无需管理员权限，无需旧密码
|
||
- 只要 JWT Token 有效（已证明当前登录身份），即可直接修改新密码
|
||
"""
|
||
try:
|
||
from app.models.system import SysUser
|
||
|
||
user_id = get_jwt_identity()
|
||
|
||
data = request.get_json()
|
||
if not data:
|
||
return jsonify({'msg': '无效的请求数据'}), 400
|
||
|
||
new_password = data.get('new_password')
|
||
confirm_password = data.get('confirm_password')
|
||
|
||
if not new_password or not confirm_password:
|
||
return jsonify({'msg': '新密码和确认新密码均不能为空'}), 400
|
||
|
||
if new_password != confirm_password:
|
||
return jsonify({'msg': '新密码与确认密码不一致'}), 400
|
||
|
||
if len(new_password) < 6:
|
||
return jsonify({'msg': '新密码长度不能少于6位'}), 400
|
||
|
||
# 超级管理员（user_id=0）使用硬编码密码，不存入数据库
|
||
if user_id == 0:
|
||
return jsonify({'msg': '超级管理员密码由系统管理员管理，当前会话无法修改'}), 200
|
||
|
||
# 普通用户：JWT 已证明身份，直接更新新密码
|
||
user = SysUser.query.get(user_id)
|
||
if not user:
|
||
return jsonify({'msg': '用户不存在'}), 404
|
||
|
||
user.set_password(new_password)
|
||
db.session.commit()
|
||
|
||
return jsonify({'msg': '密码修改成功，请使用新密码重新登录'}), 200
|
||
|
||
except Exception as e:
|
||
current_app.logger.error(f"Change Password Failed: {str(e)}")
|
||
return jsonify({'msg': f'密码修改失败: {str(e)}'}), 500
|
||
|
||
|
||
# ==============================================================================
|
||
# 自我更新邮箱
|
||
# ==============================================================================
|
||
@auth_bp.route('/me/email', methods=['PUT'])
|
||
@jwt_required()
|
||
def update_my_email():
|
||
"""
|
||
自我更新邮箱接口
|
||
- 仅更新 email 字段，与密码修改完全隔离
|
||
- 防止后端意外清空用户密码
|
||
"""
|
||
try:
|
||
from app.models.system import SysUser
|
||
|
||
user_id = get_jwt_identity()
|
||
|
||
# 超级管理员（user_id=0）不允许修改邮箱
|
||
if user_id == 0:
|
||
return jsonify({'msg': '超级管理员邮箱由系统管理员管理'}), 400
|
||
|
||
data = request.get_json()
|
||
if not data:
|
||
return jsonify({'msg': '无效的请求数据'}), 400
|
||
|
||
email = data.get('email')
|
||
if not email:
|
||
return jsonify({'msg': '邮箱不能为空'}), 400
|
||
|
||
# 简单的邮箱格式校验
|
||
import re
|
||
if not re.match(r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$', email):
|
||
return jsonify({'msg': '邮箱格式不正确'}), 400
|
||
|
||
user = SysUser.query.get(user_id)
|
||
if not user:
|
||
return jsonify({'msg': '用户不存在'}), 404
|
||
|
||
# 检查邮箱是否已被其他用户使用
|
||
existing = SysUser.query.filter(SysUser.email == email, SysUser.id != user_id).first()
|
||
if existing:
|
||
return jsonify({'msg': '该邮箱已被其他用户使用'}), 400
|
||
|
||
user.email = email
|
||
db.session.commit()
|
||
|
||
return jsonify({'msg': '邮箱更新成功'}), 200
|
||
|
||
except Exception as e:
|
||
current_app.logger.error(f"Update Email Failed: {str(e)}")
|
||
return jsonify({'msg': f'邮箱更新失败: {str(e)}'}), 500
|